Resilience API Reference

Circuit breaker and health check patterns.

financelib.resilience.CircuitBreaker(failure_threshold: int = 5, recovery_timeout: float = 30.0, name: str = 'default')

Circuit breaker for external API calls.

Tracks failures and temporarily blocks calls to failing services to prevent cascading failures and allow recovery.

Parameters:

Name	Type	Description	Default
failure_threshold	int	Number of consecutive failures before opening.	5
recovery_timeout	float	Seconds to wait before trying half-open.	30.0
name	str	Name for logging.	'default'

Example

cb = CircuitBreaker(failure_threshold=5, recovery_timeout=30) @cb ... def call_api(): ... return requests.get("https://api.example.com")

Source code in financelib/resilience.py

def __init__(
    self,
    failure_threshold: int = 5,
    recovery_timeout: float = 30.0,
    name: str = "default",
) -> None:
    self.failure_threshold = failure_threshold
    self.recovery_timeout = recovery_timeout
    self.name = name

    self._state = CircuitState.CLOSED
    self._failure_count = 0
    self._last_failure_time: float = 0.0
    self._lock = threading.Lock()

Attributes

failure_count: int property

Get the current failure count.

state: CircuitState property

Get the current circuit state.

Functions

__call__(func: F) -> F

Use as a decorator for protected function calls.

Source code in financelib/resilience.py

def __call__(self, func: F) -> F:
    """Use as a decorator for protected function calls."""
    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        current_state = self.state

        if current_state == CircuitState.OPEN:
            raise CircuitOpenError(
                f"Circuit '{self.name}' is OPEN — call blocked. "
                f"Recovery in {self.recovery_timeout - (time.time() - self._last_failure_time):.0f}s"
            )

        try:
            result = func(*args, **kwargs)
            self.record_success()
            return result
        except Exception:
            self.record_failure()
            raise

    return wrapper  # type: ignore[return-value]

record_failure() -> None

Record a failed call — may trip the circuit.

Source code in financelib/resilience.py

def record_failure(self) -> None:
    """Record a failed call — may trip the circuit."""
    with self._lock:
        self._failure_count += 1
        self._last_failure_time = time.time()
        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN
            logger.warning(
                f"Circuit '{self.name}' → OPEN "
                f"(failures: {self._failure_count}/{self.failure_threshold})"
            )

record_success() -> None

Record a successful call — resets failure count.

Source code in financelib/resilience.py

def record_success(self) -> None:
    """Record a successful call — resets failure count."""
    with self._lock:
        self._failure_count = 0
        if self._state != CircuitState.CLOSED:
            logger.info(f"Circuit '{self.name}' → CLOSED (recovered)")
        self._state = CircuitState.CLOSED

reset() -> None

Manually reset the circuit breaker.

Source code in financelib/resilience.py

def reset(self) -> None:
    """Manually reset the circuit breaker."""
    with self._lock:
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._last_failure_time = 0.0

financelib.resilience.CircuitState

Bases: Enum

Circuit breaker states.

financelib.resilience.CircuitOpenError

Bases: Exception

Raised when a call is blocked by an open circuit breaker.

financelib.resilience.HealthCheck(name: str, check_fn: Callable[[], bool], interval: float = 0)

Simple health check for external services.

Parameters:

Name	Type	Description	Default
name	str	Service name.	required
check_fn	Callable[[], bool]	Callable that returns True if service is healthy.	required
interval	float	Seconds between automatic checks (0 = manual only).	0

Example

def check_binance(): ... return requests.get("https://api.binance.com/api/v3/ping").ok health = HealthCheck("binance", check_binance) print(health.is_healthy())

Source code in financelib/resilience.py

def __init__(
    self,
    name: str,
    check_fn: Callable[[], bool],
    interval: float = 0,
) -> None:
    self.name = name
    self._check_fn = check_fn
    self.interval = interval
    self._healthy: Optional[bool] = None
    self._last_check: float = 0.0
    self._lock = threading.Lock()

Attributes

status: Dict property

Get health status as a dictionary.

Functions

is_healthy() -> bool

Check if the service is healthy.

Returns cached result if within the check interval.

Source code in financelib/resilience.py

def is_healthy(self) -> bool:
    """Check if the service is healthy.

    Returns cached result if within the check interval.
    """
    with self._lock:
        now = time.time()
        if self._healthy is None or (self.interval > 0 and now - self._last_check >= self.interval):
            try:
                self._healthy = self._check_fn()
            except Exception:
                self._healthy = False
            self._last_check = now
        return self._healthy or False