momentum devandClaude Opus 5.5
f2c7e52c79
A gradient-boosted model scores burst / micro-pullback / whole-dollar signals on
squeeze stocks from tick data, and only its top-scored regular-hours signals
are traded. Validated in data/ticksim before any live use:
- v1 on a fresh 600-day sample: real ranking (AUC 0.63), thin edge. FAIL.
- v2 (trained on Jun-Jul of samples 1+2), pre-registered, on 500 NEW Aug-Sep
symbol-days: +1.11%/trade after the spread (day-block 90% lo +0.54%),
n=332 over 38 days. PASS. It held at 2 s latency and +0.2% extra cost.
- judge/engine.py: the validated signals, execution gates and features, as one
shared source.
- judge/model.py: trains and freezes v2 (cutoff = the training top 20%).
- judge/decide.py: pure decision logic.
- judge/replay.py: a live-style replay of recorded days. Parity: 40 of 41 research
trades matched with identical model scores.
- judge/live.py: the paper trader. Watchlist from the snapshot scan with
backfill, SIP trade/quote websocket, decisions each second, a limit at the ask
with a broker OTO stop (cancelled if unfilled in 3 s), trail 6% after +1R,
20-min time stop, flat by 15:55. $20 risk and $1,000 max notional per trade.
Everything is journalled and the day's ticks are kept for paper-vs-simulation
review. Weekends and holidays exit.
- Runs from the workstation (Task Scheduler, weekdays 09:10 ET) with the
research venv. Paper only.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E3S5EFKThFVrjmN4HHsHMN
69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
"""Train + freeze the v2 judgement model exactly as validated in data/ticksim/PREREG_v2.md.
|
|
|
|
usage: python -m judge.model -> writes data/judge/model_v2.joblib (+ .json metadata)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import joblib
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.ensemble import HistGradientBoostingClassifier
|
|
|
|
from judge.engine import FEATS, SETUP_I
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
TICKSIM = ROOT / "data" / "ticksim"
|
|
OUT = ROOT / "data" / "judge"
|
|
PARAMS = dict(max_iter=200, learning_rate=0.05, max_leaf_nodes=15, min_samples_leaf=40, l2_regularization=1.0,
|
|
random_state=0)
|
|
|
|
|
|
def _prep(df: pd.DataFrame) -> pd.DataFrame:
|
|
daily = pd.read_pickle(ROOT / "data" / "ross" / "daily_all.pkl")
|
|
med = daily.sort_values("date").groupby("symbol").v.transform(lambda s: s.shift(1).rolling(20, min_periods=5).median())
|
|
daily = daily.assign(med20=med)[["symbol", "date", "gap", "med20"]]
|
|
if "gap" not in df:
|
|
df = df.merge(daily, on=["symbol", "date"], how="left")
|
|
df["cumvol_vs_med"] = df.cum_vol / df.med20
|
|
df["setup_i"] = df.setup.map(SETUP_I)
|
|
df["gap_known"] = np.where(df.rth == 1, df.gap, np.nan)
|
|
return df
|
|
|
|
|
|
def train() -> dict:
|
|
s1 = _prep(pd.read_pickle(TICKSIM / "ml_data.pkl"))
|
|
s2 = _prep(pd.read_pickle(TICKSIM / "ml_fresh_data.pkl"))
|
|
tr = pd.concat([s1, s2], ignore_index=True)
|
|
tr = tr[tr.date < "2026-08-01"]
|
|
m = HistGradientBoostingClassifier(**PARAMS).fit(tr[FEATS], (tr.r > 0).astype(int))
|
|
cutoff = float(np.quantile(m.predict_proba(tr[tr.rth == 1][FEATS])[:, 1], 0.8))
|
|
meta = dict(version="v2", trained_on="samples 1+2, signals dated < 2026-08-01", n_train=int(len(tr)),
|
|
cutoff=cutoff, features=FEATS, params=PARAMS, validation="PREREG_v2 PASS: sample-3 Aug-Sep RTH "
|
|
"top20% n=332 +1.11%/trade (day-block lo +0.54%)")
|
|
return dict(model=m, **meta)
|
|
|
|
|
|
def save(art: dict) -> Path:
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
p = OUT / "model_v2.joblib"
|
|
joblib.dump(art, p)
|
|
(OUT / "model_v2.json").write_text(json.dumps({k: v for k, v in art.items() if k != "model"}, indent=2, default=str))
|
|
return p
|
|
|
|
|
|
def load(path: Path | None = None) -> dict:
|
|
return joblib.load(path or OUT / "model_v2.joblib")
|
|
|
|
|
|
def score(art: dict, feats: dict) -> float:
|
|
x = pd.DataFrame([{k: feats.get(k, np.nan) for k in art["features"]}])
|
|
return float(art["model"].predict_proba(x)[:, 1][0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
a = train()
|
|
print("saved", save(a), "cutoff", round(a["cutoff"], 4), "n_train", a["n_train"])
|