Open catch-up flatten: close stale overnight carries at the open

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 <noreply@anthropic.com>

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))

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

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"]

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):