scripts/backtest_3way.py

#!/usr/bin/env python3
"""
3-way walk-forward backtest: SUBSET_A (40f) vs SUBSET_B (audit-filtered) vs SUBSET_C (all 85f).

Usage:
  ./venv/bin/python3 scripts/backtest_3way.py 2>/dev/null
"""

import json
import sys
import warnings
from datetime import datetime
from pathlib import Path

warnings.filterwarnings("ignore")

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 numpy as np
import pandas as pd
from scipy.stats import spearmanr
from sklearn.ensemble import RandomForestClassifier

# ── Feature subsets ────────────────────────────────────────────────────────────

SUBSET_A = [
    "return_1d", "return_5d", "return_10d", "return_20d",
    "close_ma5_ratio", "close_ma20_ratio", "ma5_ma20_ratio", "ma20_ma60_ratio",
    "rsi",
    "macd_hist", "macd_signal_ratio", "macd_hist_norm", "macd_hist_delta_1d",
    "macd_hist_slope_3d", "macd_cross_up", "macd_cross_down", "macd_above_zero",
    "bb_pct_b", "k", "d", "volume_ratio",
    "atr_ratio", "high_low_ratio", "obv_trend",
    "close_ma60_ratio", "log_volume_ratio",
    "volume_zscore", "price_volume_div", "volatility_20d",
    "taiex_return_5d", "taiex_ma20_ratio", "usdtwd_return_5d",
    "foreign_net_vol_ratio", "trust_net_vol_ratio", "dealer_net_vol_ratio",
    "institutional_net_vol_ratio", "institutional_5d_net_vol_ratio",
    "institutional_20d_zscore", "foreign_trust_alignment", "institutional_streak",
]

# SUBSET_B: SUBSET_A + audit-filtered extras (neg_count <= 2, banned-list excluded)
SUBSET_B = [
    # -- all of SUBSET_A --
    "return_1d", "return_5d", "return_10d", "return_20d",
    "close_ma5_ratio", "close_ma20_ratio", "ma5_ma20_ratio", "ma20_ma60_ratio",
    "rsi",
    "macd_hist", "macd_signal_ratio", "macd_hist_norm", "macd_hist_delta_1d",
    "macd_hist_slope_3d", "macd_cross_up", "macd_cross_down", "macd_above_zero",
    "bb_pct_b", "k", "d", "volume_ratio",
    "atr_ratio", "high_low_ratio", "obv_trend",
    "close_ma60_ratio", "log_volume_ratio",
    "volume_zscore", "price_volume_div", "volatility_20d",
    "taiex_return_5d", "taiex_ma20_ratio", "usdtwd_return_5d",
    "foreign_net_vol_ratio", "trust_net_vol_ratio", "dealer_net_vol_ratio",
    "institutional_net_vol_ratio", "institutional_5d_net_vol_ratio",
    "institutional_20d_zscore", "foreign_trust_alignment", "institutional_streak",
    # -- extras: neg_count <= 2 from audit, banned-list excluded --
    "close_ma240_ratio",       # neg=0, perm=+0.00583
    "golden_cross_5_20",       # neg=0, perm=+0.00000 (event feature, low noise)
    "death_cross_5_20",        # neg=0
    "golden_cross_20_60",      # neg=0
    "death_cross_20_60",       # neg=0
    "sox_ret_1d",              # neg=2, perm=+0.00146
    "margin_ratio",            # neg=0
    "short_margin_ratio",      # neg=0
    "margin_5d_change_pct",    # neg=0
    "margin_buy_pressure",     # neg=0
    "chip_squeeze_signal",     # neg=0
    "chip_score",              # neg=0
    "ma_bull_alignment",       # neg=1, perm=-0.00063
    "ma_bear_alignment",       # neg=0, perm=+0.00167
    "ma_alignment_days",       # neg=2, perm=+0.00146
    "bias_20",                 # neg=2, perm=+0.00729
    "vol_volatility_interact", # neg=1, perm=+0.00604
    "short_balance_5d_change", # neg=0
    "trust_vol_ratio",         # neg=0
    "return_60d",              # neg=2, perm=+0.00146
    "return_120d",             # neg=1, perm=+0.00333
    "mom_minus_reversal",      # neg=1, perm=+0.00854
]

