scripts/backtest_c27.py

#!/usr/bin/env python3
"""C27: International macro features — DXY, Nasdaq, Oil, Gold.

道瓊工業指數 was tested in C24 and conflicted with sector pooling.
This experiment adds four orthogonal international signals:

  dxy_ret_5d      — USD index 5-day return (dollar strength → TW export headwind)
  nasdaq_ret_1d   — Nasdaq prior-day return (tech sentiment, non-leaking)
  nasdaq_ret_5d   — Nasdaq 5-day return
  oil_ret_5d      — Crude oil (CL=F) 5-day return (input cost signal)
  gold_ret_5d     — Gold (GC=F) 5-day return (risk-off / safe-haven indicator)

Hypothesis:
- DXY rising → TWD weakens → TW export names benefit short-term but global risk-off
- Nasdaq (tech barometer) leads semiconductor stocks by 1 trading day
- Rising oil hurts industrial/transportation names
- Rising gold = risk-off = bearish for equities
These signals are more orthogonal to TAIEX than DJIA and should add
information the sector model doesn't already capture.
"""
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, walk_forward, 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

C27_FEATURES = [
    "dxy_ret_5d",
    "nasdaq_ret_1d",
    "nasdaq_ret_5d",
    "oil_ret_5d",
    "gold_ret_5d",
]
NEW_FEATURES = CURRENT_FEATURES + C27_FEATURES

_intl_cache: dict = {}


def _fetch_intl(start: str, end: str) -> dict[str, pd.Series | None]:
    key = f"{start}:{end}"
    if key in _intl_cache:
        return _intl_cache[key]

    tickers = ["DX-Y.NYB", "^IXIC", "CL=F", "GC=F"]
    result  = {t: None for t in tickers}
    try:
        raw = yf.download(tickers, 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")
                for t in tickers:
                    if t in close.columns:
                        result[t] = close[t].dropna()
    except Exception as exc:
        print(f"  [warn] intl fetch: {exc}")

    _intl_cache[key] = result
    return result


def _append_intl(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"
    series = _fetch_intl(start, end)

    def _align(ticker):
        s = series.get(ticker)
        if s is None or date_col is None:
            return None
        return date_col.map(s.to_dict()).astype(float)

    dxy     = _align("DX-Y.NYB")
    nasdaq  = _align("^IXIC")
    oil     = _align("CL=F")
    gold    = _align("GC=F")

    feat["dxy_ret_5d"]    = dxy.pct_change(5).values    if dxy    is not None else 0.0
    feat["nasdaq_ret_1d"] = nasdaq.pct_change(1).shift(1).values if nasdaq is not None else 0.0
    feat["nasdaq_ret_5d"] = nasdaq.pct_change(5).values if nasdaq is not None else 0.0
    feat["oil_ret_5d"]    = oil.pct_change(5).values    if oil    is not None else 0.0
    feat["gold_ret_5d"]   = gold.pct_change(5).values   if gold   is not None else 0.0

    for col in C27_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
    feat   = _build_features(df)
    feat   = _append_intl(feat, df)
    close  = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values
    labels = build_triple_barrier_labels(close)
    return feat, labels


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--extended", action="store_true")
    args = parser.parse_args()

    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=== C27: International macro features [{tag}] ===")
    print(f"  New features: {C27_FEATURES}\n")

    per_stock = {}
    base_results, c27_results = [], []

    for stock_no in stocks:
        print(f"  {stock_no}...", end=" ", flush=True)
        feat, labels = load_stock(stock_no)
        if feat is None:
            print("no data")
            continue

        m_base = walk_forward(feat, labels, CURRENT_FEATURES)
        m_c27  = walk_forward(feat, labels, NEW_FEATURES)

        per_stock[stock_no] = {"baseline": m_base, "c27": m_c27}
        base_results.append(m_base)
        c27_results.append(m_c27)

        b_dir = m_base.get("dir_accuracy", float("nan"))
        b_up  = m_base.get("up_precision", float("nan"))
        c_dir = m_c27.get("dir_accuracy",  float("nan"))
        c_up  = m_c27.get("up_precision",  float("nan"))
        print(f"base dir={b_dir}% ↑prec={b_up}%  →  c27 dir={c_dir}% ↑prec={c_up}%")

    def _avg(results, key):
        vals = [m[key] for m in results if m and not np.isnan(m.get(key, float("nan")))]
        return round(float(np.mean(vals)), 1) if vals else float("nan")

    base_agg = {"dir_accuracy": _avg(base_results, "dir_accuracy"),
                "up_precision": _avg(base_results, "up_precision")}
    c27_agg  = {"dir_accuracy": _avg(c27_results,  "dir_accuracy"),
                "up_precision": _avg(c27_results,  "up_precision")}

    passed = c27_agg["dir_accuracy"] >= PASS_DIR_ACC and c27_agg["up_precision"] >= PASS_UP_PREC

    print(f"\n  Baseline avg: dir={base_agg['dir_accuracy']}%  ↑prec={base_agg['up_precision']}%")
    print(f"  C27 avg:      dir={c27_agg['dir_accuracy']}%   ↑prec={c27_agg['up_precision']}%")
    print(f"  Gate (dir≥{PASS_DIR_ACC}% AND ↑prec≥{PASS_UP_PREC}%): {'PASS ✓' if passed else 'FAIL ✗'}")

    result = {
        "experiment":   "C27",
        "description":  "International macro: DXY, Nasdaq, Oil, Gold",
        "new_features": C27_FEATURES,
        "stocks":       stocks,
        "aggregate":    {"baseline": base_agg, "c27": c27_agg},
        "passed":       passed,
        "pass_gate":    {"dir_accuracy": PASS_DIR_ACC, "up_precision": PASS_UP_PREC},
        "per_stock":    per_stock,
    }

    out = ROOT / f"docs/c27_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())