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