gavindiazandClaude Code
be65ef441b
Polling signal loop with a state machine, circuit breaker, layered dry-run execution (subprocess mtdata-cli trade_*) and a JSONL journal. The default strategy emits no signal, so the loop is safe to run as-is; orders never reach the broker until a strategy and the live tier are explicitly wired up. Co-Authored-By: Claude Code <noreply@anthropic.com>
137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""Circuit breaker to halt the signal loop after repeated failures.
|
|
|
|
States:
|
|
CLOSED normal operation; requests allowed
|
|
OPEN too many consecutive failures; requests rejected for a cooldown
|
|
HALF_OPEN cooldown elapsed; one probe request is allowed to test recovery
|
|
|
|
Failures that count: CLI crashes, guardrail blocks, timeout, connection errors.
|
|
A single success in HALF_OPEN closes the breaker; a failure re-opens it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
class CircuitBreaker:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
failure_threshold: int = 5,
|
|
cooldown_seconds: float = 300.0,
|
|
state_file: Optional[Path] = None,
|
|
) -> None:
|
|
if failure_threshold < 1:
|
|
raise ValueError("failure_threshold must be >= 1")
|
|
self.failure_threshold = failure_threshold
|
|
self.cooldown_seconds = cooldown_seconds
|
|
self.state_file = Path(state_file) if state_file else None
|
|
self._state = "CLOSED"
|
|
self._consecutive_failures = 0
|
|
self._opened_at: Optional[float] = None
|
|
self._half_open_since: Optional[float] = None
|
|
self._last_error: Optional[str] = None
|
|
self._probe_in_flight = False
|
|
if self.state_file and self.state_file.exists():
|
|
self._load()
|
|
|
|
# --- persistence ------------------------------------------------------------
|
|
def _load(self) -> None:
|
|
try:
|
|
raw = json.loads(self.state_file.read_text(encoding="utf-8"))
|
|
self._state = raw.get("state", "CLOSED")
|
|
self._consecutive_failures = int(raw.get("consecutive_failures", 0))
|
|
self._opened_at = raw.get("opened_at")
|
|
self._half_open_since = raw.get("half_open_since")
|
|
self._last_error = raw.get("last_error")
|
|
except (json.JSONDecodeError, OSError, ValueError):
|
|
pass # corrupt file -> start fresh and closed
|
|
|
|
def _save(self) -> None:
|
|
if not self.state_file:
|
|
return
|
|
self.state_file.parent.mkdir(parents=True, exist_ok=True)
|
|
snapshot = {
|
|
"state": self._state,
|
|
"consecutive_failures": self._consecutive_failures,
|
|
"opened_at": self._opened_at,
|
|
"half_open_since": self._half_open_since,
|
|
"last_error": self._last_error,
|
|
}
|
|
self.state_file.write_text(
|
|
json.dumps(snapshot, indent=2, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
|
|
# --- gate ------------------------------------------------------------------
|
|
def allow_request(self) -> bool:
|
|
if self._state == "CLOSED":
|
|
return True
|
|
if self._state == "OPEN":
|
|
assert self._opened_at is not None
|
|
if time.time() - self._opened_at >= self.cooldown_seconds:
|
|
self._state = "HALF_OPEN"
|
|
self._half_open_since = time.time()
|
|
self._save()
|
|
return True # one probe
|
|
return False
|
|
# HALF_OPEN: allow the single probe only if no probe is in flight
|
|
return not self._probe_in_flight
|
|
|
|
def begin_probe(self) -> None:
|
|
"""Call right before issuing the request granted in HALF_OPEN."""
|
|
self._probe_in_flight = True
|
|
|
|
def record_success(self) -> None:
|
|
if self._state == "HALF_OPEN":
|
|
self._close()
|
|
else:
|
|
self._consecutive_failures = 0
|
|
self._probe_in_flight = False
|
|
self._save()
|
|
|
|
def record_failure(self, error: str) -> None:
|
|
self._consecutive_failures += 1
|
|
self._last_error = error
|
|
if self._state == "HALF_OPEN":
|
|
self._open(error)
|
|
elif self._consecutive_failures >= self.failure_threshold:
|
|
self._open(error)
|
|
self._probe_in_flight = False
|
|
self._save()
|
|
|
|
def reset(self) -> None:
|
|
self._state = "CLOSED"
|
|
self._consecutive_failures = 0
|
|
self._opened_at = None
|
|
self._half_open_since = None
|
|
self._last_error = None
|
|
self._probe_in_flight = False
|
|
self._save()
|
|
|
|
# --- internals ---------------------------------------------------------------
|
|
def _open(self, error: str) -> None:
|
|
self._state = "OPEN"
|
|
self._opened_at = time.time()
|
|
self._half_open_since = None
|
|
self._last_error = error
|
|
|
|
def _close(self) -> None:
|
|
self._state = "CLOSED"
|
|
self._consecutive_failures = 0
|
|
self._opened_at = None
|
|
self._half_open_since = None
|
|
self._last_error = None
|
|
|
|
def snapshot(self) -> dict:
|
|
return {
|
|
"state": self._state,
|
|
"consecutive_failures": self._consecutive_failures,
|
|
"opened_at": self._opened_at,
|
|
"half_open_since": self._half_open_since,
|
|
"last_error": self._last_error,
|
|
}
|