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 <noreply@anthropic.com>
This commit is contained in:
1 parent
9cc02e01ac
commit
45d5484dd7
4 files changed
+65
-21
No files matched your search
+14
-1
@@ -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
|
||||
)
|
||||
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(
|
||||
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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
|
||||
+36
-13
@@ -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,6 +284,20 @@ class TradingExecutionService:
|
||||
raise ValueError("account 'equity' field missing")
|
||||
equity = float(equity_raw)
|
||||
baseline = float(acct.get("last_equity") or equity)
|
||||
# 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
|
||||
@@ -283,10 +305,11 @@ class TradingExecutionService:
|
||||
"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
|
||||
|
||||
Reference in new issue
Block a user