scripts/backtest_c4.py

#!/usr/bin/env python3
"""C4 validation: triple-barrier labels + meta-labeling filter."""
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 numpy as np
import pandas as pd
import json
from sklearn.ensemble import RandomForestClassifier
from models.predictor import FEATURE_COLUMNS, _build_features, _build_triple_barrier_labels

STOCKS       = ["2330", "0050", "2317", "2454", "2881"]
LABEL_HORIZON = 5
MIN_TRAIN    = 240
STEP         = 21
META_SPLIT   = 0.70   # 70% of training window for primary, 30% for meta
META_THRESH  = [0.50, 0.55]
RF = dict(n_estimators=200, max_depth=6, min_samples_leaf=10,
          class_weight="balanced", random_state=42, n_jobs=-1)
RF_META = dict(n_estimators=100, max_depth=4, min_samples_leaf=5,
               random_state=42, n_jobs=-1)   # meta is simpler — avoids overfitting


def fetch_df(stock_no):
    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 fetch_cross_asset_tw
    df = _fetch_with_cache(stock_no, months=24)
    if df is None or df.empty: return None
    df = add_all_indicators(df)
    df = add_institutional_flow(df, stock_no)
    df = add_margin_flow(df, stock_no)
    start, end = str(df["date"].min()), str(df["date"].max())
    result = fetch_cross_asset_tw(start, end)
    taiex, usdtwd = result[0], result[1]
    sox = result[2] if len(result) > 2 else None
    tnx = result[3] if len(result) > 3 else None
    df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx)
    return df


def metrics(y_true, y_pred):
    if len(y_true) == 0:
        return {}
    dir_mask = y_pred != 0
    acc = (y_true == y_pred).mean() * 100
    dir_acc = (y_true[dir_mask] == y_pred[dir_mask]).mean() * 100 if dir_mask.sum() > 0 else float("nan")
    up_mask  = y_pred == 1
    up_prec  = (y_true[up_mask] == 1).mean() * 100 if up_mask.sum() > 0 else float("nan")
    dn_mask  = y_pred == -1
    dn_prec  = (y_true[dn_mask] == -1).mean() * 100 if dn_mask.sum() > 0 else float("nan")
    return {"accuracy": round(acc, 1), "dir_accuracy": round(dir_acc, 1),
            "up_precision": round(up_prec, 1), "dn_precision": round(dn_prec, 1),
            "n_signals": int(dir_mask.sum()), "n_predictions": len(y_true)}


def walk_forward_meta(feat_df, label_arr, cols):
    """
    Returns dict with keys: primary, meta_50, meta_55
    Each value is a dict of metrics.
    """
    avail = [c for c in cols if c in feat_df.columns]
    X_all = feat_df[avail].fillna(0).values
    n = len(feat_df)

    results = {f"meta_{int(t*100)}": {"y_true": [], "y_pred": []} for t in META_THRESH}
    results["primary"] = {"y_true": [], "y_pred": []}

    cutoff = MIN_TRAIN
    while cutoff + STEP + LABEL_HORIZON <= n:
        y_full = label_arr[:cutoff]
        valid_mask = ~np.isnan(y_full)

        y_valid = y_full[valid_mask]
        if len(y_valid) < 20 or len(np.unique(y_valid)) < 2:
            cutoff += STEP; continue

        # ── Primary model: train on first META_SPLIT fraction ──────────────
        split_idx = int(cutoff * META_SPLIT)
        y_prim_train = y_full[:split_idx]
        valid_prim = ~np.isnan(y_prim_train)
        if valid_prim.sum() < 10 or len(np.unique(y_prim_train[valid_prim])) < 2:
            cutoff += STEP; continue

        prim_clf = RandomForestClassifier(**RF)
        prim_clf.fit(X_all[:split_idx][valid_prim], y_prim_train[valid_prim].astype(int))

        # ── Meta training set: primary predicts on [split_idx, cutoff] ─────
        X_meta_tr = X_all[split_idx:cutoff]
        y_meta_true = y_full[split_idx:cutoff]
        valid_meta = ~np.isnan(y_meta_true)
        if valid_meta.sum() < 5:
            cutoff += STEP; continue

        X_meta_tr_v = X_meta_tr[valid_meta]
        y_meta_true_v = y_meta_true[valid_meta].astype(int)

        prim_pred_meta = prim_clf.predict(X_meta_tr_v)
        try:
            prim_proba_meta = prim_clf.predict_proba(X_meta_tr_v)  # shape (n, n_classes)
        except Exception:
            cutoff += STEP; continue

        # Meta label: 1 if primary was correct, 0 if wrong
        meta_labels = (prim_pred_meta == y_meta_true_v).astype(int)

        # Only train meta on directional predictions (skip HOLD)
        dir_mask_meta = prim_pred_meta != 0
        if dir_mask_meta.sum() < 5:
            cutoff += STEP; continue

        # Feature for meta = original features + class probabilities
        X_meta_feat = np.hstack([X_meta_tr_v[dir_mask_meta], prim_proba_meta[dir_mask_meta]])
        meta_labels_dir = meta_labels[dir_mask_meta]

        if len(np.unique(meta_labels_dir)) < 2:
            cutoff += STEP; continue

        meta_clf = RandomForestClassifier(**RF_META)
        meta_clf.fit(X_meta_feat, meta_labels_dir)

        # ── Test window ────────────────────────────────────────────────────
        test_end = min(cutoff + STEP, n - LABEL_HORIZON)
        y_te = label_arr[cutoff:test_end]
        valid_te = ~np.isnan(y_te)
        if valid_te.sum() == 0:
            cutoff += STEP; continue

        X_te = X_all[cutoff:test_end][valid_te]
        y_te_v = y_te[valid_te].astype(int)

        prim_pred_te = prim_clf.predict(X_te)
        prim_proba_te = prim_clf.predict_proba(X_te)

        # Primary-only results
        results["primary"]["y_true"].extend(y_te_v.tolist())
        results["primary"]["y_pred"].extend(prim_pred_te.tolist())

        # Meta-filtered results at each threshold
        X_meta_te = np.hstack([X_te, prim_proba_te])
        # Only run meta on directional predictions
        dir_te = prim_pred_te != 0
        meta_conf = np.zeros(len(X_te))
        if dir_te.sum() > 0:
            try:
                mc = meta_clf.predict_proba(X_meta_te[dir_te])
                # meta_clf predicts [wrong, correct]; col 1 = P(correct)
                classes = list(meta_clf.classes_)
                correct_col = classes.index(1) if 1 in classes else -1
                if correct_col >= 0:
                    meta_conf[dir_te] = mc[:, correct_col]
                else:
                    meta_conf[dir_te] = 0.0
            except Exception:
                pass

        for thresh in META_THRESH:
            key = f"meta_{int(thresh*100)}"
            final_pred = np.where(
                (prim_pred_te != 0) & (meta_conf >= thresh),
                prim_pred_te,
                0
            )
            results[key]["y_true"].extend(y_te_v.tolist())
            results[key]["y_pred"].extend(final_pred.tolist())

        cutoff += STEP

    # Compute metrics
    out = {}
    for key, data in results.items():
        if not data["y_true"]:
            out[key] = {}
        else:
            out[key] = metrics(np.array(data["y_true"]), np.array(data["y_pred"]))
    return out


def main():
    all_results = {}
    hdr = f"{'Stock':>6}  {'Method':>10}  {'Dir%':>5}  {'↑Prec%':>7}  {'Signals':>7}  {'Acc%':>5}"
    print(f"\n{hdr}\n{'-'*len(hdr)}")
    agg = {"primary": [], "meta_50": [], "meta_55": []}

    for stock_no in STOCKS:
        print(f"  computing {stock_no}...", end="\r", flush=True)
        df = fetch_df(stock_no)
        if df is None or df.empty:
            print(f"{stock_no:>6}  no data"); continue
        feat = _build_features(df)
        close = df.set_index("date")["close"] if "date" in df.columns else df["close"]
        tbl = _build_triple_barrier_labels(close)
        label_arr = tbl.values

        res = walk_forward_meta(feat, label_arr, FEATURE_COLUMNS)
        all_results[stock_no] = res

        for method in ("primary", "meta_50", "meta_55"):
            r = res.get(method, {})
            if r:
                print(f"{stock_no:>6}  {method:>10}  {r['dir_accuracy']:>5.1f}  "
                      f"{r['up_precision']:>7.1f}  {r.get('n_signals',0):>7}  {r['accuracy']:>5.1f}")
                agg[method].append(r)
        print()

    def mf(rows, f):
        vals = [r[f] for r in rows if isinstance(r.get(f), (int,float)) and not np.isnan(r.get(f, float("nan")))]
        return round(sum(vals)/len(vals), 1) if vals else float("nan")

    print("=== AGGREGATE ===")
    agg_summary = {}
    for method in ("primary", "meta_50", "meta_55"):
        rows = agg[method]
        s = {k: mf(rows, k) for k in ("accuracy","dir_accuracy","up_precision","dn_precision","n_signals")}
        agg_summary[method] = s
        print(f"  {method:>10}: dir={s['dir_accuracy']}%  ↑prec={s['up_precision']}%  signals={s['n_signals']}")

    best_method = max(("meta_50","meta_55"),
                      key=lambda m: agg_summary[m].get("up_precision", 0))
    passed = agg_summary[best_method].get("up_precision", 0) >= 54.0
    print(f"\n  Best meta: {best_method}  |  Pass: {'YES' if passed else 'NO'}")

    Path("docs").mkdir(exist_ok=True)
    with open("docs/c4_meta_label_result.json", "w") as f:
        json.dump({"results": all_results, "aggregate": agg_summary,
                   "best_meta": best_method, "passed": passed,
                   "pass_threshold": {"up_precision": 54.0}}, f, indent=2)

if __name__ == "__main__":
    main()