#!/usr/bin/env python3 """ 3-way walk-forward backtest: SUBSET_A (40f) vs SUBSET_B (audit-filtered) vs SUBSET_C (all 85f). Usage: ./venv/bin/python3 scripts/backtest_3way.py 2>/dev/null """ import json import sys import warnings from datetime import datetime from pathlib import Path warnings.filterwarnings("ignore") ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) try: from dotenv import load_dotenv load_dotenv(ROOT / ".env") except ImportError: pass import numpy as np import pandas as pd from scipy.stats import spearmanr from sklearn.ensemble import RandomForestClassifier # ── Feature subsets ──────────────────────────────────────────────────────────── SUBSET_A = [ "return_1d", "return_5d", "return_10d", "return_20d", "close_ma5_ratio", "close_ma20_ratio", "ma5_ma20_ratio", "ma20_ma60_ratio", "rsi", "macd_hist", "macd_signal_ratio", "macd_hist_norm", "macd_hist_delta_1d", "macd_hist_slope_3d", "macd_cross_up", "macd_cross_down", "macd_above_zero", "bb_pct_b", "k", "d", "volume_ratio", "atr_ratio", "high_low_ratio", "obv_trend", "close_ma60_ratio", "log_volume_ratio", "volume_zscore", "price_volume_div", "volatility_20d", "taiex_return_5d", "taiex_ma20_ratio", "usdtwd_return_5d", "foreign_net_vol_ratio", "trust_net_vol_ratio", "dealer_net_vol_ratio", "institutional_net_vol_ratio", "institutional_5d_net_vol_ratio", "institutional_20d_zscore", "foreign_trust_alignment", "institutional_streak", ] # SUBSET_B: SUBSET_A + audit-filtered extras (neg_count <= 2, banned-list excluded) SUBSET_B = [ # -- all of SUBSET_A -- "return_1d", "return_5d", "return_10d", "return_20d", "close_ma5_ratio", "close_ma20_ratio", "ma5_ma20_ratio", "ma20_ma60_ratio", "rsi", "macd_hist", "macd_signal_ratio", "macd_hist_norm", "macd_hist_delta_1d", "macd_hist_slope_3d", "macd_cross_up", "macd_cross_down", "macd_above_zero", "bb_pct_b", "k", "d", "volume_ratio", "atr_ratio", "high_low_ratio", "obv_trend", "close_ma60_ratio", "log_volume_ratio", "volume_zscore", "price_volume_div", "volatility_20d", "taiex_return_5d", "taiex_ma20_ratio", "usdtwd_return_5d", "foreign_net_vol_ratio", "trust_net_vol_ratio", "dealer_net_vol_ratio", "institutional_net_vol_ratio", "institutional_5d_net_vol_ratio", "institutional_20d_zscore", "foreign_trust_alignment", "institutional_streak", # -- extras: neg_count <= 2 from audit, banned-list excluded -- "close_ma240_ratio", # neg=0, perm=+0.00583 "golden_cross_5_20", # neg=0, perm=+0.00000 (event feature, low noise) "death_cross_5_20", # neg=0 "golden_cross_20_60", # neg=0 "death_cross_20_60", # neg=0 "sox_ret_1d", # neg=2, perm=+0.00146 "margin_ratio", # neg=0 "short_margin_ratio", # neg=0 "margin_5d_change_pct", # neg=0 "margin_buy_pressure", # neg=0 "chip_squeeze_signal", # neg=0 "chip_score", # neg=0 "ma_bull_alignment", # neg=1, perm=-0.00063 "ma_bear_alignment", # neg=0, perm=+0.00167 "ma_alignment_days", # neg=2, perm=+0.00146 "bias_20", # neg=2, perm=+0.00729 "vol_volatility_interact", # neg=1, perm=+0.00604 "short_balance_5d_change", # neg=0 "trust_vol_ratio", # neg=0 "return_60d", # neg=2, perm=+0.00146 "return_120d", # neg=1, perm=+0.00333 "mom_minus_reversal", # neg=1, perm=+0.00854 ] from models.predictor import FEATURE_COLUMNS as SUBSET_C SUBSETS = {"A": SUBSET_A, "B": SUBSET_B, "C": SUBSET_C} STOCKS = ["2330", "0050", "2317", "2454", "2881"] LABEL_HORIZON = 5 LABEL_THRESH = 0.02 MIN_TRAIN = 240 STEP = 21 RF_PARAMS = dict( n_estimators=200, max_depth=6, min_samples_leaf=10, class_weight="balanced", random_state=42, n_jobs=-1, ) def fetch_df(stock_no: str) -> pd.DataFrame: from services.predictor_service import _fetch_with_cache from indicators.technical import add_all_indicators, add_cross_asset_tw from data.institutional_flow import add_institutional_flow from data.margin_flow import add_margin_flow from data.fetcher import is_us_ticker, fetch_cross_asset_tw df = _fetch_with_cache(stock_no, months=24) if df is None or df.empty: return pd.DataFrame() df = add_all_indicators(df) if not is_us_ticker(stock_no): df = add_institutional_flow(df, stock_no) df = add_margin_flow(df, stock_no) start, end = str(df["date"].min()), str(df["date"].max()) taiex, usdtwd, sox, tnx = fetch_cross_asset_tw(start, end) df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx) return df def make_labels(close: pd.Series) -> np.ndarray: fwd = close.shift(-LABEL_HORIZON) ret = (fwd - close) / close return np.where(ret > LABEL_THRESH, 1, np.where(ret < -LABEL_THRESH, -1, 0)) def walk_forward(feat: pd.DataFrame, feature_cols: list[str]) -> dict: avail = [c for c in feature_cols if c in feat.columns] labels = feat["_label"].values X_all = feat[avail].fillna(0).values close = feat["_close"].values n = len(feat) y_true_all, y_pred_all = [], [] pnl_model = [] ic_pairs = [] # (prob_up, actual_fwd_ret) cutoff = MIN_TRAIN while cutoff + STEP + LABEL_HORIZON <= n: X_train = X_all[:cutoff] y_train = labels[:cutoff] test_end = min(cutoff + STEP, n - LABEL_HORIZON) X_test = X_all[cutoff:test_end] y_test = labels[cutoff:test_end] c_test = close[cutoff:test_end] if len(np.unique(y_train)) < 2 or len(X_test) == 0: cutoff += STEP continue clf = RandomForestClassifier(**RF_PARAMS) clf.fit(X_train, y_train) y_pred = clf.predict(X_test) proba = clf.predict_proba(X_test) classes = list(clf.classes_) prob_map = {cls: proba[:, i] for i, cls in enumerate(classes)} y_true_all.extend(y_test.tolist()) y_pred_all.extend(y_pred.tolist()) for i in range(len(X_test)): p_up = float(prob_map.get(1, np.zeros(len(X_test)))[i]) if (cutoff + i + LABEL_HORIZON) < n: fwd_ret = (close[cutoff + i + LABEL_HORIZON] - c_test[i]) / c_test[i] else: fwd_ret = 0.0 if y_pred[i] == 1: pnl_model.append(fwd_ret) elif y_pred[i] == -1: pnl_model.append(-fwd_ret) ic_pairs.append((p_up, fwd_ret)) cutoff += STEP if not y_true_all: return {} y_true = np.array(y_true_all) y_pred = np.array(y_pred_all) dir_mask = y_pred != 0 up_mask = y_pred == 1 dn_mask = y_pred == -1 y_true_dir = y_true[dir_mask] y_pred_dir = y_pred[dir_mask] acc = float((y_true == y_pred).mean() * 100) n_sig = int(dir_mask.sum()) dir_acc = float((y_true_dir == y_pred_dir).mean() * 100) if n_sig > 0 else float("nan") up_prec = float((y_true[up_mask] == 1).mean() * 100) if up_mask.sum() > 0 else float("nan") dn_prec = float((y_true[dn_mask] == -1).mean() * 100) if dn_mask.sum() > 0 else float("nan") win_rate = float((np.array(pnl_model) > 0).mean() * 100) if pnl_model else float("nan") pnl_pct = float(sum(pnl_model) * 100) n_trades = int(len(pnl_model)) # Sharpe (annualized, 5-day holding period) if len(pnl_model) > 1: arr = np.array(pnl_model) sharpe = float((arr.mean() / (arr.std() + 1e-9)) * np.sqrt(252 / 5)) else: sharpe = float("nan") # Max drawdown if pnl_model: cumulative = np.cumprod(1 + np.array(pnl_model)) rolling_max = np.maximum.accumulate(cumulative) drawdowns = (cumulative - rolling_max) / rolling_max max_dd = float(drawdowns.min()) else: max_dd = float("nan") # IC (Spearman correlation between prob_up and actual fwd return) if len(ic_pairs) > 10: ic_val, _ = spearmanr([p[0] for p in ic_pairs], [p[1] for p in ic_pairs]) ic = float(ic_val) else: ic = float("nan") return { "accuracy": round(acc, 1), "dir_accuracy": round(dir_acc, 1) if not np.isnan(dir_acc) else float("nan"), "up_precision": round(up_prec, 1) if not np.isnan(up_prec) else float("nan"), "dn_precision": round(dn_prec, 1) if not np.isnan(dn_prec) else float("nan"), "win_rate": round(win_rate, 1) if not np.isnan(win_rate) else float("nan"), "pnl_pct": round(pnl_pct, 1), "sharpe": round(sharpe, 3) if not np.isnan(sharpe) else float("nan"), "max_dd": round(max_dd, 3) if not np.isnan(max_dd) else float("nan"), "ic": round(ic, 4) if not np.isnan(ic) else float("nan"), "n_predictions": int(len(y_true)), "n_trades": n_trades, } def mean_metrics(results_list: list[dict]) -> dict: if not results_list: return {} keys = ["accuracy", "dir_accuracy", "up_precision", "dn_precision", "win_rate", "pnl_pct", "sharpe", "max_dd", "ic"] out = {} for k in keys: vals = [r[k] for r in results_list if k in r and not np.isnan(r[k])] out[k] = round(float(np.mean(vals)), 3) if vals else float("nan") return out def generate_html(results: dict, aggregate: dict, subsets_sizes: dict, winner: str) -> str: css = """ """ def fmt_val(v, is_pct=False, is_delta=False): if isinstance(v, float) and np.isnan(v): return "N/A" s = f"{v:+.1f}" if is_delta else f"{v:.1f}" cls = "" if is_delta: cls = "pos" if v > 0 else ("neg" if v < 0 else "") elif is_pct: cls = "pos" if v > 0 else ("neg" if v < 0 else "") tag = f' class="{cls}"' if cls else "" return f"{s}" def fmt_small(v): if isinstance(v, float) and np.isnan(v): return "N/A" return f"{v:.3f}" headers = ( "" "Stock/Subset" "Acc%Dir%↑Prec%↓Prec%" "Win%P&L%SharpeMaxDDIC" "TradesPreds" "" ) rows = "" for stock in STOCKS: if stock not in results: continue stock_results = results[stock] for sub_key, css_cls in [("A", "subset-a"), ("B", "subset-b"), ("C", "subset-c")]: r = stock_results.get(sub_key, {}) if not r: continue size = subsets_sizes.get(sub_key, "?") w_cls = " winner" if (stock == list(results.keys())[0] and sub_key == winner) else "" rows += f'' rows += f"{stock} {sub_key}({size}f)" rows += fmt_val(r.get("accuracy", float("nan"))) rows += fmt_val(r.get("dir_accuracy", float("nan"))) rows += fmt_val(r.get("up_precision", float("nan"))) rows += fmt_val(r.get("dn_precision", float("nan"))) rows += fmt_val(r.get("win_rate", float("nan"))) rows += fmt_val(r.get("pnl_pct", float("nan")), is_pct=True) rows += fmt_small(r.get("sharpe", float("nan"))) rows += fmt_small(r.get("max_dd", float("nan"))) rows += fmt_small(r.get("ic", float("nan"))) rows += f"{r.get('n_trades', 0)}" rows += f"{r.get('n_predictions', 0)}" rows += "" # Delta B-A and C-A rows ra = stock_results.get("A", {}) rb = stock_results.get("B", {}) rc = stock_results.get("C", {}) for label, rd in [("ΔB-A", rb), ("ΔC-A", rc)]: if ra and rd: rows += '' rows += f"  {label}" for key in ["accuracy", "dir_accuracy", "up_precision", "dn_precision", "win_rate", "pnl_pct"]: va = ra.get(key, float("nan")) vd = rd.get(key, float("nan")) if np.isnan(va) or np.isnan(vd): rows += "—" else: rows += fmt_val(vd - va, is_delta=True) rows += "" rows += "" # Aggregate rows rows += " " for sub_key, css_cls in [("A", "subset-a"), ("B", "subset-b"), ("C", "subset-c")]: r = aggregate.get(sub_key, {}) if not r: continue size = subsets_sizes.get(sub_key, "?") w_cls = " winner" if sub_key == winner else "" rows += f'' rows += f"AGG {sub_key}({size}f)" rows += fmt_val(r.get("accuracy", float("nan"))) rows += fmt_val(r.get("dir_accuracy", float("nan"))) rows += fmt_val(r.get("up_precision", float("nan"))) rows += fmt_val(r.get("dn_precision", float("nan"))) rows += fmt_val(r.get("win_rate", float("nan"))) rows += fmt_val(r.get("pnl_pct", float("nan")), is_pct=True) rows += fmt_small(r.get("sharpe", float("nan"))) rows += fmt_small(r.get("max_dd", float("nan"))) rows += fmt_small(r.get("ic", float("nan"))) rows += "" rows += "" ts = datetime.now().strftime("%Y-%m-%d %H:%M") html = f""" 3-way Backtest {ts}{css}

