diff --git a/admin_api.py b/admin_api.py index 5006e7e..ae35379 100644 --- a/admin_api.py +++ b/admin_api.py @@ -16,6 +16,7 @@ Run: python admin_api.py (ADMIN_HOST/ADMIN_PORT, default 0.0.0.0:8767) from __future__ import annotations +import hmac import http.client import json import os @@ -70,6 +71,14 @@ SEED_PROFILES = { } +def _safe_overrides() -> dict: + """Overrides for API responses with secret VALUES redacted — a /api/mode live + switch writes ALPACA_LIVE_* into overrides.env; without this they'd echo back + in every /api/status and /api/config response.""" + return {k: ("***" if any(m in k for m in SECRET_MARKERS) else v) + for k, v in env_overrides.read_overrides().items()} + + def _editable(key: str) -> bool: if key in EDITABLE_EXACT: return True @@ -194,10 +203,12 @@ class Handler(BaseHTTPRequestHandler): self.wfile.write(raw) def _authed(self) -> bool: + # FAIL CLOSED: no token configured => deny (this panel fronts the docker + # socket + the paper/live switch; an unset env var must not open it to the + # LAN). Header only — a ?token= query would leak into proxy/access logs. if not ADMIN_TOKEN: - return True # explicit opt-out: no token configured - return (self.headers.get("X-Admin-Token") == ADMIN_TOKEN - or f"token={ADMIN_TOKEN}" in (self.path.split("?", 1) + [""])[1]) + return False + return hmac.compare_digest(self.headers.get("X-Admin-Token", ""), ADMIN_TOKEN) def do_GET(self): # noqa: N802 path = self.path.split("?")[0] @@ -208,7 +219,7 @@ class Handler(BaseHTTPRequestHandler): return self._send(401, {"error": "bad or missing X-Admin-Token"}) if path == "/api/status": return self._send(200, {"containers": list_containers(), "checks": checks(), - "overrides": env_overrides.read_overrides()}) + "overrides": _safe_overrides()}) if path == "/api/config": eff = {k: v for k, v in sorted(os.environ.items()) if _editable(k)} secrets = {k: bool(os.environ.get(k)) for k in @@ -216,7 +227,7 @@ class Handler(BaseHTTPRequestHandler): "ALPACA_LIVE_SECRET_KEY", "TELEGRAM_BOT_TOKEN", "YOUTUBE_API_KEY", "FMP_API_KEY")} return self._send(200, {"effective": eff, "secrets_present": secrets, - "overrides": env_overrides.read_overrides()}) + "overrides": _safe_overrides()}) if path == "/api/profiles": return self._send(200, profiles()) return self._send(404, {"error": "unknown route"}) diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index 50c8c5c..b2d2169 100644 --- a/alpaca_paper/client.py +++ b/alpaca_paper/client.py @@ -137,8 +137,14 @@ class AlpacaPaperClient: continue raise AlpacaApiError(exc.code, exc.read().decode(errors="replace")) from exc except urllib.error.URLError: - # transient network/DNS (e.g. gaierror) — retry a couple times - if attempt >= self.max_retries: + # transient network/DNS (e.g. gaierror) — retry a couple times. + # NOT for POST: urllib wraps a lost RESPONSE (read timeout/reset) + # in URLError too, so attempt 1 may have SUCCEEDED at the broker + # with the reply lost; re-sending POST /orders then rests a + # duplicate live order (or 422s and looks like a failure while a + # real bracket rests untracked). Idempotent GET/DELETE retry; POST + # raises so the caller reconciles against broker positions. + if method == "POST" or attempt >= self.max_retries: raise _time.sleep(0.4 * (attempt + 1)) raise RuntimeError("request retries exhausted") # unreachable diff --git a/api/main.py b/api/main.py index a50d9d6..ac1e46d 100644 --- a/api/main.py +++ b/api/main.py @@ -297,7 +297,6 @@ class DashboardHandler(BaseHTTPRequestHandler): self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) - self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) except (BrokenPipeError, ConnectionResetError): @@ -346,7 +345,6 @@ class DashboardHandler(BaseHTTPRequestHandler): self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") - self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() shutdown = getattr(type(self), "shutdown_event", None) @@ -410,6 +408,13 @@ class DashboardHandler(BaseHTTPRequestHandler): self._send_json({"error": str(exc)}, 500) def do_POST(self): # noqa: N802 + # optional shared-secret gate on trade actions (approve/reject/exit/watch). + # Off by default (autonomous stack auto-approves); set DASHBOARD_TOKEN to + # require X-Dashboard-Token on every POST. CORS wildcard was removed so a + # cross-origin LAN page can't drive these either way. + _tok = os.environ.get("DASHBOARD_TOKEN") + if _tok and not hmac.compare_digest(self.headers.get("X-Dashboard-Token", ""), _tok): + return self._send_json({"error": "bad or missing X-Dashboard-Token"}, 401) try: route = urlparse(self.path).path body = self._read_body() @@ -443,7 +448,6 @@ class DashboardHandler(BaseHTTPRequestHandler): def do_OPTIONS(self): # noqa: N802 self.send_response(204) - self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() diff --git a/controlroom/app.py b/controlroom/app.py deleted file mode 100644 index fd11e6f..0000000 --- a/controlroom/app.py +++ /dev/null @@ -1,399 +0,0 @@ -"""Momentum Trading Control Room — NiceGUI live-analysis dashboard (Milestone 5, core). - -Phases 1+2 of the staged control-room plan, adapted to THIS backend: a real-time -operator view with LIVE TICKERS (SIP last-trade prices at 1s), per-symbol analysis -(state, gap, RVOL, trigger distance, quality, population-gate check), live positions -P&L, today's gate activity + attribution, and the approval queue with confirm-flow -actions. Runs as its OWN service on :8766 — the existing dashboard on :8765 stays -untouched until this reaches parity (their Phase 3-5: manual orders, settings dials, -audit trail). - -Data flow: read-only projections from the shared Postgres event store + Alpaca SIP -for live prices; fully AUTONOMOUS ops (owner directive: no approvals/interaction needed) — the only -action is the emergency panic brake (cancel all open orders, typed confirmation).""" - -from __future__ import annotations - -import json -import os -from datetime import date, datetime - -from dotenv import load_dotenv -load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) - -from nicegui import run, ui # noqa: E402 - -from alpaca_paper.client import AlpacaPaperClient # noqa: E402 -from alpaca_paper.settings import AlpacaPaperSettings # noqa: E402 -from momentum_cli import _market_clock, _triggers_snapshot, _unprotected_positions # noqa: E402 -from storage.db_pg import get_connection # noqa: E402 -from storage.event_store import EventStore # noqa: E402 -from storage.projections import query_approval_queue, query_ready_signals_snapshot # noqa: E402 - -POP_GAP, POP_VOL = 0.10, 300_000 # display mirror of the live population gate - -import threading -con = get_connection("momentum") -store = EventStore("momentum") -client = AlpacaPaperClient(AlpacaPaperSettings.from_env()) -DB_LOCK = threading.Lock() # fetchers run in io_bound THREADS; one psycopg2 conn is not thread-safe - - -# ---------------- data fetchers (sync; called via run.io_bound) ---------------- -def fetch_board() -> dict: - with DB_LOCK: - trig_rows, snap_ts, armed_n = _triggers_snapshot(con) - signals = {s["symbol"]: s for s in query_ready_signals_snapshot(store)} - rows = [] - for r in trig_rows: - sym = r.get("symbol") - sig = signals.get(sym, {}) - gap = r.get("gap") # percent units from the book - gap_frac = (gap / 100.0) if isinstance(gap, (int, float)) else sig.get("gap_pct") - vol = sig.get("cum_volume") - pop_ok = None - if gap_frac is not None: - pop_ok = bool(gap_frac >= POP_GAP and (vol is None or vol >= POP_VOL)) - rows.append({ - "symbol": sym, "state": r.get("state"), "price": r.get("price"), - "gap": gap, "rvol": r.get("rvol"), "trigger": r.get("trigger"), - "to_trig": r.get("→trigger") or r.get("to_trigger"), "stop": r.get("stop"), - "quality": sig.get("quality_score"), "vwap": sig.get("vwap"), - "pop": ("✓" if pop_ok else "✗" if pop_ok is not None else "—"), - "catalyst": (r.get("catalyst") or "")[:40], - }) - return {"rows": rows, "ts": str(snap_ts)[11:19] if snap_ts else "—", "armed": armed_n} - - -def fetch_prices(symbols: list[str]) -> dict[str, float]: - if not symbols: - return {} - try: - trades = client.get_latest_trades(symbols) - return {s: float(t.get("p")) for s, t in (trades or {}).items() if t and t.get("p")} - except Exception: # noqa: BLE001 - return {} - - -def fetch_account() -> dict: - try: - a = client.get_account() - eq = float(a.get("equity") or 0) - last = float(a.get("last_equity") or 0) - return {"equity": eq, "day_pnl": (eq - last) if last else 0.0} - except Exception: # noqa: BLE001 - return {} - - -def fetch_positions() -> list[dict]: - try: - pos = client.get_positions(fresh=True) - with DB_LOCK: - naked = {n["symbol"] for n in (_unprotected_positions(con) or [])} - return [{ - "symbol": p.get("symbol"), "qty": p.get("qty"), - "avg": float(p.get("avg_entry_price") or 0), - "price": float(p.get("current_price") or 0), - "pnl": float(p.get("unrealized_pl") or 0), - "protected": "⚠ NAKED" if p.get("symbol") in naked else "✓", - } for p in pos] - except Exception: # noqa: BLE001 - return [] - - -def fetch_gates() -> list[dict]: - today = date.today().isoformat() - hist: dict[str, int] = {} - with DB_LOCK: - rows = con.execute( - "SELECT payload_json FROM events WHERE event_type='risk_rule_triggered' " - "AND timestamp >= ?", [today]).fetchall() - for (pj,) in rows: - rt = (json.loads(pj or "{}").get("rule_type")) or "?" - hist[rt] = hist.get(rt, 0) + 1 - return [{"gate": k, "cuts": v} for k, v in sorted(hist.items(), key=lambda x: -x[1])] - - -def fetch_attribution() -> list[dict]: - from research.attribution import compute_attribution - att = compute_attribution(days=7) - out = [] - for d in sorted(att["days"], reverse=True): - st = att["days"][d] - c, r = st.get("counts", {}), st.get("rating", {}) - out.append({"day": d, "disc": c.get("symbol_discovered", 0), - "ready": c.get("signal_ready", 0), - "early": f"{r.get('signals_early_pct', '—')}%", - "trades": st.get("outcomes", {}).get("trades", 0), - "win": f"{r.get('win_rate_pct', '—')}%", - "pnl": r.get("pnl")}) - return out - - -def fetch_activity(limit: int = 40) -> list[dict]: - """Merged live event feed for autonomous ops: entries, exits, fills, gate cuts.""" - kinds = ("position_closed", "order_submitted", "order_cancelled", "signal_ready", - "risk_rule_triggered", "order_filled") - ph = ",".join(["?"] * len(kinds)) - with DB_LOCK: - rows = con.execute( - f"SELECT timestamp, event_type, message FROM events " - f"WHERE event_type IN ({ph}) ORDER BY timestamp DESC LIMIT ?", - [*kinds, limit]).fetchall() - return [{"time": str(ts)[11:19], "type": et, "message": (m or "")[:140]} - for ts, et, m in rows] - - -def fetch_summary() -> dict: - today = date.today().isoformat() - with DB_LOCK: - closes = con.execute( - "SELECT payload_json FROM events WHERE event_type='position_closed' " - "AND timestamp >= ?", [today]).fetchall() - subs = con.execute( - "SELECT count(*) FROM events WHERE event_type='order_submitted' " - "AND timestamp >= ?", [today]).fetchone()[0] - cuts = con.execute( - "SELECT count(*) FROM events WHERE event_type='risk_rule_triggered' " - "AND timestamp >= ?", [today]).fetchone()[0] - hb = con.execute( - "SELECT max(timestamp) FROM events WHERE event_type='module_tick'").fetchone()[0] - pnls = [] - for (pj,) in closes: - try: - pnls.append(float(json.loads(pj or "{}").get("realized_pnl") or 0)) - except Exception: # noqa: BLE001 - pass - age = (datetime.now() - hb).total_seconds() if hb else None - return {"trades": len(pnls), "wins": sum(1 for x in pnls if x > 0), - "realized": round(sum(pnls), 2), "submissions": subs, "cuts": cuts, - "loop_age": age} - - -def cancel_all_orders() -> str: - return str(client._trading("DELETE", "/orders"))[:200] - - -# ---------------- UI ---------------- -@ui.page("/") -def main_page(): - ui.add_head_html(""" - """) - prices: dict[str, float] = {} - board_syms: list[str] = [] - - with ui.header().classes("items-center justify-between px-4 bg-blue-900 text-white"): - with ui.row().classes("items-center gap-3"): - ui.label("Momentum Control Room").classes("text-lg font-bold") - ui.badge("🔵 PAPER — Alpaca (Simulated)").classes("text-sm px-2") - with ui.row().classes("items-center gap-4"): - phase_lbl = ui.label("—").classes("text-sm") - eq_lbl = ui.label("equity —").classes("font-mono text-sm") - hb_lbl = ui.label("♥ —").classes("font-mono text-xs") - - async def panic(): - with ui.dialog() as d, ui.card().classes("p-6"): - ui.label("⛔ Cancel ALL open orders?").classes("text-lg font-bold") - ui.label('Type "STOP" to confirm:').classes("mt-2") - typed = ui.input("") - with ui.row().classes("justify-end gap-2 mt-4"): - btn = ui.button("CANCEL ALL", on_click=lambda: d.submit(True)) \ - .props("color=negative") - btn.bind_enabled_from(typed, "value", backward=lambda v: v == "STOP") - ui.button("Back", on_click=lambda: d.submit(False)) - if await d: - res = await run.io_bound(cancel_all_orders) - ui.notify(f"cancel-all sent: {res}", type="warning") - ui.button("⛔ STOP ALL", on_click=panic).props("color=negative outline") \ - .classes("font-bold") - clock_lbl = ui.label("--:--:--").classes("font-mono") - - with ui.tabs().classes("w-full") as tabs: - t_live = ui.tab("Live") - t_pos = ui.tab("Positions") - t_gates = ui.tab("Gates & Attribution") - t_act = ui.tab("Activity") - - with ui.tab_panels(tabs, value=t_live).classes("w-full"): - # ---- LIVE: real-time tickers + analysis ---- - with ui.tab_panel(t_live): - with ui.row().classes("gap-4 mb-2"): - def _card(title): - with ui.card().classes("p-3 min-w-32"): - ui.label(title).classes("text-xs text-grey-6") - return ui.label("—").classes("text-lg font-bold font-mono") - c_pnl = _card("Day P&L") - c_trades = _card("Trades today") - c_win = _card("Win rate") - c_subs = _card("Submissions") - c_cuts = _card("Gate cuts") - c_loop = _card("Loop ♥") - - async def refresh_summary(): - sm = await run.io_bound(fetch_summary) - acct = await run.io_bound(fetch_account) - pnl = acct.get("day_pnl", 0.0) - c_pnl.text = f"${pnl:+,.0f}" - c_pnl.classes(replace="text-lg font-bold font-mono " + - ("text-green" if pnl >= 0 else "text-red")) - c_trades.text = str(sm["trades"]) - c_win.text = (f"{100*sm['wins']/sm['trades']:.0f}%" if sm["trades"] else "—") - c_subs.text = str(sm["submissions"]) - c_cuts.text = str(sm["cuts"]) - age = sm.get("loop_age") - c_loop.text = f"{age:.0f}s" if age is not None else "—" - c_loop.classes(replace="text-lg font-bold font-mono " + - ("text-green" if age is not None and age < 180 else "text-red")) - ui.timer(5.0, refresh_summary) - - with ui.row().classes("items-center gap-4"): - ui.label("Live tickers").classes("text-lg font-bold") - snap_lbl = ui.label("snapshot —").classes("text-xs font-mono text-grey") - armed_lbl = ui.label("").classes("text-xs font-mono") - live_grid = ui.aggrid({ - "columnDefs": [ - {"headerName": "Sym", "field": "symbol", "pinned": "left", "width": 90}, - {"headerName": "State", "field": "state", "width": 95, "cellClassRules": { - "text-green": 'value === "armed" || value === "fired"', - }}, - {"headerName": "Price", "field": "price", "width": 95, - "cellClass": "font-mono", "cellClassRules": { - "flash-green": "data._d > 0", "flash-red": "data._d < 0"}}, - {"headerName": "Gap%", "field": "gap", "width": 80, "cellClass": "font-mono"}, - {"headerName": "RVOL", "field": "rvol", "width": 80, "cellClass": "font-mono"}, - {"headerName": "Trigger", "field": "trigger", "width": 95, "cellClass": "font-mono"}, - {"headerName": "→Trig", "field": "to_trig", "width": 85, "cellClass": "font-mono"}, - {"headerName": "Stop", "field": "stop", "width": 90, "cellClass": "font-mono"}, - {"headerName": "Qual", "field": "quality", "width": 75, "cellClassRules": { - "text-green": "value >= 0.5", "text-red": "value < 0.5"}}, - {"headerName": "Pop", "field": "pop", "width": 65, "cellClassRules": { - "text-green": 'value === "✓"', "text-red": 'value === "✗"'}}, - {"headerName": "VWAP", "field": "vwap", "width": 90, "cellClass": "font-mono"}, - {"headerName": "Catalyst", "field": "catalyst", "flex": 1}, - ], - "rowData": [], "domLayout": "autoHeight", "animateRows": True, - }).classes("w-full") - - async def refresh_board(): - data = await run.io_bound(fetch_board) - board_syms[:] = [r["symbol"] for r in data["rows"] if r.get("symbol")] - for r in data["rows"]: - lp = prices.get(r["symbol"]) - if lp: - old = r.get("price") - r["_d"] = (lp - old) if isinstance(old, (int, float)) else 0 - r["price"] = lp - live_grid.options["rowData"] = data["rows"] - live_grid.update() - snap_lbl.text = f"loop snapshot {data['ts']}" - armed_lbl.text = f"{data['armed']} armed · {len(data['rows'])} tracked" - - async def refresh_prices(): - if board_syms: - fresh = await run.io_bound(fetch_prices, list(board_syms)) - if fresh: - prices.update(fresh) - rows = live_grid.options["rowData"] - for r in rows: - lp = prices.get(r["symbol"]) - if lp and isinstance(r.get("price"), (int, float)): - r["_d"] = round(lp - r["price"], 4) - r["price"] = lp - live_grid.update() - - ui.timer(3.0, refresh_board) - ui.timer(1.0, refresh_prices) - - # ---- POSITIONS ---- - with ui.tab_panel(t_pos): - pos_grid = ui.aggrid({ - "columnDefs": [ - {"headerName": "Sym", "field": "symbol", "width": 90}, - {"headerName": "Qty", "field": "qty", "width": 80, "cellClass": "font-mono"}, - {"headerName": "Avg", "field": "avg", "width": 95, "cellClass": "font-mono"}, - {"headerName": "Price", "field": "price", "width": 95, "cellClass": "font-mono"}, - {"headerName": "uP&L $", "field": "pnl", "width": 110, "cellClass": "font-mono", - "cellClassRules": {"text-green": "value >= 0", "text-red": "value < 0"}}, - {"headerName": "Stop", "field": "protected", "width": 110, "cellClassRules": { - "text-red": 'value !== "✓"'}}, - ], "rowData": [], "domLayout": "autoHeight", - }).classes("w-full") - - async def refresh_pos(): - pos_grid.options["rowData"] = await run.io_bound(fetch_positions) - pos_grid.update() - ui.timer(2.0, refresh_pos) - - # ---- GATES + ATTRIBUTION ---- - with ui.tab_panel(t_gates): - ui.label("Today's gate activity").classes("text-lg font-bold") - gates_grid = ui.aggrid({"columnDefs": [ - {"headerName": "Gate", "field": "gate", "flex": 1}, - {"headerName": "Cuts today", "field": "cuts", "width": 120}], - "rowData": [], "domLayout": "autoHeight"}).classes("w-1/2") - ui.label("Component attribution (7 days)").classes("text-lg font-bold mt-6") - att_grid = ui.aggrid({"columnDefs": [ - {"headerName": "Day", "field": "day", "width": 110}, - {"headerName": "Disc", "field": "disc", "width": 80}, - {"headerName": "Ready", "field": "ready", "width": 80}, - {"headerName": "Early%", "field": "early", "width": 90}, - {"headerName": "Trades", "field": "trades", "width": 80}, - {"headerName": "Win%", "field": "win", "width": 80}, - {"headerName": "P&L", "field": "pnl", "width": 100, "cellClass": "font-mono", - "cellClassRules": {"text-green": "value >= 0", "text-red": "value < 0"}}], - "rowData": [], "domLayout": "autoHeight"}).classes("w-full") - - async def refresh_gates(): - gates_grid.options["rowData"] = await run.io_bound(fetch_gates) - gates_grid.update() - ui.timer(5.0, refresh_gates) - - async def load_attr(): - att_grid.options["rowData"] = await run.io_bound(fetch_attribution) - att_grid.update() - ui.button("Load / refresh attribution", on_click=load_attr).props("outline") - - # ---- ACTIVITY: autonomous-ops live feed ---- - with ui.tab_panel(t_act): - ui.label("Live activity (autonomous — no input needed)").classes("text-lg font-bold") - act_grid = ui.aggrid({"columnDefs": [ - {"headerName": "Time", "field": "time", "width": 100, "cellClass": "font-mono"}, - {"headerName": "Event", "field": "type", "width": 170, "cellClassRules": { - "text-green": 'value === "position_closed" || value === "order_filled"', - "text-red": 'value === "risk_rule_triggered"'}}, - {"headerName": "Detail", "field": "message", "flex": 1}], - "rowData": [], "domLayout": "autoHeight"}).classes("w-full") - - async def refresh_act(): - act_grid.options["rowData"] = await run.io_bound(fetch_activity) - act_grid.update() - ui.timer(4.0, refresh_act) - - # ---- global header timers ---- - async def refresh_header(): - acct = await run.io_bound(fetch_account) - if acct: - eq_lbl.text = f"equity ${acct['equity']:,.0f}" - phase, detail, _ = _market_clock() - phase_lbl.text = f"{phase} {detail}" - hb_lbl.text = f"♥ {datetime.now().strftime('%H:%M:%S')}" - ui.timer(2.0, refresh_header) - - def tick_clock(): - clock_lbl.text = datetime.now().strftime("%H:%M:%S") - ui.timer(1.0, tick_clock) - - -if __name__ in {"__main__", "__mp_main__"}: - ui.run(host="0.0.0.0", port=int(os.environ.get("CONTROLROOM_PORT", "8766")), - title="Momentum Control Room", dark=True, show=False, reload=False, - favicon="🎯") diff --git a/core/contracts.py b/core/contracts.py deleted file mode 100644 index d8ce52e..0000000 --- a/core/contracts.py +++ /dev/null @@ -1,259 +0,0 @@ -"""Module contracts and interfaces for the momentum trading system. - -Every module must implement the ModuleContract protocol, defining: -- input_schema: What the module expects -- output_schema: What the module produces -- run(): Execute the module -- validate_input(): Ensure data meets contracts -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from dataclasses import dataclass, field -from datetime import datetime, date -from typing import Any, Protocol, TypeVar - -T_Input = TypeVar("T_Input") -T_Output = TypeVar("T_Output") - - -@dataclass -class ModuleMetrics: - """Standard metrics every module should track.""" - - module_name: str - timestamp: datetime - duration_ms: float - input_count: int = 0 - output_count: int = 0 - error_count: int = 0 - warnings: list[str] = field(default_factory=list) - custom_metrics: dict[str, Any] = field(default_factory=dict) - - def to_dict(self) -> dict: - return { - "module_name": self.module_name, - "timestamp": self.timestamp.isoformat(), - "duration_ms": self.duration_ms, - "input_count": self.input_count, - "output_count": self.output_count, - "error_count": self.error_count, - "warnings": self.warnings, - "custom_metrics": self.custom_metrics, - } - - -@dataclass -class ValidationFailure: - """Single validation failure with context.""" - - field: str - value: Any - reason: str - severity: str = "error" # "error" or "warning" - - -@dataclass -class ModuleResult: - """Base class for all module results.""" - - success: bool - timestamp: datetime - metrics: ModuleMetrics - validation_failures: list[ValidationFailure] = field(default_factory=list) - - def has_errors(self) -> bool: - return bool(self.validation_failures or not self.success) - - def error_summary(self) -> str: - """Human-readable error summary.""" - if not self.validation_failures: - return "No errors" - errors = [f for f in self.validation_failures if f.severity == "error"] - if not errors: - return f"{len(self.validation_failures)} warnings only" - return f"{len(errors)} error(s), {len([f for f in self.validation_failures if f.severity == 'warning'])} warning(s)" - - -class ModuleContract(Protocol): - """Contract every module must fulfill.""" - - def run(self, input: Any) -> ModuleResult: - """Execute the module. - - Args: - input: Module-specific input conforming to input_schema - - Returns: - Module-specific result inheriting from ModuleResult - """ - ... - - def validate_input(self, input: Any) -> tuple[bool, list[ValidationFailure]]: - """Validate input before processing. - - Args: - input: Data to validate - - Returns: - (is_valid, failures) - """ - ... - - def get_metrics(self) -> ModuleMetrics: - """Get last execution metrics.""" - ... - - -# ============================================================================ -# INGESTION MODULE CONTRACTS -# ============================================================================ - - -@dataclass -class BarIngestRequest: - """Input contract for ingestion module.""" - - symbols: list[str] - session_date: date - lookback_minutes: int = 120 # how many minutes of history to fetch - source: str = "alpaca" # "alpaca", "test", "file" - - -@dataclass -class BarIngestResult(ModuleResult): - """Output contract for ingestion module.""" - - minute_rows_inserted: int = 0 - symbols_updated: list[str] = field(default_factory=list) - source_health: dict[str, str] = field(default_factory=dict) # {source: "healthy"|"error"} - - -# ============================================================================ -# RESEARCH MODULE CONTRACTS -# ============================================================================ - - -@dataclass -class ResearchRequest: - """Input contract for research module.""" - - symbols: list[str] - session_date: date - lookback_days: int = 30 # for daily history - - -@dataclass -class ResearchResult(ModuleResult): - """Output contract for research module.""" - - symbols_researched: list[str] = field(default_factory=list) - symbols_with_data: list[str] = field(default_factory=list) - symbols_missing_data: list[str] = field(default_factory=list) - total_bars_available: int = 0 - - -# ============================================================================ -# EVALUATION MODULE CONTRACTS -# ============================================================================ - - -@dataclass -class CriterionDetail: - """Result of evaluating one criterion.""" - - name: str - passed: bool - value: float # actual metric value - threshold: float | None = None # minimum to pass - reason: str = "" - confidence: float = 1.0 # 0-1, how confident in this result - - -@dataclass -class EvaluationRequest: - """Input contract for evaluation module.""" - - symbol: str - bars_df: Any # pd.DataFrame with {timestamp, open, high, low, close, volume} - previous_close: float - avg_daily_volume: float - session_date: date - - -@dataclass -class EvaluationResult(ModuleResult): - """Output contract for evaluation module.""" - - symbol: str = "" - status: str = "late" # "ready", "blocked", "late", "error" - score: float = 0.0 # 0-100 - criteria: list[CriterionDetail] = field(default_factory=list) - entry_price: float | None = None - stop_loss: float | None = None - target_price: float | None = None - quality_score: float = 0.0 # 0-1 - blocking_reason: str | None = None - metrics_dict: dict[str, float] = field(default_factory=dict) # gap%, rvol, etc. - - -# ============================================================================ -# EXECUTION MODULE CONTRACTS -# ============================================================================ - - -@dataclass -class ExecutionRequest: - """Input contract for execution module.""" - - symbol: str - entry_price: float - stop_price: float - target_price: float - score: float - criteria_count: int - account_equity: float - current_positions: int - today_realized_pnl: float - - -@dataclass -class ExecutionResult(ModuleResult): - """Output contract for execution module.""" - - symbol: str = "" - status: str = "" # "approved", "rejected", "pending_approval", "error" - reason: str = "" # why it was approved/rejected - order_id: str | None = None - shares: int | None = None - risk_dollars: float | None = None - risk_checks_applied: list[str] = field(default_factory=list) - - -# ============================================================================ -# BROKER MODULE CONTRACTS -# ============================================================================ - - -@dataclass -class BrokerRequest: - """Input contract for broker module.""" - - symbol: str - order_id: str - entry_price: float - stop_price: float - target_price: float - shares: int - order_type: str = "limit" # "limit" or "market" - - -@dataclass -class BrokerResult(ModuleResult): - """Output contract for broker module.""" - - symbol: str = "" - status: str = "" # "submitted", "rejected", "error" - broker_order_id: str | None = None - submitted_at: datetime | None = None diff --git a/core/telemetry.py b/core/telemetry.py deleted file mode 100644 index 3ef5382..0000000 --- a/core/telemetry.py +++ /dev/null @@ -1,274 +0,0 @@ -"""Telemetry collection and event emission for module observability.""" - -from __future__ import annotations - -import time -import logging -from dataclasses import dataclass, field -from datetime import datetime -from typing import TYPE_CHECKING, Any, Callable - -if TYPE_CHECKING: - from core.contracts import ModuleMetrics - -logger = logging.getLogger(__name__) - - -@dataclass -class TelemetryCollector: - """Collect metrics during module execution.""" - - module_name: str - start_time: float = field(default_factory=time.time) - input_count: int = 0 - output_count: int = 0 - error_count: int = 0 - warnings: list[str] = field(default_factory=list) - custom_metrics: dict[str, Any] = field(default_factory=dict) - - def record_input(self, count: int = 1) -> None: - """Record input items processed.""" - self.input_count += count - - def record_output(self, count: int = 1) -> None: - """Record output items produced.""" - self.output_count += count - - def record_error(self) -> None: - """Record an error occurred.""" - self.error_count += 1 - - def record_warning(self, message: str) -> None: - """Record a warning.""" - self.warnings.append(message) - - def record_metric(self, name: str, value: float | int | str) -> None: - """Record a custom metric.""" - self.custom_metrics[name] = value - - def finalize(self) -> ModuleMetrics: - """Finalize and return metrics.""" - duration_ms = (time.time() - self.start_time) * 1000 - return ModuleMetrics( - module_name=self.module_name, - timestamp=datetime.now(), - duration_ms=duration_ms, - input_count=self.input_count, - output_count=self.output_count, - error_count=self.error_count, - warnings=self.warnings, - custom_metrics=self.custom_metrics, - ) - - -class TelemetryEmitter: - """Emit telemetry events to event store and observability systems.""" - - def __init__(self, event_store=None): - """Initialize emitter. - - Args: - event_store: Optional EventStore for persisting telemetry events - """ - self.event_store = event_store - self.callbacks: list[Callable[[dict], None]] = [] - - def add_callback(self, callback: Callable[[dict], None]) -> None: - """Add a callback to receive telemetry events.""" - self.callbacks.append(callback) - - def emit_module_tick( - self, - module_name: str, - stage: str, - metrics: ModuleMetrics | None = None, - errors: list[dict] | None = None, - correlation_id: str | None = None, - ) -> None: - """Emit a module tick event. - - Args: - module_name: Name of module (e.g., "ingestion", "evaluation") - stage: Stage name (e.g., "started", "processing", "completed") - metrics: Optional ModuleMetrics object - errors: Optional list of error details - correlation_id: Correlation ID linking to signal_ready event - """ - event = { - "type": "module_tick", - "timestamp": datetime.now().isoformat(), - "module": module_name, - "stage": stage, - "metrics": metrics.to_dict() if metrics else None, - "errors": errors or [], - "correlation_id": correlation_id, - } - self._emit(event) - - def emit_data_validation( - self, - module_name: str, - validation_type: str, - valid: bool, - details: dict, - correlation_id: str | None = None, - ) -> None: - """Emit a data validation event. - - Args: - module_name: Name of module performing validation - validation_type: Type of validation ("schema", "checksum", "business_rules") - valid: Whether validation passed - details: Validation details (errors, warnings, statistics) - correlation_id: Optional correlation ID - """ - event = { - "type": "data_validation", - "timestamp": datetime.now().isoformat(), - "module": module_name, - "validation_type": validation_type, - "valid": valid, - "details": details, - "correlation_id": correlation_id, - } - self._emit(event) - - def emit_symbol_evaluation_detail( - self, - symbol: str, - stage: str, - metrics: dict[str, Any], - passed: bool, - blocking_reason: str | None = None, - correlation_id: str | None = None, - ) -> None: - """Emit detailed evaluation event for a symbol. - - Args: - symbol: Symbol being evaluated - stage: Evaluation stage (e.g., "bars_loaded", "gap_checked", "rvol_calculated") - metrics: Metrics at this stage - passed: Whether this stage passed - blocking_reason: If not passed, why - correlation_id: Optional correlation ID - """ - event = { - "type": "symbol_evaluation_detail", - "timestamp": datetime.now().isoformat(), - "symbol": symbol, - "stage": stage, - "metrics": metrics, - "passed": passed, - "blocking_reason": blocking_reason, - "correlation_id": correlation_id, - } - self._emit(event) - - def _emit(self, event: dict) -> None: - """Emit event to all registered callbacks and event store.""" - logger.debug(f"Telemetry: {event['type']} - {event.get('module', event.get('symbol', 'n/a'))}") - - # Call all registered callbacks - for callback in self.callbacks: - try: - callback(event) - except Exception as e: - logger.error(f"Telemetry callback failed: {e}") - - # Emit to event store if available - if self.event_store: - try: - # Import here to avoid circular dependency - from storage.event_schema import ModuleTickEvent, DataValidationEvent - - if event["type"] == "module_tick": - self.event_store.emit( - ModuleTickEvent( - timestamp=datetime.now(), - module=event["module"], - stage=event["stage"], - duration_ms=event["metrics"]["duration_ms"] - if event["metrics"] - else 0, - input_count=event["metrics"]["input_count"] - if event["metrics"] - else 0, - output_count=event["metrics"]["output_count"] - if event["metrics"] - else 0, - metrics=event["metrics"]["custom_metrics"] - if event["metrics"] - else {}, - errors=event["errors"], - correlation_id=event.get("correlation_id"), - ) - ) - elif event["type"] == "data_validation": - self.event_store.emit( - DataValidationEvent( - timestamp=datetime.now(), - module=event["module"], - validation_type=event["validation_type"], - valid=event["valid"], - details=event["details"], - correlation_id=event.get("correlation_id"), - ) - ) - except Exception as e: - logger.error(f"Failed to emit telemetry event to store: {e}") - - -# Global instance (can be overridden in tests) -_telemetry_emitter: TelemetryEmitter | None = None - - -def set_telemetry_emitter(emitter: TelemetryEmitter) -> None: - """Set the global telemetry emitter.""" - global _telemetry_emitter - _telemetry_emitter = emitter - - -def get_telemetry_emitter() -> TelemetryEmitter: - """Get the global telemetry emitter.""" - global _telemetry_emitter - if _telemetry_emitter is None: - _telemetry_emitter = TelemetryEmitter() - return _telemetry_emitter - - -def emit_module_tick( - module_name: str, - stage: str, - metrics: ModuleMetrics | None = None, - errors: list[dict] | None = None, - correlation_id: str | None = None, -) -> None: - """Convenience function to emit module tick.""" - get_telemetry_emitter().emit_module_tick(module_name, stage, metrics, errors, correlation_id) - - -def emit_data_validation( - module_name: str, - validation_type: str, - valid: bool, - details: dict, - correlation_id: str | None = None, -) -> None: - """Convenience function to emit data validation.""" - get_telemetry_emitter().emit_data_validation( - module_name, validation_type, valid, details, correlation_id - ) - - -def emit_symbol_evaluation_detail( - symbol: str, - stage: str, - metrics: dict[str, Any], - passed: bool, - blocking_reason: str | None = None, - correlation_id: str | None = None, -) -> None: - """Convenience function to emit symbol evaluation detail.""" - get_telemetry_emitter().emit_symbol_evaluation_detail( - symbol, stage, metrics, passed, blocking_reason, correlation_id - ) diff --git a/core/validation.py b/core/validation.py deleted file mode 100644 index 8422917..0000000 --- a/core/validation.py +++ /dev/null @@ -1,284 +0,0 @@ -"""Data validation utilities for module boundaries.""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Callable - -if TYPE_CHECKING: - from core.contracts import ValidationFailure - - -@dataclass -class SchemaValidator: - """Validate data against a schema.""" - - required_fields: dict[str, type] # {field_name: expected_type} - optional_fields: dict[str, type] | None = None - - def __post_init__(self): - if self.optional_fields is None: - self.optional_fields = {} - - def validate(self, data: dict) -> tuple[bool, list[dict]]: - """Validate data against schema. - - Args: - data: Dictionary to validate - - Returns: - (is_valid, list of failure dicts) - """ - failures = [] - - # Check required fields - for field_name, expected_type in self.required_fields.items(): - if field_name not in data: - failures.append( - { - "field": field_name, - "value": None, - "reason": "Required field missing", - "severity": "error", - } - ) - elif not isinstance(data[field_name], expected_type): - failures.append( - { - "field": field_name, - "value": data[field_name], - "reason": f"Expected {expected_type.__name__}, got {type(data[field_name]).__name__}", - "severity": "error", - } - ) - - # Check optional fields if present - if self.optional_fields: - for field_name, expected_type in self.optional_fields.items(): - if field_name in data and not isinstance(data[field_name], expected_type): - failures.append( - { - "field": field_name, - "value": data[field_name], - "reason": f"Expected {expected_type.__name__}, got {type(data[field_name]).__name__}", - "severity": "warning", - } - ) - - return len([f for f in failures if f.severity == "error"]) == 0, failures - - -class RangeValidator: - """Validate numeric values are within expected ranges.""" - - def __init__(self, min_val: float | None = None, max_val: float | None = None): - self.min_val = min_val - self.max_val = max_val - - def validate(self, value: float, field_name: str = "value") -> tuple[bool, dict | None]: - """Validate value is in range. - - Args: - value: Numeric value to validate - field_name: Name of field for error message - - Returns: - (is_valid, failure dict or None) - """ - if self.min_val is not None and value < self.min_val: - return False, { - "field": field_name, - "value": value, - "reason": f"Value {value} below minimum {self.min_val}", - "severity": "error", - } - - if self.max_val is not None and value > self.max_val: - return False, { - "field": field_name, - "value": value, - "reason": f"Value {value} above maximum {self.max_val}", - "severity": "error", - } - - return True, None - - -class ChecksumValidator: - """Validate data integrity using checksums.""" - - @staticmethod - def compute_checksum(data: dict | list) -> str: - """Compute simple checksum for data. - - Args: - data: Dictionary or list to checksum - - Returns: - Hex checksum string - """ - import hashlib - import json - - data_str = json.dumps(data, sort_keys=True, default=str) - return hashlib.md5(data_str.encode()).hexdigest() - - @staticmethod - def validate_checksum( - data: dict | list, expected_checksum: str - ) -> tuple[bool, dict | None]: - """Validate data matches checksum. - - Args: - data: Data to validate - expected_checksum: Expected checksum value - - Returns: - (is_valid, failure dict or None) - """ - actual_checksum = ChecksumValidator.compute_checksum(data) - if actual_checksum != expected_checksum: - return False, { - "field": "checksum", - "value": actual_checksum, - "reason": f"Checksum mismatch: expected {expected_checksum}, got {actual_checksum}", - "severity": "error", - } - return True, None - - -class BusinessRuleValidator: - """Validate data against business logic rules.""" - - def __init__(self): - self.rules: dict[str, Callable[[Any], tuple[bool, str]]] = {} - - def add_rule(self, rule_name: str, rule_fn: Callable[[Any], tuple[bool, str]]) -> None: - """Add a business rule. - - Args: - rule_name: Name of the rule - rule_fn: Function that takes data and returns (is_valid, reason) - """ - self.rules[rule_name] = rule_fn - - def validate(self, data: Any) -> tuple[bool, list[dict]]: - """Apply all business rules. - - Args: - data: Data to validate - - Returns: - (is_valid, list of failure dicts) - """ - failures = [] - for rule_name, rule_fn in self.rules.items(): - try: - is_valid, reason = rule_fn(data) - if not is_valid: - failures.append( - { - "field": "business_rule", - "value": None, - "reason": f"{rule_name}: {reason}", - "severity": "error", - } - ) - except Exception as e: - failures.append( - { - "field": "business_rule", - "value": None, - "reason": f"{rule_name} raised exception: {e}", - "severity": "error", - } - ) - - return len(failures) == 0, failures - - -# Concrete validators for common cases - -class BarDataValidator: - """Validate OHLCV bar data.""" - - @staticmethod - def validate_bar(bar: dict) -> tuple[bool, list[dict]]: - """Validate a single bar. - - Args: - bar: Bar dictionary with {timestamp, open, high, low, close, volume} - - Returns: - (is_valid, list of failure dicts) - """ - failures = [] - - # Check required fields - required = {"timestamp", "open", "high", "low", "close", "volume"} - for field in required: - if field not in bar: - failures.append( - { - "field": field, - "value": None, - "reason": "Missing required field", - "severity": "error", - } - ) - - if failures: - return False, failures - - # Check value consistency - open_val = bar["open"] - high_val = bar["high"] - low_val = bar["low"] - close_val = bar["close"] - volume_val = bar["volume"] - - # High must be >= all others - if high_val < open_val or high_val < close_val or high_val < low_val: - failures.append( - { - "field": "high", - "value": high_val, - "reason": f"High ({high_val}) must be >= open ({open_val}), close ({close_val}), low ({low_val})", - "severity": "error", - } - ) - - # Low must be <= all others - if low_val > open_val or low_val > close_val or low_val > high_val: - failures.append( - { - "field": "low", - "value": low_val, - "reason": f"Low ({low_val}) must be <= open ({open_val}), close ({close_val}), high ({high_val})", - "severity": "error", - } - ) - - # Volume should be non-negative - if volume_val < 0: - failures.append( - { - "field": "volume", - "value": volume_val, - "reason": "Volume cannot be negative", - "severity": "error", - } - ) - - # Warn on zero volume - if volume_val == 0: - failures.append( - { - "field": "volume", - "value": volume_val, - "reason": "Zero volume bar", - "severity": "warning", - } - ) - - return len([f for f in failures if f.severity == "error"]) == 0, failures diff --git a/deploy/systemd/momentum-controlroom.service b/deploy/systemd/momentum-controlroom.service deleted file mode 100644 index 26d928e..0000000 --- a/deploy/systemd/momentum-controlroom.service +++ /dev/null @@ -1,17 +0,0 @@ -[Unit] -Description=Momentum Trading Control Room (NiceGUI live-analysis dashboard, :8766) -StartLimitIntervalSec=0 - -[Service] -Type=simple -WorkingDirectory=/mnt/c/code/momentum -Environment=PYTHONPATH=/mnt/c/code/momentum -ExecStart=/home/philip/.venvs/momentum/bin/python -u controlroom/app.py -Restart=always -RestartSec=10 -StandardOutput=append:/mnt/c/code/momentum/data/controlroom.log -StandardError=append:/mnt/c/code/momentum/data/controlroom.log -SyslogIdentifier=momentum-controlroom - -[Install] -WantedBy=default.target diff --git a/ARCHITECTURE_VISUAL.html b/docs/history/ARCHITECTURE_VISUAL.html similarity index 100% rename from ARCHITECTURE_VISUAL.html rename to docs/history/ARCHITECTURE_VISUAL.html diff --git a/QUICK_REFERENCE.txt b/docs/history/QUICK_REFERENCE.txt similarity index 100% rename from QUICK_REFERENCE.txt rename to docs/history/QUICK_REFERENCE.txt diff --git a/SYSTEM_FLOW.txt b/docs/history/SYSTEM_FLOW.txt similarity index 100% rename from SYSTEM_FLOW.txt rename to docs/history/SYSTEM_FLOW.txt diff --git a/models/__init__.py b/models/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/models/base.py b/models/base.py deleted file mode 100644 index 62448b2..0000000 --- a/models/base.py +++ /dev/null @@ -1,26 +0,0 @@ -"""Models for Milestone 2. - -Event and data models for the event-first architecture. -""" - -from pydantic import BaseModel, Field - - -class EventMode(str): - """Runtime modes for events.""" - - REPLAY = "replay" - PAPER = "paper" - LIVE = "live" - - -class BaseEvent(BaseModel): - """Base event with common fields.""" - - timestamp: str = Field(description="Event timestamp (ISO format)") - mode: EventMode = Field(description="Runtime mode (replay/paper/live)") - event_type: str = Field(description="Event type identifier") - correlation_id: str | None = Field( - default=None, description="Correlation ID for causal chain" - ) - message: str = Field(description="Human-readable message") diff --git a/package-lock.json b/package-lock.json deleted file mode 100644 index 68e2e2a..0000000 --- a/package-lock.json +++ /dev/null @@ -1,18 +0,0 @@ -{ - "name": "momentum", - "lockfileVersion": 3, - "requires": true, - "packages": { - "": { - "dependencies": { - "claude": "^0.1.1" - } - }, - "node_modules/claude": { - "version": "0.1.1", - "resolved": "https://registry.npmjs.org/claude/-/claude-0.1.1.tgz", - "integrity": "sha512-j7oSibqQdIODNhkI1sEJzHMiPsF43L/GqNbcA+eDDyGM10+x2sH9NW/PK6vM3z0J2tLDKMBcc5ZjVaoRinhuCA==", - "license": "ISC" - } - } -} diff --git a/package.json b/package.json deleted file mode 100644 index 1353701..0000000 --- a/package.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "dependencies": { - "claude": "^0.1.1" - } -} diff --git a/run_live_paper.py b/run_live_paper.py index 674d771..dbf46dd 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -411,13 +411,21 @@ def _session_open(con, symbol: str, session_date) -> float | None: "AND is_regular_hours ORDER BY timestamp LIMIT 1", [symbol, session_date] ).fetchall() if r: - # trust it ONLY if it is really the opening bar (<= 09:33 ET = 13:33Z): - # mid-day-discovered names briefly have bars starting mid-session, and - # treating that first bar as "the open" is how TVRD/FRTT armed as fake - # 20% gappers on 2026-07-08. - ts = str(r[0][1]) - hhmm = ts[11:16] - if "13:30" <= hhmm <= "13:33": + # trust it ONLY if it is really the 09:30 ET opening bar: mid-day- + # discovered names briefly have bars starting mid-session, and treating + # that first bar as "the open" is how TVRD/FRTT armed as fake 20% + # gappers on 2026-07-08. minute_bars.timestamp is UTC-naive, so the + # open is 13:30Z in summer (EDT) but 14:30Z in winter (EST) — convert + # to ET instead of a fixed UTC window, else the whole fast path arms + # nothing all winter. + from datetime import datetime, timezone + from zoneinfo import ZoneInfo + raw = r[0][1] + ts = raw if isinstance(raw, datetime) else datetime.fromisoformat(str(raw)) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + et_hhmm = ts.astimezone(ZoneInfo("America/New_York")).strftime("%H:%M") + if "09:30" <= et_hhmm <= "09:33": _SESSION_OPEN_CACHE[key] = float(r[0][0]) return _SESSION_OPEN_CACHE[key] except Exception: # noqa: BLE001 — fail-open; gap falls back to window/day_open diff --git a/runtime/providers/__init__.py b/runtime/providers/__init__.py deleted file mode 100644 index d6eeb5f..0000000 --- a/runtime/providers/__init__.py +++ /dev/null @@ -1,10 +0,0 @@ -"""Market-data provider registry.""" - -from runtime.providers.registry import ( - ProviderKind, - ProviderConfig, - ProviderRegistry, - default_registry, -) - -__all__ = ["ProviderKind", "ProviderConfig", "ProviderRegistry", "default_registry"] diff --git a/runtime/providers/registry.py b/runtime/providers/registry.py deleted file mode 100644 index 9365c1c..0000000 --- a/runtime/providers/registry.py +++ /dev/null @@ -1,112 +0,0 @@ -"""Provider registry for market-data sources. - -(Repaired: ProviderConfig is a proper top-level dataclass and the registry -initializes its mapping correctly — the original nested the dataclass inside -the enum and used an invalid `self.providers: {}` annotation-as-assignment.) -""" - -from __future__ import annotations - -import os -from dataclasses import dataclass, field -from enum import Enum - - -class ProviderKind(str, Enum): - """Supported market-data providers.""" - - ALPACA = "alpaca" - SCHWAB = "schwab" - POLYGON = "polygon" - FINNHUB = "finnhub" - SYNTHETIC = "synthetic" - - -@dataclass -class ProviderConfig: - """Configuration for one provider.""" - - kind: ProviderKind - enabled: bool = True - priority: int = 100 # lower = preferred - rate_limit_per_minute: int = 200 - supports_minute_bars: bool = True - supports_quotes: bool = True - env_keys: tuple[str, ...] = field(default_factory=tuple) - - def is_configured(self, env: dict[str, str] | None = None) -> bool: - """A provider is configured when all of its env keys are present.""" - values = dict(os.environ) - if env is not None: - values.update(env) - return all(values.get(key) for key in self.env_keys) - - -class ProviderRegistry: - """Registry of provider configs with priority-based selection.""" - - def __init__(self) -> None: - self.providers: dict[ProviderKind, ProviderConfig] = {} - - def register(self, config: ProviderConfig) -> None: - self.providers[config.kind] = config - - def get(self, kind: ProviderKind | str) -> ProviderConfig | None: - if isinstance(kind, str): - try: - kind = ProviderKind(kind) - except ValueError: - return None - return self.providers.get(kind) - - def enabled_providers(self) -> list[ProviderConfig]: - return sorted( - (p for p in self.providers.values() if p.enabled), - key=lambda p: p.priority, - ) - - def best_configured( - self, env: dict[str, str] | None = None - ) -> ProviderConfig | None: - for provider in self.enabled_providers(): - if provider.is_configured(env): - return provider - return None - - -def default_registry() -> ProviderRegistry: - """Registry with the standard provider stack.""" - registry = ProviderRegistry() - registry.register( - ProviderConfig( - kind=ProviderKind.ALPACA, - priority=10, - env_keys=("ALPACA_API_KEY", "ALPACA_SECRET_KEY"), - ) - ) - registry.register( - ProviderConfig( - kind=ProviderKind.SCHWAB, - priority=20, - env_keys=("SCHWAB_MARKET_DATA_APP_KEY", "SCHWAB_MARKET_DATA_APP_SECRET"), - ) - ) - registry.register( - ProviderConfig( - kind=ProviderKind.POLYGON, priority=30, env_keys=("POLYGON_API_KEY",) - ) - ) - registry.register( - ProviderConfig( - kind=ProviderKind.FINNHUB, priority=40, env_keys=("FINNHUB_API_KEY",) - ) - ) - registry.register( - ProviderConfig( - kind=ProviderKind.SYNTHETIC, - priority=1000, - env_keys=(), - supports_quotes=False, - ) - ) - return registry diff --git a/storage/replay.py b/storage/replay.py deleted file mode 100644 index 2e4d36a..0000000 --- a/storage/replay.py +++ /dev/null @@ -1,148 +0,0 @@ -import json - -"""Replay/rebuild tooling for Milestone 2. - -Reconstruct state from event store for analysis and testing. -""" - -import logging -from typing import TYPE_CHECKING - -from .event_store import EventStore - -if TYPE_CHECKING: - pass - -logger = logging.getLogger(__name__) - - -class StateRebuilder: - """Rebuild trading state from event store.""" - - def __init__(self, store: EventStore): - self.store = store - - def rebuild_session(self, session_id: str) -> dict: - """Rebuild complete state for a session from events.""" - logger.info(f"Rebuilding session {session_id}") - - events = self.store.query_events(session_id=session_id) - - state = { - "session_id": session_id, - "symbols": {}, - "orders": {}, - "positions": {}, - "events": [], - } - - for event in events: - event_type = event["event_type"] - payload = json.loads(event.get("payload_json", "{}")) - symbol = payload.get("symbol") - - self._process_event(state, event_type, symbol, payload, event) - - logger.info(f"Rebuilt state for {len(events)} events") - return state - - def _process_event( - self, - state: dict, - event_type: str, - symbol: str | None, - payload: dict, - event: dict, - ): - """Process individual event and update state.""" - if event_type == "symbol_discovered": - state["symbols"][symbol] = { - "state": "discovered", - "discovered_at": event["timestamp"], - } - - elif event_type == "symbol_state_changed": - if symbol not in state["symbols"]: - state["symbols"][symbol] = { - "state": payload.get("previous_state") or "discovered", - "discovered_at": event["timestamp"], - } - state["symbols"][symbol]["state"] = payload["new_state"] - state["symbols"][symbol]["state_history"] = state["symbols"][symbol].get( - "state_history", [] - ) - state["symbols"][symbol]["state_history"].append( - { - "state": payload["new_state"], - "timestamp": event["timestamp"], - "reason": payload.get("state_reason"), - } - ) - - elif event_type == "signal_ready": - if symbol in state["symbols"]: - state["symbols"][symbol]["signals"] = state["symbols"][symbol].get( - "signals", [] - ) - state["symbols"][symbol]["signals"].append( - { - "signal_type": payload["signal_type"], - "confidence": payload["confidence"], - "timestamp": event["timestamp"], - } - ) - - elif event_type == "position_opened": - position_id = payload["position_id"] - state["positions"][position_id] = { - "symbol": symbol, - "status": "open", - "entry_price": payload["entry_price"], - "quantity": payload["quantity"], - "stop_loss_price": payload["stop_loss_price"], - "opened_at": event["timestamp"], - } - - elif event_type == "position_closed": - position_id = payload["position_id"] - if position_id in state["positions"]: - state["positions"][position_id].update( - { - "status": "closed", - "exit_price": payload["exit_price"], - "exit_reason": payload["exit_reason"], - "realized_pnl": payload["realized_pnl"], - "closed_at": event["timestamp"], - } - ) - - elif event_type == "order_filled": - state["events"].append( - { - "type": event_type, - "order_id": payload["order_id"], - "symbol": symbol, - "timestamp": event["timestamp"], - } - ) - - elif event_type == "order_submitted": - order_id = payload["order_id"] - state["orders"][order_id] = { - "symbol": symbol, - "status": "submitted", - "side": payload.get("side"), - "quantity": payload.get("quantity"), - "price": payload.get("price"), - "updated_at": event["timestamp"], - } - - elif event_type == "order_cancelled": - order_id = payload["order_id"] - state["orders"][order_id] = { - **state["orders"].get(order_id, {}), - "symbol": symbol, - "status": "cancelled", - "cancel_reason": payload.get("cancel_reason"), - "updated_at": event["timestamp"], - } diff --git a/strategy/evaluation/llm_integration.py b/strategy/evaluation/llm_integration.py deleted file mode 100644 index 291d7b8..0000000 --- a/strategy/evaluation/llm_integration.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Optional LLM (Ollama) enrichment hooks. Disabled unless configured.""" - -from __future__ import annotations - -import json -import logging -import urllib.request - -logger = logging.getLogger(__name__) - - -def summarize_setup_with_llm( - setup: dict, - host: str = "http://localhost:11434", - model: str = "mistral", - timeout: int = 15, -) -> str | None: - """Ask a local Ollama model to summarize a setup. Returns None on failure.""" - prompt = ( - "Summarize this momentum trading setup in one sentence:\n" - + json.dumps(setup, default=str) - ) - try: - req = urllib.request.Request( - f"{host}/api/generate", - data=json.dumps( - {"model": model, "prompt": prompt, "stream": False} - ).encode(), - headers={"Content-Type": "application/json"}, - ) - with urllib.request.urlopen(req, timeout=timeout) as resp: - body = json.loads(resp.read().decode()) - return body.get("response") - except Exception: - logger.debug("LLM enrichment unavailable", exc_info=True) - return None diff --git a/strategy/evaluation/storage.py b/strategy/evaluation/storage.py deleted file mode 100644 index bd98314..0000000 --- a/strategy/evaluation/storage.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Persist setup evaluations into the research DB (setup_events table).""" - -from __future__ import annotations - -import uuid -from datetime import datetime - - -def store_setup_event(con, symbol: str, session_date, result: dict) -> str: - """Insert one evaluation result into setup_events. Returns setup_id.""" - setup_id = str(uuid.uuid4()) - setups = result.get("setups") or [] - best = setups[0] if setups else {} - con.execute( - """ - INSERT INTO setup_events - (setup_id, symbol, setup_time, session_date, setup_name, - entry_reference_price, invalidation_price, gap_pct, relative_volume) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - [ - setup_id, - symbol, - datetime.now(), - session_date, - best.get("setup_type") or "first_pullback", - best.get("entry_price"), - best.get("stop_loss_price"), - result.get("gap_pct"), - result.get("relative_volume"), - ], - ) - return setup_id diff --git a/strategy/evaluation/tracking.py b/strategy/evaluation/tracking.py deleted file mode 100644 index cc7929c..0000000 --- a/strategy/evaluation/tracking.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Lightweight in-memory tracking of evaluation history per symbol.""" - -from __future__ import annotations - -from collections import defaultdict - - -class EvaluationTracker: - """Tracks recent evaluation results per symbol for debounce/analysis.""" - - def __init__(self, max_history: int = 50): - self.max_history = max_history - self._history: dict[str, list[dict]] = defaultdict(list) - - def record(self, symbol: str, result: dict) -> None: - history = self._history[symbol] - history.append(result) - if len(history) > self.max_history: - del history[: len(history) - self.max_history] - - def last(self, symbol: str) -> dict | None: - history = self._history.get(symbol) - return history[-1] if history else None - - def consecutive_status(self, symbol: str, status: str) -> int: - count = 0 - for result in reversed(self._history.get(symbol, [])): - if result.get("status") == status: - count += 1 - else: - break - return count diff --git a/telemetry/__init__.py b/telemetry/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/telemetry/events.py b/telemetry/events.py deleted file mode 100644 index d2149ba..0000000 --- a/telemetry/events.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Telemetry event emission and querying for runtime observability.""" - -from __future__ import annotations - -import json -import logging -import uuid -from enum import Enum - -logger = logging.getLogger(__name__) - - -class TelemetryEventType(str, Enum): - SOURCE_DEGRADED = "SOURCE_DEGRADED" - SOURCE_RECOVERED = "SOURCE_RECOVERED" - SOURCE_COOLDOWN = "SOURCE_COOLDOWN" - WATCHLIST_ADD = "WATCHLIST_ADD" - WATCHLIST_DROP = "WATCHLIST_DROP" - CANDIDATE_STATE_CHANGE = "CANDIDATE_STATE_CHANGE" - FETCH_ATTEMPT = "FETCH_ATTEMPT" - ENRICHMENT_QUEUED = "ENRICHMENT_QUEUED" - ENRICHMENT_DONE = "ENRICHMENT_DONE" - ENRICHMENT_FAILED = "ENRICHMENT_FAILED" - - -def emit_event( - con, - event_type: TelemetryEventType | str, - session_id: str | None = None, - symbol: str | None = None, - source: str | None = None, - from_state: str | None = None, - to_state: str | None = None, - reason: str | None = None, - metadata: dict | None = None, -) -> None: - event_id = str(uuid.uuid4()) - event_type_val = ( - event_type.value - if isinstance(event_type, TelemetryEventType) - else str(event_type) - ) - metadata_json = json.dumps(metadata) if metadata is not None else None - try: - con.execute( - """ - INSERT INTO telemetry_events - (id, event_type, session_id, symbol, source, from_state, to_state, reason, metadata_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - [ - event_id, - event_type_val, - session_id, - symbol, - source, - from_state, - to_state, - reason, - metadata_json, - ], - ) - except Exception: - logger.exception( - "telemetry emit_event failed — event_type=%s symbol=%s", - event_type_val, - symbol, - ) - - -def query_events( - con, - session_id: str | None = None, - symbol: str | None = None, - source: str | None = None, - limit: int = 100, -) -> list[dict]: - clauses: list[str] = [] - params: list = [] - - if session_id is not None: - clauses.append("session_id = ?") - params.append(session_id) - if symbol is not None: - clauses.append("symbol = ?") - params.append(symbol) - if source is not None: - clauses.append("source = ?") - params.append(source) - - where = ("WHERE " + " AND ".join(clauses)) if clauses else "" - sql = f"SELECT * FROM telemetry_events {where} ORDER BY created_at DESC LIMIT ?" - params.append(limit) - - rows = con.execute(sql, params).fetchall() - cols = [ - desc[0] - for desc in con.execute( - f"SELECT * FROM telemetry_events {where} LIMIT 0", params[:-1] - ).description - ] - return [dict(zip(cols, row)) for row in rows] diff --git a/tests/unit/alpaca_paper/test_execution_and_sync.py b/tests/unit/alpaca_paper/test_execution_and_sync.py index d8c62e8..ab09915 100644 --- a/tests/unit/alpaca_paper/test_execution_and_sync.py +++ b/tests/unit/alpaca_paper/test_execution_and_sync.py @@ -231,3 +231,31 @@ def test_reconcile_seeds_from_persisted_snapshot_after_restart(store): rows = _closed(store) assert len(rows) == 1 and rows[0]["symbol"] == "REST" assert rows[0]["realized_pnl"] == 5.0 # (2.5 - 2.0) * 10 + + +def test_post_not_retried_on_urlerror(monkeypatch): + """Lost-response URLError on POST /orders must NOT re-send (duplicate live + order, 2026-07-06). GET/DELETE still retry.""" + import urllib.error + import urllib.request + from alpaca_paper.client import AlpacaPaperClient + from alpaca_paper.settings import AlpacaPaperSettings + cli = AlpacaPaperClient.__new__(AlpacaPaperClient) + cli.timeout = 1 + cli.max_retries = 2 + cli._headers = lambda: {} + calls = {"n": 0} + + def boom(req, timeout=None): + calls["n"] += 1 + raise urllib.error.URLError("connection reset") + + monkeypatch.setattr(urllib.request, "urlopen", boom) + import pytest + with pytest.raises(urllib.error.URLError): + cli._request("POST", "https://x/v2/orders", body={"symbol": "AAA"}) + assert calls["n"] == 1, "POST must not retry a lost-response URLError" + calls["n"] = 0 + with pytest.raises(urllib.error.URLError): + cli._request("GET", "https://x/v2/account") + assert calls["n"] > 1, "idempotent GET should still retry" diff --git a/core/__init__.py b/tests/unit/test_alpaca_client.py similarity index 100% rename from core/__init__.py rename to tests/unit/test_alpaca_client.py diff --git a/tests/unit/test_trading_execution.py b/tests/unit/test_trading_execution.py index 2e594d1..085f82a 100644 --- a/tests/unit/test_trading_execution.py +++ b/tests/unit/test_trading_execution.py @@ -875,3 +875,59 @@ def test_session_gate_bars_after_hours_entries(): assert svc._outside_session(datetime(2026, 7, 9, 9, 30, tzinfo=ET)) is None # the open svc.settings = ExecutionSettings(session_only=False) assert svc._outside_session(datetime(2026, 7, 9, 20, 9, tzinfo=ET)) is None # opt-out works + + +def test_flatten_all_closes_broker_position_missed_by_snapshot(store): + """Naked-through-breaker fix: a position that filled in the sync gap isn't in + the event-store held set, so _flatten_all's step-2 skips it. Step 3 (moved out + of close_session so the daily-loss breaker gets it too) closes it from fresh + broker positions.""" + from alpaca_paper.execution import AlpacaPaperExecutor + + class _Cli: + def __init__(self): + self.closed = [] + self.pos = [{"symbol": "GAP", "qty": "100"}] # filled, snapshot missed it + + def get_positions(self, fresh=False): + return list(self.pos) + + def close_position(self, symbol, qty=None, percentage=None): + self.closed.append(symbol) + self.pos = [p for p in self.pos if p["symbol"] != symbol] + return {"id": "c"} + + def get_orders(self, **kw): + return [] + + def cancel_order(self, oid): + pass + + cli = _Cli() + svc = TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=cli), + settings=ExecutionSettings(auto_approve=True), session_id="t") + result = svc._flatten_all("daily_loss_circuit_breaker") + assert "GAP" in cli.closed, "breaker flatten must close broker positions the snapshot missed" + assert "GAP" in result["closed_positions"] + + +def test_daily_state_rebuilds_from_events_on_restart(store): + """Mid-session restart must not reset the entry cap to 0 or forget resting + orders (re-fire => duplicate order, 2026-07-06). A new service instance + rebuilds fresh-fill count, requested symbols, and losing benches from today.""" + from storage.event_schema import OrderSubmittedEvent, OrderFilledEvent, PositionClosedEvent + now = datetime.now() + store.emit(OrderSubmittedEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t", + message="sub", order_id="o1", symbol="AAA", side="buy", quantity=10, price=5.0, + broker_order_id="b1")) + store.emit(OrderFilledEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t", + message="fill", order_id="o1", symbol="AAA", fill_price=5.0, fill_quantity=10, + broker_order_id="b1")) + store.emit(PositionClosedEvent(timestamp=now, mode=EventMode.PAPER, correlation_id="t", + message="closed", position_id="p1", symbol="AAA", side="buy", quantity=10, + entry_price=5.0, exit_price=4.5, realized_pnl=-5.0, exit_reason="stop")) + svc = make_service(store) # boot AFTER the events land + assert svc._fresh_entries["n"] == 1, "cap count must survive restart" + assert "AAA" in svc._requested_symbols, "resting-order dedup must survive restart" + assert "AAA" in svc._exited_today, "losing bench must survive restart" diff --git a/trading_execution.py b/trading_execution.py index 32661d2..fa1a0b8 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -403,6 +403,59 @@ class TradingExecutionService: ) # serializes state mutations across the main loop and the trigger thread self._lock = threading.RLock() + # rebuild the memory-only daily state from today's events so a mid-session + # restart doesn't reset the entry cap to 0, un-bench stopped-out names, or + # forget which symbols already have a resting order (re-fire => duplicate + # order, the 2026-07-06 incident). Best-effort; a fresh DB is fine. + self._rebuild_daily_state() + + def _rebuild_daily_state(self) -> None: + """Restore per-day counters/benches from today's event stream on boot. + + ponytail: rebuilds the SLOW-path dedup + cap + bench only. The fast + trigger book (run_live_paper) is separately memory-only; the standing + broker-open-orders diff after a restart is still the backstop for a + resting fast-path order (see momentum-live-loop-architecture memory). + """ + try: + import json as _json + today = self._now().date() + since = today.isoformat() + + def _payload(e): + raw = e.get("payload") or e.get("payload_json") + if isinstance(raw, str): + try: + return _json.loads(raw) + except ValueError: + return {} + return raw or {} + + # order_filled carries no side; map fills back to their BUY submit so + # sell/exit fills don't inflate the fresh-entry cap + buy_order_ids = set() + for e in self.store.query_events(event_type="order_submitted", since=since, limit=500): + p = _payload(e) + sym, oid = p.get("symbol"), p.get("order_id") + if sym: + self._requested_symbols.add(str(sym)) + if oid and str(p.get("side", "buy")).lower() == "buy": + buy_order_ids.add(str(oid)) + n_fresh = sum( + 1 for e in self.store.query_events(event_type="order_filled", since=since, limit=500) + if str(_payload(e).get("order_id")) in buy_order_ids) + self._fresh_entries = {"date": today, "n": n_fresh} + for e in self.store.query_events(event_type="position_closed", since=since, limit=500): + p = _payload(e) + sym = p.get("symbol") + # bench only names that closed at a LOSS (mirror the live bench rule) + if sym and float(p.get("realized_pnl") or 0) < 0: + self._exited_today.add(str(sym)) + if n_fresh or self._requested_symbols or self._exited_today: + logger.info("restored daily state: %d fresh fills, %d requested, %d benched", + n_fresh, len(self._requested_symbols), len(self._exited_today)) + except Exception as exc: # noqa: BLE001 — a rebuild failure must not block boot + logger.warning("daily-state rebuild skipped: %s", exc) # -- pipeline ----------------------------------------------------------- @@ -499,10 +552,18 @@ class TradingExecutionService: block the rest. Resting bracket legs are cancelled before each close so the held quantity is freed (else close_position 403s).""" result: dict = {"cancelled_entries": 0, "closed_positions": [], "errors": []} + client = getattr(self.executor, "client", None) + # broker truth up front: an entry that FILLED in the sync gap is still in + # _armed; cancelling its bracket parent would strip the stop/TP legs and + # leave the position naked. Never cancel a name the broker holds — flatten + # it in step 2 instead. None => positions unreadable, stay conservative. + broker_held = self._broker_held_symbols() or set() # 1) cancel unfilled (armed) entries we are tracking for order_id in list(self._armed): armed = self._armed.pop(order_id) symbol = armed.get("symbol", "") + if symbol in broker_held: + continue # filled — leave the bracket intact; step 2/3 closes it try: self.executor.cancel_entry( order_id, armed.get("broker_order_id"), symbol, reason @@ -511,8 +572,7 @@ class TradingExecutionService: except Exception as exc: # noqa: BLE001 result["errors"].append(f"cancel {symbol}: {exc}") self._requested_symbols.discard(symbol) - # 2) market-close open positions - client = getattr(self.executor, "client", None) + # 2) market-close open positions from the store snapshot for symbol in sorted(self._held_symbols()): if client is None or not hasattr(client, "close_position"): result["errors"].append(f"close {symbol}: no broker client to flatten") @@ -522,6 +582,22 @@ class TradingExecutionService: result["closed_positions"].append(symbol) # only on a confirmed close except Exception as exc: # noqa: BLE001 result["errors"].append(f"close {symbol}: {exc}") + # 3) belt-and-suspenders: close any BROKER position the store snapshot + # missed (filled within the ~60s sync lag). Was in close_session only, so + # the daily-loss breaker flatten skipped it — moved here so EVERY caller + # (breaker + EOD) gets it. This is the naked-through-breaker fix. + if client is not None and hasattr(client, "get_positions"): + try: + for p in client.get_positions(fresh=True) or []: + sym = p.get("symbol") + if sym and sym not in result["closed_positions"]: + try: + self._release_and_close(client, sym) + result["closed_positions"].append(sym) + except Exception as exc: # noqa: BLE001 + result["errors"].append(f"close {sym}: {exc}") + except Exception as exc: # noqa: BLE001 + result["errors"].append(f"positions: {exc}") return result @_locked @@ -1396,21 +1472,7 @@ class TradingExecutionService: halt new entries for the rest of the session. """ self._session_closed = True - result = self._flatten_all(reason) - # belt-and-suspenders: close any broker position the store snapshot missed - client = getattr(self.executor, "client", None) - if client is not None and hasattr(client, "get_positions"): - try: - for p in client.get_positions(): - sym = p.get("symbol") - if sym and sym not in result["closed_positions"]: - try: - self._release_and_close(client, sym) - result["closed_positions"].append(sym) - except Exception as exc: # noqa: BLE001 - result["errors"].append(f"close {sym}: {exc}") - except Exception as exc: # noqa: BLE001 - result["errors"].append(f"positions: {exc}") + result = self._flatten_all(reason) # fresh-broker belt now lives inside self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, @@ -1576,6 +1638,7 @@ class TradingExecutionService: and self._fresh_entries["n"] > 0): self._fresh_entries["n"] -= 1 + @_locked def expire_stale_entries(self) -> list[dict]: """Back out of unfilled entries that timed out or broke their trigger.