diff --git a/run_live_paper.py b/run_live_paper.py index bf66f8f..7dd3597 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -28,6 +28,7 @@ import argparse import json import os import sys +import threading import time import uuid from datetime import datetime, timezone @@ -340,11 +341,33 @@ def main(argv: list[str] | None = None) -> int: + (f" (errors: {daily.errors})" if daily.errors else "")) lookback = int(os.environ.get("LIVE_BARS_LOOKBACK_MINUTES", "240")) + refresh_lb = int(os.environ.get("LIVE_BARS_REFRESH_MINUTES", "20")) + _ingest_first = {"v": True} + + def _first_pass_lookback() -> int: + # The first ingest must cover the whole session-so-far so the + # opening-range bars land in the DB (even on a mid-session restart); + # after that we only pull the last few minutes — closed bars never + # change, and re-fetching a 4-hour window for 80 symbols every minute + # was the ~46s step that froze the loop. + try: + from zoneinfo import ZoneInfo + now_et = datetime.now(ZoneInfo("America/New_York")) + open_et = now_et.replace(hour=9, minute=30, second=0, microsecond=0) + since_open = max(0.0, (now_et - open_et).total_seconds() / 60.0) + except Exception: # noqa: BLE001 + since_open = 0.0 + return int(max(lookback, since_open + 15)) def step_ingest(): if not client: return "skipped (no keys)" - res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lookback) + if _ingest_first["v"]: + lb = _first_pass_lookback() + _ingest_first["v"] = False + else: + lb = refresh_lb + res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb) rt["store"].emit( ModuleTickEvent( timestamp=datetime.now(), @@ -360,7 +383,7 @@ def main(argv: list[str] | None = None) -> int: errors=[{"error": e} for e in res.errors], ) ) - return f"{res.minute_rows} minute rows" + (f", errors={res.errors}" if res.errors else "") + return f"{res.minute_rows} rows (lookback {lb}m)" + (f", errors={res.errors}" if res.errors else "") def step_watch(): res = rt["watcher"].tick(_now_session_date()) @@ -425,7 +448,6 @@ def main(argv: list[str] | None = None) -> int: # --- fast armed-trigger path ------------------------------------------- book = rt["book"] - latest_price = rt["latest_price"] price_min_v = float(os.environ.get("WATCHER_PRICE_MIN", "1")) price_max_v = float(os.environ.get("WATCHER_PRICE_MAX", "20")) orb_bars_v = int(os.environ.get("ORB_BARS", "5")) @@ -471,20 +493,43 @@ def main(argv: list[str] | None = None) -> int: armed = sum(1 for t in book.triggers.values() if t.state == "armed") return f"tracking={len(book.triggers)} armed={armed}" - def step_trigger(): - """Fast tick: refresh live prices for the armed names, fire any that have - crossed their trigger, and publish the snapshot the live board renders.""" - if not book.triggers: - return None - for sym in list(book.triggers): - book.update_price(sym, latest_price(sym) if client else None) + # The trigger runs on its OWN thread, not the scheduler, so a slow ingest + # can never freeze breakout detection. It only touches the (thread-safe) + # book, execution service, and event store — never research_con. + trigger_interval = float(os.environ.get("TRIGGER_INTERVAL_SECONDS", "4")) + stop_event = threading.Event() + + def _batch_prices(syms): + """One batched last-trade call for all armed names (8 syms ≈ 1 sym cost).""" + if client is None or not syms: + return {} + try: + trades = client.get_latest_trades(list(syms)) + except Exception: # noqa: BLE001 + return {} + out = {} + for s in syms: + tr = trades.get(s) or trades.get(s.upper()) + if tr and tr.get("p") is not None: + try: + out[s] = float(tr["p"]) + except (TypeError, ValueError): + pass + return out + + def _trigger_pass(): + snap = book.snapshot() + syms = [r["symbol"] for r in snap] + if not syms: + return + for s, px in _batch_prices(syms).items(): + book.update_price(s, px) + fired: list[str] = [] if rt["execution"] is not None: try: book.mark_filled(rt["execution"]._held_symbols()) except Exception: # noqa: BLE001 pass - fired: list[str] = [] - if rt["execution"] is not None: for t in book.fires(): res = rt["execution"].submit_breakout_now( t.symbol, t.trigger, t.stop, last_price=t.price @@ -492,23 +537,27 @@ def main(argv: list[str] | None = None) -> int: if res.get("ok"): book.mark_fired(t.symbol) fired.append(t.symbol) - armed_n = sum(1 for t in book.triggers.values() if t.state == "armed") + snap = book.snapshot() + armed_n = sum(1 for r in snap if r["state"] == "armed") rt["store"].emit( ModuleTickEvent( - timestamp=datetime.now(), - mode=EventMode.PAPER, + timestamp=datetime.now(), mode=EventMode.PAPER, correlation_id=rt["session_id"], message=f"triggers armed={armed_n} fired={len(fired)}", - module="triggers", - stage="snapshot", - duration_ms=0.0, - input_count=len(book.triggers), - output_count=len(fired), - metrics={"triggers": book.snapshot(), "armed": armed_n}, - errors=[], + module="triggers", stage="snapshot", duration_ms=0.0, + input_count=len(syms), output_count=len(fired), + metrics={"triggers": snap, "armed": armed_n}, errors=[], ) ) - return ("FIRED " + ", ".join(fired)) if fired else None + if fired: + print(f"[{datetime.now():%H:%M:%S}] TRIGGER FIRED {', '.join(fired)}", flush=True) + + def trigger_loop(): + while not stop_event.wait(trigger_interval): + try: + _trigger_pass() + except Exception as exc: # noqa: BLE001 + print(f"[trigger] error: {exc}", flush=True) scheduler = Scheduler() scheduler.add("discover", step_discover, @@ -520,8 +569,6 @@ def main(argv: list[str] | None = None) -> int: enabled=_flag("WATCHER_ENABLED", "1")) scheduler.add("arm", step_arm, float(os.environ.get("TRIGGER_ARM_INTERVAL_SECONDS", "20")), enabled=_flag("TRIGGER_ENABLED", "1")) - scheduler.add("trigger", step_trigger, float(os.environ.get("TRIGGER_INTERVAL_SECONDS", "4")), - enabled=_flag("TRIGGER_ENABLED", "1")) scheduler.add("sync", step_sync, float(os.environ.get("ALPACA_PAPER_SYNC_INTERVAL_SECONDS", "60")), enabled=_flag("ALPACA_PAPER_SYNC_ENABLED", "1")) scheduler.add("execute", step_execute, float(os.environ.get("TRADING_EXECUTION_INTERVAL_SECONDS", "30")), @@ -534,7 +581,7 @@ def main(argv: list[str] | None = None) -> int: for task in scheduler.tasks: task.last_run = -10**9 results = {} - for name in ("discover", "ingest", "watch", "arm", "trigger", "sync", "execute"): + for name in ("discover", "ingest", "watch", "arm", "sync", "execute"): for task in scheduler.tasks: if task.name == name and task.enabled: results.update(scheduler_run_one(task)) @@ -561,6 +608,12 @@ def main(argv: list[str] | None = None) -> int: return 0 print("[boot] entering live loop (Ctrl-C to stop)") + # start the fast trigger on its own thread so slow ingest can't freeze it + trigger_thread = None + if _flag("TRIGGER_ENABLED", "1") and client is not None: + trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True) + trigger_thread.start() + print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)") try: # one immediate full pass, then interval-driven run_pass() @@ -577,6 +630,9 @@ def main(argv: list[str] | None = None) -> int: except KeyboardInterrupt: print("\n[stop] shutting down") finally: + stop_event.set() + if trigger_thread is not None: + trigger_thread.join(timeout=5) if server: server.shutdown() rt["store"].close() diff --git a/runtime/triggers.py b/runtime/triggers.py index f710519..9487fb1 100644 --- a/runtime/triggers.py +++ b/runtime/triggers.py @@ -23,6 +23,8 @@ State machine per symbol: from __future__ import annotations +import functools +import threading from dataclasses import dataclass, field WAITING = "waiting" @@ -36,6 +38,17 @@ FILLED = "filled" _COMMITTED = (FIRED, FILLED) +def _synced(method): + """Serialize a book method behind self._lock so the main loop's arm() and + the trigger thread's poll/fire never iterate ``triggers`` while it's being + rebuilt.""" + @functools.wraps(method) + def wrapper(self, *args, **kwargs): + with self._lock: + return method(self, *args, **kwargs) + return wrapper + + @dataclass class ArmedTrigger: symbol: str @@ -83,6 +96,7 @@ class ArmedTriggerBook: rvol_min: float = 2.0 # min relative volume to be eligible to fire min_range_pct: float = 0.004 # opening range must be at least this wide triggers: dict[str, ArmedTrigger] = field(default_factory=dict) + _lock: object = field(default_factory=threading.RLock, repr=False, compare=False) def _eligible(self, t: ArmedTrigger) -> bool: return ( @@ -94,6 +108,7 @@ class ArmedTriggerBook: and t.range_pct >= self.min_range_pct ) + @_synced def arm(self, candidates: list[dict]) -> None: """Refresh the book from a ranked candidate list. @@ -131,11 +146,13 @@ class ArmedTriggerBook: new[sym] = t self.triggers = new + @_synced def update_price(self, symbol: str, price: float | None) -> None: t = self.triggers.get(symbol) if t is not None and price is not None: t.price = float(price) + @_synced def fires(self) -> list[ArmedTrigger]: """Armed triggers whose live price has reached or crossed the trigger.""" return [ @@ -147,11 +164,13 @@ class ArmedTriggerBook: and t.price >= t.trigger ] + @_synced def mark_fired(self, symbol: str) -> None: t = self.triggers.get(symbol) if t is not None: t.state = FIRED + @_synced def mark_filled(self, held_symbols) -> None: """Promote tracked symbols that are now open broker positions to filled.""" held = set(held_symbols or ()) @@ -159,6 +178,7 @@ class ArmedTriggerBook: if sym in held: t.state = FILLED + @_synced def snapshot(self) -> list[dict]: """Board-ready rows, ordered armed -> fired -> filled -> waiting -> weak.""" order = {ARMED: 0, FIRED: 1, FILLED: 2, WAITING: 3, WEAK: 4} diff --git a/storage/event_store.py b/storage/event_store.py index 397f604..15e15c8 100644 --- a/storage/event_store.py +++ b/storage/event_store.py @@ -12,6 +12,7 @@ from .db import get_connection from uuid import uuid4 import json import logging +import threading logger = logging.getLogger(__name__) @@ -22,6 +23,9 @@ class EventStore: def __init__(self, db_path: str = "./data/momentum_events.duckdb"): """Initialize event store with DuckDB connection.""" self.con = get_connection(db_path) + # serialize DB access: the fast trigger thread and the main loop share + # this one psycopg2 connection, which is not safe for concurrent use. + self._lock = threading.RLock() self._ensure_schema() def _ensure_schema(self): @@ -64,26 +68,27 @@ class EventStore: event_id = str(uuid4()) payload_json = json.dumps(event.model_dump(mode="json", exclude_unset=True)) - cursor = self.con.cursor() - cursor.execute( - """ - INSERT INTO events - (id, timestamp, mode, event_type, correlation_id, message, payload_json, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, current_timestamp) - """, - [ - event_id, - event.timestamp, - event.mode.value, - event.event_type.value - if isinstance(event.event_type, EventType) - else str(event.event_type), - event.correlation_id, - event.message, - payload_json, - ], - ) - self.con.commit() + with self._lock: + cursor = self.con.cursor() + cursor.execute( + """ + INSERT INTO events + (id, timestamp, mode, event_type, correlation_id, message, payload_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, current_timestamp) + """, + [ + event_id, + event.timestamp, + event.mode.value, + event.event_type.value + if isinstance(event.event_type, EventType) + else str(event.event_type), + event.correlation_id, + event.message, + payload_json, + ], + ) + self.con.commit() logger.debug( f"Emitted event {event_id}: {event.event_type.value} - {event.message}" @@ -152,9 +157,10 @@ class EventStore: {limit_clause} """ - cursor = self.con.cursor() - cursor.execute(query, params) - rows = cursor.fetchall() + with self._lock: + cursor = self.con.cursor() + cursor.execute(query, params) + rows = cursor.fetchall() results = [] for row in rows: @@ -175,9 +181,10 @@ class EventStore: def count_events(self) -> int: """Total number of events. Cheap change-detection signal for SSE.""" - cursor = self.con.cursor() - cursor.execute("SELECT COUNT(*) FROM events") - row = cursor.fetchone() + with self._lock: + cursor = self.con.cursor() + cursor.execute("SELECT COUNT(*) FROM events") + row = cursor.fetchone() return int(row[0]) if row else 0 def get_symbol_timeline(self, symbol: str) -> list[dict]: diff --git a/tests/unit/test_triggers.py b/tests/unit/test_triggers.py index 56b8619..8f95dfa 100644 --- a/tests/unit/test_triggers.py +++ b/tests/unit/test_triggers.py @@ -108,22 +108,27 @@ def test_mark_filled_promotes_held(): # -- live fire (submit_breakout_now) -------------------------------------- class _FakeClient: - def __init__(self, resp=None, account=None): + def __init__(self, resp=None, account=None, positions=None): self.resp = resp or {"id": "o1", "status": "new", "filled_qty": "0"} self.account = account or {"equity": "100000", "last_equity": "100000"} + self._positions = positions or [] self.last = None self.closed = [] + self.canceled = [] def submit_order(self, **kw): self.last = kw return self.resp def cancel_order(self, order_id): - pass + self.canceled.append(order_id) def get_account(self): return self.account + def get_positions(self): + return self._positions + def close_position(self, symbol): self.closed.append(symbol) return {"id": "c"} @@ -170,3 +175,24 @@ def test_breakout_now_rejects_bad_geometry(): store, svc = _svc(_FakeClient()) res = svc.submit_breakout_now("AAA", 10.0, 10.5, last_price=10.0) # stop > entry assert res["ok"] is False and res["skipped"] == "bad_geometry" + + +def test_guard_never_cancels_a_filled_position(): + """The naked-stop fix: a stale-looking armed entry whose position is actually + OPEN at the broker must NOT be cancelled (that would strip its stop/TP).""" + from datetime import datetime, timedelta + store = EventStore(":memory:") + client = _FakeClient(positions=[{"symbol": "AAA", "qty": "100"}]) # AAA filled + svc = TradingExecutionService( + store, executor=AlpacaPaperExecutor(store, client=client), + settings=ExecutionSettings(entry_timeout_bars=1, entry_invalidate_pct=0.0, + max_daily_loss_pct=0.5), + session_id="t", price_provider=lambda s: 0.5, # below trigger -> would invalidate + ) + # an entry that LOOKS stale (armed 5 min ago, price below trigger) + svc._armed["o1"] = {"symbol": "AAA", "entry_price": 1.0, "broker_order_id": "b1", + "armed_at": datetime.now() - timedelta(minutes=5), "checks": 0} + backed = svc.expire_stale_entries() + assert backed == [] # not backed out + assert client.canceled == [] # stop/TP legs NOT stripped + assert "o1" not in svc._armed # but we stop tracking the filled entry diff --git a/trading_execution.py b/trading_execution.py index 994ba45..e0bb9c5 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -14,9 +14,11 @@ Exit orders close existing broker positions on demand. from __future__ import annotations +import functools import json import logging import os +import threading from dataclasses import dataclass from datetime import datetime @@ -46,6 +48,22 @@ from trading_mode import TradingModeSettings logger = logging.getLogger(__name__) +def _locked(method): + """Serialize a mutating service method behind self._lock (a re-entrant lock). + + The fast trigger thread (submit_breakout_now) and the main loop (tick, + approvals) both mutate execution state (_armed, _requested_symbols) and emit + events; without this they would race on the shared psycopg2 connection and + the in-memory sets. RLock so a public method may call another (tick -> + expire_stale_entries) without deadlocking. + """ + @functools.wraps(method) + def wrapper(self, *args, **kwargs): + with self._lock: + return method(self, *args, **kwargs) + return wrapper + + @dataclass class ExecutionSettings: enabled: bool = True @@ -154,6 +172,8 @@ class TradingExecutionService: self._halted = False # consecutive equity-read failures; after a few we fail CLOSED (halt) self._equity_read_failures = 0 + # serializes state mutations across the main loop and the trigger thread + self._lock = threading.RLock() # -- pipeline ----------------------------------------------------------- @@ -177,6 +197,27 @@ class TradingExecutionService: if p.get("symbol") } + def _broker_held_symbols(self) -> set[str] | None: + """Symbols the broker reports as OPEN POSITIONS right now (truth). + + Used to guard entry cancellation: our own order_filled events lag the + actual fill (they arrive via the 60s account sync), so the entry-timeout + guard once cancelled an entry it thought was unfilled AFTER it had really + filled — and cancelling a bracket parent cancels its stop/take-profit + legs, leaving the position NAKED. This queries the broker directly so a + filled name is never cancelled. None if positions can't be read (caller + then falls back to the synced snapshot and stays conservative). + """ + client = getattr(self.executor, "client", None) + if client is None or not hasattr(client, "get_positions"): + return None + try: + positions = client.get_positions() or [] + return {str(p.get("symbol")) for p in positions if p.get("symbol")} + except Exception as exc: # noqa: BLE001 + logger.warning("broker positions unreadable in entry guard: %s", exc) + return None + def _flatten_all(self, reason: str) -> dict: """Cancel unfilled entries and market-close every open position. @@ -211,6 +252,7 @@ class TradingExecutionService: result["errors"].append(f"close {symbol}: {exc}") return result + @_locked def _daily_loss_breach(self) -> bool: """True once the day's loss hits the circuit-breaker threshold. @@ -306,6 +348,7 @@ class TradingExecutionService: ) return self._halted + @_locked def request_approvals_for_ready_signals(self) -> list[str]: """Turn fresh ready signals into approval requests. Returns order ids.""" if not self.settings.enabled: @@ -395,6 +438,7 @@ class TradingExecutionService: return created + @_locked def approve_order( self, order_id: str, approved_by: str = "dashboard", notes: str | None = None ) -> dict: @@ -442,6 +486,7 @@ class TradingExecutionService: "error": result.error, } + @_locked def submit_breakout_now( self, symbol: str, @@ -569,6 +614,7 @@ class TradingExecutionService: "stop": stop_f, } + @_locked def reject_order( self, order_id: str, rejected_by: str = "dashboard", reason: str = "manual" ) -> dict: @@ -591,6 +637,7 @@ class TradingExecutionService: self._requested_symbols.discard(entry.get("symbol") or "") return {"ok": True, "order_id": order_id} + @_locked def process_auto_approvals(self) -> list[dict]: """Auto-approve any pending requests marked approval_mode=auto.""" results = [] @@ -601,6 +648,7 @@ class TradingExecutionService: ) return results + @_locked def submit_exit_order(self, symbol: str, reason: str = "manual_exit") -> dict: """Close an open broker position for symbol with a market sell.""" held = self._held_symbols() @@ -658,6 +706,7 @@ class TradingExecutionService: ids.add(str(payload["order_id"])) return ids + @_locked def expire_stale_entries(self) -> list[dict]: """Back out of unfilled entries that timed out or broke their trigger. @@ -677,6 +726,12 @@ class TradingExecutionService: filled = self._filled_order_ids() cancelled = self._cancelled_order_ids() + # broker truth: never cancel an entry whose position is actually open + # (cancelling its bracket parent would strip the protective stop/TP). + # Fall back to the synced snapshot if the broker can't be read. + broker_held = self._broker_held_symbols() + if broker_held is None: + broker_held = self._held_symbols() invalidate_pct = self.settings.entry_invalidate_pct timeout_bars = self.settings.entry_timeout_bars now = self._now() @@ -694,6 +749,11 @@ class TradingExecutionService: self._armed.pop(order_id, None) self._requested_symbols.discard(symbol) continue + # FILLED at the broker (position open) -> stop tracking, NEVER cancel + # (this is the naked-stop fix: cancelling here killed the stop leg). + if symbol in broker_held: + self._armed.pop(order_id, None) + continue armed["checks"] += 1 reason = None @@ -746,6 +806,7 @@ class TradingExecutionService: return actions + @_locked def tick(self) -> dict: """One execution pass: expire stale entries, request approvals, auto-execute.""" backed_out = self.expire_stale_entries()