scripts/backtest_c25.py

#!/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())