PaperTrade / research /ml_backtest.py
Khanna, Videh Rakesh Rakesh
ML intraday-high guard, yfinance thread-safety, validation consistency
0602ac7
Raw
History Blame Contribute Delete
22.5 kB
#!/usr/bin/env python3
"""research/ml_backtest.py β€” accuracy backtest for the ml_predictor model.
Evaluates the trained quantile model on the TRUE out-of-sample rows (dates on/after
manifest.holdout_start) already present in ml_predictor/training_data.csv β€” those rows
carry both the point-in-time features AND the realized forward excursions
(up_INTRADAY/dn_INTRADAY/up_1D/dn_1D/up_3D/dn_3D), so grading needs no re-fetch.
It reuses the model's OWN derivation (MLPredictor._predict_tf, the exact production path,
called with a normalized price=100) and grades with _graded_hit / _evaluate_intraday_hit β€”
copied verbatim from research/backtest.py so ML numbers are directly comparable to the LLM's.
Metrics per timeframe (and per confidence bucket):
β€’ direction accuracy (predicted 3-class == realized 3-class)
β€’ direction_hit / graded hit rate (MIDPOINT_HIT / RANGE_HIT / MISS)
β€’ estimated-high MAE & bias (up_q90 vs realized max-up)
β€’ quantile calibration/coverage (up_q90 β‰ˆ90%, down_q10 β‰ˆ90% below) + pinball loss
β€’ dip-level-reached % (bearish) and stop-would-hit %
Usage (after dataset.py + train.py):
python research/ml_backtest.py
python research/ml_backtest.py --csv ml_predictor/training_data.csv
"""
from __future__ import annotations
import argparse
import os
import sys
import numpy as np
import pandas as pd
_PROJ_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _PROJ_ROOT not in sys.path:
sys.path.insert(0, _PROJ_ROOT)
from ml_predictor.features import FEATURE_COLUMNS, TIMEFRAMES # noqa: E402
from ml_predictor.infer import MLPredictor # noqa: E402
DEFAULT_CSV = os.path.join(_PROJ_ROOT, "ml_predictor", "training_data.csv")
OUT_CSV = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ml_backtest_results.csv")
_UP = {"INTRADAY": "up_INTRADAY", "1D": "up_1D", "3D": "up_3D"}
_DN = {"INTRADAY": "dn_INTRADAY", "1D": "dn_1D", "3D": "dn_3D"}
_DIRC = {"INTRADAY": "dir_INTRADAY", "1D": "dir_1D", "3D": "dir_3D"}
# ── Grading β€” copied verbatim from research/backtest.py (kept in sync) ─────────
def _evaluate_intraday_hit(direction, price, target_lo, target_hi, max_up, min_down, tf_label, ret_for_tf):
direction = (direction or "NEUTRAL").upper()
try:
target_point = (float(target_lo) + float(target_hi)) / 2.0
req_move = (target_point / float(price) - 1.0) * 100.0
except Exception:
req_move = float("nan")
if direction == "BULLISH":
direction_hit = max_up > 0
target_hit = direction_hit and (not pd.isna(req_move)) and (req_move >= 0) and (max_up >= req_move)
return direction_hit, target_hit
if direction == "BEARISH":
direction_hit = min_down < 0
target_hit = direction_hit and (not pd.isna(req_move)) and (req_move <= 0) and (min_down <= req_move <= max_up)
return direction_hit, target_hit
if direction == "NEUTRAL":
neutral_caps = {"INTRADAY": 0.90, "1D": 1.0, "3D": 1.0, "5D": 1.0}
cap = neutral_caps.get(tf_label, 1.0)
direction_hit = (abs(ret_for_tf) <= cap)
target_hit = (direction_hit and (not pd.isna(req_move)) and (abs(req_move) <= cap / 3.0)
and (min_down <= req_move <= max_up))
return direction_hit, target_hit
return False, False
def _graded_hit(direction, price, target_lo, target_hi, max_up, min_down) -> str:
try:
d = (direction or "NEUTRAL").upper()
lo_pct = (float(target_lo) / float(price) - 1.0) * 100.0
hi_pct = (float(target_hi) / float(price) - 1.0) * 100.0
mid_pct = (lo_pct + hi_pct) / 2.0
except Exception:
return "MISS"
if d in ("BULLISH", "SLIGHTLY BULLISH"):
if max_up >= mid_pct:
return "MIDPOINT_HIT"
if max_up >= lo_pct:
return "RANGE_HIT"
return "MISS"
if d in ("BEARISH", "SLIGHTLY BEARISH"):
if min_down <= mid_pct:
return "MIDPOINT_HIT"
if min_down <= hi_pct:
return "RANGE_HIT"
return "MISS"
if min_down <= hi_pct and max_up >= lo_pct:
return "MIDPOINT_HIT"
return "MISS"
def _pinball(y, q_pred, tau):
d = y - q_pred
return float(np.mean(np.maximum(tau * d, (tau - 1) * d)))
def run(csv_path: str = DEFAULT_CSV, sweep: bool = False, target_winrate: float = 0.85) -> pd.DataFrame:
predictor = MLPredictor()
if not predictor.available:
raise SystemExit("ml_predictor model not loaded β€” run `python ml_predictor/train.py` first.")
df = pd.read_csv(csv_path)
df["date"] = pd.to_datetime(df["date"])
holdout_start = predictor.manifest.get("holdout_start")
if holdout_start:
oos = df[df["date"] >= pd.to_datetime(holdout_start)].copy()
print(f" Out-of-sample rows (date β‰₯ {holdout_start}): {len(oos):,}")
else:
oos = df
print(f" (no holdout_start in manifest β€” grading all {len(oos):,} rows)")
if oos.empty:
raise SystemExit("No out-of-sample rows to evaluate.")
feat_mat = oos[FEATURE_COLUMNS].to_numpy(dtype=float)
atr_arr = oos["atr_pct"].to_numpy(dtype=float) # price normalized to 100 β†’ atr14 β‰ˆ atr_pct
records = []
for tf in TIMEFRAMES:
# Batch-predict all rows through the estimators ONCE (fast), then derive per row.
q, proba_m, classes = predictor._raw_predict(tf, feat_mat)
median_w = float(predictor.manifest.get("tf", {}).get(tf, {}).get("median_train_width", 1.5)) or 1.5
up_col, dn_col = oos[_UP[tf]].to_numpy(float), oos[_DN[tf]].to_numpy(float)
ret_col = (np.zeros(len(oos)) if tf == "INTRADAY" else oos[f"ret_{tf}"].to_numpy(float))
for pos in range(len(oos)):
row = oos.iloc[pos]
up_real, dn_real, ret_real = float(up_col[pos]), float(dn_col[pos]), float(ret_col[pos])
atr14 = float(atr_arr[pos]) if np.isfinite(atr_arr[pos]) else None
row_q = {k: float(v[pos]) for k, v in q.items()}
pred = predictor._derive(
row_q, proba_m[pos], classes, tf, price=100.0, atr14=atr14, median_w=median_w,
live_price=None, today_high=None, news_score=0, anchor_close=100.0)
tlo, thi = pred["target_price_lo"], pred["target_price_hi"]
direction = pred["direction"]
dir_hit, tgt_hit = _evaluate_intraday_hit(direction, 100.0, tlo, thi, up_real, dn_real, tf, ret_real)
grade = _graded_hit(direction, 100.0, tlo, thi, up_real, dn_real)
pq = pred["quantiles"]
records.append({
"date": row["date"].strftime("%Y-%m-%d"), "ticker": row["ticker"], "tf": tf,
"direction": direction, "confidence": pred["confidence"],
"true_direction": str(row[_DIRC[tf]]),
"up_real": up_real, "dn_real": dn_real, "ret_real": ret_real,
"up_q10": pq["up_q10"], "up_q50": pq["up_q50"], "up_q90": pq["up_q90"],
"down_q10": pq["down_q10"], "down_q50": pq["down_q50"], "down_q90": pq["down_q90"],
"stop_loss_pct": pred["stop_loss_pct"],
# expected (median) take-profit return %, entry normalized to price=100.
# range-bound NEUTRAL now returns expected_target_price=None β†’ 0% expected move.
"expected_ret_pct": round((pred["expected_target_price"] or 100.0) - 100.0, 3),
"should_buy": int(pred["should_buy"]),
"direction_hit": int(dir_hit), "target_hit": int(tgt_hit),
"hit_grade": grade,
"graded_hit": int(grade in ("MIDPOINT_HIT", "RANGE_HIT")),
"midpoint_hit": int(grade == "MIDPOINT_HIT"),
"dir_correct": int(direction == str(row[_DIRC[tf]])),
})
res = pd.DataFrame(records)
res.to_csv(OUT_CSV, index=False)
_summary(res)
_pnl_simulation(res)
if sweep:
_pnl_sweep(res, target_winrate=target_winrate)
_dip_entry_compare(res)
print(f"\n βœ“ Wrote per-row results β†’ {OUT_CSV}")
return res
# ── P&L trade simulation (doc-1 metrics: total return, win rate, max DD, PF, #trades) ──
ROUND_TRIP_COST_PCT = 0.30 # 0.10% fees + 0.05% slippage per side Γ— 2 (matches research/db_backtest.py)
def _simulate_trade(tp_pct, stop_pct, up_real, dn_real, ret_real, order: str = "stop_first"):
"""Long trade, market entry (price=100). Exit at take-profit, stop, or window close.
Excursion-only data (daily bars) cannot tell us whether the target or the stop was
touched FIRST when both lie inside the horizon's range. So we bracket the truth:
β€’ order="stop_first" β†’ pessimistic: assume the stop fills first (lower bound)
β€’ order="target_first"β†’ optimistic: assume the target fills first (upper bound)
The real P&L is guaranteed to sit between the two. Returns net-of-cost P&L in %.
"""
hit_stop = bool(stop_pct and dn_real <= -stop_pct)
hit_tp = bool(tp_pct and tp_pct > 0 and up_real >= tp_pct)
if order == "target_first":
if hit_tp:
gross = tp_pct
elif hit_stop:
gross = -stop_pct
else:
gross = ret_real
else: # stop_first (default, pessimistic)
if hit_stop:
gross = -stop_pct
elif hit_tp:
gross = tp_pct
else:
gross = ret_real # exit at window close (INTRADAY retβ‰ˆ0 β†’ β‰ˆflat)
return gross - ROUND_TRIP_COST_PCT
def _equity_metrics(daily_rets_pct):
"""(total_return%, max_drawdown%) from a DAILY equity curve: each element is the
equal-weighted mean P&L of all signals on one date, compounded day-over-day. This
reflects concurrent cross-sectional positions β€” unlike compounding 20k trades one at
a time (which explodes), this stays a sane, portfolio-like figure."""
eq = 1.0
peak = 1.0
max_dd = 0.0
for p in daily_rets_pct:
eq *= (1 + p / 100.0)
peak = max(peak, eq)
max_dd = min(max_dd, (eq - peak) / peak)
return (eq - 1.0) * 100.0, max_dd * 100.0
def _pnl_simulation(res: pd.DataFrame):
"""Simulate taking the model's LONG (BULLISH / should_buy) calls, net of fees.
Reports a PESSIMISTIC↔OPTIMISTIC bracket: since daily-bar excursions can't tell us
whether the stop or the target was touched first, we show both orderings. The real
P&L sits between them. WinRate/Expect columns are the pessimistic (honest lower) bound.
"""
print("\n" + "=" * 78)
print(" P&L TRADE SIMULATION β€” long-only BULLISH signals, market entry")
print(f" Take-profit = expected(median) target Β· stop = model stop Β· cost = {ROUND_TRIP_COST_PCT}% round-trip")
print(" Bracket = [stop-first pessimistic … target-first optimistic]; truth sits between.")
print("=" * 78)
hdr = f" {'TF':<9}{'#Trades':>8}{'WinRate':>9}{'Expect↓':>9}{'Expect↑':>9}{'AvgWin':>9}{'AvgLoss':>9}"
hdr += f"{'PF':>7}{'DayRet':>9}{'MaxDD':>9}"
print(hdr)
print(" " + "-" * 84)
for tf in TIMEFRAMES:
d = res[(res["tf"] == tf) & (res["should_buy"] == 1)].copy()
if d.empty:
print(f" {tf:<9}{'0':>8} (no BULLISH signals)")
continue
pess = np.array([_simulate_trade(r.expected_ret_pct, r.stop_loss_pct, r.up_real, r.dn_real, r.ret_real, "stop_first")
for r in d.itertuples()])
opt = np.array([_simulate_trade(r.expected_ret_pct, r.stop_loss_pct, r.up_real, r.dn_real, r.ret_real, "target_first")
for r in d.itertuples()])
d["pnl"] = pess
wins, losses = pess[pess > 0], pess[pess < 0]
pf = (wins.sum() / abs(losses.sum())) if losses.sum() < 0 else 99.0
# Equal-weighted daily equity curve (mean P&L per date, compounded across dates).
daily = d.groupby("date")["pnl"].mean().sort_index()
total_ret, max_dd = _equity_metrics(daily.tolist())
print(f" {tf:<9}{len(pess):>8}{np.mean(pess > 0):>9.0%}{np.mean(pess):>+9.2f}{np.mean(opt):>+9.2f}"
f"{(np.mean(wins) if len(wins) else 0):>+9.2f}{(np.mean(losses) if len(losses) else 0):>+9.2f}"
f"{min(pf, 99.0):>7.2f}{total_ret:>+9.1f}{max_dd:>9.1f}")
print("\n Expect↓ = pessimistic (stop-first) mean net P&L/trade β€” the honest headline.")
print(" Expect↑ = optimistic (target-first) upper bound. DayRet uses Expect↓.")
# ── Stop Γ— take-profit sweep β€” find the profit-maximizing exit at each accuracy level ──
# TP candidates are the model's own up-move quantiles (how ambitious the target is); a
# LOWER quantile is reached more often (higher win rate) but banks a smaller gain.
_TP_LEVELS = [("q10", "up_q10"), ("q50", "up_q50"), ("q90", "up_q90")]
_STOP_MULTS = [0.5, 0.7, 1.0, 1.3] # Γ—model stop; <1 = "bring the stop down" (tighter)
def _pnl_sweep(res: pd.DataFrame, target_winrate: float = 0.85):
"""Grid-search take-profit level Γ— stop multiplier per TF to maximize profit.
For each config, WinRate + Expect↓ (pessimistic) and Expect↑ (optimistic) are shown.
Then per TF we highlight (a) the max-expectancy config and (b) the best config whose
win rate β‰₯ target_winrate β€” so the profit↔accuracy trade-off is explicit.
"""
print("\n" + "=" * 90)
print(f" STOP Γ— TAKE-PROFIT SWEEP β€” long BULLISH, net {ROUND_TRIP_COST_PCT}% cost Β· "
f"target win rate β‰₯ {target_winrate:.0%}")
print(" TP = model up-move quantile (lower = hit more often, banks less). StopΓ— scales the model stop.")
print("=" * 90)
for tf in TIMEFRAMES:
d = res[(res["tf"] == tf) & (res["should_buy"] == 1)].copy()
if d.empty:
print(f"\n {tf}: (no BULLISH signals)")
continue
print(f"\n {tf} (n={len(d)})")
print(f" {'TP':>5}{'StopΓ—':>7}{'WinRate':>9}{'Expect↓':>9}{'Expect↑':>9}{'AvgWin':>8}{'AvgLoss':>9}{'PF':>7}")
print(" " + "-" * 62)
rows = []
for tp_name, tp_col in _TP_LEVELS:
for sm in _STOP_MULTS:
tp = d[tp_col].to_numpy(float)
stop = d["stop_loss_pct"].to_numpy(float) * sm
pess = np.array([_simulate_trade(tp[i], stop[i], d.iloc[i].up_real, d.iloc[i].dn_real,
d.iloc[i].ret_real, "stop_first") for i in range(len(d))])
opt = np.array([_simulate_trade(tp[i], stop[i], d.iloc[i].up_real, d.iloc[i].dn_real,
d.iloc[i].ret_real, "target_first") for i in range(len(d))])
wins, losses = pess[pess > 0], pess[pess < 0]
pf = (wins.sum() / abs(losses.sum())) if losses.sum() < 0 else 99.0
wr = float(np.mean(pess > 0))
rows.append({"tp": tp_name, "sm": sm, "wr": wr, "exp_lo": float(np.mean(pess)),
"exp_hi": float(np.mean(opt)), "avg_win": float(np.mean(wins)) if len(wins) else 0.0,
"avg_loss": float(np.mean(losses)) if len(losses) else 0.0, "pf": min(pf, 99.0)})
print(f" {tp_name:>5}{sm:>7.1f}{wr:>9.0%}{np.mean(pess):>+9.2f}{np.mean(opt):>+9.2f}"
f"{(np.mean(wins) if len(wins) else 0):>+8.2f}{(np.mean(losses) if len(losses) else 0):>+9.2f}{min(pf,99.0):>7.2f}")
best = max(rows, key=lambda r: r["exp_lo"])
constrained = [r for r in rows if r["wr"] >= target_winrate]
best_c = max(constrained, key=lambda r: r["exp_lo"]) if constrained else None
print(f" β†’ MAX PROFIT: TP={best['tp']} StopΓ—{best['sm']} β†’ Expect↓ {best['exp_lo']:+.2f}% "
f"(win {best['wr']:.0%}, PF {best['pf']:.2f})")
if best_c:
print(f" β†’ BEST @β‰₯{target_winrate:.0%} win: TP={best_c['tp']} StopΓ—{best_c['sm']} β†’ "
f"Expect↓ {best_c['exp_lo']:+.2f}% (win {best_c['wr']:.0%}, PF {best_c['pf']:.2f})")
else:
print(f" β†’ No config reaches {target_winrate:.0%} win rate at any stop/TP "
f"(max win {max(r['wr'] for r in rows):.0%}).")
# ── Market-entry vs DIP-entry (limit buy at the modeled pullback) ─────────────
# The model already outputs buy_price_suggestion = price*(1+down_q50/100) β€” i.e. buy on the
# modeled median dip instead of at market. Entering lower gives a cushion above the stop AND
# puts the target closer, so filled trades should win more often. The cost: you only get
# filled when the price actually dips (fill-rate < 100%), and you miss runaway winners that
# never pulled back. This measures the trade-off with the same excursion data.
def _dip_entry_compare(res: pd.DataFrame):
print("\n" + "=" * 90)
print(" MARKET-ENTRY vs DIP-ENTRY (limit buy at model's suggested dip = down_q50)")
print(" DIP fills only if price reached the dip; win/expectancy shown for FILLED trades,")
print(" plus expectancy PER SIGNAL (unfilled = 0, since you couldn't deploy capital).")
print("=" * 90)
for tf in TIMEFRAMES:
d = res[(res["tf"] == tf) & (res["should_buy"] == 1)].copy()
if d.empty:
print(f"\n {tf}: (no BULLISH signals)")
continue
up_real = d["up_real"].to_numpy(float)
dn_real = d["dn_real"].to_numpy(float)
ret_real = d["ret_real"].to_numpy(float)
dip = np.minimum(d["down_q50"].to_numpy(float), 0.0) # modeled dip, % (<0)
print(f"\n {tf} (n={len(d)}) β€” dip median {np.median(dip):+.2f}%")
print(f" {'TP':>5}{'StopΓ—':>7}{'Entry':>8}{'Fill%':>7}{'Win%':>7}{'Exp/fill':>9}{'Exp/sig':>9}{'PF':>7}")
print(" " + "-" * 60)
for tp_name, tp_col in _TP_LEVELS:
for sm in (1.0, 1.3):
tp = d[tp_col].to_numpy(float)
stop = d["stop_loss_pct"].to_numpy(float) * sm
# MARKET entry: entry=close(0), excursions as-is, all rows filled.
mkt = np.array([_simulate_trade(tp[i], stop[i], up_real[i], dn_real[i], ret_real[i], "stop_first")
for i in range(len(d))])
# DIP entry: fill only if dn_real <= dip; shift excursions by -dip (cushion).
filled = dn_real <= dip
u = up_real - dip # larger upside from lower entry
dd = dn_real - dip # smaller (<=0) downside from entry
rc = ret_real - dip
dip_pnl = np.array([_simulate_trade(tp[i], stop[i], u[i], dd[i], rc[i], "stop_first")
for i in range(len(d))])
for label, pnl, fillmask in (("mkt", mkt, np.ones(len(d), bool)), ("dip", dip_pnl, filled)):
sub = pnl[fillmask]
if not len(sub):
continue
wins, losses = sub[sub > 0], sub[sub < 0]
pf = (wins.sum() / abs(losses.sum())) if losses.sum() < 0 else 99.0
fill_rate = float(fillmask.mean())
exp_fill = float(np.mean(sub))
exp_sig = float(np.sum(sub) / len(d)) # per signal (unfilled=0)
print(f" {tp_name:>5}{sm:>7.1f}{label:>8}{fill_rate:>7.0%}{np.mean(sub > 0):>7.0%}"
f"{exp_fill:>+9.2f}{exp_sig:>+9.2f}{min(pf, 99.0):>7.2f}")
def _summary(res: pd.DataFrame):
print("\n" + "=" * 78)
print(" ML MODEL BACKTEST β€” out-of-sample")
print("=" * 78)
hdr = f" {'TF':<9}{'N':>6}{'DirAcc':>8}{'DirHit':>8}{'Graded':>8}{'MidHit':>8}"
hdr += f"{'up90cov':>9}{'dn10cov':>9}{'HighMAE':>9}{'HighBias':>9}"
print(hdr)
print(" " + "-" * 76)
for tf in TIMEFRAMES:
d = res[res["tf"] == tf]
if d.empty:
continue
up_cov = float(np.mean(d["up_real"] <= d["up_q90"])) # target β‰ˆ .90
dn_cov = float(np.mean(d["dn_real"] >= d["down_q10"])) # target β‰ˆ .90
high_mae = float(np.mean(np.abs(d["up_q90"] - d["up_real"])))
high_bias = float(np.mean(d["up_q90"] - d["up_real"]))
print(f" {tf:<9}{len(d):>6}{d['dir_correct'].mean():>8.0%}{d['direction_hit'].mean():>8.0%}"
f"{d['graded_hit'].mean():>8.0%}{d['midpoint_hit'].mean():>8.0%}"
f"{up_cov:>9.0%}{dn_cov:>9.0%}{high_mae:>9.2f}{high_bias:>+9.2f}")
# Pinball loss per quantile
print("\n Pinball loss (lower=better):")
for tf in TIMEFRAMES:
d = res[res["tf"] == tf]
if d.empty:
continue
pb = {
"up_q50": _pinball(d["up_real"].values, d["up_q50"].values, 0.50),
"up_q90": _pinball(d["up_real"].values, d["up_q90"].values, 0.90),
"dn_q50": _pinball(d["dn_real"].values, d["down_q50"].values, 0.50),
"dn_q10": _pinball(d["dn_real"].values, d["down_q10"].values, 0.10),
}
print(f" {tf:<9} " + " ".join(f"{k}={v:.3f}" for k, v in pb.items()))
# Graded hit by confidence bucket
print("\n Graded hit rate by confidence:")
for conf in ("HIGH", "MEDIUM", "LOW"):
d = res[res["confidence"] == conf]
if len(d) >= 10:
print(f" {conf:<7} N={len(d):>5} graded={d['graded_hit'].mean():.0%} "
f"dir_correct={d['dir_correct'].mean():.0%}")
# Dip-level-reached % (bearish) and stop-would-hit %
print("\n Directional diagnostics:")
for tf in TIMEFRAMES:
d = res[res["tf"] == tf]
bear = d[d["direction"] == "BEARISH"]
dip_reached = float(np.mean(bear["dn_real"] <= bear["down_q50"])) if len(bear) else float("nan")
# stop would hit if realized worst-down over window breaches -stop_loss_pct
stop_hit = float(np.mean(d["dn_real"] <= -d["stop_loss_pct"])) if len(d) else float("nan")
n_bear = len(bear)
print(f" {tf:<9} bearish N={n_bear:>4} dip_reached={dip_reached:.0%}"
f" stop_would_hit={stop_hit:.0%}")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--csv", default=DEFAULT_CSV)
ap.add_argument("--sweep", action="store_true", help="run the stopΓ—take-profit profit-maximization grid")
ap.add_argument("--target-winrate", type=float, default=0.85, help="win-rate floor for the sweep's constrained best")
args = ap.parse_args()
run(args.csv, sweep=args.sweep, target_winrate=args.target_winrate)
if __name__ == "__main__":
main()