from models.predictor import FEATURE_COLUMNS as SUBSET_C

SUBSETS = {"A": SUBSET_A, "B": SUBSET_B, "C": SUBSET_C}

STOCKS = ["2330", "0050", "2317", "2454", "2881"]
LABEL_HORIZON  = 5
LABEL_THRESH   = 0.02
MIN_TRAIN      = 240
STEP           = 21

RF_PARAMS = 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: str) -> pd.DataFrame:
    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 is_us_ticker, fetch_cross_asset_tw

    df = _fetch_with_cache(stock_no, months=24)
    if df is None or df.empty:
        return pd.DataFrame()

    df = add_all_indicators(df)
    if not is_us_ticker(stock_no):
        df = add_institutional_flow(df, stock_no)
        df = add_margin_flow(df, stock_no)
        start, end = str(df["date"].min()), str(df["date"].max())
        taiex, usdtwd, sox, tnx = fetch_cross_asset_tw(start, end)
        df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx)
    return df


def make_labels(close: pd.Series) -> np.ndarray:
    fwd = close.shift(-LABEL_HORIZON)
    ret = (fwd - close) / close
    return np.where(ret > LABEL_THRESH, 1, np.where(ret < -LABEL_THRESH, -1, 0))


def walk_forward(feat: pd.DataFrame, feature_cols: list[str]) -> dict:
    avail = [c for c in feature_cols if c in feat.columns]
    labels = feat["_label"].values
    X_all  = feat[avail].fillna(0).values
    close  = feat["_close"].values
    n      = len(feat)

    y_true_all, y_pred_all = [], []
    pnl_model = []
    ic_pairs = []  # (prob_up, actual_fwd_ret)

    cutoff = MIN_TRAIN
    while cutoff + STEP + LABEL_HORIZON <= n:
        X_train = X_all[:cutoff]
        y_train = labels[:cutoff]

        test_end = min(cutoff + STEP, n - LABEL_HORIZON)
        X_test   = X_all[cutoff:test_end]
        y_test   = labels[cutoff:test_end]
        c_test   = close[cutoff:test_end]

        if len(np.unique(y_train)) < 2 or len(X_test) == 0:
            cutoff += STEP
            continue

        clf = RandomForestClassifier(**RF_PARAMS)
        clf.fit(X_train, y_train)
        y_pred = clf.predict(X_test)
        proba  = clf.predict_proba(X_test)
        classes = list(clf.classes_)
        prob_map = {cls: proba[:, i] for i, cls in enumerate(classes)}

        y_true_all.extend(y_test.tolist())
        y_pred_all.extend(y_pred.tolist())

        for i in range(len(X_test)):
            p_up = float(prob_map.get(1, np.zeros(len(X_test)))[i])
            if (cutoff + i + LABEL_HORIZON) < n:
                fwd_ret = (close[cutoff + i + LABEL_HORIZON] - c_test[i]) / c_test[i]
            else:
                fwd_ret = 0.0

            if y_pred[i] == 1:
                pnl_model.append(fwd_ret)
            elif y_pred[i] == -1:
                pnl_model.append(-fwd_ret)

            ic_pairs.append((p_up, fwd_ret))

        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
    up_mask  = y_pred == 1
    dn_mask  = y_pred == -1
    y_true_dir = y_true[dir_mask]
    y_pred_dir = y_pred[dir_mask]

    acc     = float((y_true == y_pred).mean() * 100)
    n_sig   = int(dir_mask.sum())
    dir_acc = float((y_true_dir == y_pred_dir).mean() * 100) if n_sig > 0 else float("nan")
    up_prec = float((y_true[up_mask] == 1).mean() * 100) if up_mask.sum() > 0 else float("nan")
    dn_prec = float((y_true[dn_mask] == -1).mean() * 100) if dn_mask.sum() > 0 else float("nan")
    win_rate = float((np.array(pnl_model) > 0).mean() * 100) if pnl_model else float("nan")
    pnl_pct  = float(sum(pnl_model) * 100)
    n_trades = int(len(pnl_model))

    # Sharpe (annualized, 5-day holding period)
    if len(pnl_model) > 1:
        arr = np.array(pnl_model)
        sharpe = float((arr.mean() / (arr.std() + 1e-9)) * np.sqrt(252 / 5))
    else:
        sharpe = float("nan")

    # Max drawdown
    if pnl_model:
        cumulative = np.cumprod(1 + np.array(pnl_model))
        rolling_max = np.maximum.accumulate(cumulative)
        drawdowns = (cumulative - rolling_max) / rolling_max
        max_dd = float(drawdowns.min())
    else:
        max_dd = float("nan")

    # IC (Spearman correlation between prob_up and actual fwd return)
    if len(ic_pairs) > 10:
        ic_val, _ = spearmanr([p[0] for p in ic_pairs], [p[1] for p in ic_pairs])
        ic = float(ic_val)
    else:
        ic = float("nan")

    return {
        "accuracy":       round(acc, 1),
        "dir_accuracy":   round(dir_acc, 1) if not np.isnan(dir_acc) else float("nan"),
        "up_precision":   round(up_prec, 1) if not np.isnan(up_prec) else float("nan"),
        "dn_precision":   round(dn_prec, 1) if not np.isnan(dn_prec) else float("nan"),
        "win_rate":       round(win_rate, 1) if not np.isnan(win_rate) else float("nan"),
        "pnl_pct":        round(pnl_pct, 1),
        "sharpe":         round(sharpe, 3) if not np.isnan(sharpe) else float("nan"),
        "max_dd":         round(max_dd, 3) if not np.isnan(max_dd) else float("nan"),
        "ic":             round(ic, 4) if not np.isnan(ic) else float("nan"),
        "n_predictions":  int(len(y_true)),
        "n_trades":       n_trades,
    }


