feat: judge.review compares each paper trade with the simulator

For a paper day, every order is lined up with what the validated simulator does on
the same recorded ticks: the sim exit from the real fill (stop, 6% trail after +1R,
20-min time stop, sell at the bid 300 ms after the trigger) vs the paper exit,
plus the sim R of orders that did not fill. Trades are flagged when R differs by
more than 0.3R, the exit reason differs, or the entry slipped. The report goes to
data/judge/reviews/<date>.md.

On 2026-09-30: paper -8.25R vs sim -3.97R on the same six fills. The gap is two
trades where Alpaca's paper engine filled a triggered stop 6-60 s late (CMCT,
PWCM; PWCM's own exit was also blocked by the stop-id bug fixed in f3c8be1).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86

judge/review.py

@@ -0,0 +1,160 @@
+"""Per-trade review of a judge paper-trading day against the simulator, on the day's recorded ticks.
+
+For every order the trader sent, it lines up what the paper account did with what the validated simulator
+(data/ticksim/sim.py rules) does on the same ticks:
+  - filled orders: the sim exit from the actual fill (stop / 6% trail after +1R / 20-min time stop, sell at the bid
+    300 ms after the trigger) vs the paper exit. Flagged when R differs by > FLAG_R or the exit reason differs.
+  - unfilled orders: the R the sim would have made from the order's limit price (the paper-only cost of missing).
+The day summary is the evidence for the paper-vs-simulation question before any real money.
+
+usage: python -m judge.review [YYYY-MM-DD ...]      (default: the latest journal with saved ticks)
+writes data/judge/reviews/<date>.md
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from judge import engine
+
+ROOT = Path(__file__).resolve().parent.parent
+OUT = ROOT / "data" / "judge"
+FLAG_R = 0.3
+LAT = np.timedelta64(int(engine.EXEC["lat_ms"]), "ms")
+
+
+def sim_exit(T: pd.DataFrame, Q: pd.DataFrame, t0, entry: float, stop: float) -> dict | None:
+    """The simulator's exit for a long entered at `entry` at time t0 with initial `stop` (sim.py exit loop)."""
+    if stop >= entry or T.empty or Q.empty:
+        return None
+    tt, tp = T.t.values, T.p.values
+    qt = Q.t.values
+    P = engine.EXEC
+    risk = entry - stop
+    k = np.searchsorted(tt, np.datetime64(t0), side="right")
+    deadline = np.datetime64(t0) + np.timedelta64(int(P["hold_s"]), "s")
+    hw, cur = entry, stop
+    reason, trig = "eod", tt[-1]
+    while k < len(tt):
+        px, tk = tp[k], tt[k]
+        if tk >= deadline:
+            reason, trig = "time", tk
+            break
+        hw = max(hw, px)
+        if (hw - entry) / risk >= P["trail_after_r"]:
+            cur = max(cur, hw * (1 - P["trail_pct"]), entry)
+        if px <= cur:
+            reason, trig = ("stop" if cur <= stop else "trail"), tk
+            break
+        k += 1
+    qj = np.searchsorted(qt, trig + LAT, side="right") - 1
+    px = float(Q.bp.iat[max(qj, 0)])
+    return dict(exit=px, r=(px - entry) / risk, reason=reason, at=pd.Timestamp(trig))
+
+
+def load_ticks(day: str, sym: str):
+    f = OUT / "ticks" / f"{day}_{sym}.pkl"
+    if not f.exists():
+        return None, None
+    d = pd.read_pickle(f)
+    T = d["trades"].sort_values("t").reset_index(drop=True)
+    Q = d["quotes"].sort_values("t").reset_index(drop=True)
+    return T, Q[(Q.ap > 0) & (Q.bp > 0) & (Q.ap >= Q.bp)].reset_index(drop=True)
+
+
+def review(day: str) -> tuple[pd.DataFrame, str]:
+    rows = []
+    for line in (OUT / "journal" / f"{day}.jsonl").read_text(encoding="utf-8").splitlines():
+        try:
+            rows.append(json.loads(line))
+        except ValueError:
+            pass
+    out, open_ = [], {}
+    for r in rows:
+        k = r.get("kind")
+        if k == "order":
+            open_[r["symbol"]] = dict(order=r, fill=None)
+        elif k == "unfilled" and r["symbol"] in open_:
+            o = open_.pop(r["symbol"])["order"]
+            out.append(_row(day, o, None, None))
+        elif k == "fill" and r["symbol"] in open_:
+            open_[r["symbol"]]["fill"] = r
+        elif k == "exit" and r["symbol"] in open_ and open_[r["symbol"]]["fill"]:
+            o = open_.pop(r["symbol"])
+            out.append(_row(day, o["order"], o["fill"], r))
+    df = pd.DataFrame(out)
+    return df, _markdown(day, df)
+
+
+def _row(day, o, f, x) -> dict:
+    sym = o["symbol"]
+    T, Q = load_ticks(day, sym)
+    base = dict(sym=sym, setup=o.get("setup"), p=round(float(o.get("p") or 0), 3), order_at=o["at"][11:19],
+                limit=o.get("limit"), stop=o.get("stop"))
+    if f is None:
+        s = sim_exit(T, Q, pd.Timestamp(o["at"]), float(o["limit"]), float(o["stop"])) if T is not None else None
+        return {**base, "status": "unfilled", "sim_r": None if s is None else round(s["r"], 2),
+                "sim_reason": None if s is None else s["reason"]}
+    entry, stop = float(f["price"]), float(f["stop"])
+    risk = entry - stop
+    paper_px = x.get("price") if x.get("price") is not None else x.get("last")
+    paper_r = (float(paper_px) - entry) / risk if paper_px is not None and risk > 0 else None
+    s = sim_exit(T, Q, pd.Timestamp(f["at"]), entry, stop) if T is not None else None
+    row = {**base, "status": "filled", "fill": entry, "entry_slip": round(entry - float(o["limit"]), 4),
+           "paper_r": None if paper_r is None else round(paper_r, 2), "paper_reason": x.get("reason"),
+           "exit_at": x["at"][11:19],
+           "sim_r": None if s is None else round(s["r"], 2), "sim_reason": None if s is None else s["reason"],
+           "sim_exit_at": None if s is None else str(s["at"])[11:19]}
+    flags = []
+    if T is None:
+        flags.append("no ticks")
+    elif s and paper_r is not None:
+        if abs(paper_r - s["r"]) > FLAG_R:
+            flags.append(f"R gap {paper_r - s['r']:+.2f}")
+        if (x.get("reason") or "") not in (s["reason"], "flat_at_broker") and not (
+                x.get("reason") == "stop" and s["reason"] == "trail"):
+            flags.append(f"reason {x.get('reason')} vs sim {s['reason']}")
+    if abs(row["entry_slip"]) > 0.005:
+        flags.append(f"entry slip {row['entry_slip']:+.3f}")
+    row["flags"] = "; ".join(flags)
+    return row
+
+
+def _markdown(day: str, df: pd.DataFrame) -> str:
+    if df.empty:
+        return f"# Judge paper review {day}\n\nNo orders.\n"
+    fl = df[df.status == "filled"]
+    uf = df[df.status == "unfilled"]
+    lines = [f"# Judge paper review {day}", "",
+             f"- orders {len(df)}: filled {len(fl)}, unfilled {len(uf)}",
+             f"- filled: paper R sum {fl.paper_r.sum():+.2f} (mean {fl.paper_r.mean():+.2f}) | "
+             f"sim on the same fills {fl.sim_r.sum():+.2f} (mean {fl.sim_r.mean():+.2f})" if len(fl) else "- filled: none",
+             f"- unfilled: the sim would have made {uf.sim_r.sum():+.2f}R on them" if len(uf) else "- unfilled: none",
+             f"- flagged trades: {int((fl['flags'] != '').sum()) if len(fl) else 0}", "",
+             _table(df.fillna("")), ""]
+    return "\n".join(lines)
+
+
+def _table(df: pd.DataFrame) -> str:
+    cols = list(df.columns)
+    out = ["| " + " | ".join(cols) + " |", "|" + "---|" * len(cols)]
+    out += ["| " + " | ".join(str(v) for v in r) + " |" for r in df.itertuples(index=False)]
+    return "\n".join(out)
+
+
+def main(argv=None) -> None:
+    days = (argv if argv is not None else sys.argv[1:]) or [
+        sorted({p.stem.split("_")[0] for p in (OUT / "ticks").glob("*.pkl")})[-1]]
+    (OUT / "reviews").mkdir(parents=True, exist_ok=True)
+    for day in days:
+        df, md = review(day)
+        (OUT / "reviews" / f"{day}.md").write_text(md, encoding="utf-8")
+        print(md)
+
+
+if __name__ == "__main__":
+    main()

