#!/usr/bin/env python3 """C9 validation: single-stock training vs multi-stock training universe.""" import sys from pathlib import Path 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 warnings; warnings.filterwarnings("ignore") import json import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier from scripts.improvement_harness import ( BASELINE_FEATURES, fetch_df, build_triple_barrier_labels, compute_metrics, RF_PARAMS, MIN_TRAIN, STEP, LABEL_HORIZON, walk_forward, ) from models.predictor import _build_features TEST_STOCKS = ["2330", "0050", "2317", "2454", "2881"] UNIVERSE = list(dict.fromkeys([ "2330", "0050", "2317", "2454", "2881", "2382", "2303", "2308", "3034", "2412", "6505", "6669", "2886", "2891", "3008", "4904", "2882", "2885", "1301", "2912" ])) def walk_forward_multi( test_feat_df, test_labels, all_pool_rows, test_dates, cols=BASELINE_FEATURES, ): """Walk-forward where training pulls from full universe pool.""" avail = [c for c in cols if c in test_feat_df.columns] X_test_all = test_feat_df.reindex(columns=avail, fill_value=0).fillna(0).values n = len(test_feat_df) y_true_all, y_pred_all = [], [] cutoff = MIN_TRAIN while cutoff + STEP + LABEL_HORIZON <= n: cutoff_date = test_dates[cutoff] if cutoff < len(test_dates) else None X_pool, y_pool = [], [] for (pool_feat_df, pool_labels, pool_dates) in all_pool_rows: pool_avail = [c for c in cols if c in pool_feat_df.columns] if len(pool_avail) < len(avail) * 0.8: continue X_p = pool_feat_df.reindex(columns=avail, fill_value=0).fillna(0).values if cutoff_date is not None: mask_date = pool_dates < cutoff_date else: mask_date = np.ones(len(pool_labels), dtype=bool) valid = ~np.isnan(pool_labels) & mask_date if valid.sum() < 5: continue X_pool.append(X_p[valid]) y_pool.append(pool_labels[valid]) if not X_pool: cutoff += STEP continue X_tr = np.vstack(X_pool) y_tr = np.concatenate(y_pool).astype(int) if len(np.unique(y_tr)) < 2: cutoff += STEP continue clf = RandomForestClassifier(**RF_PARAMS) clf.fit(X_tr, y_tr) test_end = min(cutoff + STEP, n - LABEL_HORIZON) y_te = test_labels[cutoff:test_end] valid_te = ~np.isnan(y_te) if valid_te.sum() == 0: cutoff += STEP continue y_pred = clf.predict(X_test_all[cutoff:test_end][valid_te]) y_true_all.extend(y_te[valid_te].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(): print("Fetching universe data...") pool_data = {} # stock_no -> (feat_df, labels, dates_array) for stock_no in UNIVERSE: print(f" {stock_no}...", end=" ", flush=True) try: df = fetch_df(stock_no) if df is None or df.empty: print("no data") continue feat = _build_features(df) date_col = df["date"] if "date" in df.columns else df.index.to_series() dates = pd.to_datetime(date_col).values # numpy datetime64 array close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values labels = build_triple_barrier_labels(close) pool_data[stock_no] = (feat, labels, dates) print(f"rows={len(feat)}") except Exception as e: print(f"error: {e}") stocks_fetched = len(pool_data) print(f"\nFetched {stocks_fetched}/{len(UNIVERSE)} stocks") all_pool_rows = list(pool_data.values()) print("\nRunning walk-forward...") results = {} for stock_no in TEST_STOCKS: if stock_no not in pool_data: print(f" {stock_no}: no data, skipping") continue feat_df, labels, dates = pool_data[stock_no] print(f" {stock_no} single...", end=" ", flush=True) single = walk_forward(feat_df, labels, BASELINE_FEATURES) print(f"dir={single.get('dir_accuracy','?')} up={single.get('up_precision','?')}", end=" ") print(f"multi...", end=" ", flush=True) multi = walk_forward_multi(feat_df, labels, all_pool_rows, dates) print(f"dir={multi.get('dir_accuracy','?')} up={multi.get('up_precision','?')}") results[stock_no] = {"single": single, "multi": multi} # Aggregate def _mean(dicts, key): vals = [d[key] for d in dicts if isinstance(d.get(key), (int, float)) and not np.isnan(d.get(key, float("nan")))] return round(sum(vals) / len(vals), 1) if vals else float("nan") single_rows = [v["single"] for v in results.values() if v["single"]] multi_rows = [v["multi"] for v in results.values() if v["multi"]] agg_keys = ["accuracy", "dir_accuracy", "up_precision", "dn_precision", "n_signals"] agg = { "single": {k: _mean(single_rows, k) for k in agg_keys}, "multi": {k: _mean(multi_rows, k) for k in agg_keys}, } print("\n=== AGGREGATE ===") for mode in ("single", "multi"): s = agg[mode] print(f" {mode:>6}: acc={s['accuracy']}% dir={s['dir_accuracy']}% " f"↑prec={s['up_precision']}% signals={s['n_signals']}") multi_dir = agg["multi"]["dir_accuracy"] multi_prec = agg["multi"]["up_precision"] passed = ( isinstance(multi_dir, float) and multi_dir >= 44.0 and isinstance(multi_prec, float) and multi_prec >= 54.0 ) print(f"\n Pass (dir≥44.0 AND up_prec≥54.0): {'YES' if passed else 'NO'}") output = { "universe_size": len(UNIVERSE), "stocks_fetched": stocks_fetched, "results": results, "aggregate": agg, "passed": passed, "pass_criterion": "dir_accuracy >= 44.0 AND up_precision >= 54.0", } out_path = ROOT / "docs" / "c9_multi_result.json" class _NumpyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, (np.integer,)): return int(obj) if isinstance(obj, (np.floating,)): return float(obj) if isinstance(obj, (np.bool_,)): return bool(obj) if isinstance(obj, np.ndarray): return obj.tolist() return super().default(obj) with open(out_path, "w") as f: json.dump(output, f, indent=2, cls=_NumpyEncoder) print(f"\nWrote {out_path}") return output if __name__ == "__main__": main()