scripts/backtest_c7.py

#!/usr/bin/env python3
"""C7 validation: BASELINE vs BASELINE + calendar features + 36-month lookback."""
import sys
import json
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

from scripts.improvement_harness import (
    BASELINE_FEATURES,
    build_triple_barrier_labels,
    compute_metrics,
    walk_forward,
    RF_PARAMS,
    DEFAULT_STOCKS,
)

CALENDAR_FEATURES = [
    "dow_sin", "dow_cos",
    "month_sin", "month_cos",
    "is_options_expiry_week",
    "is_earnings_season",
]

C7_FEATURES = BASELINE_FEATURES + CALENDAR_FEATURES


def fetch_df_36(stock_no: str):
    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=36)
    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 _mean(rows, field):
    vals = [r[field] for r in rows
            if isinstance(r.get(field), (int, float)) and not np.isnan(r.get(field, float("nan")))]
    return round(sum(vals) / len(vals), 1) if vals else float("nan")


def main():
    from models.predictor import _build_features

    feature_sets = [("baseline", BASELINE_FEATURES), ("calendar", C7_FEATURES)]
    per_stock = {}
    agg = {"baseline": [], "calendar": []}

    hdr = f"{'Stock':>6}  {'Set':>12}  {'Acc%':>5}  {'Dir%':>5}  {'↑Prec%':>7}  {'Signals':>7}"
    print(f"\n{hdr}\n{'-' * len(hdr)}")

    for stock_no in DEFAULT_STOCKS:
        print(f"  computing {stock_no}...", end="\r", flush=True)
        df = fetch_df_36(stock_no)
        if df is None or df.empty:
            print(f"{stock_no:>6}  no data")
            continue

        feat = _build_features(df)
        # Merge calendar columns from df into feat so walk_forward can use them
        for col in CALENDAR_FEATURES:
            if col in df.columns:
                feat[col] = df[col].values
        close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values
        labels = build_triple_barrier_labels(close)

        per_stock[stock_no] = {}
        for name, cols in feature_sets:
            r = walk_forward(feat, labels, cols)
            per_stock[stock_no][name] = r
            if r:
                agg[name].append(r)
                prefix = f"{stock_no:>6}" if name == "baseline" else f"{'':>6}"
                print(f"{prefix}  {name:>12}  {r['accuracy']:>5.1f}  {r['dir_accuracy']:>5.1f}  "
                      f"{r['up_precision']:>7.1f}  {r.get('n_signals', 0):>7}")

        r0 = per_stock[stock_no].get("baseline", {})
        r1 = per_stock[stock_no].get("calendar", {})
        if r0 and r1:
            dd = r1["dir_accuracy"] - r0["dir_accuracy"]
            dp = r1["up_precision"] - r0["up_precision"]
            print(f"{'':>6}  {'Δ':>12}  {'':>5}  {dd:>+5.1f}  {dp:>+7.1f}")
        print()

    print("=== AGGREGATE ===")
    agg_summary = {}
    for name, rows in agg.items():
        s = {k: _mean(rows, k) for k in ("accuracy", "dir_accuracy", "up_precision", "dn_precision", "n_signals")}
        agg_summary[name] = s
        print(f"  {name:>12}: acc={s['accuracy']}%  dir={s['dir_accuracy']}%  "
              f"↑prec={s['up_precision']}%  signals={s['n_signals']}")

    PASS_DIR_ACC = 43.0
    PASS_UP_PREC = 54.0
    cal = agg_summary.get("calendar", {})
    bas = agg_summary.get("baseline", {})

    cal_dir = cal.get("dir_accuracy", 0)
    cal_prec = cal.get("up_precision", 0)
    bas_dir = bas.get("dir_accuracy", 0)
    bas_prec = bas.get("up_precision", 0)

    meets_threshold = (cal_dir >= PASS_DIR_ACC) or (cal_prec >= PASS_UP_PREC)
    no_regression = (
        (cal_dir >= bas_dir - 1.0) and
        (cal_prec >= bas_prec - 1.0)
    )
    passed = meets_threshold and no_regression

    print(f"\n  calendar dir={cal_dir}%  ↑prec={cal_prec}%")
    print(f"  Pass (dir≥{PASS_DIR_ACC} OR ↑prec≥{PASS_UP_PREC}, no regression>1pp): {'YES' if passed else 'NO'}")

    result = {
        "results": per_stock,
        "aggregate": agg_summary,
        "passed": passed,
        "pass_criterion": f"dir_accuracy >= {PASS_DIR_ACC} OR up_precision >= {PASS_UP_PREC}",
    }

    def _jsonify(obj):
        if isinstance(obj, dict):
            return {k: _jsonify(v) for k, v in obj.items()}
        if isinstance(obj, list):
            return [_jsonify(v) for v in 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)
        return obj

    out = ROOT / "docs" / "c7_calendar_result.json"
    out.parent.mkdir(exist_ok=True)
    with open(out, "w") as f:
        json.dump(_jsonify(result), f, indent=2)
    print(f"\n  Written: {out}")


if __name__ == "__main__":
    main()