tests/unit/test_judge_review.py

@@ -0,0 +1,31 @@
+"""judge.review.sim_exit reproduces the simulator's exit rules on synthetic ticks."""
+import pandas as pd
+
+from judge.review import sim_exit
+
+T0 = pd.Timestamp("2026-09-30 10:00:00")
+
+
+def tape(prices, step_s=1):
+    t = [T0 + pd.Timedelta(seconds=step_s * (i + 1)) for i in range(len(prices))]
+    T = pd.DataFrame({"t": t, "p": prices, "s": 100})
+    Q = pd.DataFrame({"t": t, "bp": [p - 0.01 for p in prices], "bs": 100, "ap": [p + 0.01 for p in prices], "as": 100})
+    return T, Q
+
+
+def test_stop_sells_at_the_bid_after_the_breach():
+    T, Q = tape([10.0, 9.9, 9.5, 9.4])
+    s = sim_exit(T, Q, T0, 10.0, 9.5)
+    assert s["reason"] == "stop" and round(s["exit"], 2) == 9.49 and round(s["r"], 2) == -1.02
+
+
+def test_trail_after_one_r_gives_back_six_percent():
+    T, Q = tape([10.5, 11.0, 12.0, 11.2, 11.0])     # risk 0.5: +1R at 10.5; trail 6% under 12.0 = 11.28
+    s = sim_exit(T, Q, T0, 10.0, 9.5)
+    assert s["reason"] == "trail" and round(s["exit"], 2) == 11.19
+
+
+def test_time_stop_after_twenty_minutes():
+    T, Q = tape([10.1] * 30, step_s=60)
+    s = sim_exit(T, Q, T0, 10.0, 9.5)
+    assert s["reason"] == "time"