momentum-system/research/eod_reconcile.py

"""Nightly cage count — reconcile BROKER truth against EVENT-STORE truth (plan 003 step 5).

After the close, diff what Alpaca says (open positions, today's orders) against what
the event store recorded (today's submits/fills/closes). Any discrepancy — a position
the store doesn't know about, a broker order with no submit event, an unjournaled fill —
emits a loud ``eod_reconcile`` event and a Telegram alert. READ-ONLY: it reports, it
never auto-fixes (a duplicate resting order after a crash-restart is the canonical case,
2026-07-06 — the human decides what to cancel).

    python -m research.eod_reconcile        # run after the close

The diff core (``compute_reconcile``) is pure so it unit-tests without a broker or DB.
"""

from __future__ import annotations

import json
import os
from datetime import datetime
from zoneinfo import ZoneInfo

ET = ZoneInfo("America/New_York")

# broker order statuses that are still WORKING (an open resting order the store
# should have a matching submit for); terminal statuses are expected to differ.
_LIVE_ORDER_STATUSES = {"new", "accepted", "pending_new", "partially_filled",
                        "held", "accepted_for_bidding"}


def _payload(e: dict) -> dict:
    raw = e.get("payload") or e.get("payload_json")
    if isinstance(raw, str):
        try:
            return json.loads(raw)
        except ValueError:
            return {}
    return raw or {}


def compute_reconcile(broker_positions: list[dict], broker_orders: list[dict],
                      events: dict[str, list[dict]]) -> dict:
    """Pure diff. ``events`` maps event_type -> today's events (order_submitted,
    order_filled, position_closed). Returns {ok, mismatches:[...], summary:{...}}."""
    # broker truth
    broker_pos_syms = {str(p.get("symbol")) for p in broker_positions if p.get("symbol")}
    live_broker_orders = [o for o in broker_orders
                          if str(o.get("status")) in _LIVE_ORDER_STATUSES]
    broker_order_syms = {str(o.get("symbol")) for o in live_broker_orders if o.get("symbol")}

    # event-store truth
    submitted_syms = {str(_payload(e).get("symbol"))
                      for e in events.get("order_submitted", []) if _payload(e).get("symbol")}
    closed_syms = {str(_payload(e).get("symbol"))
                   for e in events.get("position_closed", []) if _payload(e).get("symbol")}

    mismatches: list[str] = []
    # A held position the store never saw submitted, or already saw closed, is a drift.
    for sym in sorted(broker_pos_syms):
        if sym not in submitted_syms:
            mismatches.append(f"broker holds {sym} but no order_submitted event today")
        elif sym in closed_syms:
            mismatches.append(f"broker still holds {sym} but the store recorded it closed")
    # A live broker order with no matching submit event = an untracked resting order
    # (the duplicate-after-restart case).
    for sym in sorted(broker_order_syms):
        if sym not in submitted_syms:
            mismatches.append(f"live broker order for {sym} with no order_submitted event today")

    summary = {
        "broker_positions": sorted(broker_pos_syms),
        "live_broker_orders": sorted(broker_order_syms),
        "submitted_today": sorted(submitted_syms),
        "closed_today": sorted(closed_syms),
        "n_mismatches": len(mismatches),
    }
    return {"ok": not mismatches, "mismatches": mismatches, "summary": summary}


def run(store, client, *, now_et: datetime | None = None, notify=None) -> dict:
    """Fetch broker + store truth, diff, emit an event, alert on mismatch. Returns
    the reconcile dict. ``notify`` is an optional send(text) callable (Telegram)."""
    from storage.event_schema import EventMode, RiskRuleTriggeredEvent

    now_et = now_et or datetime.now(ET)
    since = now_et.date().isoformat()
    day_start = f"{since}T00:00:00-04:00"

    broker_positions = client.get_positions() or []
    broker_orders = client.get_orders(status="all", after=day_start) or []
    events = {
        et: list(store.query_events(event_type=et, since=since, limit=None))
        for et in ("order_submitted", "order_filled", "position_closed")
    }
    result = compute_reconcile(broker_positions, broker_orders, events)

    rule = "eod_reconcile_ok" if result["ok"] else "eod_reconcile_mismatch"
    msg = ("EOD reconcile OK — broker and event store agree"
           if result["ok"]
           else f"EOD RECONCILE MISMATCH ({result['summary']['n_mismatches']}): "
                + "; ".join(result["mismatches"]))
    store.emit(RiskRuleTriggeredEvent(
        timestamp=datetime.now(), mode=EventMode.PAPER,
        correlation_id="eod_reconcile", message=msg,
        rule_type=rule, rule_value=float(result["summary"]["n_mismatches"]),
        current_state=result["summary"], action_taken="reported",
    ))
    if not result["ok"] and notify is not None:
        try:
            notify(f"⚠️ {msg}")
        except Exception:  # noqa: BLE001 — an alert failure must not break the job
            pass
    print(f"[eod_reconcile] {msg}", flush=True)
    return result


def main() -> int:
    from dotenv import load_dotenv
    load_dotenv()
    from runtime.env_overrides import load as _load_overrides
    _load_overrides()

    from storage.event_store import EventStore
    from tradier_broker.factory import make_trading_client

    # reconcile against the ACTIVE execution broker (BROKER env), never a
    # hardwired one — drift-checking the wrong broker is silently meaningless
    client, label = make_trading_client()
    if client is None:
        print(f"[eod_reconcile] {label} — skipping", flush=True)
        return 0
    print(f"[eod_reconcile] reconciling against {label}", flush=True)
    store = EventStore(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum"))
    try:
        from runtime.notifier import send_telegram
    except Exception:  # noqa: BLE001
        send_telegram = None
    result = run(store, client, notify=send_telegram)
    return 0 if result["ok"] else 1


if __name__ == "__main__":
    raise SystemExit(main())