def mean_metrics(results_list: list[dict]) -> dict:
    if not results_list:
        return {}
    keys = ["accuracy", "dir_accuracy", "up_precision", "dn_precision",
            "win_rate", "pnl_pct", "sharpe", "max_dd", "ic"]
    out = {}
    for k in keys:
        vals = [r[k] for r in results_list if k in r and not np.isnan(r[k])]
        out[k] = round(float(np.mean(vals)), 3) if vals else float("nan")
    return out


def generate_html(results: dict, aggregate: dict, subsets_sizes: dict, winner: str) -> str:
    css = """
    <style>
      body { background: #1a1a2e; color: #e0e0e0; font-family: monospace; padding: 20px; }
      h1, h2 { color: #e0e0ff; }
      table { border-collapse: collapse; width: 100%; margin-bottom: 24px; }
      th { background: #16213e; color: #a0c4ff; padding: 8px 12px; text-align: right; }
      th:first-child { text-align: left; }
      td { padding: 6px 12px; text-align: right; border-bottom: 1px solid #2a2a4e; }
      td:first-child { text-align: left; }
      tr.subset-a { background: #1e2a3a; }
      tr.subset-b { background: #1a2e2e; }
      tr.subset-c { background: #2a1e2e; }
      tr.delta { color: #888; font-size: 0.88em; }
      tr.winner { font-weight: bold; color: #ffe066; }
      tr.agg-row { background: #0f3460; color: #ffffff; font-weight: bold; }
      .pos { color: #66ff99; }
      .neg { color: #ff6666; }
      .label { color: #a0c4ff; font-weight: bold; }
    </style>
    """

    def fmt_val(v, is_pct=False, is_delta=False):
        if isinstance(v, float) and np.isnan(v):
            return "<td>N/A</td>"
        s = f"{v:+.1f}" if is_delta else f"{v:.1f}"
        cls = ""
        if is_delta:
            cls = "pos" if v > 0 else ("neg" if v < 0 else "")
        elif is_pct:
            cls = "pos" if v > 0 else ("neg" if v < 0 else "")
        tag = f' class="{cls}"' if cls else ""
        return f"<td{tag}>{s}</td>"

    def fmt_small(v):
        if isinstance(v, float) and np.isnan(v):
            return "<td>N/A</td>"
        return f"<td>{v:.3f}</td>"

    headers = (
        "<tr>"
        "<th>Stock/Subset</th>"
        "<th>Acc%</th><th>Dir%</th><th>↑Prec%</th><th>↓Prec%</th>"
        "<th>Win%</th><th>P&L%</th><th>Sharpe</th><th>MaxDD</th><th>IC</th>"
        "<th>Trades</th><th>Preds</th>"
        "</tr>"
    )

    rows = ""
    for stock in STOCKS:
        if stock not in results:
            continue
        stock_results = results[stock]
        for sub_key, css_cls in [("A", "subset-a"), ("B", "subset-b"), ("C", "subset-c")]:
            r = stock_results.get(sub_key, {})
            if not r:
                continue
            size = subsets_sizes.get(sub_key, "?")
            w_cls = " winner" if (stock == list(results.keys())[0] and sub_key == winner) else ""
            rows += f'<tr class="{css_cls}{w_cls}">'
            rows += f"<td><span class='label'>{stock}</span> {sub_key}({size}f)</td>"
            rows += fmt_val(r.get("accuracy", float("nan")))
            rows += fmt_val(r.get("dir_accuracy", float("nan")))
            rows += fmt_val(r.get("up_precision", float("nan")))
            rows += fmt_val(r.get("dn_precision", float("nan")))
            rows += fmt_val(r.get("win_rate", float("nan")))
            rows += fmt_val(r.get("pnl_pct", float("nan")), is_pct=True)
            rows += fmt_small(r.get("sharpe", float("nan")))
            rows += fmt_small(r.get("max_dd", float("nan")))
            rows += fmt_small(r.get("ic", float("nan")))
            rows += f"<td>{r.get('n_trades', 0)}</td>"
            rows += f"<td>{r.get('n_predictions', 0)}</td>"
            rows += "</tr>"

        # Delta B-A and C-A rows
        ra = stock_results.get("A", {})
        rb = stock_results.get("B", {})
        rc = stock_results.get("C", {})
        for label, rd in [("ΔB-A", rb), ("ΔC-A", rc)]:
            if ra and rd:
                rows += '<tr class="delta">'
                rows += f"<td>&nbsp;&nbsp;{label}</td>"
                for key in ["accuracy", "dir_accuracy", "up_precision", "dn_precision", "win_rate", "pnl_pct"]:
                    va = ra.get(key, float("nan"))
                    vd = rd.get(key, float("nan"))
                    if np.isnan(va) or np.isnan(vd):
                        rows += "<td>—</td>"
                    else:
                        rows += fmt_val(vd - va, is_delta=True)
                rows += "<td colspan='5'></td>"
                rows += "</tr>"

    # Aggregate rows
    rows += "<tr><td colspan='12'>&nbsp;</td></tr>"
    for sub_key, css_cls in [("A", "subset-a"), ("B", "subset-b"), ("C", "subset-c")]:
        r = aggregate.get(sub_key, {})
        if not r:
            continue
        size = subsets_sizes.get(sub_key, "?")
        w_cls = " winner" if sub_key == winner else ""
        rows += f'<tr class="agg-row{w_cls}">'
        rows += f"<td>AGG {sub_key}({size}f)</td>"
        rows += fmt_val(r.get("accuracy", float("nan")))
        rows += fmt_val(r.get("dir_accuracy", float("nan")))
        rows += fmt_val(r.get("up_precision", float("nan")))
        rows += fmt_val(r.get("dn_precision", float("nan")))
        rows += fmt_val(r.get("win_rate", float("nan")))
        rows += fmt_val(r.get("pnl_pct", float("nan")), is_pct=True)
        rows += fmt_small(r.get("sharpe", float("nan")))
        rows += fmt_small(r.get("max_dd", float("nan")))
        rows += fmt_small(r.get("ic", float("nan")))
        rows += "<td colspan='2'></td>"
        rows += "</tr>"

    ts = datetime.now().strftime("%Y-%m-%d %H:%M")
    html = f"""<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>3-way Backtest {ts}</title>{css}</head>
<body>
<h1>3-Way Feature Subset Backtest</h1>
<p>Walk-forward | MIN_TRAIN={MIN_TRAIN} | STEP={STEP}d | LABEL=5d ±2% | RF(200est,d6,leaf10)</p>
<p>Generated: {ts} | Winner: <strong>{winner} ({subsets_sizes[winner]}f)</strong></p>
<table>
{headers}
{rows}
</table>
</body>
</html>"""
    return html


