#!/usr/bin/env python3 """C26: VIX-conditional sector pooling (regime-gated C20). Root cause from C25: adding VIX as a raw feature to the sector model caused a 14pp precision collapse on Fubon. The conflict is that sector peer correlation and macro fear signal pull the model in opposite directions. Fix: use VIX to *gate* sector pooling rather than as a feature. - Per walk-forward window, compute avg VIX during training period. - If avg VIX > VIX_HIGH_THRESHOLD (25): disable peer augmentation — macro fear regime disrupts sector correlations → train solo. - If avg VIX <= threshold: enable C20 sector pooling as normal. This decouples the two signals: structural (sector) vs macro (fear). """ import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) import warnings; warnings.filterwarnings("ignore") import json import argparse import numpy as np import pandas as pd import yfinance as yf from sklearn.ensemble import RandomForestClassifier from sklearn.preprocessing import StandardScaler from scripts.improvement_harness import ( fetch_df, build_triple_barrier_labels, compute_metrics, CURRENT_FEATURES, DEFAULT_STOCKS, EXTENDED_STOCKS, PASS_DIR_ACC, PASS_UP_PREC, MIN_TRAIN, STEP, LABEL_HORIZON, RF_PARAMS, ) from models.predictor import _build_features # ── Sector config (same as C20) ────────────────────────────────────────────── SECTOR_MAP = { "semis": ["2330", "2454", "2303"], "electronics": ["2317", "2382", "2308"], "financials": ["2881", "2882", "2886"], "etfs": ["0050", "0056"], "telecom": ["2412"], } NO_POOL_SECTORS = {"electronics", "telecom"} STOCK_SECTOR = {s: sec for sec, stocks in SECTOR_MAP.items() for s in stocks} ALL_STOCKS = list(dict.fromkeys(DEFAULT_STOCKS + EXTENDED_STOCKS)) VIX_HIGH_THRESHOLD = 25.0 # disable pooling above this fear level _vix_cache: dict = {} def _fetch_vix_series(start: str, end: str) -> dict[str, float]: """Return dict of {date_str: vix_close}.""" key = f"{start}:{end}" if key in _vix_cache: return _vix_cache[key] result: dict[str, float] = {} try: raw = yf.download("^VIX", start=start, end=end, auto_adjust=True, progress=False) if not raw.empty: close = raw["Close"] if "Close" in raw.columns else raw["close"] if hasattr(close, "iloc"): close = close.iloc[:, 0] if close.ndim == 2 else close close.index = pd.to_datetime(close.index).strftime("%Y-%m-%d") result = close.dropna().to_dict() except Exception as exc: print(f" [warn] VIX fetch: {exc}") _vix_cache[key] = result return result def load_stock(stock_no: str): df = fetch_df(stock_no) if df is None or df.empty: return None, None, None, None feat = _build_features(df) close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values labels = build_triple_barrier_labels(close) dates = df["date"].values if "date" in df.columns else None return feat, labels, dates, df def walk_forward_vix_gated(target_feat, target_labels, target_dates, peers, cols, vix_dict: dict): avail = [c for c in cols if c in target_feat.columns] X_tgt = target_feat[avail].fillna(0).values n = len(target_feat) y_true_all, y_pred_all = [], [] windows_pooled = windows_solo = 0 cutoff = MIN_TRAIN while cutoff + STEP + LABEL_HORIZON <= n: train_end = cutoff - LABEL_HORIZON if train_end < MIN_TRAIN - LABEL_HORIZON: cutoff += STEP; continue y_tr = target_labels[:train_end] valid = ~np.isnan(y_tr) y_v = y_tr[valid].astype(int) if len(y_v) < 10 or len(np.unique(y_v)) < 2: cutoff += STEP; continue # Compute avg VIX over training window to gate sector pooling if target_dates is not None and vix_dict: window_dates = target_dates[:train_end] vix_vals = [vix_dict[d] for d in window_dates if d in vix_dict] avg_vix = float(np.mean(vix_vals)) if vix_vals else 0.0 else: avg_vix = 0.0 use_pooling = (avg_vix <= VIX_HIGH_THRESHOLD) and len(peers) > 0 X_train_list = [X_tgt[:train_end][valid]] y_train_list = [y_v] if use_pooling: windows_pooled += 1 cutoff_date = target_dates[train_end - 1] if (target_dates is not None and train_end > 0) else None for (peer_feat, peer_labels, peer_dates) in peers: peer_avail = [c for c in cols if c in peer_feat.columns] if cutoff_date is not None and peer_dates is not None: peer_end = int((peer_dates <= cutoff_date).sum()) else: peer_end = int(train_end * len(peer_feat) / n) peer_end = min(peer_end, len(peer_feat) - LABEL_HORIZON) if peer_end < 15: continue y_p = peer_labels[:peer_end] valid_p = ~np.isnan(y_p) y_vp = y_p[valid_p].astype(int) if len(y_vp) < 5 or len(np.unique(y_vp)) < 2: continue X_p = peer_feat[peer_avail].fillna(0).values[:peer_end][valid_p] X_train_list.append(X_p) y_train_list.append(y_vp) else: windows_solo += 1 X_train = np.vstack(X_train_list) y_train = np.concatenate(y_train_list) if len(np.unique(y_train)) < 2: cutoff += STEP; continue scaler = StandardScaler() X_train_s = scaler.fit_transform(X_train) rf = RandomForestClassifier(**RF_PARAMS) rf.fit(X_train_s, y_train) test_end = min(cutoff + STEP, n - LABEL_HORIZON) y_te = target_labels[cutoff:test_end] valid_te = ~np.isnan(y_te) if valid_te.sum() == 0: cutoff += STEP; continue X_te_s = scaler.transform(X_tgt[cutoff:test_end][valid_te]) y_pred = rf.predict(X_te_s) y_true_all.extend(y_te[valid_te].astype(int).tolist()) y_pred_all.extend(y_pred.tolist()) cutoff += STEP if not y_true_all: return {}, 0, 0 return compute_metrics(np.array(y_true_all), np.array(y_pred_all)), windows_pooled, windows_solo def main(): parser = argparse.ArgumentParser() parser.add_argument("--extended", action="store_true") parser.add_argument("--vix-threshold", type=float, default=VIX_HIGH_THRESHOLD) args = parser.parse_args() eval_stocks = EXTENDED_STOCKS if args.extended else DEFAULT_STOCKS tag = "12-stock" if args.extended else "5-stock" suffix = "_12stock" if args.extended else "" threshold = args.vix_threshold print(f"\n=== C26: VIX-gated sector pooling [{tag}] ===") print(f" VIX threshold: >{threshold} → solo training, <={threshold} → sector pooling\n") print(" Loading all stock data...", flush=True) stock_data: dict = {} for s in ALL_STOCKS: feat, labels, dates, df = load_stock(s) if feat is None: print(f" {s}: no data") continue # Fetch VIX for this stock's date range start = str(df["date"].min()) if "date" in df.columns else "2020-01-01" end = str(df["date"].max()) if "date" in df.columns else "2025-01-01" vix_dict = _fetch_vix_series(start, end) stock_data[s] = {"feat": feat, "labels": labels, "dates": dates, "vix": vix_dict} print(f" {s}: {len(feat)} rows, vix_dates={len(vix_dict)}") print() per_stock, metrics_list = {}, [] for target_no in eval_stocks: if target_no not in stock_data: print(f" {target_no}: missing, skip") continue sector = STOCK_SECTOR.get(target_no, "unknown") if sector in NO_POOL_SECTORS: peer_stocks = [] else: peer_stocks = [s for s in SECTOR_MAP.get(sector, []) if s != target_no and s in stock_data] peers = [ (stock_data[p]["feat"], stock_data[p]["labels"], stock_data[p]["dates"]) for p in peer_stocks ] td = stock_data[target_no] print(f" {target_no} [{sector}, peers={peer_stocks}]...", end=" ", flush=True) m, n_pooled, n_solo = walk_forward_vix_gated( td["feat"], td["labels"], td["dates"], peers, CURRENT_FEATURES, td["vix"], ) per_stock[target_no] = {**m, "sector": sector, "peers": peer_stocks, "windows_pooled": n_pooled, "windows_solo": n_solo} metrics_list.append(m) if m: print(f"dir={m['dir_accuracy']}% ↑prec={m['up_precision']}% " f"[pooled={n_pooled} solo={n_solo}]") else: print("no output") def _avg(key): vals = [m[key] for m in metrics_list if m and not np.isnan(m.get(key, float("nan")))] return round(float(np.mean(vals)), 1) if vals else float("nan") avg_dir = _avg("dir_accuracy") avg_up = _avg("up_precision") passed = avg_dir >= PASS_DIR_ACC and avg_up >= PASS_UP_PREC print(f"\n Avg: dir={avg_dir}% ↑prec={avg_up}%") print(f" Gate (dir≥{PASS_DIR_ACC}% AND ↑prec≥{PASS_UP_PREC}%): {'PASS ✓' if passed else 'FAIL ✗'}") result = { "experiment": "C26", "description": f"VIX-gated sector pooling (threshold={threshold})", "vix_threshold": threshold, "stocks": eval_stocks, "aggregate": {"dir_accuracy": avg_dir, "up_precision": avg_up}, "per_stock": per_stock, "passed": passed, "pass_gate": {"dir_accuracy": PASS_DIR_ACC, "up_precision": PASS_UP_PREC}, } out = ROOT / f"docs/c26_result{suffix}.json" out.parent.mkdir(exist_ok=True) out.write_text(json.dumps(result, indent=2)) print(f"\n Saved: {out}") return 0 if passed else 1 if __name__ == "__main__": sys.exit(main())