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
80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
"""Parity check: feed recorded tick days through judge.decide in live-like 1-second steps and compare the
|
|
scored signals with the research pipeline (data/ticksim/ml_s3_data.pkl + the v2 model).
|
|
|
|
usage: python -m judge.replay [n_days]
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from judge import engine
|
|
from judge.decide import DayContext, scan
|
|
from judge.model import load, ROOT
|
|
|
|
TICKSIM = ROOT / "data" / "ticksim"
|
|
|
|
|
|
def ctx_for(path: Path, d: dict) -> DayContext:
|
|
date, sym = path.stem.split("_", 1)
|
|
daily = _daily()
|
|
row = daily[(daily.symbol == sym) & (daily.date == date)]
|
|
gap = float(row.gap.iloc[0]) if len(row) else None
|
|
med20 = float(row.med20.iloc[0]) if len(row) and pd.notna(row.med20.iloc[0]) else None
|
|
ws = [[pd.Timestamp(s), pd.Timestamp(e)] for s, e in d["windows"]]
|
|
return DayContext(sym, date, float(d["prev_close"]), gap, med20, ws)
|
|
|
|
|
|
_D = None
|
|
|
|
|
|
def _daily():
|
|
global _D
|
|
if _D is None:
|
|
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())
|
|
_D = daily.assign(med20=med)[["symbol", "date", "gap", "med20"]]
|
|
return _D
|
|
|
|
|
|
def replay_day(path: Path, art: dict, step_s: int = 1) -> list:
|
|
d = pd.read_pickle(path)
|
|
T = d["trades"].sort_values("t").reset_index(drop=True)
|
|
Q = d["quotes"].sort_values("t").reset_index(drop=True)
|
|
Q = Q[(Q.ap > 0) & (Q.bp > 0) & (Q.ap >= Q.bp)].reset_index(drop=True)
|
|
ctx = ctx_for(path, d)
|
|
start = max(pd.Timestamp(f"{ctx.date} 09:30"), T.t.min())
|
|
end = T.t.max()
|
|
seen = {}
|
|
now = start
|
|
last = None
|
|
while now <= end:
|
|
k = np.searchsorted(T.t.values, np.datetime64(now), side="right")
|
|
kq = np.searchsorted(Q.t.values, np.datetime64(now), side="right")
|
|
for dcs in scan(T.iloc[:k], Q.iloc[:kq], ctx, art, now=now, after=last):
|
|
key = (dcs.setup, str(dcs.ts))
|
|
if key not in seen:
|
|
seen[key] = dcs
|
|
last = now - pd.Timedelta(seconds=10) # re-check a short overlap (partial-bar signals)
|
|
now += pd.Timedelta(seconds=step_s)
|
|
return list(seen.values())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
n = int(sys.argv[1]) if len(sys.argv) > 1 else 3
|
|
art = load()
|
|
research = pd.read_pickle(TICKSIM / "ml_s3_data.pkl")
|
|
files = sorted(TICKSIM.glob("ticks/*.pkl"))
|
|
s3 = set(research.date + "_" + research.symbol)
|
|
files = [f for f in files if f.stem in s3][:n]
|
|
for f in files:
|
|
date, sym = f.stem.split("_", 1)
|
|
live = replay_day(f, art, step_s=5)
|
|
r = research[(research.date == date) & (research.symbol == sym) & (research.rth == 1)]
|
|
live_take = sorted((x.setup, round(x.p, 4)) for x in live if x.take)
|
|
print(f"{f.stem}: live-style signals {len(live)} taken {len(live_take)} | research executed RTH trades {len(r)}")
|
|
print(" live taken:", live_take[:8])
|