scripts/backtest_c9.py

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