def main():
    from models.predictor import _build_features

    docs = ROOT / "docs"
    docs.mkdir(exist_ok=True)

    all_results: dict[str, dict[str, dict]] = {}
    agg_lists: dict[str, list] = {"A": [], "B": [], "C": []}

    for stock_no in STOCKS:
        print(f"  [{stock_no}] loading...", file=sys.stderr)
        try:
            df = fetch_df(stock_no)
        except Exception as e:
            print(f"  [{stock_no}] fetch error: {e}", file=sys.stderr)
            continue

        if df is None or df.empty or len(df) < MIN_TRAIN + STEP + LABEL_HORIZON:
            print(f"  [{stock_no}] insufficient data", file=sys.stderr)
            continue

        feat = _build_features(df)
        close = df["close"] if "date" not in df.columns else df.set_index("date")["close"]
        labels = make_labels(close.reset_index(drop=True))
        feat["_label"] = labels
        feat["_close"] = close.reset_index(drop=True).values

        stock_res = {}
        for sub_key, sub_cols in SUBSETS.items():
            print(f"  [{stock_no}] backtest {sub_key}({len(sub_cols)}f)...", file=sys.stderr)
            try:
                r = walk_forward(feat, sub_cols)
                if r:
                    stock_res[sub_key] = r
                    agg_lists[sub_key].append(r)
            except Exception as e:
                print(f"  [{stock_no}] {sub_key} error: {e}", file=sys.stderr)

        if stock_res:
            all_results[stock_no] = stock_res

    # Aggregate
    aggregate = {k: mean_metrics(v) for k, v in agg_lists.items()}

    # Winner: highest mean dir_accuracy
    winner = max(
        ["A", "B", "C"],
        key=lambda k: aggregate.get(k, {}).get("dir_accuracy", float("-inf")),
    )
    winner_vals = {k: aggregate.get(k, {}).get("dir_accuracy", float("nan")) for k in ["A", "B", "C"]}
    winner_reason = (
        f"Subset {winner} has highest mean dir_accuracy "
        f"(A={winner_vals['A']:.1f}%, B={winner_vals['B']:.1f}%, C={winner_vals['C']:.1f}%)"
    )

    subsets_sizes = {"A": len(SUBSET_A), "B": len(SUBSET_B), "C": len(SUBSET_C)}

    output = {
        "subsets": subsets_sizes,
        "results": all_results,
        "aggregate": aggregate,
        "winner": winner,
        "winner_reason": winner_reason,
    }

    json_path = docs / "backtest_3way.json"
    with open(json_path, "w") as f:
        json.dump(output, f, indent=2)
    print(f"Wrote {json_path}", file=sys.stderr)

    html = generate_html(all_results, aggregate, subsets_sizes, winner)
    html_path = docs / "backtest_3way.html"
    with open(html_path, "w") as f:
        f.write(html)
    print(f"Wrote {html_path}", file=sys.stderr)

    # Console summary
    print(f"\n{'='*60}")
    print(f"Winner: {winner} ({subsets_sizes[winner]}f)")
    print(f"{winner_reason}")
    print(f"\nAggregate metrics (mean across {len(all_results)} stocks):")
    hdr = f"  {'Subset':<8}  {'Acc%':>5}  {'Dir%':>5}  {'↑Prec':>6}  {'↓Prec':>6}  {'Win%':>5}  {'P&L%':>6}  {'Sharpe':>6}  {'MaxDD':>6}  {'IC':>6}"
    print(hdr)
    print("  " + "-" * (len(hdr) - 2))
    for k in ["A", "B", "C"]:
        r = aggregate.get(k, {})
        def f(v):
            return f"{v:>6.1f}" if not np.isnan(v) else "   N/A"
        def fs(v):
            return f"{v:>6.3f}" if not np.isnan(v) else "   N/A"
        print(
            f"  {k}({subsets_sizes[k]}f)  "
            f"{f(r.get('accuracy', float('nan')))}  "
            f"{f(r.get('dir_accuracy', float('nan')))}  "
            f"{f(r.get('up_precision', float('nan')))}  "
            f"{f(r.get('dn_precision', float('nan')))}  "
            f"{f(r.get('win_rate', float('nan')))}  "
            f"{f(r.get('pnl_pct', float('nan')))}  "
            f"{fs(r.get('sharpe', float('nan')))}  "
            f"{fs(r.get('max_dd', float('nan')))}  "
            f"{fs(r.get('ic', float('nan')))}"
        )


if __name__ == "__main__":
    main()