3-Way Feature Subset Backtest

Walk-forward | MIN_TRAIN={MIN_TRAIN} | STEP={STEP}d | LABEL=5d ±2% | RF(200est,d6,leaf10)

Generated: {ts} | Winner: {winner} ({subsets_sizes[winner]}f)

{headers} {rows}
""" return html def main(): from models.predictor import _build_features docs = ROOT / "docs" docs.mkdir(exist_ok=True) all_results: dict[str, dict[str, dict]] = {} agg_lists: dict[str, list] = {"A": [], "B": [], "C": []} for stock_no in STOCKS: print(f" [{stock_no}] loading...", file=sys.stderr) try: df = fetch_df(stock_no) except Exception as e: print(f" [{stock_no}] fetch error: {e}", file=sys.stderr) continue if df is None or df.empty or len(df) < MIN_TRAIN + STEP + LABEL_HORIZON: print(f" [{stock_no}] insufficient data", file=sys.stderr) continue feat = _build_features(df) close = df["close"] if "date" not in df.columns else df.set_index("date")["close"] labels = make_labels(close.reset_index(drop=True)) feat["_label"] = labels feat["_close"] = close.reset_index(drop=True).values stock_res = {} for sub_key, sub_cols in SUBSETS.items(): print(f" [{stock_no}] backtest {sub_key}({len(sub_cols)}f)...", file=sys.stderr) try: r = walk_forward(feat, sub_cols) if r: stock_res[sub_key] = r agg_lists[sub_key].append(r) except Exception as e: print(f" [{stock_no}] {sub_key} error: {e}", file=sys.stderr) if stock_res: all_results[stock_no] = stock_res # Aggregate aggregate = {k: mean_metrics(v) for k, v in agg_lists.items()} # Winner: highest mean dir_accuracy winner = max( ["A", "B", "C"], key=lambda k: aggregate.get(k, {}).get("dir_accuracy", float("-inf")), ) winner_vals = {k: aggregate.get(k, {}).get("dir_accuracy", float("nan")) for k in ["A", "B", "C"]} winner_reason = ( f"Subset {winner} has highest mean dir_accuracy " f"(A={winner_vals['A']:.1f}%, B={winner_vals['B']:.1f}%, C={winner_vals['C']:.1f}%)" ) subsets_sizes = {"A": len(SUBSET_A), "B": len(SUBSET_B), "C": len(SUBSET_C)} output = { "subsets": subsets_sizes, "results": all_results, "aggregate": aggregate, "winner": winner, "winner_reason": winner_reason, } json_path = docs / "backtest_3way.json" with open(json_path, "w") as f: json.dump(output, f, indent=2) print(f"Wrote {json_path}", file=sys.stderr) html = generate_html(all_results, aggregate, subsets_sizes, winner) html_path = docs / "backtest_3way.html" with open(html_path, "w") as f: f.write(html) print(f"Wrote {html_path}", file=sys.stderr) # Console summary print(f"\n{'='*60}") print(f"Winner: {winner} ({subsets_sizes[winner]}f)") print(f"{winner_reason}") print(f"\nAggregate metrics (mean across {len(all_results)} stocks):") hdr = f" {'Subset':<8} {'Acc%':>5} {'Dir%':>5} {'↑Prec':>6} {'↓Prec':>6} {'Win%':>5} {'P&L%':>6} {'Sharpe':>6} {'MaxDD':>6} {'IC':>6}" print(hdr) print(" " + "-" * (len(hdr) - 2)) for k in ["A", "B", "C"]: r = aggregate.get(k, {}) def f(v): return f"{v:>6.1f}" if not np.isnan(v) else " N/A" def fs(v): return f"{v:>6.3f}" if not np.isnan(v) else " N/A" print( f" {k}({subsets_sizes[k]}f) " f"{f(r.get('accuracy', float('nan')))} " f"{f(r.get('dir_accuracy', float('nan')))} " f"{f(r.get('up_precision', float('nan')))} " f"{f(r.get('dn_precision', float('nan')))} " f"{f(r.get('win_rate', float('nan')))} " f"{f(r.get('pnl_pct', float('nan')))} " f"{fs(r.get('sharpe', float('nan')))} " f"{fs(r.get('max_dd', float('nan')))} " f"{fs(r.get('ic', float('nan')))}" ) if __name__ == "__main__": main()