money-path: daily self-restart + SIGTERM + positive-evidence exit gate

Plan 003 steps 1-3.
- Loop exit(0) after a SUCCESSFUL EOD flatten (TRADING_EXIT_AFTER_EOD_FLATTEN,
  default on); compose restarts fresh. Restart-churn guard: _rebuild_daily_state
  detects today's eod_flatten event and seeds _session_closed/_eod_done so a
  post-flatten restart neither re-flattens nor re-exits. SIGTERM = clean drain.
- Exit manager: _FETCH_FAILED sentinel distinguishes a failed get_orders from a
  genuinely-empty snapshot. A failed fetch skips naked-stop enforcement (catastrophe
  still exits, re-fetching); an empty-but-successful snapshot needs TWO consecutive
  reads before 'naked' is trusted (Alpaca silently returns [] for symbols=). Costs
  ~one 12s pass on the all-naked case only; isolated naked positions unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
momentum devandClaude Fable 5 committed 2026-07-13 22:20:37 -04:00
1 parent af8137dfd7
commit d16eb85ee9
7 files changed
+322 -9

No files matched your search

+4
View File
@@ -143,6 +143,10 @@ TRADING_LIQUIDITY_MAX_VOLUME_PCT=0.025
# never carries overnight gap risk (a day-trading strategy shouldn't hold). # never carries overnight gap risk (a day-trading strategy shouldn't hold).
TRADING_EOD_FLATTEN_ENABLED=1 TRADING_EOD_FLATTEN_ENABLED=1
TRADING_EOD_FLATTEN_TIME=15:55 TRADING_EOD_FLATTEN_TIME=15:55
# The process is one trading session: after a SUCCESSFUL EOD flatten the loop
# exits 0 and compose restarts it fresh, so memory-only daily state (entry cap,
# benches, session-closed flags) can never go stale across a date rollover.
TRADING_EXIT_AFTER_EOD_FLATTEN=1
TRADING_MAX_ORDERS_PER_TICK=1 TRADING_MAX_ORDERS_PER_TICK=1
TRADING_ENTRY_ORDER_TYPE=limit # rest at the breakout trigger TRADING_ENTRY_ORDER_TYPE=limit # rest at the breakout trigger
# --- managed exit (swept + out-of-sample verified: trailLow_a1) ------------- # --- managed exit (swept + out-of-sample verified: trailLow_a1) -------------
+52 -2
View File
@@ -27,6 +27,7 @@ from __future__ import annotations
import argparse import argparse
import json import json
import os import os
import signal
import sys import sys
import threading import threading
import time import time
@@ -82,6 +83,36 @@ def _flag(name: str, default: str = "1") -> bool:
return os.environ.get(name, default).strip().lower() in {"1", "true", "yes", "on"} return os.environ.get(name, default).strip().lower() in {"1", "true", "yes", "on"}
# The process is ONE trading session. A dozen memory-only daily flags (_eod_done,
# _session_closed, entry caps, benches...) are set once and never reset on date
# rollover, so an uninterrupted day-2 of uptime would take zero entries and never
# EOD-flatten. Instead of resetting them all, the loop EXITS cleanly after a
# successful EOD flatten (and on SIGTERM) — compose's `restart: unless-stopped`
# brings back a fresh process whose boot recovery (_rebuild_daily_state) restores
# any state that matters. Set from the scheduler thread and the signal handler;
# consumed by the main loop so the existing `finally:` cleanup runs.
_clean_exit_requested = threading.Event()
def _request_exit_after_eod(ok: bool) -> bool:
"""Request a clean process exit after a SUCCESSFUL EOD flatten.
Gated on TRADING_EXIT_AFTER_EOD_FLATTEN (default on). A FAILED flatten must
keep retrying inside the window, so ``ok=False`` never requests exit.
Returns whether exit was requested."""
if not ok or not _flag("TRADING_EXIT_AFTER_EOD_FLATTEN", "1"):
return False
_clean_exit_requested.set()
return True
def _sigterm_handler(signum, frame): # noqa: ARG001 — signal API shape
"""SIGTERM (docker stop) = clean shutdown: drain through the main loop's
``finally:`` (threads joined, DB connections closed) instead of dying mid-pass."""
print("[stop] SIGTERM received — draining for clean shutdown", flush=True)
_clean_exit_requested.set()
def _now_session_date(): def _now_session_date():
session_date, _, _, _ = classify_session(datetime.now(timezone.utc)) session_date, _, _, _ = classify_session(datetime.now(timezone.utc))
return session_date return session_date
@@ -1061,7 +1092,14 @@ def main(argv: list[str] | None = None) -> int:
eod_h, eod_m = 15, 55 eod_h, eod_m = 15, 55
_normal_lead = max(1, 16 * 60 - (eod_h * 60 + eod_m)) # 15:55 -> 5 _normal_lead = max(1, 16 * 60 - (eod_h * 60 + eod_m)) # 15:55 -> 5
_pre_lead = int(os.environ.get("TRADING_PRE_CLOSURE_FLATTEN_LEAD_MIN", "20")) _pre_lead = int(os.environ.get("TRADING_PRE_CLOSURE_FLATTEN_LEAD_MIN", "20"))
_eod_done = {"v": False} # Restart-churn guard: the daily-restart exit fires INSIDE the EOD window and
# compose restarts within seconds — the fresh process must see today's
# already-successful flatten (event-store truth) or it would flatten-nothing,
# exit again, and churn until the window closes.
_eod_done = {"v": rt["execution"] is not None
and rt["execution"].eod_flatten_done_today()}
if _eod_done["v"]:
print("[boot] EOD flatten already succeeded today — no re-flatten, no re-exit")
def step_eod_flatten(): def step_eod_flatten():
if not eod_enabled or _eod_done["v"] or rt["execution"] is None: if not eod_enabled or _eod_done["v"] or rt["execution"] is None:
@@ -1103,6 +1141,13 @@ def main(argv: list[str] | None = None) -> int:
query_session_pnl(rt["store"])) query_session_pnl(rt["store"]))
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
pass pass
# the process is the session: a successful flatten (including "already
# flat, nothing to close") ends it — exit cleanly, compose restarts fresh
if _request_exit_after_eod(ok):
msg = ("♻️ EOD flatten complete — exiting for daily restart "
"(compose brings the loop back fresh).")
print(f"[{datetime.now():%H:%M:%S}] {msg}", flush=True)
send_telegram(msg)
return f"{tag} closed [{closed}]" return f"{tag} closed [{closed}]"
# --- open catch-up flatten: a day-trading book should be FLAT by the open; # --- open catch-up flatten: a day-trading book should be FLAT by the open;
@@ -1243,10 +1288,13 @@ def main(argv: list[str] | None = None) -> int:
if ollama_cfg.enabled and ollama_cfg.enrichment_enabled: if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
threading.Thread(target=enrich_loop, name="enrich", daemon=True).start() threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()
print(f"[boot] news enrichment thread started (every {ollama_cfg.enrichment_interval_seconds}s, model {ollama_cfg.model})") print(f"[boot] news enrichment thread started (every {ollama_cfg.enrichment_interval_seconds}s, model {ollama_cfg.model})")
# SIGTERM (docker stop) drains through the loop's finally: instead of a
# mid-pass kill — leaves the book protected by broker-side stop legs.
signal.signal(signal.SIGTERM, _sigterm_handler)
try: try:
# one immediate full pass, then interval-driven # one immediate full pass, then interval-driven
run_pass() run_pass()
while True: while not _clean_exit_requested.is_set():
results = scheduler.run_pending() results = scheduler.run_pending()
if results: if results:
stamp = datetime.now().strftime("%H:%M:%S") stamp = datetime.now().strftime("%H:%M:%S")
@@ -1256,6 +1304,8 @@ def main(argv: list[str] | None = None) -> int:
) )
print(f"[{stamp}] {line}") print(f"[{stamp}] {line}")
time.sleep(1.0) time.sleep(1.0)
# clean exit (EOD daily restart or SIGTERM): fall through to finally
print("[stop] clean exit — shutting down (compose restarts the loop fresh)")
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n[stop] shutting down") print("\n[stop] shutting down")
finally: finally:
+42 -4
View File
@@ -24,6 +24,10 @@ from strategy.exits import ExitConfig, TRAIL_NONE, catastrophe_triggered, manage
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Sentinel for "the orders fetch FAILED" — distinct from None / an empty list so
# a transient broker/API error is never mistaken for "every position is naked".
_FETCH_FAILED = object()
@dataclass @dataclass
class _Tracked: class _Tracked:
@@ -53,6 +57,7 @@ class LiveExitManager:
self._tracked: dict[str, _Tracked] = {} self._tracked: dict[str, _Tracked] = {}
self._naked_passes: dict[str, int] = {} # consecutive passes a held position had no live stop self._naked_passes: dict[str, int] = {} # consecutive passes a held position had no live stop
self._open_anchored: set[str] = set() # breakeven carries re-anchored to today's opening range self._open_anchored: set[str] = set() # breakeven carries re-anchored to today's opening range
self._empty_snapshot_streak = 0 # consecutive empty-but-successful order snapshots (Alpaca can silently return [])
# -- broker helpers ---------------------------------------------------- # -- broker helpers ----------------------------------------------------
@@ -65,19 +70,26 @@ class LiveExitManager:
# scope to the held symbols — Alpaca's `limit` keeps only the most-recent # scope to the held symbols — Alpaca's `limit` keeps only the most-recent
# orders, so on a busy day an unscoped query drops older positions' stop # orders, so on a busy day an unscoped query drops older positions' stop
# legs and we'd misread a protected position as naked. # legs and we'd misread a protected position as naked.
# POSITIVE-EVIDENCE CONTRACT: a FAILED call returns the _FETCH_FAILED
# sentinel (NOT None), so it's distinguishable from a genuinely-empty
# but successful response — the caller must never read a fetch failure
# as "every position is naked" and market-flatten a protected book.
try: try:
return self.client.get_orders(status="all", limit=500, nested=True, return self.client.get_orders(status="all", limit=500, nested=True,
symbols=list(symbols) if symbols else None) symbols=list(symbols) if symbols else None)
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
return None return _FETCH_FAILED
def _stop_leg(self, symbol: str, orders): def _stop_leg(self, symbol: str, orders):
"""(order_id, stop_price) of the live sell-stop protecting symbol. """(order_id, stop_price) of the live sell-stop protecting symbol.
Uses pre-fetched ``orders`` (status="all" — the protective leg is a Uses pre-fetched ``orders`` (status="all" — the protective leg is a
'held' child of the FILLED entry, which status="open" wouldn't surface). 'held' child of the FILLED entry, which status="open" wouldn't surface).
A failed fetch (``_FETCH_FAILED``) or ``None`` yields (None, None) — the
pass loop guards naked enforcement on the sentinel separately, so this
only ever means "no stop found in what we could read".
""" """
if orders is None: if orders is None or orders is _FETCH_FAILED:
return (None, None) return (None, None)
for o in orders: for o in orders:
for c in [o, *(o.get("legs") or [])]: for c in [o, *(o.get("legs") or [])]:
@@ -184,6 +196,23 @@ class LiveExitManager:
orders = self._all_orders(held.keys()) orders = self._all_orders(held.keys())
mkt_open = self._market_open() # once per pass — gates flatten-vs-rest + trailing mkt_open = self._market_open() # once per pass — gates flatten-vs-rest + trailing
# POSITIVE-EVIDENCE GATE — never let an ambiguous read command a naked flatten:
# - fetch FAILED (sentinel): skip naked enforcement this pass entirely; the
# catastrophe stop still runs but flattens with orders=None so the helper
# re-fetches instead of consuming the sentinel.
# - fetch SUCCEEDED but empty while positions are held (Alpaca's documented
# silently-empty symbols= response): require TWO consecutive empty reads
# before trusting "naked" — one blip can't flatten a protected book.
fetch_failed = orders is _FETCH_FAILED
orders_for_flatten = None if fetch_failed else orders
if fetch_failed:
logger.warning("orders fetch failed — skipping naked-stop enforcement this pass")
else:
self._empty_snapshot_streak = (self._empty_snapshot_streak + 1) if not orders else 0
empty_untrusted = (not fetch_failed) and (not orders) and self._empty_snapshot_streak < 2
# naked enforcement is suppressed when we can't positively confirm nakedness
naked_ok = not fetch_failed and not empty_untrusted
actions: list[str] = [] actions: list[str] = []
for sym, p in held.items(): for sym, p in held.items():
entry = float(p.get("avg_entry_price") or 0.0) entry = float(p.get("avg_entry_price") or 0.0)
@@ -202,7 +231,11 @@ class LiveExitManager:
loss_pct = (entry - cur) / entry * 100 loss_pct = (entry - cur) / entry * 100
try: try:
if mkt_open: if mkt_open:
self._flatten(sym, orders) # orders_for_flatten is None on a failed fetch so the helper
# re-fetches rather than acting on the sentinel; a true
# catastrophe still exits even when stop status is unknowable
# (a deep-drawdown position is worse than a redundant cancel).
self._flatten(sym, orders_for_flatten)
self._emit(sym, f"CATASTROPHE exit {sym} @~{cur:.4f} " self._emit(sym, f"CATASTROPHE exit {sym} @~{cur:.4f} "
f"({loss_pct:.1f}% below entry)", "exit_catastrophe", f"({loss_pct:.1f}% below entry)", "exit_catastrophe",
{"price": cur, "entry": entry, "stop": cur_stop}) {"price": cur, "entry": entry, "stop": cur_stop})
@@ -231,7 +264,7 @@ class LiveExitManager:
# Flatten after a short grace (consecutive passes) so a just-filled # Flatten after a short grace (consecutive passes) so a just-filled
# bracket's legs can attach first. # bracket's legs can attach first.
grace = self.cfg.enforce_stop_grace_passes grace = self.cfg.enforce_stop_grace_passes
if grace and entry > 0 and leg_id is None: if grace and entry > 0 and leg_id is None and naked_ok:
n = self._naked_passes.get(sym, 0) + 1 n = self._naked_passes.get(sym, 0) + 1
self._naked_passes[sym] = n self._naked_passes[sym] = n
if n >= grace: if n >= grace:
@@ -259,6 +292,11 @@ class LiveExitManager:
logger.warning("naked-stop enforcement failed %s: %s", sym, exc) logger.warning("naked-stop enforcement failed %s: %s", sym, exc)
self._naked_passes.pop(sym, None) self._naked_passes.pop(sym, None)
continue # don't run trail logic on a naked position continue # don't run trail logic on a naked position
# Couldn't positively confirm the stop this pass (failed/untrusted-empty
# fetch) and no leg was found: skip the symbol — don't reset the naked
# counter on unverified evidence, and don't trail against a None stop.
if leg_id is None and not naked_ok:
continue
self._naked_passes.pop(sym, None) # has a live stop self._naked_passes.pop(sym, None) # has a live stop
# TRAILING runs only while the session is OPEN: after-hours there are no new # TRAILING runs only while the session is OPEN: after-hours there are no new
+88
View File
@@ -0,0 +1,88 @@
"""Daily self-restart: the process is ONE trading session (plan 003, steps 1-2).
After a SUCCESSFUL EOD flatten the loop requests a clean exit and compose
restarts it fresh; a post-flatten restart must neither re-flatten nor re-exit
(today's eod_flatten event in the store is the truth); a FAILED flatten keeps
retrying inside the window.
"""
from datetime import datetime
import run_live_paper as rlp
from alpaca_paper.execution import AlpacaPaperExecutor
from storage.event_schema import EventMode, RiskRuleTriggeredEvent
from storage.event_store import EventStore
from trading_execution import ExecutionSettings, TradingExecutionService
class _FakeClient:
def submit_order(self, **kw):
return {}
def get_account(self):
return {"equity": "100000", "last_equity": "100000"}
def _svc(store):
return TradingExecutionService(
store, executor=AlpacaPaperExecutor(store, client=_FakeClient()),
settings=ExecutionSettings(), session_id="t",
)
def _emit_eod_flatten(store, errors):
store.emit(RiskRuleTriggeredEvent(
timestamp=datetime.now(), mode=EventMode.PAPER, correlation_id="t",
message="End-of-day flatten (eod_flatten)",
rule_type="eod_flatten", rule_value=0.0,
current_state={"cancelled_entries": 0,
"closed_positions": [] if errors else ["AAA"],
"errors": errors},
action_taken="closed_session",
))
def test_eod_exit_flag(monkeypatch):
"""_request_exit_after_eod: ok+flag-on -> exit requested; failed flatten or
flag-off -> not (a failed flatten must keep retrying inside the window)."""
rlp._clean_exit_requested.clear()
monkeypatch.setenv("TRADING_EXIT_AFTER_EOD_FLATTEN", "1")
assert rlp._request_exit_after_eod(True) is True
assert rlp._clean_exit_requested.is_set()
rlp._clean_exit_requested.clear()
assert rlp._request_exit_after_eod(False) is False
assert not rlp._clean_exit_requested.is_set()
monkeypatch.setenv("TRADING_EXIT_AFTER_EOD_FLATTEN", "0")
assert rlp._request_exit_after_eod(True) is False
assert not rlp._clean_exit_requested.is_set()
rlp._clean_exit_requested.clear()
def test_boot_after_eod_flatten():
"""Restart-churn guard: with today's SUCCESSFUL eod_flatten event in the
store, a fresh service boots session-closed (no re-entry) and reports
done-today (the loop then neither re-flattens nor re-exits)."""
store = EventStore(":memory:")
_emit_eod_flatten(store, errors=[])
svc = _svc(store)
assert svc._session_closed is True
assert svc.eod_flatten_done_today() is True
def test_boot_after_failed_eod_flatten_still_retries():
"""A FAILED flatten blocks re-entry but must NOT read as done — the fresh
process retries the flatten inside the window exactly as before."""
store = EventStore(":memory:")
_emit_eod_flatten(store, errors=["close AAA: 403"])
svc = _svc(store)
assert svc._session_closed is True
assert svc.eod_flatten_done_today() is False
def test_boot_fresh_day_not_closed():
store = EventStore(":memory:")
svc = _svc(store)
assert svc._session_closed is False
assert svc.eod_flatten_done_today() is False
+10 -3
View File
@@ -125,9 +125,14 @@ def test_naked_position_flattened_after_grace_when_market_open():
cfg=ExitConfig(enforce_stop_grace_passes=2, catastrophe_pct=0.10), cfg=ExitConfig(enforce_stop_grace_passes=2, catastrophe_pct=0.10),
session_id="t", session_id="t",
) )
mgr.manage() # pass 1: naked but within grace # The _NakedBroker returns an EMPTY snapshot until a stop is rested, so the
# positive-evidence gate spends pass 1 CONFIRMING the empty read (Alpaca can
# silently return [] — one blip must not flatten). Naked passes then count.
mgr.manage() # pass 1: empty snapshot untrusted -> no count
assert broker.closed == [] assert broker.closed == []
mgr.manage() # pass 2: grace reached -> flatten mgr.manage() # pass 2: empty confirmed; naked pass 1/2
assert broker.closed == []
mgr.manage() # pass 3: naked pass 2/2 = grace -> flatten
assert broker.closed == [("AAA", None)] assert broker.closed == [("AAA", None)]
assert broker.submitted == [] # flattened, not rest-a-stop assert broker.submitted == [] # flattened, not rest-a-stop
@@ -144,7 +149,9 @@ def test_naked_position_rests_stop_when_market_closed():
cfg=ExitConfig(enforce_stop_grace_passes=2, catastrophe_pct=0.10), cfg=ExitConfig(enforce_stop_grace_passes=2, catastrophe_pct=0.10),
session_id="t", session_id="t",
) )
mgr.manage(); mgr.manage() # grace reached while CLOSED # +1 pass vs the pre-gate timing: the empty snapshot is confirmed once before
# naked passes count (positive-evidence gate). Three passes reach grace=2.
mgr.manage(); mgr.manage(); mgr.manage() # grace reached while CLOSED
assert broker.closed == [] # NOT flattened (would never fill) assert broker.closed == [] # NOT flattened (would never fill)
assert len(broker.submitted) == 1 # rested a protective stop instead assert len(broker.submitted) == 1 # rested a protective stop instead
o = broker.submitted[0] o = broker.submitted[0]
+97
View File
@@ -0,0 +1,97 @@
"""Exit-manager positive-evidence gate (plan 003, step 3).
No destructive naked-stop flatten may fire on an AMBIGUOUS orders read:
- a FAILED get_orders (sentinel) suppresses naked enforcement, but the
catastrophe backstop still runs (flattening via a re-fetch, not the sentinel);
- a genuinely-empty-but-successful snapshot needs TWO consecutive empty reads
before it is trusted (Alpaca can silently return [] for a symbols= query);
- a real naked position with a non-empty snapshot is still enforced (regression).
"""
import pandas as pd
from runtime.exit_manager import LiveExitManager
from storage.event_store import EventStore
from strategy.exits import ExitConfig
class _Broker:
def __init__(self, positions, orders_result, market_open=True):
self._positions = positions
self._orders_result = orders_result # a list, or an Exception to raise
self.market_open = market_open
self.closed = []
self.canceled = []
self.get_orders_calls = 0
def get_positions(self):
return self._positions
def get_orders(self, status="open", limit=200, nested=True, symbols=None):
self.get_orders_calls += 1
if isinstance(self._orders_result, Exception):
raise self._orders_result
return list(self._orders_result)
def get_clock(self):
return {"is_open": self.market_open}
def cancel_order(self, order_id):
self.canceled.append(order_id)
def close_position(self, symbol, qty=None, percentage=None):
self.closed.append((symbol, qty))
return {"id": "c"}
def _cfg():
# grace=1 isolates the positive-evidence gate from the multi-pass grace
return ExitConfig(enforce_stop_grace_passes=1, catastrophe_pct=0.10)
def _pos(sym, entry, current):
return {"symbol": sym, "avg_entry_price": str(entry), "qty": "100",
"current_price": str(current), "side": "long"}
def test_fetch_failed_skips_naked_but_catastrophe_still_exits():
store = EventStore(":memory:")
# AAA is a mild naked position (should NOT flatten on a failed fetch);
# BBB is catastrophically underwater (MUST flatten regardless of stop status).
broker = _Broker([_pos("AAA", 10.0, 9.9), _pos("BBB", 10.0, 8.5)],
orders_result=RuntimeError("api down"))
mgr = LiveExitManager(broker, store, lambda s: pd.DataFrame(), cfg=_cfg(), session_id="t")
mgr.manage()
closed = {s for s, _ in broker.closed}
assert "AAA" not in closed, "naked enforcement must be suppressed on a failed fetch"
assert "BBB" in closed, "catastrophe backstop must still exit a deep-drawdown position"
def test_stop_leg_returns_none_for_sentinel():
from runtime.exit_manager import _FETCH_FAILED
store = EventStore(":memory:")
mgr = LiveExitManager(_Broker([], []), store, lambda s: pd.DataFrame(),
cfg=_cfg(), session_id="t")
assert mgr._stop_leg("AAA", _FETCH_FAILED) == (None, None)
assert mgr._stop_leg("AAA", None) == (None, None)
def test_empty_snapshot_needs_two_consecutive_reads():
store = EventStore(":memory:")
broker = _Broker([_pos("AAA", 10.0, 9.95)], orders_result=[]) # empty-but-successful
mgr = LiveExitManager(broker, store, lambda s: pd.DataFrame(), cfg=_cfg(), session_id="t")
mgr.manage() # pass 1: first empty read — untrusted, no flatten
assert broker.closed == [], "one empty snapshot must not flatten a protected book"
mgr.manage() # pass 2: second consecutive empty read — trusted, naked enforced
assert ("AAA", None) in broker.closed, "two consecutive empty reads confirm naked"
def test_real_naked_still_enforced():
store = EventStore(":memory:")
# a non-empty snapshot with a leg for a DIFFERENT symbol: AAA is genuinely naked
other_leg = {"id": "s1", "symbol": "ZZZ", "type": "stop", "side": "sell",
"stop_price": "9.0", "status": "held"}
broker = _Broker([_pos("AAA", 10.0, 9.95)], orders_result=[other_leg])
mgr = LiveExitManager(broker, store, lambda s: pd.DataFrame(), cfg=_cfg(), session_id="t")
mgr.manage()
assert ("AAA", None) in broker.closed, "a confirmed-naked position must still be flattened"
+29
View File
@@ -396,6 +396,10 @@ class TradingExecutionService:
self._data_halt = False self._data_halt = False
# session closed (end-of-day flatten): no new entries for the rest of the day # session closed (end-of-day flatten): no new entries for the rest of the day
self._session_closed = False self._session_closed = False
# today's EOD flatten already SUCCEEDED (rebuilt from events on boot) —
# the daily-restart loop reads this so a post-flatten fresh process
# neither re-flattens nor re-exits (restart-churn guard)
self._eod_flatten_ok_today = False
# consecutive equity-read failures; after the limit we pause (recoverable) # consecutive equity-read failures; after the limit we pause (recoverable)
self._equity_read_failures = 0 self._equity_read_failures = 0
self._equity_fail_limit = int( self._equity_fail_limit = int(
@@ -451,12 +455,37 @@ class TradingExecutionService:
# bench only names that closed at a LOSS (mirror the live bench rule) # bench only names that closed at a LOSS (mirror the live bench rule)
if sym and float(p.get("realized_pnl") or 0) < 0: if sym and float(p.get("realized_pnl") or 0) < 0:
self._exited_today.add(str(sym)) self._exited_today.add(str(sym))
# session-closed detection: close_session emits rule_type="eod_flatten".
# ANY such event today = the EOD window already began, so re-entry
# stays blocked; a SUCCESSFUL one (no errors) also tells the loop not
# to re-flatten/re-exit after the daily restart. A FAILED flatten
# deliberately does NOT set the ok flag — the fresh process retries
# inside the window exactly as before. limit=None: the eod_flatten
# event lands LAST in the day and query order is ASC, so a capped
# query on a busy risk-event day could silently drop it.
for e in self.store.query_events(event_type="risk_rule_triggered",
since=since, limit=None):
p = _payload(e)
if p.get("rule_type") != "eod_flatten":
continue
self._session_closed = True
if not (p.get("current_state") or {}).get("errors"):
self._eod_flatten_ok_today = True
if self._session_closed:
logger.info("restored session-closed state: EOD flatten already "
"emitted today (succeeded=%s)", self._eod_flatten_ok_today)
if n_fresh or self._requested_symbols or self._exited_today: if n_fresh or self._requested_symbols or self._exited_today:
logger.info("restored daily state: %d fresh fills, %d requested, %d benched", logger.info("restored daily state: %d fresh fills, %d requested, %d benched",
n_fresh, len(self._requested_symbols), len(self._exited_today)) n_fresh, len(self._requested_symbols), len(self._exited_today))
except Exception as exc: # noqa: BLE001 — a rebuild failure must not block boot except Exception as exc: # noqa: BLE001 — a rebuild failure must not block boot
logger.warning("daily-state rebuild skipped: %s", exc) logger.warning("daily-state rebuild skipped: %s", exc)
def eod_flatten_done_today(self) -> bool:
"""True when today's EOD flatten already SUCCEEDED (from the event stream
at boot). The loop seeds its _eod_done flag from this so a post-flatten
restart neither re-flattens an already-flat book nor exit-churns."""
return self._eod_flatten_ok_today
# -- pipeline ----------------------------------------------------------- # -- pipeline -----------------------------------------------------------
def _open_position_count(self) -> int: def _open_position_count(self) -> int: