# 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
```