scripts/backtest_c3.py

#!/usr/bin/env python3
"""C3 validation: fixed-horizon labels vs triple-barrier labels."""
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

STOCKS = ["2330", "0050", "2317", "2454", "2881"]
LABEL_HORIZON = 5
FIXED_THRESH  = 0.02
PT_SL_RATIO   = 1.0   # symmetric barriers; scale = 1 * daily_vol
VOL_LOOKBACK  = 20
MIN_TRAIN     = 240
STEP          = 21
RF = 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):
    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 fixed_labels(close: np.ndarray) -> np.ndarray:
    """5-day forward return ±2% threshold."""
    n = len(close)
    labels = np.zeros(n, dtype=float)
    for i in range(n - LABEL_HORIZON):
        ret = (close[i + LABEL_HORIZON] - close[i]) / close[i]
        labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0)
    labels[-LABEL_HORIZON:] = np.nan  # no forward data
    return labels


def triple_barrier_labels(close: np.ndarray) -> np.ndarray:
    """
    For each row t:
      - compute daily_vol = rolling std of log returns over VOL_LOOKBACK days
      - upper = close[t] * (1 + daily_vol * PT_SL_RATIO)
      - lower = close[t] * (1 - daily_vol * PT_SL_RATIO)
      - scan close[t+1 .. t+LABEL_HORIZON], find which barrier is hit first
      - label: 1 (upper), -1 (lower), 0 (neither = vertical barrier expired)
    """
    n = len(close)
    log_ret = np.diff(np.log(close + 1e-9))  # length n-1
    labels = np.zeros(n, dtype=float)

    for i in range(n - LABEL_HORIZON):
        # Daily vol: std of log_ret in [i-VOL_LOOKBACK, i)
        start_vol = max(0, i - VOL_LOOKBACK)
        window = log_ret[start_vol:i]
        if len(window) < 5:
            # Not enough data — fall back to fixed threshold
            ret = (close[i + LABEL_HORIZON] - close[i]) / close[i]
            labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0)
            continue

        daily_vol = float(np.std(window))
        if daily_vol < 0.001:
            daily_vol = 0.001  # floor

        upper = close[i] * (1.0 + daily_vol * PT_SL_RATIO)
        lower = close[i] * (1.0 - daily_vol * PT_SL_RATIO)

        label = 0  # default: vertical barrier (HOLD)
        for j in range(1, LABEL_HORIZON + 1):
            c = close[i + j]
            if c >= upper:
                label = 1
                break
            if c <= lower:
                label = -1
                break

        labels[i] = label

    labels[-LABEL_HORIZON:] = np.nan
    return labels


def walk_forward(feat_df, label_arr, cols):
    avail = [c for c in cols if c in feat_df.columns]
    X_all = feat_df[avail].fillna(0).values
    n = len(feat_df)
    y_true_all, y_pred_all = [], []
    cutoff = MIN_TRAIN
    while cutoff + STEP + LABEL_HORIZON <= n:
        y_tr = label_arr[:cutoff]
        valid = ~np.isnan(y_tr)
        if valid.sum() < 2 or len(np.unique(y_tr[valid])) < 2:
            cutoff += STEP; continue
        clf = RandomForestClassifier(**RF)
        clf.fit(X_all[:cutoff][valid], y_tr[valid].astype(int))
        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
        y_pred = clf.predict(X_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 {}
    y_true = np.array(y_true_all)
    y_pred = np.array(y_pred_all)
    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_predictions": len(y_true), "n_features": len(avail),
            "label_up_pct": round((y_true == 1).mean() * 100, 1),
            "label_dn_pct": round((y_true == -1).mean() * 100, 1)}


def main():
    results = {}
    hdr = f"{'Stock':>6}  {'Label':>12}  {'Acc%':>5}  {'Dir%':>5}  {'↑Prec%':>7}  {'↓Prec%':>7}  {'Preds':>5}"
    print(f"\n{hdr}\n{'-'*len(hdr)}")
    agg = {"fixed": [], "triple": []}

    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"].values if "date" in df.columns else df["close"].values

        fl = fixed_labels(close)
        tbl = triple_barrier_labels(close)

        results[stock_no] = {}
        for name, labels in [("fixed", fl), ("triple", tbl)]:
            r = walk_forward(feat, labels, FEATURE_COLUMNS)
            results[stock_no][name] = r
            if r:
                print(f"{stock_no:>6}  {name:>12}  {r['accuracy']:>5.1f}  {r['dir_accuracy']:>5.1f}  "
                      f"{r['up_precision']:>7.1f}  {r['dn_precision']:>7.1f}  {r['n_predictions']:>5}")
                agg[name].append(r)

        if results[stock_no].get("fixed") and results[stock_no].get("triple"):
            rf, rt = results[stock_no]["fixed"], results[stock_no]["triple"]
            dd = rt["dir_accuracy"] - rf["dir_accuracy"]
            dp = rt["up_precision"] - rf["up_precision"]
            print(f"{'':>6}  {'Δ':>12}  {'':>5}  {dd:>+5.1f}  {dp:>+7.1f}")
        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 name in ("fixed", "triple"):
        rows = agg[name]
        s = {k: mf(rows, k) for k in ("accuracy", "dir_accuracy", "up_precision", "dn_precision")}
        agg_summary[name] = s
        print(f"  {name:>12}: acc={s['accuracy']}%  dir={s['dir_accuracy']}%  ↑prec={s['up_precision']}%  ↓prec={s['dn_precision']}%")

    tb = agg_summary["triple"]
    passed = tb["dir_accuracy"] >= 31.5 or tb["up_precision"] >= 43.2
    winner = "triple" if (tb["dir_accuracy"] > agg_summary["fixed"]["dir_accuracy"] or
                          tb["up_precision"] > agg_summary["fixed"]["up_precision"]) else "fixed"
    print(f"\n  Winner: {winner}  |  Pass: {'YES' if passed else 'NO'}")

    Path("docs").mkdir(exist_ok=True)
    with open("docs/c3_triple_barrier_result.json", "w") as f:
        json.dump({"results": results, "aggregate": agg_summary,
                   "winner": winner, "passed": passed,
                   "pass_threshold": {"dir_accuracy": 31.5, "up_precision": 43.2}}, f, indent=2)

if __name__ == "__main__":
    main()