From 1be2dfd042eb8feffeb3869c3680a05293298ebb Mon Sep 17 00:00:00 2001 From: momentum dev Date: Mon, 22 Jun 2026 12:08:09 -0400 Subject: [PATCH] Open catch-up flatten: close stale overnight carries at the open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backstop behind the EOD-flatten fix: even with the EOD flatten working, a carry can survive (loop down at EOD, or the held-qty 403 before it was fixed) and bleed for days — ATPC sat 4 days over the Juneteenth gap, -$651. This closes any position carried from a PRIOR session, at the open, before it bleeds. - runtime/flatten.find_overnight_carries (pure, tested): open positions with no BUY filled today (ET) = carried overnight. Today's fresh names have a buy today, so they're never touched. - ExecutionService.flatten_overnight_carries: detects carries from the broker (positions + filled buys) and closes them via the robust cancel-legs-then-close path — WITHOUT halting the session (unlike close_session), so today's trading continues. Emits an open_catchup_flatten risk event. - run_live_paper step_open_catchup_flatten: once per session, during RTH, gated by TRADING_OPEN_CATCHUP_FLATTEN_ENABLED (def 1). Loud log when it flattens a carry. Dry-run against the live book: correctly targets ONLY ATPC (6/18 carry), leaves today's DAMD/ECX. 38 runtime/safety tests pass. Needs a loop restart to activate. Co-Authored-By: Claude Opus 4.8 --- run_live_paper.py | 36 ++++++++++++- runtime/flatten.py | 37 +++++++++++++ tests/unit/runtime/test_flatten_carries.py | 48 +++++++++++++++++ trading_execution.py | 60 +++++++++++++++++++++- 4 files changed, 178 insertions(+), 3 deletions(-) create mode 100644 tests/unit/runtime/test_flatten_carries.py diff --git a/run_live_paper.py b/run_live_paper.py index b4793bd..d22cb1e 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -59,7 +59,6 @@ from runtime.market_calendar import ( days_to_next_session, eod_flatten_status, is_regular_hours, - session_close_hm, ) from runtime.triggers import ArmedTriggerBook from runtime.watcher import Watcher, WatcherConfig @@ -859,6 +858,36 @@ def main(argv: list[str] | None = None) -> int: return f"{sev}{tag} closed [{closed}] errors={res['errors']} (will retry)" return f"{tag} closed [{closed}]" + # --- open catch-up flatten: a day-trading book should be FLAT by the open; + # close anything carried from a prior session (a missed/failed EOD flatten) + # before it bleeds for days — the ATPC-over-Juneteenth backstop. Does NOT + # halt the session; today's fresh names are left to trade normally. + catchup_enabled = _flag("TRADING_OPEN_CATCHUP_FLATTEN_ENABLED", "1") + _catchup_done = {"v": False} + + def step_open_catchup_flatten(): + if not catchup_enabled or _catchup_done["v"] or rt["execution"] is None: + return None + try: + from zoneinfo import ZoneInfo + now_et = datetime.now(ZoneInfo("America/New_York")) + except Exception: # noqa: BLE001 + return None + if not is_regular_hours(now_et): # only act while the market is open + return None + res = rt["execution"].flatten_overnight_carries(now_et.date(), "open_catchup") + if res["errors"]: + print(f"[{datetime.now():%H:%M:%S}] ‼ OPEN CATCH-UP could not flatten carry: " + f"{res['errors']}", flush=True) + return f"OPEN CATCH-UP carries={res['carries']} closed={res['closed']} errors={res['errors']} (will retry)" + # clean (no carries) or all carries closed -> done for the session + _catchup_done["v"] = True + if res["carries"]: + print(f"[{datetime.now():%H:%M:%S}] OPEN CATCH-UP flattened stale carries: " + f"{res['closed']}", flush=True) + return f"OPEN CATCH-UP flattened overnight carries {res['closed']}" + return None + def step_news(): """Land latest Alpaca/Benzinga news for the tracked universe into raw_news_items (feeds recent_news_map catalyst prioritisation + the Ollama @@ -915,13 +944,16 @@ def main(argv: list[str] | None = None) -> int: enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")), enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) + scheduler.add("catchup", step_open_catchup_flatten, + float(os.environ.get("TRADING_OPEN_CATCHUP_INTERVAL_SECONDS", "30")), + enabled=_flag("TRADING_EXECUTION_ENABLED", "1")) def run_pass(): # force every task due, in pipeline order for task in scheduler.tasks: task.last_run = -10**9 results = {} - for name in ("discover", "ingest", "watch", "arm", "sync", "execute", "exits", "eod"): + for name in ("discover", "ingest", "watch", "arm", "sync", "execute", "exits", "eod", "catchup"): for task in scheduler.tasks: if task.name == name and task.enabled: results.update(scheduler_run_one(task)) diff --git a/runtime/flatten.py b/runtime/flatten.py index c7fc66d..30b484e 100644 --- a/runtime/flatten.py +++ b/runtime/flatten.py @@ -21,9 +21,46 @@ from __future__ import annotations import logging import time +from datetime import date, datetime, timezone +from zoneinfo import ZoneInfo logger = logging.getLogger(__name__) +_ET = ZoneInfo("America/New_York") + + +def _fill_et_date(iso) -> date | None: + """ET calendar date of a broker fill timestamp (stored UTC), or None.""" + try: + dt = datetime.fromisoformat(str(iso).replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(_ET).date() + except (TypeError, ValueError): + return None + + +def find_overnight_carries(open_symbols, buy_fills, today_et: date) -> list[str]: + """Open positions whose lot was NOT entered today (ET) — i.e. carried across + a prior session. A day-trading book should be flat by the open, so anything + still held that wasn't bought today survived overnight (the EOD flatten was + missed/failed) and should be caught up at the open before it bleeds for days + (the ATPC-over-Juneteenth case). + + Args: + open_symbols: iterable of currently-open position symbols. + buy_fills: iterable of {"symbol", "filled_at"} for FILLED BUY orders. + today_et: today's ET date. + Returns sorted symbols open today with no buy filled today. + """ + bought_today: set[str] = set() + for f in buy_fills or []: + if _fill_et_date(f.get("filled_at")) == today_et: + sym = f.get("symbol") + if sym: + bought_today.add(sym) + return sorted(s for s in (open_symbols or ()) if s and s not in bought_today) + # Order states in which a sell order still RESERVES position quantity. While an # order is in one of these states, that many shares show as ``held_for_orders`` # and are unavailable to a separate close. (A ``filled``/``canceled`` leg no diff --git a/tests/unit/runtime/test_flatten_carries.py b/tests/unit/runtime/test_flatten_carries.py new file mode 100644 index 0000000..7c10ebe --- /dev/null +++ b/tests/unit/runtime/test_flatten_carries.py @@ -0,0 +1,48 @@ +"""Overnight-carry detection for the open catch-up flatten.""" + +from datetime import date + +from runtime.flatten import find_overnight_carries + +TODAY = date(2026, 6, 22) + + +def _buy(sym, filled_at): + return {"symbol": sym, "filled_at": filled_at} + + +def test_position_bought_today_is_not_a_carry(): + # 14:00Z == 10:00 ET on 6/22 -> bought today + carries = find_overnight_carries({"DAMD"}, [_buy("DAMD", "2026-06-22T14:00:00Z")], TODAY) + assert carries == [] + + +def test_position_with_no_buy_today_is_a_carry(): + # only a prior-day buy -> carried overnight + carries = find_overnight_carries({"ATPC"}, [_buy("ATPC", "2026-06-18T14:26:00Z")], TODAY) + assert carries == ["ATPC"] + + +def test_mixed_book_returns_only_carries(): + open_syms = {"ATPC", "DAMD", "NIXX"} + fills = [ + _buy("ATPC", "2026-06-18T14:26:00Z"), # carry + _buy("DAMD", "2026-06-22T14:34:00Z"), # today + _buy("NIXX", "2026-06-22T14:00:00Z"), # today + ] + assert find_overnight_carries(open_syms, fills, TODAY) == ["ATPC"] + + +def test_late_utc_fill_maps_to_prior_et_day(): + # 01:00Z on 6/22 == 21:00 ET on 6/21 -> a prior session, so a carry + carries = find_overnight_carries({"XYZ"}, [_buy("XYZ", "2026-06-22T01:00:00Z")], TODAY) + assert carries == ["XYZ"] + + +def test_no_open_positions_no_carries(): + assert find_overnight_carries(set(), [_buy("AAA", "2026-06-18T14:00:00Z")], TODAY) == [] + + +def test_open_with_no_fills_at_all_is_a_carry(): + # open but we have zero buy-fill records for it -> treat as carried (fail safe) + assert find_overnight_carries({"GHOST"}, [], TODAY) == ["GHOST"] diff --git a/trading_execution.py b/trading_execution.py index eb94e82..59618e9 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -22,7 +22,7 @@ import threading from dataclasses import dataclass from datetime import datetime, timedelta -from runtime.flatten import cancel_protective_and_close +from runtime.flatten import cancel_protective_and_close, find_overnight_carries from alpaca_paper.execution import ( AlpacaPaperExecutor, ExecutionRequest, @@ -1023,6 +1023,64 @@ class TradingExecutionService: ) return result + def flatten_overnight_carries(self, today_et, reason: str = "open_catchup") -> dict: + """Catch-up flatten: close positions CARRIED from a prior session. + + A day-trading book should never hold overnight, but a missed/failed EOD + flatten (loop down, or the held-qty 403 before that was fixed) can leave a + stale carry bleeding for days (ATPC over the 4-day Juneteenth gap). On + each session this closes any position not entered today — WITHOUT halting + the session (unlike ``close_session``); today's fresh names are untouched. + Best-effort: errors are collected, not raised. Returns + {carries, closed, errors}. + """ + result: dict = {"carries": [], "closed": [], "errors": []} + client = getattr(self.executor, "client", None) + if client is None or not hasattr(client, "close_position"): + return result + try: + positions = client.get_positions() + open_syms = {p.get("symbol") for p in positions + if abs(float(p.get("qty") or 0)) > 0} + except Exception as exc: # noqa: BLE001 + result["errors"].append(f"positions: {exc}") + return result + if not open_syms: + return result + try: + orders = client.get_orders(status="all", limit=500, nested=True, + symbols=sorted(open_syms)) + except Exception as exc: # noqa: BLE001 + result["errors"].append(f"orders: {exc}") + return result + buy_fills = [] + for o in orders or []: + for x in [o, *(o.get("legs") or [])]: + if (str(x.get("side")) == "buy" and x.get("status") == "filled" + and x.get("filled_at")): + buy_fills.append({"symbol": x.get("symbol") or o.get("symbol"), + "filled_at": x.get("filled_at")}) + carries = find_overnight_carries(open_syms, buy_fills, today_et) + result["carries"] = carries + for symbol in carries: + try: + self._release_and_close(client, symbol) + result["closed"].append(symbol) + except Exception as exc: # noqa: BLE001 + result["errors"].append(f"close {symbol}: {exc}") + if carries: + self.store.emit( + RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, correlation_id=self.session_id, + message=(f"Open catch-up flatten ({reason}): carried {carries}; " + f"closed {result['closed']}" + + (f"; ERRORS {result['errors']}" if result['errors'] else "")), + rule_type="open_catchup_flatten", rule_value=0.0, + current_state=result, action_taken="flatten_carries", + ) + ) + return result + def _filled_order_ids(self) -> set[str]: ids: set[str] = set() for e in self.store.query_events(event_type="order_filled", limit=None):