Advanced Python & SQL Script Suite

# Advanced Data Engineering, ML & Analytics Script Suite

This document contains full production-grade, executable **Python (Polars, Pandas, Scipy)** and **SQL (PostgreSQL / Snowflake)** implementations for the 8 architectural patterns featured in the **Data & Analytics Architectural Playbook**.

---

## 1. CUPED Variance Reduction & A/B Testing Engine (Python)

```python import numpy as np import pandas as pd from scipy import stats

def compute_cuped_adjustment(df: pd.DataFrame, y_pre_col: str, y_post_col: str, group_col: str): """ Computes Controlled-experiments Using Pre-Experiment Data (CUPED) variance reduction. """ # Compute Covariance and Variance cov_matrix = np.cov(df[y_pre_col], df[y_post_col]) covariance = cov_matrix[0, 1] variance_pre = np.var(df[y_pre_col], ddof=1)

# Calculate Optimal Theta theta = covariance / variance_pre y_pre_mean = df[y_pre_col].mean()

# Apply CUPED adjustment: Y_cuped = Y_post - theta * (Y_pre - Mean(Y_pre)) df['Y_cuped'] = df[y_post_col] - theta * (df[y_pre_col] - y_pre_mean)

# Variance comparison var_unadjusted = df[y_post_col].var() var_cuped = df['Y_cuped'].var() variance_reduction_pct = (1 - (var_cuped / var_unadjusted)) * 100

# Statistical t-test on CUPED adjusted metric control_cuped = df[df[group_col] == 'Control']['Y_cuped'] treatment_cuped = df[df[group_col] == 'Treatment']['Y_cuped'] t_stat, p_val = stats.ttest_ind(treatment_cuped, control_cuped)

return { 'theta': theta, 'unadjusted_variance': var_unadjusted, 'cuped_variance': var_cuped, 'variance_reduction_pct': variance_reduction_pct, 'p_value': p_val, 'df_adjusted': df } ```

---

## 2. Polars & Apache Arrow Ingestion Engine (Python)

```python import polars as pl

def execute_lazy_analytics_pipeline(file_path: str): """ Executes lazy evaluation over large datasets using Polars and Apache Arrow vectorization. """ lazy_df = pl.scan_csv(file_path)

query = ( lazy_df .filter(pl.col("status") == "COMPLETED") .with_columns([ (pl.col("gross_amount") * (1 - pl.col("discount_rate"))).alias("net_revenue"), pl.col("order_timestamp").str.to_datetime().dt.year().alias("order_year") ]) .groupby(["order_year", "customer_segment"]) .agg([ pl.col("net_revenue").sum().alias("total_net_revenue"), pl.col("net_revenue").mean().alias("avg_order_value"), pl.col("order_id").count().alias("transaction_count") ]) .sort("total_net_revenue", descending=True) )

# Execute query engine with streaming result_df = query.collect(streaming=True) return result_df ```

---

## 3. SQL Index Optimization & Partition Pruning (PostgreSQL)

```sql -- Create B-Tree Index with Included Columns (Covering Index) CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_customer_date ON sales.orders (customer_id, order_date DESC) INCLUDE (net_amount, order_status);

-- Optimized Query using Index Seek + Partition Pruning EXPLAIN ANALYZE SELECT customer_id, DATE_TRUNC('month', order_date) AS order_month, SUM(net_amount) AS monthly_spend, COUNT(order_id) AS total_orders FROM sales.orders WHERE order_date >= '2026-01-01' AND order_status = 'COMPLETED' GROUP BY customer_id, DATE_TRUNC('month', order_date) ORDER BY monthly_spend DESC; ```

---

## 4. Real-Time Feature Store Entity Vector Pipeline (Python + Redis)

```python import redis import json import pandas as pd

class FeatureStorePipeline: def __init__(self, redis_host='localhost', redis_port=6379): self.client = redis.StrictRedis(host=redis_host, port=redis_port, decode_responses=True)

def write_online_features(self, entity_df: pd.DataFrame, entity_key: str): """ Pushes batch features to online low-latency Redis cache. """ for _, row in entity_df.iterrows(): key = f"entity_feature:{row[entity_key]}" feature_vector = { "days_since_active": int(row["days_since_active"]), "frequency_30d": int(row["frequency_30d"]), "churn_score": float(row["churn_score"]) } self.client.setex(key, time=86400, value=json.dumps(feature_vector))

def get_online_feature(self, entity_id: str): key = f"entity_feature:{entity_id}" val = self.client.get(key) return json.loads(val) if val else None ```