april done. built a signal quality gate with LightGBM.

from typing import TypedDict


class SignalFeatures(TypedDict):
    # IV environment at signal time
    iv_rank_30d: float          # 0-100 percentile rank vs trailing 30 days
    iv_percentile_252d: float   # annual horizon for comparison
    iv_vs_rv_ratio: float       # implied vol / realized vol ratio — >1.0 means IV rich

    # Market state
    vix_spot: float             # VIX level at signal time
    vix_term_ratio: float       # VIX3M / VIX1M (term structure slope)
    spy_vol_vs_20d_avg: float   # today's realized vol as ratio vs 20-day average

    # Signal specifics
    dte: int                    # days to expiration
    delta_short_strike: float   # delta of the short leg
    bid_ask_pct: float          # spread as % of mid (execution quality proxy)
    time_of_day_sin: float      # cyclical encoding of hour-of-day (sin component)
    time_of_day_cos: float      # cyclical encoding of hour-of-day (cos component)

    # Strategy momentum
    strategy_7d_win_rate: float      # this strategy's win rate over last 7 days
    strategy_14d_pnl_zscore: float   # z-score of strategy P&L vs historical mean

    # Contract characteristics
    open_interest_ratio: float  # OI at this strike / total OI at expiry
    moneyness: float            # (strike - spot) / spot for the short leg

import lightgbm as lgb
import numpy as np
import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import roc_auc_score
from typing import Tuple, List


FEATURE_COLS = list(SignalFeatures.__annotations__.keys())


def prepare_quality_labels(
    fills_df: pd.DataFrame,
    min_cohort_size: int = 40,
) -> pd.DataFrame:
    """
    Label top/bottom quartile within (strategy_id, expiration) cohorts.
    Drop middle 50% — cleaner binary signal.

    Args:
        fills_df: DataFrame with columns: strategy_id, expiration,
                  realized_pnl, signal_timestamp, + all FEATURE_COLS
        min_cohort_size: Skip cohorts with fewer fills (insufficient stats)

    Returns:
        Labeled DataFrame sorted by signal_timestamp, with 'quality_label' column.
    """
    labeled: List[pd.DataFrame] = []

    for (strategy, expiry), group in fills_df.groupby(["strategy_id", "expiration"]):
        if len(group) < min_cohort_size:
            continue

        q25 = group["realized_pnl"].quantile(0.25)
        q75 = group["realized_pnl"].quantile(0.75)

        top_q = group[group["realized_pnl"] >= q75].copy()
        top_q["quality_label"] = 1

        bot_q = group[group["realized_pnl"] <= q25].copy()
        bot_q["quality_label"] = 0

        labeled.extend([top_q, bot_q])

    if not labeled:
        raise ValueError("no cohorts met min_cohort_size — check data or lower threshold")

    return pd.concat(labeled).sort_values("signal_timestamp").reset_index(drop=True)


def train_quality_classifier(
    X: pd.DataFrame,
    y: pd.Series,
    n_splits: int = 5,
) -> Tuple[lgb.Booster, float]:
    """
    Binary quality classifier with TimeSeriesSplit cross-validation.

    TimeSeriesSplit is non-negotiable here. Standard k-fold bleeds
    future fills into training folds and gives optimistically wrong OOF scores.
    gap=200 adds a buffer of 200 observations between train/val to reduce
    leakage from strategy autocorrelation.

    Returns:
        (best_model, mean_oof_auc) where best_model = highest-AUC fold
    """
    tscv = TimeSeriesSplit(n_splits=n_splits, gap=200)

    params = {
        "objective": "binary",
        "metric": "auc",
        "num_leaves": 31,
        "learning_rate": 0.05,
        "feature_fraction": 0.8,
        "bagging_fraction": 0.8,
        "bagging_freq": 5,
        "min_child_samples": 20,
        "reg_alpha": 0.1,
        "reg_lambda": 0.1,
        "verbose": -1,
    }

    fold_aucs: List[float] = []
    fold_models: List[lgb.Booster] = []

    for fold_idx, (train_idx, val_idx) in enumerate(tscv.split(X)):
        X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
        y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]

        dtrain = lgb.Dataset(X_train, label=y_train)
        dval = lgb.Dataset(X_val, label=y_val, reference=dtrain)

        model = lgb.train(
            params,
            dtrain,
            num_boost_round=600,
            valid_sets=[dval],
            callbacks=[
                lgb.early_stopping(stopping_rounds=50, verbose=False),
                lgb.log_evaluation(period=-1),
            ],
        )

        val_preds = model.predict(X_val)
        auc = roc_auc_score(y_val, val_preds)
        fold_aucs.append(auc)
        fold_models.append(model)

    mean_auc = float(np.mean(fold_aucs))
    best_idx = int(np.argmax(fold_aucs))

    return fold_models[best_idx], mean_auc

