From 45d5484dd75060ee06265f29a5c73910125f6e8b Mon Sep 17 00:00:00 2001 From: momentum dev Date: Wed, 17 Jun 2026 15:03:35 -0400 Subject: [PATCH] Breaker: recoverable data-halt + network retries (DNS blip ended the day) A transient WSL DNS outage made 3 consecutive equity reads fail, and the fail-CLOSED breaker then PERMANENTLY halted new entries for the whole session even after DNS recovered seconds later. Fixes: - split the breaker into a permanent _halted (real daily-loss breach) and a recoverable _data_halt (equity unreadable) that clears on the next good read; raise the pause threshold to 5 (TRADING_EQUITY_FAIL_LIMIT). - AlpacaPaperClient._request retries transient URLError/gaierror (not HTTP 4xx/5xx) a couple times, so brief blips don't even register as failures. - sync logs network failures as one-line warnings, not full tracebacks (the log was spamming stacks). Co-Authored-By: Claude Opus 4.8 --- alpaca_paper/client.py | 27 +++++++++++---- alpaca_paper/sync.py | 6 ++-- tests/unit/test_execution_safety.py | 16 ++++++--- trading_execution.py | 51 +++++++++++++++++++++-------- 4 files changed, 72 insertions(+), 28 deletions(-) diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index 3f501e7..eb2f5c0 100644 --- a/alpaca_paper/client.py +++ b/alpaca_paper/client.py @@ -25,9 +25,13 @@ class AlpacaApiError(RuntimeError): class AlpacaPaperClient: - def __init__(self, settings: AlpacaPaperSettings | None = None, timeout: int = 15): + def __init__(self, settings: AlpacaPaperSettings | None = None, timeout: int = 15, + max_retries: int = 2): self.settings = settings or AlpacaPaperSettings.from_env() self.timeout = timeout + # retry transient network/DNS errors (not HTTP 4xx/5xx) a couple times; + # DNS failures fail fast, so this smooths brief blips without much latency + self.max_retries = max_retries # -- low level ------------------------------------------------------- @@ -50,12 +54,21 @@ class AlpacaPaperClient: req = urllib.request.Request( url, data=data, headers=self._headers(), method=method ) - try: - with urllib.request.urlopen(req, timeout=self.timeout) as resp: - raw = resp.read().decode() - return json.loads(raw) if raw else {} - except urllib.error.HTTPError as exc: - raise AlpacaApiError(exc.code, exc.read().decode(errors="replace")) from exc + import time as _time + for attempt in range(self.max_retries + 1): + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + raw = resp.read().decode() + return json.loads(raw) if raw else {} + except urllib.error.HTTPError as exc: + # a real API response (4xx/5xx) — do not retry, surface it + 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: + raise + _time.sleep(0.4 * (attempt + 1)) + raise RuntimeError("request retries exhausted") # unreachable def _trading(self, method: str, path: str, **kw): return self._request( diff --git a/alpaca_paper/sync.py b/alpaca_paper/sync.py index 9e9f3b6..0b86fe3 100644 --- a/alpaca_paper/sync.py +++ b/alpaca_paper/sync.py @@ -56,7 +56,7 @@ class AlpacaPaperSync: try: account = self.client.get_account() except Exception as exc: - logger.exception("alpaca account sync failed") + logger.warning("alpaca account sync failed: %s", exc) self._emit_health(f"account: {exc}") return None account_id = str(account.get("account_number") or account.get("id") or "paper") @@ -81,7 +81,7 @@ class AlpacaPaperSync: try: raw = self.client.get_positions() except Exception as exc: - logger.exception("alpaca positions sync failed") + logger.warning("alpaca positions sync failed: %s", exc) self._emit_health(f"positions: {exc}") return None positions = [ @@ -114,7 +114,7 @@ class AlpacaPaperSync: try: raw = self.client.get_orders(status=status, limit=limit) except Exception as exc: - logger.exception("alpaca orders sync failed") + logger.warning("alpaca orders sync failed: %s", exc) self._emit_health(f"orders: {exc}") return None orders = [ diff --git a/tests/unit/test_execution_safety.py b/tests/unit/test_execution_safety.py index 91b5d8b..7f7115c 100644 --- a/tests/unit/test_execution_safety.py +++ b/tests/unit/test_execution_safety.py @@ -58,15 +58,23 @@ def test_partial_fill_is_recorded(): assert len(store.query_events(event_type="order_filled", limit=None)) == 1 -def test_breaker_fails_closed_when_equity_unreadable(): +def test_breaker_pauses_then_recovers_when_equity_unreadable(): + """Equity-read outage PAUSES new entries (recoverable), not a permanent halt: + a transient DNS/network blip must not end the trading day.""" store = EventStore(":memory:") + client = _FakeClient(raise_account=True) svc = TradingExecutionService( store, - executor=AlpacaPaperExecutor(store, client=_FakeClient(raise_account=True)), + executor=AlpacaPaperExecutor(store, client=client), settings=ExecutionSettings(max_daily_loss_pct=0.03), session_id="cb", ) + svc._equity_fail_limit = 3 # tighten for the test assert svc._daily_loss_breach() is False # 1 failure assert svc._daily_loss_breach() is False # 2 failures - assert svc._daily_loss_breach() is True # 3 -> fail CLOSED - assert svc._halted is True + assert svc._daily_loss_breach() is True # 3 -> PAUSE (recoverable) + assert svc._data_halt is True and svc._halted is False # NOT a permanent halt + # equity readable again -> resume + client.raise_account = False + assert svc._daily_loss_breach() is False + assert svc._data_halt is False diff --git a/trading_execution.py b/trading_execution.py index e0bb9c5..d26d553 100644 --- a/trading_execution.py +++ b/trading_execution.py @@ -169,9 +169,17 @@ class TradingExecutionService: # order_id -> {symbol, entry_price, broker_order_id, armed_at, checks} self._armed: dict[str, dict] = {} # daily-loss circuit breaker: once tripped, no new entries this session + # (a REAL loss is permanent for the session — a bounce can't re-open it) self._halted = False - # consecutive equity-read failures; after a few we fail CLOSED (halt) + # transient data-halt: equity unreadable (network/DNS). Blocks new entries + # WHILE unreadable but RECOVERS on the next good read — a DNS blip must not + # end the trading day the way a real loss does. + self._data_halt = False + # consecutive equity-read failures; after the limit we pause (recoverable) self._equity_read_failures = 0 + self._equity_fail_limit = int( + os.environ.get("TRADING_EQUITY_FAIL_LIMIT", "5") + ) # serializes state mutations across the main loop and the trigger thread self._lock = threading.RLock() @@ -254,12 +262,12 @@ class TradingExecutionService: @_locked def _daily_loss_breach(self) -> bool: - """True once the day's loss hits the circuit-breaker threshold. + """True while new entries should be blocked. - Compares broker account equity to the prior-session close - (account.last_equity). Once tripped, stays tripped for the session so a - bounce can't re-open the floodgates. If equity can't be read, does NOT - block (fail-open on a data hiccup, not on a real loss). + Two distinct halts: a REAL daily-loss breach (``_halted``) is permanent + for the session so a bounce can't re-open the floodgates; an equity-read + outage (``_data_halt``) blocks WHILE unreadable but recovers on the next + good read — a transient DNS/network blip must not end the trading day. """ if self._halted: return True @@ -276,17 +284,32 @@ class TradingExecutionService: raise ValueError("account 'equity' field missing") equity = float(equity_raw) baseline = float(acct.get("last_equity") or equity) - self._equity_read_failures = 0 + # good read -> clear the transient data-halt and resume + if self._data_halt or self._equity_read_failures: + if self._data_halt: + self.store.emit( + RiskRuleTriggeredEvent( + timestamp=datetime.now(), mode=self.mode, + correlation_id=self.session_id, + message="Equity readable again — resuming new entries", + rule_type="equity_recovered", rule_value=0.0, + current_state={"equity": equity}, + action_taken="resumed_new_entries", + ) + ) + self._data_halt = False + self._equity_read_failures = 0 except Exception as exc: # noqa: BLE001 self._equity_read_failures += 1 logger.warning( "equity read failed (%d consecutive): %s", self._equity_read_failures, exc, ) - # FAIL CLOSED: if equity is unreadable repeatedly, stop opening risk - # rather than silently disabling the most important safety control. - if self._equity_read_failures >= 3 and not self._halted: - self._halted = True + # PAUSE (recoverable): block new entries while equity is unreadable, + # rather than silently disabling the most important safety control — + # but DO NOT permanently halt the session for a transient outage. + if self._equity_read_failures >= self._equity_fail_limit and not self._data_halt: + self._data_halt = True self.store.emit( RiskRuleTriggeredEvent( timestamp=datetime.now(), @@ -294,15 +317,15 @@ class TradingExecutionService: correlation_id=self.session_id, message=( f"Equity unreadable {self._equity_read_failures}x — " - "failing CLOSED, halting new entries" + "pausing new entries until it recovers" ), rule_type="equity_unreadable", rule_value=float(self._equity_read_failures), current_state={"failures": self._equity_read_failures}, - action_taken="halted_new_entries", + action_taken="paused_new_entries", ) ) - return self._halted + return self._data_halt if baseline <= 0: return False pnl_pct = (equity - baseline) / baseline