feat: weekly execution tallies — midday-entry net + market_exit post-run

Read-only nightly-scheduler job (Friday-gated): buckets realized P&L by entry
time (open/late-morning/midday) and, for each market_exit close, computes the
post-exit run-up capturable under a breakeven stop (0 if price undercuts entry
first; intrabar ties count as stopped). Emits a weekly_tally event + Telegram
summary. Evidence accumulator for the dead no_new_entries_after cutoff and the
weakness-exit winner-clipping question — no live trading path touched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

research/weekly_tallies.py

@@ -0,0 +1,206 @@
+"""Weekly execution tallies — two standing stats the 2026-07-24 trade review flagged.
+
+READ-ONLY: queries the event store + minute bars, emits one summary event, optional
+Telegram. Never touches live trading state.
+
+  MIDDAY NET    — realized P&L bucketed by ENTRY time (open ≤10:30 ET, late-morning
+                  ≤12:00, midday after). Accumulating evidence for wiring the dead
+                  no_new_entries_after cutoff: if midday stays net-negative for a few
+                  weeks, wire it; until then just count.
+  EXIT POST-RUN — for each market_exit (weakness exit) close: how much run-up was
+                  CAPTURABLE after the exit, i.e. the high reached before price first
+                  dipped below entry (a breakeven stop survives until then; a dip below
+                  entry first means the "money left" was never reachable → 0).
+
+Entry time = first order_submitted event for (day, symbol); re-entries on the same
+symbol/day share that first submit time (close enough for a bucket tally).
+
+CLI:  python -m research.weekly_tallies [--days 7] [--force]
+Runs only on Friday (ET) unless --force — the nightly scheduler chain calls it daily.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from datetime import datetime, time, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+ET = ZoneInfo("America/New_York")
+
+OPEN_CUTOFF = time(10, 30)
+MIDDAY_CUTOFF = time(12, 0)
+
+
+def _bucket(entry_et: time) -> str:
+    if entry_et <= OPEN_CUTOFF:
+        return "open"
+    if entry_et <= MIDDAY_CUTOFF:
+        return "late_morning"
+    return "midday"
+
+
+def capturable_left_pct(bars: list[dict], exit_ts_et: datetime,
+                        entry_price: float, exit_price: float) -> float:
+    """Run-up (% of exit price) to the highest high after exit BEFORE the first bar
+    whose low undercuts entry price. ``bars`` are chronological dicts with tz-aware
+    ET ``ts``, ``high``, ``low``. Dip below entry before any new high -> 0.0.
+    Intrabar order is unknowable from minute bars, so a bar that both undercuts
+    entry and makes a new high counts as stopped first (pessimistic on purpose —
+    'capturable' must not depend on winning an intrabar coin flip)."""
+    best = exit_price
+    for b in bars:
+        if b["ts"] <= exit_ts_et:
+            continue
+        if b["low"] < entry_price:
+            break
+        if b["high"] > best:
+            best = b["high"]
+    return max(0.0, (best - exit_price) / exit_price * 100.0)
+
+
+def compute_tallies(closes: list[dict], bars_for) -> dict:
+    """Pure core. ``closes``: dicts with symbol, day (date), entry_ts_et /
+    exit_ts_et (tz-aware datetimes), entry_price, exit_price, exit_reason,
+    realized_pnl. ``bars_for(symbol, day)`` -> chronological bar dicts
+    (tz-aware ET ts, high, low). Returns the tally dict."""
+    buckets: dict[str, dict] = {k: {"n": 0, "net_pnl": 0.0}
+                                for k in ("open", "late_morning", "midday")}
+    post_runs: list[dict] = []
+    for c in closes:
+        b = _bucket(c["entry_ts_et"].timetz().replace(tzinfo=None))
+        buckets[b]["n"] += 1
+        buckets[b]["net_pnl"] = round(buckets[b]["net_pnl"] + c["realized_pnl"], 2)
+        if c["exit_reason"] == "market_exit":
+            left = capturable_left_pct(bars_for(c["symbol"], c["day"]),
+                                       c["exit_ts_et"], c["entry_price"], c["exit_price"])
+            post_runs.append({"symbol": c["symbol"], "day": str(c["day"]),
+                              "pnl": c["realized_pnl"], "left_pct": round(left, 1)})
+    clipped = [p for p in post_runs if p["left_pct"] >= 5.0]
+    return {
+        "n_trades": len(closes),
+        "entry_buckets": buckets,
+        "market_exit_post_runs": post_runs,
+        "n_market_exits": len(post_runs),
+        "n_clipped_ge_5pct": len(clipped),
+        "median_left_pct": (sorted(p["left_pct"] for p in post_runs)[len(post_runs) // 2]
+                            if post_runs else 0.0),
+    }
+
+
+def _load_closes(con, since_et: datetime) -> list[dict]:
+    """events.timestamp is stored ET-naive (matches the live loop's wall clock)."""
+    rows = con.execute(
+        "SELECT timestamp, payload_json FROM events "
+        "WHERE event_type = 'position_closed' AND timestamp >= ? ORDER BY timestamp",
+        [since_et.replace(tzinfo=None)],
+    ).fetchall()
+    submits = con.execute(
+        "SELECT timestamp, payload_json FROM events "
+        "WHERE event_type = 'order_submitted' AND timestamp >= ? ORDER BY timestamp",
+        [since_et.replace(tzinfo=None)],
+    ).fetchall()
+    first_submit: dict[tuple, datetime] = {}
+    for ts, pj in submits:
+        p = pj if isinstance(pj, dict) else json.loads(pj or "{}")
+        key = (ts.date(), p.get("symbol"))
+        first_submit.setdefault(key, ts)
+    closes = []
+    for ts, pj in rows:
+        p = pj if isinstance(pj, dict) else json.loads(pj or "{}")
+        sym = p.get("symbol")
+        if not sym or p.get("realized_pnl") is None:
+            continue
+        sub = first_submit.get((ts.date(), sym)) or ts
+        closes.append({
+            "symbol": sym, "day": ts.date(),
+            "entry_ts_et": sub.replace(tzinfo=ET),
+            "exit_ts_et": ts.replace(tzinfo=ET),
+            "entry_price": float(p.get("entry_price") or 0),
+            "exit_price": float(p.get("exit_price") or 0),
+            "exit_reason": p.get("exit_reason") or "",
+            "realized_pnl": float(p["realized_pnl"]),
+        })
+    return closes
+
+
+def _bars_for(con):
+    """minute_bars.timestamp is stored UTC-naive — convert to tz-aware ET."""
+    def fetch(symbol: str, day) -> list[dict]:
+        rows = con.execute(
+            "SELECT timestamp, high, low FROM minute_bars "
+            "WHERE symbol = ? AND session_date = ? AND is_regular_hours "
+            "ORDER BY timestamp",
+            [symbol, day],
+        ).fetchall()
+        return [{"ts": t.replace(tzinfo=timezone.utc).astimezone(ET),
+                 "high": float(h), "low": float(l)} for t, h, l in rows]
+    return fetch
+
+
+def _summary_text(t: dict) -> str:
+    eb = t["entry_buckets"]
+    lines = [f"Weekly tallies ({t['n_trades']} trades):"]
+    for k in ("open", "late_morning", "midday"):
+        lines.append(f"  entry {k}: {eb[k]['n']} trades net ${eb[k]['net_pnl']:+.0f}")
+    lines.append(f"  market_exit post-run: {t['n_market_exits']} exits, "
+                 f"{t['n_clipped_ge_5pct']} clipped >=5% capturable, "
+                 f"median left {t['median_left_pct']:.1f}%")
+    return "\n".join(lines)
+
+
+def run(con, store, *, days: int = 7, now_et: datetime | None = None, notify=None) -> dict:
+    from storage.event_schema import EventMode, RiskRuleTriggeredEvent
+
+    now_et = now_et or datetime.now(ET)
+    since = now_et - timedelta(days=days)
+    closes = _load_closes(con, since)
+    tallies = compute_tallies(closes, _bars_for(con))
+    msg = _summary_text(tallies)
+    store.emit(RiskRuleTriggeredEvent(
+        timestamp=datetime.now(), mode=EventMode.PAPER,
+        correlation_id="weekly_tallies", message=msg.splitlines()[0],
+        rule_type="weekly_tally", rule_value=float(tallies["n_trades"]),
+        current_state=tallies, action_taken="reported",
+    ))
+    if notify is not None:
+        try:
+            notify(f"\U0001F4CA {msg}")
+        except Exception:  # noqa: BLE001 — an alert failure must not break the job
+            pass
+    print(f"[weekly_tallies] {msg}", flush=True)
+    return tallies
+
+
+def main() -> int:
+    from dotenv import load_dotenv
+    load_dotenv()
+    from runtime.env_overrides import load as _load_overrides
+    _load_overrides()
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--days", type=int, default=7)
+    parser.add_argument("--force", action="store_true",
+                        help="run even when it is not Friday ET")
+    args = parser.parse_args()
+
+    now_et = datetime.now(ET)
+    if now_et.weekday() != 4 and not args.force:
+        print("[weekly_tallies] not Friday — skipping (use --force to override)", flush=True)
+        return 0
+
+    from storage.db import get_connection
+    from storage.event_store import EventStore
+    con = get_connection(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum"))
+    store = EventStore(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum"))
+    try:
+        from runtime.notifier import send_telegram
+    except Exception:  # noqa: BLE001
+        send_telegram = None
+    run(con, store, days=args.days, now_et=now_et, notify=send_telegram)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())