feat: judgement-model (v2) paper trader on the Alpaca "Momentum V2" account
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
This commit is contained in:
momentum devandClaude Opus 5.5 committed
1 parent
1cc15f5ea7
commit
f2c7e52c79
8 files changed
+873
No files matched your search
@@ -0,0 +1,4 @@
|
||||
"""Learned-judgement scalping strategy (v2): a gradient-boosted model scores burst / micro-pullback /
|
||||
whole-dollar signals on squeeze stocks from tick data, and only the top-scored ones are traded.
|
||||
Research and validation: data/ticksim (PREREG_v2.md, results.md). Paper trial first; never live money
|
||||
without a matched paper-vs-simulation review."""
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Pure decision logic: trades/quotes up to `now` -> the scored, tradeable signals.
|
||||
Shared by the live paper trader and the replay/parity check."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from judge import engine
|
||||
from judge.model import score
|
||||
|
||||
|
||||
@dataclass
|
||||
class DayContext:
|
||||
symbol: str
|
||||
date: str # YYYY-MM-DD
|
||||
prev_close: float
|
||||
gap: float | None # today's open vs prior close (known after 09:30)
|
||||
med20: float | None # median daily volume over the prior 20 sessions
|
||||
windows: list = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Decision:
|
||||
setup: str
|
||||
ts: object # signal time (exact crossing trade for micropb)
|
||||
sp: float # signal price
|
||||
stop: float
|
||||
p: float
|
||||
take: bool
|
||||
reason: str
|
||||
feats: dict
|
||||
|
||||
|
||||
def window_trades(T: pd.DataFrame, ctx: DayContext) -> pd.DataFrame:
|
||||
"""Trades inside the model's data windows. Features and bars must see exactly what training saw."""
|
||||
if T is None or T.empty:
|
||||
return T
|
||||
m = np.zeros(len(T), dtype=bool)
|
||||
for s, e in ctx.windows:
|
||||
m |= (T.t.values >= np.datetime64(s)) & (T.t.values < np.datetime64(e))
|
||||
return T[m].reset_index(drop=True)
|
||||
|
||||
|
||||
def scan(T: pd.DataFrame, Q: pd.DataFrame, ctx: DayContext, art: dict, now=None, after=None,
|
||||
include_partial: bool = True) -> list[Decision]:
|
||||
"""Every signal known by `now` (and newer than `after`), scored.
|
||||
T and Q hold ALL buffered trades/quotes; windowing happens here. `now` defaults to the last trade."""
|
||||
Tw = window_trades(T, ctx)
|
||||
if Tw is None or len(Tw) < 50 or Q is None or len(Q) < 50:
|
||||
return []
|
||||
now = np.datetime64(now) if now is not None else Tw.t.values[-1]
|
||||
Tw = Tw[Tw.t.values <= now].reset_index(drop=True)
|
||||
b_all = engine.bars5(Tw)
|
||||
closed = b_all[b_all.tc.values <= now].reset_index(drop=True)
|
||||
tt = Tw.t.values; tp = Tw.p.values.astype(float)
|
||||
out = []
|
||||
for setup in engine.SETUPS:
|
||||
P = dict(engine.EXEC, setup=setup)
|
||||
bars = b_all if (setup == "micropb" and include_partial) else closed
|
||||
for sig in engine.signals(bars, P):
|
||||
ts, sp, stop = sig[0], float(sig[1]), float(sig[2])
|
||||
if len(sig) > 3 and sig[3] == "cross":
|
||||
ts = engine.cross_time(tt, tp, ts, sp)
|
||||
if ts is None:
|
||||
continue
|
||||
if ts > now or (after is not None and ts <= np.datetime64(after)):
|
||||
continue
|
||||
tsn = pd.Timestamp(ts)
|
||||
if tsn.time() < pd.Timestamp("09:30").time(): # the v2 rule trades regular hours only
|
||||
continue
|
||||
f = engine.feats_at(Tw, Q, ts, ctx.prev_close)
|
||||
if f is None:
|
||||
continue
|
||||
f = engine.finish_features(f, setup, ctx.gap, ctx.med20)
|
||||
p = score(art, f)
|
||||
take = p >= art["cutoff"]
|
||||
out.append(Decision(setup, ts, sp, stop, p, take, "score>=cutoff" if take else "score<cutoff", f))
|
||||
out.sort(key=lambda d: d.ts)
|
||||
return out
|
||||
+179
@@ -0,0 +1,179 @@
|
||||
"""Judgement-model engine: signals, features and data windows. This is the single source shared by the
|
||||
research simulator (data/ticksim) and the live paper trader (judge/live.py).
|
||||
|
||||
It is a verbatim port of data/ticksim/sim.py (signals, execution checks) and data/ticksim/ml_dataset.py
|
||||
(features) as validated by PREREG_v2 on 2026-09-28. Changing anything here changes the model's inputs, so
|
||||
retrain and re-validate after any edit.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
TICK = 0.01
|
||||
# the execution config the v2 model was trained and tested under (ml_dataset.EXEC)
|
||||
EXEC = dict(setup="burst", lat_ms=300, max_spread=0.03, max_chase=0.01, lookback_s=120, vol_mult=4.0,
|
||||
burst_pct=0.02, stop_lookback_s=30, min_stop_pct=0.01, max_stop_pct=0.08, run_pct=0.05, run_s=60,
|
||||
max_pb=3, lvl_stop_pct=0.015, target_r=0.0, trail_after_r=1.0, trail_pct=0.06, hold_s=1200,
|
||||
cooldown_s=30, max_trades=6, price_min=1.0, price_max=20.0)
|
||||
SETUPS = ("burst", "micropb", "dollar")
|
||||
FEATS = ["ret_10s", "ret_30s", "ret_60s", "ret_300s", "vol_surge_10s", "vol_surge_60s", "trades_10s", "trades_60s",
|
||||
"rv_60s", "off_hod", "vs_vwap", "gain_prev", "up_from_5m_low", "spread", "imb", "ask_size_usd", "min_since_4am",
|
||||
"rth", "price", "cumvol_vs_med", "gap_known", "setup_i"]
|
||||
SETUP_I = {"burst": 0, "micropb": 1, "dollar": 2}
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- data windows
|
||||
def windows(session_date: str, first20_ts: pd.Timestamp | None) -> list[list[pd.Timestamp]]:
|
||||
"""The tick windows the model was trained on: 09:30-10:30, plus [first +20% minute - 2 min, +60 min]."""
|
||||
w = [(pd.Timestamp(f"{session_date} 09:30"), pd.Timestamp(f"{session_date} 10:30"))]
|
||||
if first20_ts is not None:
|
||||
w.append((max(first20_ts - pd.Timedelta(minutes=2), pd.Timestamp(f"{session_date} 04:00")),
|
||||
first20_ts + pd.Timedelta(minutes=60)))
|
||||
w.sort()
|
||||
merged = [list(w[0])]
|
||||
for s, e in w[1:]:
|
||||
if s <= merged[-1][1]:
|
||||
merged[-1][1] = max(merged[-1][1], e)
|
||||
else:
|
||||
merged.append([s, e])
|
||||
return merged
|
||||
|
||||
|
||||
def in_windows(ts, ws) -> bool:
|
||||
return any(s <= ts < e for s, e in ws)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- bars + signals
|
||||
def bars5(T: pd.DataFrame) -> pd.DataFrame:
|
||||
b = T.set_index("t").resample("5s").agg(o=("p", "first"), h=("p", "max"), l=("p", "min"), c=("p", "last"),
|
||||
v=("s", "sum"))
|
||||
b = b.dropna(subset=["c"])
|
||||
b["tc"] = b.index + pd.Timedelta(seconds=5)
|
||||
return b.reset_index()
|
||||
|
||||
|
||||
def signals(b: pd.DataFrame, P: dict) -> list[tuple]:
|
||||
"""[(t, signal_price, stop[, 'cross'])]. burst/dollar: t = the bar close. micropb: t = the breakout bar START
|
||||
(resolve the exact crossing trade with cross_time)."""
|
||||
out = []
|
||||
n = len(b)
|
||||
if n < 30:
|
||||
return out
|
||||
c = b.c.values; h = b.h.values; l = b.l.values; o = b.o.values; v = b.v.values.astype(float)
|
||||
t = b.t.values; tc = b.tc.values
|
||||
med = pd.Series(v).rolling(24, min_periods=6).median().shift(1).values
|
||||
lb = max(1, P["lookback_s"] // 5)
|
||||
if P["setup"] == "burst":
|
||||
prior_hi = pd.Series(h).rolling(lb, min_periods=1).max().shift(1).values
|
||||
c30 = pd.Series(c).shift(6).values
|
||||
for i in range(24, n):
|
||||
if (h[i] > prior_hi[i] and med[i] > 0 and v[i] >= P["vol_mult"] * med[i] and c30[i] > 0
|
||||
and c[i] / c30[i] - 1 >= P["burst_pct"]):
|
||||
k = max(0, i - P["stop_lookback_s"] // 5)
|
||||
out.append((tc[i], c[i], float(l[k:i + 1].min())))
|
||||
elif P["setup"] == "micropb":
|
||||
rs = max(1, P["run_s"] // 5)
|
||||
i = rs
|
||||
while i < n - 2:
|
||||
lo = l[i - rs:i + 1].min(); hi = h[i - rs:i + 1].max()
|
||||
if lo > 0 and hi / lo - 1 >= P["run_pct"] and h[i] == hi:
|
||||
j = i + 1; pb_low = None
|
||||
while j < n and j - i <= P["max_pb"]:
|
||||
if h[j] > hi:
|
||||
break
|
||||
if c[j] < o[j] or c[j] < c[j - 1]:
|
||||
pb_low = l[j] if pb_low is None else min(pb_low, l[j])
|
||||
if hi - pb_low > 0.5 * (hi - lo):
|
||||
pb_low = None; break
|
||||
if j + 1 < n and h[j + 1] > h[j]:
|
||||
out.append((t[j + 1], float(h[j] + TICK), float(pb_low - TICK), "cross"))
|
||||
i = j + 1
|
||||
break
|
||||
j += 1
|
||||
i += 1
|
||||
elif P["setup"] == "dollar":
|
||||
for i in range(24, n):
|
||||
for step in (1.0, 0.5):
|
||||
lvl = np.floor(c[i] / step) * step
|
||||
if lvl > 0 and c[i - 1] < lvl <= h[i] and med[i] > 0 and v[i] >= P["vol_mult"] * med[i]:
|
||||
out.append((tc[i], c[i], lvl * (1 - P["lvl_stop_pct"])))
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def cross_time(tt: np.ndarray, tp: np.ndarray, bar_start, level: float):
|
||||
"""The first trade at/above `level` inside the 5-s bar starting at bar_start (no look-ahead)."""
|
||||
k0 = np.searchsorted(tt, bar_start, side="left")
|
||||
k1 = np.searchsorted(tt, bar_start + np.timedelta64(5, "s"), side="left")
|
||||
hit = np.nonzero(tp[k0:k1] >= level)[0]
|
||||
return tt[k0 + hit[0]] if len(hit) else None
|
||||
|
||||
|
||||
def entry_check(bid: float, ask: float, sp: float, stop: float, P: dict = EXEC):
|
||||
"""Execution gates from the simulator. Returns (entry, stop) or None."""
|
||||
if bid <= 0 or ask <= 0 or ask < bid:
|
||||
return None
|
||||
mid = (bid + ask) / 2
|
||||
if (ask - bid) / mid > P["max_spread"] or ask > sp * (1 + P["max_chase"]):
|
||||
return None
|
||||
if not (P["price_min"] <= sp <= P["price_max"]):
|
||||
return None
|
||||
entry = ask
|
||||
stop = min(stop, entry * (1 - P["min_stop_pct"]))
|
||||
if (entry - stop) / entry > P["max_stop_pct"] or stop >= entry:
|
||||
return None
|
||||
return entry, stop
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- features
|
||||
def feats_at(T: pd.DataFrame, Q: pd.DataFrame, ts, prev_close: float) -> dict | None:
|
||||
"""Features from trades/quotes strictly before ts (verbatim from ml_dataset.feats_at)."""
|
||||
tt = T.t.values; k = np.searchsorted(tt, ts, side="left")
|
||||
if k < 20:
|
||||
return None
|
||||
p = T.p.values[:k].astype(float); s = T.s.values[:k].astype(float); t = tt[:k]
|
||||
last = p[-1]
|
||||
|
||||
def back(sec):
|
||||
return np.searchsorted(t, ts - np.timedelta64(sec, "s"), side="left")
|
||||
|
||||
f = {}
|
||||
for sec in (10, 30, 60, 300):
|
||||
j = back(sec)
|
||||
f[f"ret_{sec}s"] = last / p[j] - 1 if j < k and p[j] > 0 else 0.0
|
||||
f[f"vol_{sec}s"] = s[j:].sum()
|
||||
f[f"trades_{sec}s"] = k - j
|
||||
j120 = back(120); j300 = back(300)
|
||||
base = s[j300:j120].sum() / 180.0 if j120 > j300 else np.nan
|
||||
f["vol_surge_10s"] = (f["vol_10s"] / 10.0) / base if base and base > 0 else np.nan
|
||||
f["vol_surge_60s"] = (f["vol_60s"] / 60.0) / base if base and base > 0 else np.nan
|
||||
j60 = back(60)
|
||||
lr = np.diff(np.log(p[j60:])) if k - j60 > 3 else np.array([0.0])
|
||||
f["rv_60s"] = float(np.std(lr)) * np.sqrt(max(1, len(lr)))
|
||||
hod = p.max(); f["off_hod"] = last / hod - 1
|
||||
vw = (p * s).sum() / s.sum() if s.sum() else last; f["vs_vwap"] = last / vw - 1
|
||||
f["gain_prev"] = last / prev_close - 1
|
||||
lo300 = p[j300:].min() if k > j300 else last; f["up_from_5m_low"] = last / lo300 - 1
|
||||
qt = Q.t.values; qi = np.searchsorted(qt, ts, side="right") - 1
|
||||
if qi >= 0:
|
||||
bp, ap = float(Q.bp.iat[qi]), float(Q.ap.iat[qi]); bs, as_ = float(Q.bs.iat[qi]), float(Q["as"].iat[qi])
|
||||
mid = (bp + ap) / 2
|
||||
f["spread"] = (ap - bp) / mid if mid > 0 else np.nan
|
||||
f["imb"] = bs / (bs + as_) if bs + as_ > 0 else np.nan
|
||||
f["ask_size_usd"] = as_ * ap
|
||||
tsn = pd.Timestamp(ts)
|
||||
f["min_since_4am"] = (tsn - tsn.normalize() - pd.Timedelta(hours=4)).total_seconds() / 60
|
||||
f["rth"] = float(tsn.time() >= pd.Timestamp("09:30").time())
|
||||
f["price"] = last
|
||||
f["cum_vol"] = s.sum()
|
||||
return f
|
||||
|
||||
|
||||
def finish_features(f: dict, setup: str, gap: float | None, med20: float | None) -> dict:
|
||||
"""The derived features (ml_fresh.prep): setup code, gap only after 09:30, cumulative volume vs the 20-day median."""
|
||||
f = dict(f)
|
||||
f["setup_i"] = SETUP_I[setup]
|
||||
f["gap_known"] = gap if (f.get("rth") == 1.0 and gap is not None) else np.nan
|
||||
f["cumvol_vs_med"] = f["cum_vol"] / med20 if med20 else np.nan
|
||||
return f
|
||||
+422
@@ -0,0 +1,422 @@
|
||||
"""Live PAPER trader for the v2 judgement model (Alpaca paper account "Momentum V2").
|
||||
|
||||
usage: python -m judge.live (runs one session: ~09:15 ET until 16:00 ET, then exits)
|
||||
env: ALPACA_V2_KEY / ALPACA_V2_SECRET (paper only), JUDGE_RISK_USD=20, JUDGE_MAX_NOTIONAL=1000,
|
||||
JUDGE_MAX_OPEN=4, JUDGE_MAX_TRADES_PER_SYMBOL=6, JUDGE_SCAN_INTERVAL_S=30, JUDGE_DRY_RUN=0
|
||||
|
||||
It mirrors the validated simulator:
|
||||
- decisions come from judge.decide.scan on tick data inside the model's windows (regular hours only);
|
||||
- the entry is a limit at the ask with a broker-side stop (OTO); unfilled after 3 s it is cancelled;
|
||||
- exits: the stop, a 6% trail under the high-water once +1R, a 20-min time stop, and flat by 15:55 ET.
|
||||
Every scored signal, order, fill and exit goes to data/judge/journal/<date>.jsonl, and the day's ticks go to
|
||||
data/judge/ticks/<date>_<sym>.pkl for a paper-vs-simulation comparison.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
from dotenv import load_dotenv # noqa: E402
|
||||
|
||||
load_dotenv(ROOT / ".env")
|
||||
from alpaca_paper.client import AlpacaPaperClient # noqa: E402
|
||||
from alpaca_paper.settings import AlpacaPaperSettings # noqa: E402
|
||||
from judge import engine # noqa: E402
|
||||
from judge.decide import DayContext, scan # noqa: E402
|
||||
from judge.model import load as load_model # noqa: E402
|
||||
from research.ingestion.snapshot_scan import SnapshotScanConfig, SnapshotScanner # noqa: E402
|
||||
|
||||
ET = ZoneInfo("America/New_York")
|
||||
OUT = ROOT / "data" / "judge"
|
||||
RISK_USD = float(os.environ.get("JUDGE_RISK_USD", "20"))
|
||||
MAX_NOTIONAL = float(os.environ.get("JUDGE_MAX_NOTIONAL", "1000"))
|
||||
MAX_OPEN = int(os.environ.get("JUDGE_MAX_OPEN", "4"))
|
||||
MAX_TRADES_SYM = int(os.environ.get("JUDGE_MAX_TRADES_PER_SYMBOL", "6"))
|
||||
SCAN_S = float(os.environ.get("JUDGE_SCAN_INTERVAL_S", "30"))
|
||||
DRY = os.environ.get("JUDGE_DRY_RUN", "0") == "1"
|
||||
|
||||
|
||||
def now_et() -> pd.Timestamp:
|
||||
return pd.Timestamp(datetime.now(ET).replace(tzinfo=None))
|
||||
|
||||
|
||||
def to_et(iso: str) -> np.datetime64:
|
||||
return np.datetime64(pd.Timestamp(iso).tz_convert(ET).tz_localize(None))
|
||||
|
||||
|
||||
class Journal:
|
||||
def __init__(self, day: str):
|
||||
(OUT / "journal").mkdir(parents=True, exist_ok=True)
|
||||
self.f = open(OUT / "journal" / f"{day}.jsonl", "a", encoding="utf-8")
|
||||
|
||||
def log(self, kind: str, **kw):
|
||||
kw = {k: (str(v) if isinstance(v, (pd.Timestamp, np.datetime64, datetime)) else v) for k, v in kw.items()}
|
||||
self.f.write(json.dumps(dict(at=str(now_et()), kind=kind, **kw), default=str) + "\n")
|
||||
self.f.flush()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Book:
|
||||
ctx: DayContext
|
||||
tt: list = field(default_factory=list); tp: list = field(default_factory=list); ts_: list = field(default_factory=list)
|
||||
qt: list = field(default_factory=list); qbp: list = field(default_factory=list); qbs: list = field(default_factory=list)
|
||||
qap: list = field(default_factory=list); qas: list = field(default_factory=list)
|
||||
ready: bool = False
|
||||
last_scan: pd.Timestamp | None = None
|
||||
last_len: int = 0
|
||||
seen: set = field(default_factory=set)
|
||||
n_trades: int = 0
|
||||
cooldown_until: pd.Timestamp | None = None
|
||||
pending_stream: list = field(default_factory=list)
|
||||
|
||||
def T(self) -> pd.DataFrame:
|
||||
return pd.DataFrame({"t": np.array(self.tt, dtype="datetime64[ns]"), "p": np.array(self.tp, dtype="float64"),
|
||||
"s": np.array(self.ts_, dtype="int64")})
|
||||
|
||||
def Q(self) -> pd.DataFrame:
|
||||
return pd.DataFrame({"t": np.array(self.qt, dtype="datetime64[ns]"), "bp": self.qbp, "bs": self.qbs,
|
||||
"ap": self.qap, "as": self.qas})
|
||||
|
||||
def add_trade(self, t, p, s):
|
||||
self.tt.append(t); self.tp.append(float(p)); self.ts_.append(int(s))
|
||||
|
||||
def add_quote(self, t, bp, bs, ap, as_):
|
||||
if ap > 0 and bp > 0 and ap >= bp:
|
||||
self.qt.append(t); self.qbp.append(float(bp)); self.qbs.append(int(bs)); self.qap.append(float(ap)); self.qas.append(int(as_))
|
||||
|
||||
|
||||
@dataclass
|
||||
class Pos:
|
||||
symbol: str
|
||||
qty: int
|
||||
entry: float
|
||||
stop: float
|
||||
risk: float
|
||||
fill_time: pd.Timestamp
|
||||
order_id: str
|
||||
leg_id: str | None
|
||||
hw: float
|
||||
setup: str
|
||||
p: float
|
||||
|
||||
|
||||
class Trader:
|
||||
def __init__(self):
|
||||
key, sec = os.environ["ALPACA_V2_KEY"], os.environ["ALPACA_V2_SECRET"]
|
||||
self.key, self.sec = key, sec
|
||||
self.client = AlpacaPaperClient(AlpacaPaperSettings.from_env(
|
||||
{"ALPACA_API_KEY": key, "ALPACA_SECRET_KEY": sec, "ALPACA_DATA_FEED": "sip",
|
||||
"APCA_API_BASE_URL": "https://paper-api.alpaca.markets"}))
|
||||
self.art = load_model()
|
||||
self.day = now_et().strftime("%Y-%m-%d")
|
||||
self.j = Journal(self.day)
|
||||
cfg = SnapshotScanConfig.from_env()
|
||||
cfg.price_min, cfg.price_max, cfg.min_pct, cfg.max_per_sweep = 1.0, 20.0, 20.0, 10
|
||||
self.scanner = SnapshotScanner(self.client, cfg)
|
||||
self.books: dict[str, Book] = {}
|
||||
self.pos: dict[str, Pos] = {}
|
||||
self.ws = None
|
||||
self.subscribed: set = set()
|
||||
self.j.log("start", cutoff=self.art["cutoff"], risk_usd=RISK_USD, max_notional=MAX_NOTIONAL, dry=DRY)
|
||||
|
||||
# --------------------------------------------------------------------- discovery + backfill
|
||||
def _paged(self, path, key, sym, start_utc, end_utc):
|
||||
rows, tok = [], None
|
||||
while True:
|
||||
p = self.client._data("GET", path, params={"symbols": sym, "start": start_utc, "end": end_utc, "limit": 10000,
|
||||
"feed": "sip", "page_token": tok})
|
||||
rows += (p.get(key) or {}).get(sym, [])
|
||||
tok = p.get("next_page_token")
|
||||
if not tok:
|
||||
return rows
|
||||
|
||||
def _context(self, sym: str, prev_close: float) -> DayContext:
|
||||
"""first +20% minute (-> windows), today's open gap and the 20-day median daily volume."""
|
||||
d = self.day
|
||||
mb = self.client.get_minute_bars([sym], start_iso=f"{d}T08:00:00Z",
|
||||
end_iso=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), feed="sip").get(sym) or []
|
||||
first20 = None; open_px = None
|
||||
for b in mb:
|
||||
t = pd.Timestamp(b["t"]).tz_convert(ET).tz_localize(None)
|
||||
if first20 is None and b["h"] >= prev_close * 1.20:
|
||||
first20 = t.floor("min")
|
||||
if open_px is None and t.time() >= pd.Timestamp("09:30").time():
|
||||
open_px = b["o"]
|
||||
start = (pd.Timestamp(d) - pd.Timedelta(days=45)).strftime("%Y-%m-%d")
|
||||
daily = self.client.get_daily_bars([sym], start_iso=f"{start}T00:00:00Z", feed="sip").get(sym) or []
|
||||
vols = [b["v"] for b in daily if b["t"][:10] < d][-20:]
|
||||
med20 = float(np.median(vols)) if len(vols) >= 5 else None
|
||||
gap = (open_px / prev_close - 1) if open_px else None
|
||||
return DayContext(sym, d, prev_close, gap, med20, engine.windows(d, first20))
|
||||
|
||||
async def add_symbol(self, sym: str, prev_close: float):
|
||||
if sym in self.books:
|
||||
return
|
||||
ctx = await asyncio.to_thread(self._context, sym, prev_close)
|
||||
bk = Book(ctx); self.books[sym] = bk
|
||||
await self.subscribe([sym])
|
||||
start = min(s for s, _ in ctx.windows)
|
||||
s_utc = start.tz_localize(ET).tz_convert("UTC").strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
e_utc = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
tr = await asyncio.to_thread(self._paged, "/v2/stocks/trades", "trades", sym, s_utc, e_utc)
|
||||
qu = await asyncio.to_thread(self._paged, "/v2/stocks/quotes", "quotes", sym, s_utc, e_utc)
|
||||
for x in tr:
|
||||
bk.add_trade(to_et(x["t"]), x["p"], x["s"])
|
||||
for x in qu:
|
||||
bk.add_quote(to_et(x["t"]), x["bp"], x["bs"], x["ap"], x["as"])
|
||||
last_t = bk.tt[-1] if bk.tt else np.datetime64("1970-01-01")
|
||||
last_q = bk.qt[-1] if bk.qt else np.datetime64("1970-01-01")
|
||||
for kind, row in bk.pending_stream: # stream data that arrived during the backfill
|
||||
if kind == "t" and row[0] > last_t:
|
||||
bk.add_trade(*row)
|
||||
elif kind == "q" and row[0] > last_q:
|
||||
bk.add_quote(*row)
|
||||
bk.pending_stream.clear(); bk.ready = True
|
||||
bk.last_scan = now_et() - pd.Timedelta(seconds=10)
|
||||
self.j.log("add", symbol=sym, prev_close=prev_close, gap=ctx.gap, med20=ctx.med20,
|
||||
windows=[[str(a), str(b)] for a, b in ctx.windows], backfill_trades=len(tr), backfill_quotes=len(qu))
|
||||
|
||||
async def scan_loop(self):
|
||||
while True:
|
||||
n = now_et()
|
||||
if n.time() >= pd.Timestamp("15:30").time():
|
||||
return
|
||||
if n.time() >= pd.Timestamp("09:15").time():
|
||||
try:
|
||||
res = await asyncio.to_thread(self.scanner.sweep, datetime.now(timezone.utc), set(self.books))
|
||||
for h in res.hits:
|
||||
if h.symbol not in self.books and len(self.books) < 30:
|
||||
await self.add_symbol(h.symbol, h.prev_close)
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.j.log("error", where="scan", err=str(e)[:300])
|
||||
await asyncio.sleep(SCAN_S)
|
||||
|
||||
# --------------------------------------------------------------------- streaming
|
||||
async def subscribe(self, syms):
|
||||
new = [s for s in syms if s not in self.subscribed]
|
||||
if not new:
|
||||
return
|
||||
self.subscribed.update(new)
|
||||
if self.ws is not None:
|
||||
await self.ws.send(json.dumps({"action": "subscribe", "trades": new, "quotes": new}))
|
||||
|
||||
async def stream_loop(self):
|
||||
import websockets
|
||||
url = "wss://stream.data.alpaca.markets/v2/sip"
|
||||
while True:
|
||||
try:
|
||||
async with websockets.connect(url, max_size=2 ** 24, ping_interval=20) as ws:
|
||||
await ws.send(json.dumps({"action": "auth", "key": self.key, "secret": self.sec}))
|
||||
self.ws = ws
|
||||
if self.subscribed:
|
||||
s = sorted(self.subscribed)
|
||||
await ws.send(json.dumps({"action": "subscribe", "trades": s, "quotes": s}))
|
||||
self.j.log("stream", status="connected", symbols=len(self.subscribed))
|
||||
async for raw in ws:
|
||||
for m in json.loads(raw):
|
||||
T = m.get("T"); sym = m.get("S")
|
||||
bk = self.books.get(sym)
|
||||
if bk is None:
|
||||
continue
|
||||
if T == "t":
|
||||
row = (to_et(m["t"]), m["p"], m["s"])
|
||||
(bk.add_trade(*row) if bk.ready else bk.pending_stream.append(("t", row)))
|
||||
elif T == "q":
|
||||
row = (to_et(m["t"]), m["bp"], m["bs"], m["ap"], m["as"])
|
||||
(bk.add_quote(*row) if bk.ready else bk.pending_stream.append(("q", row)))
|
||||
elif T in ("error", "subscription"):
|
||||
self.j.log("stream", msg=m)
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.ws = None
|
||||
self.j.log("error", where="stream", err=str(e)[:300])
|
||||
await asyncio.sleep(3)
|
||||
|
||||
# --------------------------------------------------------------------- decisions + entries
|
||||
async def decide_loop(self):
|
||||
while True:
|
||||
n = now_et()
|
||||
if n.time() >= pd.Timestamp("15:50").time():
|
||||
return
|
||||
for sym, bk in list(self.books.items()):
|
||||
if not bk.ready or len(bk.tt) == bk.last_len:
|
||||
continue
|
||||
if not (n.time() >= pd.Timestamp("09:30").time() and engine.in_windows(n, bk.ctx.windows)):
|
||||
bk.last_len = len(bk.tt); continue
|
||||
bk.last_len = len(bk.tt)
|
||||
try:
|
||||
decs = await asyncio.to_thread(scan, bk.T(), bk.Q(), bk.ctx, self.art, n,
|
||||
bk.last_scan - pd.Timedelta(seconds=10))
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.j.log("error", where="scan_signals", symbol=sym, err=str(e)[:300]); continue
|
||||
bk.last_scan = n
|
||||
for d in decs:
|
||||
key = (d.setup, str(d.ts))
|
||||
if key in bk.seen:
|
||||
continue
|
||||
bk.seen.add(key)
|
||||
age = (n - pd.Timestamp(d.ts)).total_seconds()
|
||||
self.j.log("signal", symbol=sym, setup=d.setup, ts=d.ts, sp=d.sp, stop=d.stop, p=round(d.p, 4),
|
||||
take=d.take, age_s=age, feats={k: (None if isinstance(v, float) and math.isnan(v) else v)
|
||||
for k, v in d.feats.items()})
|
||||
if d.take and age <= 5:
|
||||
await self.enter(sym, bk, d)
|
||||
await asyncio.sleep(1.0)
|
||||
|
||||
def _order(self, oid):
|
||||
return self.client._trading("GET", f"/orders/{oid}", params={"nested": "true"})
|
||||
|
||||
async def enter(self, sym, bk, d):
|
||||
why = None
|
||||
if sym in self.pos:
|
||||
why = "already in position"
|
||||
elif len(self.pos) >= MAX_OPEN:
|
||||
why = "max open positions"
|
||||
elif bk.n_trades >= MAX_TRADES_SYM:
|
||||
why = "max trades for symbol"
|
||||
elif bk.cooldown_until is not None and now_et() < bk.cooldown_until:
|
||||
why = "cooldown"
|
||||
if why:
|
||||
self.j.log("skip", symbol=sym, setup=d.setup, why=why); return
|
||||
if not bk.qt:
|
||||
return
|
||||
bid, ask = bk.qbp[-1], bk.qap[-1]
|
||||
chk = engine.entry_check(bid, ask, d.sp, d.stop)
|
||||
if chk is None:
|
||||
self.j.log("skip", symbol=sym, setup=d.setup, why="entry_check (spread/chase/stop)", bid=bid, ask=ask, sp=d.sp); return
|
||||
entry, stop = chk
|
||||
qty = int(min(RISK_USD / (entry - stop), MAX_NOTIONAL / entry))
|
||||
if qty < 1:
|
||||
self.j.log("skip", symbol=sym, why="qty<1"); return
|
||||
bk.n_trades += 1
|
||||
coid = f"jv2-{sym}-{now_et():%H%M%S}"
|
||||
self.j.log("order", symbol=sym, setup=d.setup, qty=qty, limit=round(entry, 2), stop=round(stop, 2), p=d.p, coid=coid)
|
||||
if DRY:
|
||||
return
|
||||
try:
|
||||
o = await asyncio.to_thread(self.client.submit_order, sym, qty, "buy", "limit", "day", round(entry, 2), None,
|
||||
round(stop, 2), None, coid)
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.j.log("error", where="submit", symbol=sym, err=str(e)[:300]); return
|
||||
oid = o.get("id")
|
||||
filled = None
|
||||
for _ in range(12):
|
||||
await asyncio.sleep(0.25)
|
||||
try:
|
||||
st = await asyncio.to_thread(self._order, oid)
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
if st.get("status") == "filled":
|
||||
filled = st; break
|
||||
if filled is None:
|
||||
try:
|
||||
await asyncio.to_thread(self.client.cancel_order, oid)
|
||||
st = await asyncio.to_thread(self._order, oid)
|
||||
if float(st.get("filled_qty") or 0) > 0:
|
||||
filled = st
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
if filled is None:
|
||||
self.j.log("unfilled", symbol=sym, coid=coid); return
|
||||
fq = int(float(filled["filled_qty"])); fpx = float(filled["filled_avg_price"])
|
||||
legs = filled.get("legs") or []
|
||||
leg_id = legs[0]["id"] if legs else None
|
||||
self.pos[sym] = Pos(sym, fq, fpx, stop, fpx - stop, now_et(), oid, leg_id, fpx, d.setup, d.p)
|
||||
self.j.log("fill", symbol=sym, qty=fq, price=fpx, stop=stop, leg_id=leg_id, slippage_vs_ask=fpx - entry)
|
||||
|
||||
# --------------------------------------------------------------------- exits
|
||||
async def exit_pos(self, pz: Pos, reason: str, last: float):
|
||||
try:
|
||||
if pz.leg_id:
|
||||
await asyncio.to_thread(self.client.cancel_order, pz.leg_id)
|
||||
await asyncio.to_thread(self.client.close_position, pz.symbol)
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.j.log("error", where="exit", symbol=pz.symbol, err=str(e)[:300]); return
|
||||
self.j.log("exit", symbol=pz.symbol, reason=reason, last=last, entry=pz.entry, r_est=(last - pz.entry) / pz.risk)
|
||||
self.pos.pop(pz.symbol, None)
|
||||
bk = self.books.get(pz.symbol)
|
||||
if bk:
|
||||
bk.cooldown_until = now_et() + pd.Timedelta(seconds=engine.EXEC["cooldown_s"])
|
||||
|
||||
async def manage_loop(self):
|
||||
tick = 0
|
||||
while True:
|
||||
n = now_et(); tick += 1
|
||||
for sym, pz in list(self.pos.items()):
|
||||
bk = self.books[sym]
|
||||
k = np.searchsorted(np.array(bk.tt[-5000:], dtype="datetime64[ns]"), np.datetime64(pz.fill_time), side="left")
|
||||
recent = bk.tp[-5000:][k:]
|
||||
last = recent[-1] if recent else pz.entry
|
||||
if recent:
|
||||
pz.hw = max(pz.hw, max(recent))
|
||||
if n.time() >= pd.Timestamp("15:55").time():
|
||||
await self.exit_pos(pz, "eod", last); continue
|
||||
if (n - pz.fill_time).total_seconds() >= engine.EXEC["hold_s"]:
|
||||
await self.exit_pos(pz, "time", last); continue
|
||||
if (pz.hw - pz.entry) / pz.risk >= engine.EXEC["trail_after_r"]:
|
||||
new = max(pz.stop, pz.hw * (1 - engine.EXEC["trail_pct"]), pz.entry)
|
||||
if new >= last:
|
||||
await self.exit_pos(pz, "trail", last); continue
|
||||
if new > pz.stop + 0.01 and pz.leg_id:
|
||||
try:
|
||||
await asyncio.to_thread(self.client.replace_order, pz.leg_id, stop_price=round(new, 2))
|
||||
self.j.log("stop_move", symbol=sym, frm=pz.stop, to=round(new, 2)); pz.stop = new
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.j.log("error", where="replace", symbol=sym, err=str(e)[:200])
|
||||
if tick % 5 == 0 and pz.leg_id:
|
||||
try:
|
||||
leg = await asyncio.to_thread(self._order, pz.leg_id)
|
||||
if leg.get("status") == "filled":
|
||||
self.j.log("exit", symbol=sym, reason="stop", price=float(leg.get("filled_avg_price") or 0),
|
||||
entry=pz.entry, r_est=(float(leg.get("filled_avg_price") or 0) - pz.entry) / pz.risk)
|
||||
self.pos.pop(sym, None)
|
||||
bk.cooldown_until = n + pd.Timedelta(seconds=engine.EXEC["cooldown_s"])
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
if n.time() >= pd.Timestamp("16:00").time() and not self.pos:
|
||||
return
|
||||
await asyncio.sleep(1.0)
|
||||
|
||||
def save_ticks(self):
|
||||
(OUT / "ticks").mkdir(parents=True, exist_ok=True)
|
||||
for sym, bk in self.books.items():
|
||||
pd.to_pickle(dict(trades=bk.T(), quotes=bk.Q(), windows=bk.ctx.windows, prev_close=bk.ctx.prev_close,
|
||||
gap=bk.ctx.gap, med20=bk.ctx.med20), OUT / "ticks" / f"{self.day}_{sym}.pkl")
|
||||
|
||||
def trading_day(self) -> bool:
|
||||
"""False on weekends/holidays: the broker clock's next open is not today and the market is closed."""
|
||||
try:
|
||||
c = self.client.get_clock()
|
||||
if c.get("is_open"):
|
||||
return True
|
||||
nxt = pd.Timestamp(c.get("next_open")).tz_convert(ET).strftime("%Y-%m-%d")
|
||||
return nxt == self.day
|
||||
except Exception: # noqa: BLE001
|
||||
return now_et().weekday() < 5
|
||||
|
||||
async def run(self):
|
||||
if not self.trading_day():
|
||||
self.j.log("stop", why="not a trading day"); return
|
||||
tasks = [asyncio.create_task(self.stream_loop()), asyncio.create_task(self.scan_loop()),
|
||||
asyncio.create_task(self.decide_loop())]
|
||||
await self.manage_loop()
|
||||
for t in tasks:
|
||||
t.cancel()
|
||||
self.save_ticks()
|
||||
self.j.log("stop", books=len(self.books))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(Trader().run())
|
||||
@@ -0,0 +1,68 @@
|
||||
"""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"])
|
||||
@@ -0,0 +1,79 @@
|
||||
"""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])
|
||||
@@ -0,0 +1,10 @@
|
||||
@echo off
|
||||
rem Paper trial of the v2 judgement model (Alpaca paper "Momentum V2"). Started by Task Scheduler at 08:10 local
|
||||
rem (09:10 ET); the process exits by itself after 16:00 ET. Log: data\judge\logs\live_YYYY-MM-DD.log
|
||||
setlocal
|
||||
cd /d "%~dp0.."
|
||||
set PYTHONIOENCODING=utf-8
|
||||
if not exist data\judge\logs mkdir data\judge\logs
|
||||
for /f %%i in ('powershell -NoProfile -Command "Get-Date -Format yyyy-MM-dd"') do set RUNDATE=%%i
|
||||
data\ross\.venv\Scripts\python.exe -m judge.live >> "data\judge\logs\live_%RUNDATE%.log" 2>&1
|
||||
exit /b %ERRORLEVEL%
|
||||
@@ -0,0 +1,30 @@
|
||||
"""judge.engine: data windows, execution gates and exact-cross timing (the look-ahead fix)."""
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from judge import engine
|
||||
|
||||
|
||||
def test_windows_merge_open_and_squeeze_window():
|
||||
ws = engine.windows("2026-09-28", pd.Timestamp("2026-09-28 10:00"))
|
||||
assert ws == [[pd.Timestamp("2026-09-28 09:30"), pd.Timestamp("2026-09-28 11:00")]]
|
||||
ws = engine.windows("2026-09-28", pd.Timestamp("2026-09-28 06:03"))
|
||||
assert ws[0] == [pd.Timestamp("2026-09-28 06:01"), pd.Timestamp("2026-09-28 07:03")]
|
||||
assert engine.in_windows(pd.Timestamp("2026-09-28 09:45"), ws) and not engine.in_windows(pd.Timestamp("2026-09-28 08:00"), ws)
|
||||
|
||||
|
||||
def test_entry_check_gates():
|
||||
assert engine.entry_check(4.99, 5.00, 5.00, 4.80) == (5.00, 4.80)
|
||||
assert engine.entry_check(4.70, 5.00, 5.00, 4.80) is None # spread 6% > 3%
|
||||
assert engine.entry_check(5.09, 5.10, 5.00, 4.80) is None # ask > signal price + 1% chase
|
||||
assert engine.entry_check(4.99, 5.00, 5.00, 4.00) is None # stop 20% away > 8% max
|
||||
entry, stop = engine.entry_check(4.99, 5.00, 5.00, 4.99) # stop inside 1% -> widened
|
||||
assert stop == 5.00 * (1 - engine.EXEC["min_stop_pct"])
|
||||
|
||||
|
||||
def test_cross_time_uses_the_first_trade_through_the_level_not_the_bar_start():
|
||||
base = np.datetime64("2026-09-28T09:40:00")
|
||||
tt = np.array([base + np.timedelta64(ms, "ms") for ms in (0, 1500, 3200, 4100)])
|
||||
tp = np.array([4.95, 4.99, 5.01, 5.05])
|
||||
assert engine.cross_time(tt, tp, base, 5.00) == base + np.timedelta64(3200, "ms")
|
||||
assert engine.cross_time(tt, tp, base, 6.00) is None
|
||||
Reference in new issue
Block a user