#!/usr/bin/env python3 """C25: C20 sector-conditional model + C24 DJIA/VIX macro features (combined). C20 showed sector pooling raises 12-stock up_prec to 54.2% (gate pass). C24 showed DJIA+VIX raises TSMC up_prec from 54% → 61% but pulls MediaTek/Fubon without sector pooling. Combining both should keep TSMC's DJIA boost while C20's sector training floors MediaTek and Fubon precision. New features over CURRENT_FEATURES: djia_ret_1d — prior-day DJIA return (non-leaking) djia_ret_5d — DJIA 5-day return djia_ma20_ratio — DJIA / 20d MA vix_level — VIX spot level vix_change_5d — 5-day VIX change """ 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)) # ── C24 feature extensions ─────────────────────────────────────────────────── C24_FEATURES = ["djia_ret_1d", "djia_ret_5d", "djia_ma20_ratio", "vix_level", "vix_change_5d"] NEW_FEATURES = CURRENT_FEATURES + C24_FEATURES _macro_cache: dict = {} def _fetch_macro(start: str, end: str): key = f"{start}:{end}" if key in _macro_cache: return _macro_cache[key] djia = vix = None try: raw = yf.download(["^DJI", "^VIX"], start=start, end=end, auto_adjust=True, progress=False) if not raw.empty: close = raw["Close"] if "Close" in raw.columns else raw.get("close") if close is not None and not close.empty: close.index = pd.to_datetime(close.index).strftime("%Y-%m-%d") if "^DJI" in close.columns: djia = close["^DJI"].dropna() if "^VIX" in close.columns: vix = close["^VIX"].dropna() except Exception as exc: print(f" [warn] macro fetch: {exc}") result = (djia, vix) _macro_cache[key] = result return result def _append_macro(feat: pd.DataFrame, df: pd.DataFrame) -> pd.DataFrame: date_col = df["date"].astype(str) if "date" in df.columns else None 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" djia, vix = _fetch_macro(start, end) nan_col = pd.Series(np.nan, index=feat.index) if djia is not None and date_col is not None: d = djia.to_dict() aligned = date_col.map(d).astype(float) feat["djia_ret_1d"] = aligned.pct_change(1).shift(1).values feat["djia_ret_5d"] = aligned.pct_change(5).values ma20 = aligned.rolling(20, min_periods=20).mean() feat["djia_ma20_ratio"] = (aligned / ma20.replace(0, np.nan)).values else: feat["djia_ret_1d"] = feat["djia_ret_5d"] = feat["djia_ma20_ratio"] = 0.0 if vix is not None and date_col is not None: v = vix.to_dict() aligned_v = date_col.map(v).astype(float) feat["vix_level"] = aligned_v.values feat["vix_change_5d"] = aligned_v.diff(5).values else: feat["vix_level"] = feat["vix_change_5d"] = 0.0 for col in C24_FEATURES: feat[col] = pd.Series(feat[col], index=feat.index).ffill().bfill().fillna(0.0) return feat def load_stock(stock_no: str): df = fetch_df(stock_no) if df is None or df.empty: return None, None, None feat = _build_features(df) feat = _append_macro(feat, 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 def walk_forward_sector(target_feat, target_labels, target_dates, peers, cols): 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 = [], [] 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 X_train_list = [X_tgt[:train_end][valid]] y_train_list = [y_v] 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) 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 {} return compute_metrics(np.array(y_true_all), np.array(y_pred_all)) def main(): parser = argparse.ArgumentParser() parser.add_argument("--extended", action="store_true") 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 "" print(f"\n=== C25: Sector model + DJIA/VIX features [{tag}] ===") print(f" Features: {len(NEW_FEATURES)} ({len(CURRENT_FEATURES)} base + {len(C24_FEATURES)} macro)\n") print(" Loading all stock data...", flush=True) stock_data: dict = {} for s in ALL_STOCKS: feat, labels, dates = load_stock(s) if feat is None: print(f" {s}: no data") continue stock_data[s] = {"feat": feat, "labels": labels, "dates": dates} print(f" {s}: {len(feat)} rows, sector={STOCK_SECTOR.get(s, '?')}") 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 = walk_forward_sector(td["feat"], td["labels"], td["dates"], peers, NEW_FEATURES) per_stock[target_no] = {**m, "sector": sector, "peers": peer_stocks} metrics_list.append(m) if m: print(f"dir={m['dir_accuracy']}% ↑prec={m['up_precision']}%") 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": "C25", "description": "C20 sector-conditional model + C24 DJIA/VIX macro features", "new_features": C24_FEATURES, "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/c25_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())