import redis
import pickle
import logging
from dataclasses import dataclass
from typing import Any


logger = logging.getLogger(__name__)

GATE_THRESHOLD = 0.35


@dataclass
class GateResult:
    score: float
    decision: str          # "pass" | "filter" | "error"
    threshold: float
    features_available: int


class SignalQualityGate:
    """
    Production inference wrapper for the LightGBM quality classifier.

    Design principle: gate errors open. if inference fails for any reason,
    signal passes through unchanged. i don't want the classifier blocking
    execution when Redis is slow or model bytes are corrupt.

    Model refreshed from Redis every 24h via background task.
    """

    MODEL_REDIS_KEY = "quality_gate:lgbm_v1"

    def __init__(self, redis_client: redis.Redis) -> None:
        self.redis = redis_client
        self.model: Any = None
        self._load_model()

    def _load_model(self) -> None:
        raw = self.redis.get(self.MODEL_REDIS_KEY)
        if raw is None:
            logger.error("quality gate model missing from Redis — gate will error-open")
            return
        self.model = pickle.loads(raw)
        logger.info("quality gate model loaded (%d bytes)", len(raw))

    def evaluate(self, features: dict) -> GateResult:
        if self.model is None:
            return GateResult(
                score=0.5, decision="error",
                threshold=GATE_THRESHOLD, features_available=0
            )

        try:
            vec = [features.get(k, 0.0) for k in FEATURE_COLS]
            n_available = sum(1 for k in FEATURE_COLS if k in features)

            score = float(self.model.predict([vec])[0])
            decision = "pass" if score >= GATE_THRESHOLD else "filter"

            return GateResult(
                score=round(score, 4),
                decision=decision,
                threshold=GATE_THRESHOLD,
                features_available=n_available,
            )

        except Exception as exc:
            logger.exception("quality gate inference failed: %s", exc)
            return GateResult(
                score=0.5, decision="error",
                threshold=GATE_THRESHOLD, features_available=0
            )

-- 7-day rolling strategy metrics
-- refreshed every 30 minutes by TimescaleDB background worker
CREATE MATERIALIZED VIEW quality_gate_strategy_metrics
WITH (timescaledb.continuous) AS
SELECT
    strategy_id,
    time_bucket('1 day', signal_timestamp)                          AS bucket,
    COUNT(*) FILTER (WHERE realized_pnl > 0) * 1.0
        / NULLIF(COUNT(*), 0)                                       AS win_rate_7d,
    AVG(realized_pnl)                                               AS avg_pnl,
    STDDEV(realized_pnl)                                            AS std_pnl,
    COUNT(*)                                                        AS fill_count
FROM trade_fills
GROUP BY strategy_id, time_bucket('1 day', signal_timestamp)
WITH NO DATA;

SELECT add_continuous_aggregate_policy(
    'quality_gate_strategy_metrics',
    start_offset      => INTERVAL '14 days',
    end_offset        => INTERVAL '30 minutes',
    schedule_interval => INTERVAL '30 minutes'
);

-- feature computation at signal time (fast JOIN against pre-computed aggregate)
WITH recent_14d AS (
    SELECT
        strategy_id,
        SUM(avg_pnl * fill_count) / NULLIF(SUM(fill_count), 0)           AS rolling_mean,
        SQRT(SUM(std_pnl * std_pnl * fill_count) / NULLIF(SUM(fill_count), 0)) AS rolling_std
    FROM quality_gate_strategy_metrics
    WHERE bucket >= CURRENT_DATE - INTERVAL '14 days'
    GROUP BY strategy_id
),
today_metrics AS (
    SELECT strategy_id, win_rate_7d, avg_pnl
    FROM quality_gate_strategy_metrics
    WHERE bucket = CURRENT_DATE
)
SELECT
    t.strategy_id,
    t.win_rate_7d                                               AS strategy_7d_win_rate,
    CASE
        WHEN r.rolling_std > 0
        THEN (t.avg_pnl - r.rolling_mean) / r.rolling_std
        ELSE 0.0
    END                                                         AS strategy_14d_pnl_zscore
FROM today_metrics t
JOIN recent_14d r USING (strategy_id);