The Night Everything Went Wrong
At 2:47 AM on March 15, 2020, a quant trader we'll call Marcus woke up to 47 notification alerts. His mean-reversion strategy—backtested over six years and running flawlessly for eight months—had executed 312 orders in the preceding six hours. Every single one was a loss. The portfolio had shed 23% of its equity in a single night.
The cause was not a flawed strategy. The cause was a regime change. COVID-19 had shattered every correlation model in the strategy's training set. But the strategy had no mechanism to detect that it was operating outside its designed conditions. It kept trading.
Marcus did not lose money because his strategy was bad. He lost money because his strategy had no circuit breaker.
This article is about fixing that gap. We will design and implement a production-grade circuit breaker using a state machine architecture—a design pattern borrowed from distributed systems engineering that translates surprisingly well to automated trading. The circuit breaker will handle three critical failure modes: consecutive loss counting, single-day drawdown threshold monitoring, and a human override interface for edge cases.
Why Consecutive Losses Are a Different Threat
Most traders monitor absolute drawdown. If your account is down 10%, you might receive an alert. But consecutive losses create a more insidious problem: volatility clustering in the wrong direction.
When a strategy enters a drawdown period, it often does so in streaks. The reasons vary:
- Regime shifts (like March 2020) invalidate the statistical assumptions baked into the strategy.
- Liquidity crises cause spreads to widen beyond what the strategy's cost model anticipated.
- Data corruption (stale quotes, misaligned timestamps) generates phantom signals.
- External shocks (news events, policy changes) create directional pressure that counter-trend strategies cannot survive.
In each case, the danger is not the individual loss. It is the accumulation of losses while the strategy remains active. A strategy that loses 1% per trade for ten consecutive trades is not experiencing ten isolated events—it is experiencing a systemic failure. The circuit breaker exists to interrupt that accumulation.
The Mathematics of Streak Risk
Consider a strategy with a 55% win rate. Each trade is independently assumed to be profitable with probability 0.55. The probability of experiencing five consecutive losses is:
P(5 losses) = (1 - 0.55)^5 = 0.45^5 ≈ 0.018%
Negligible. But a 55% win rate assumes stationarity—the strategy's edge is constant over time. During a regime shift, the win rate might collapse to 30%. At that point:
P(5 losses) = 0.70^5 ≈ 16.8%
A one-in-six chance of five consecutive losses becomes likely within days. The circuit breaker exists to catch this regime shift before the account is devastated.
The State Machine Architecture
A state machine is a computational model with a finite number of states, transitions between those states, and actions triggered by those transitions. For a circuit breaker, the state machine manages the strategy's operational mode based on its recent performance.
State Definitions
| State | Description | Strategy behavior |
|---|---|---|
| NORMAL | Strategy performing within expected parameters | Fully operational, all signals processed |
| WATCH | Consecutive losses detected, but within tolerance | Signals processed, but monitoring intensified |
| CIRCUIT_OPEN | Circuit breaker triggered | All new signals blocked, existing positions managed per exit rules |
| MANUAL_OVERRIDE | Human trader has taken control | Strategy remains paused, human manages positions |
| RECOVERY | Cooldown period completed, testing resumed | Strategy processes signals at reduced size or in paper mode |
State Transition Diagram
┌─────────────────────────────────────────┐
│ │
▼ │
┌──────────┐ consecutive_loss >= N ┌────────┐
┌────────▶│ NORMAL │────────────────────────────────▶│ WATCH │
│ └──────────┘ └────────┘
│ ▲ │
│ │ ▼
│ │ consecutive_loss >= 2N ┌─────────────────┐
│ │◀─────────────────────────────────────│ CIRCUIT_OPEN │
│ │ └─────────────────┘
│ │ │
│ │ cooldown_elapsed │
│ │◀─────────────────────────────────────────┤
│ │ │
│ │ ▼
│ │ ┌───────────┐
│ │ drawdown < threshold │ RECOVERY │
└───────────────┴────────────────────────────────────▶└───────────┘
│ │
│ manual_resume() │
└────────────────────────────────────────────┘
(from any state except NORMAL)
Transition Triggers
| Transition | Trigger condition | Action |
|---|---|---|
| NORMAL → WATCH | consecutive_loss >= 1 |
Increment watch counter, log warning |
| WATCH → NORMAL | consecutive_profit >= 2 |
Reset counters, log recovery |
| WATCH → CIRCUIT_OPEN | consecutive_loss >= N or daily_drawdown >= D |
Block all new signals, fire alert |
| CIRCUIT_OPEN → RECOVERY | cooldown_seconds_elapsed |
Allow paper-trading mode, resume at reduced size |
| RECOVERY → NORMAL | recovery_trades_profitable |
Restore full operation |
| ANY → MANUAL_OVERRIDE | manual_override_triggered() |
Freeze state machine, transfer control to human |
Production-Grade Implementation
The following implementation uses Python with a clean state machine abstraction. It is designed to be asset-class agnostic—you can integrate it with equity, futures, crypto, or forex strategies by passing the appropriate data feed.
Core State Machine Class
import os
import time
import json
import logging
import threading
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, Any
from datetime import datetime, timedelta
from collections import deque
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("circuit_breaker")
class CircuitState(Enum):
"""Enumeration of valid circuit breaker states."""
NORMAL = auto()
WATCH = auto()
CIRCUIT_OPEN = auto()
MANUAL_OVERRIDE = auto()
RECOVERY = auto()
@dataclass
class CircuitBreakerConfig:
"""
Configuration parameters for the circuit breaker.
All thresholds are expressed in strategy-native units (e.g.,
percentage of equity, number of contracts, etc.).
"""
consecutive_loss_threshold: int = 5 # N: losses before WATCH
severe_loss_threshold: int = 10 # 2N: losses before CIRCUIT_OPEN
daily_drawdown_threshold: float = 0.10 # D: 10% daily drawdown triggers breaker
cooldown_seconds: int = 3600 # 1 hour minimum cooldown
recovery_trade_count: int = 3 # Successful trades to exit RECOVERY
recovery_size_factor: float = 0.5 # Position size during RECOVERY (50%)
manual_override_timeout: int = 86400 # 24 hours before auto-resume from override
@dataclass
class CircuitMetrics:
"""Live metrics tracked by the circuit breaker."""
consecutive_loss_count: int = 0
consecutive_profit_count: int = 0
daily_pnl: float = 0.0
daily_peak_equity: float = 0.0
current_equity: float = 0.0
watch_enter_time: Optional[datetime] = None
circuit_open_time: Optional[datetime] = None
last_reset_time: datetime = field(default_factory=datetime.now)
recent_trades: deque = field(default_factory=lambda: deque(maxlen=50))
class CircuitBreaker:
"""
State machine-based circuit breaker for trading strategies.
This class monitors strategy performance and automatically halts
trading when loss thresholds are breached. It supports manual
override, configurable cooldown periods, and recovery modes.
Thread-safe for use in multi-threaded trading environments.
"""
def __init__(
self,
config: CircuitBreakerConfig,
alert_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None,
api_key: Optional[str] = None
):
self.config = config
self.metrics = CircuitMetrics()
self.state = CircuitState.NORMAL
self._lock = threading.RLock()
self._alert_callback = alert_callback
self._api_key = api_key or os.environ.get("TICKDB_API_KEY")
self._last_state_change = datetime.now()
self._recovery_trade_count = 0
logger.info(
f"CircuitBreaker initialized. "
f"Consecutive loss threshold: {config.consecutive_loss_threshold}, "
f"Drawdown threshold: {config.daily_drawdown_threshold:.1%}"
)
def record_trade(self, pnl: float, trade_id: str) -> bool:
"""
Record a completed trade and update circuit state.
Args:
pnl: Profit or loss from the trade (positive = profit, negative = loss)
trade_id: Unique identifier for the trade
Returns:
True if trading is permitted, False if blocked by circuit breaker
"""
with self._lock:
# Update metrics
self.metrics.daily_pnl += pnl
self.metrics.current_equity += pnl
self.metrics.peak_equity = max(
self.metrics.peak_equity,
self.metrics.current_equity
)
self.metrics.recent_trades.append({
"trade_id": trade_id,
"pnl": pnl,
"timestamp": datetime.now()
})
# Process based on current state
if pnl >= 0:
return self._handle_profit()
else:
return self._handle_loss()
def _handle_profit(self) -> bool:
"""Process a profitable trade."""
self.metrics.consecutive_profit_count += 1
self.metrics.consecutive_loss_count = 0
logger.debug(
f"Profit recorded. Consecutive profit count: "
f"{self.metrics.consecutive_profit_count}"
)
# Recovery from WATCH state
if self.state == CircuitState.WATCH:
if self.metrics.consecutive_profit_count >= 2:
self._transition_to(CircuitState.NORMAL)
return True
# Recovery from RECOVERY state
if self.state == CircuitState.RECOVERY:
self._recovery_trade_count += 1
if self._recovery_trade_count >= self.config.recovery_trade_count:
self._transition_to(CircuitState.NORMAL)
return True
return True
def _handle_loss(self) -> bool:
"""Process a losing trade."""
self.metrics.consecutive_loss_count += 1
self.metrics.consecutive_profit_count = 0
logger.warning(
f"Loss recorded. Consecutive loss count: "
f"{self.metrics.consecutive_loss_count}"
)
if self.state == CircuitState.NORMAL:
if self.metrics.consecutive_loss_count >= self.config.consecutive_loss_threshold:
self._transition_to(CircuitState.WATCH)
elif self.state == CircuitState.WATCH:
if self.metrics.consecutive_loss_count >= self.config.severe_loss_threshold:
self._transition_to(CircuitState.CIRCUIT_OPEN)
elif self.state == CircuitState.RECOVERY:
# Any loss during recovery immediately reopens circuit
self._transition_to(CircuitState.CIRCUIT_OPEN)
return self.state not in (CircuitState.CIRCUIT_OPEN, CircuitState.MANUAL_OVERRIDE)
def check_drawdown(self) -> bool:
"""
Evaluate whether the daily drawdown threshold has been breached.
Should be called at regular intervals (e.g., every minute).
Returns:
True if trading is permitted, False if blocked
"""
with self._lock:
if self.metrics.peak_equity == 0:
return True
current_drawdown = (
(self.metrics.peak_equity - self.metrics.current_equity)
/ self.metrics.peak_equity
)
if current_drawdown >= self.config.daily_drawdown_threshold:
logger.critical(
f"Daily drawdown threshold breached: {current_drawdown:.2%} "
f"(threshold: {self.config.daily_drawdown_threshold:.2%})"
)
self._transition_to(CircuitState.CIRCUIT_OPEN)
return False
return True
def manual_override(self, reason: str) -> None:
"""
Transition to MANUAL_OVERRIDE state, freezing automated trading.
Args:
reason: Human-readable explanation for the override
"""
with self._lock:
logger.warning(f"Manual override activated. Reason: {reason}")
self._transition_to(CircuitState.MANUAL_OVERRIDE)
self._send_alert(
"MANUAL_OVERRIDE",
{
"reason": reason,
"timestamp": datetime.now().isoformat(),
"metrics_snapshot": self._get_metrics_snapshot()
}
)
def resume_automated(self) -> bool:
"""
Attempt to resume automated trading from MANUAL_OVERRIDE.
Returns:
True if transition successful, False if not in override state
"""
with self._lock:
if self.state != CircuitState.MANUAL_OVERRIDE:
logger.error("Cannot resume: not in MANUAL_OVERRIDE state")
return False
logger.info("Manual override released, resuming circuit breaker")
self._transition_to(CircuitState.RECOVERY)
return True
def _transition_to(self, new_state: CircuitState) -> None:
"""Internal state transition handler."""
old_state = self.state
self.state = new_state
self._last_state_change = datetime.now()
# Reset recovery counter on new cycle
if new_state == CircuitState.RECOVERY:
self._recovery_trade_count = 0
# Log transition
transition_log = {
"from": old_state.name,
"to": new_state.name,
"timestamp": self._last_state_change.isoformat(),
"metrics": self._get_metrics_snapshot()
}
logger.info(f"State transition: {old_state.name} → {new_state.name}")
logger.debug(f"Transition details: {json.dumps(transition_log)}")
# Fire alert on critical transitions
if new_state in (CircuitState.CIRCUIT_OPEN, CircuitState.MANUAL_OVERRIDE):
self._send_alert(
f"STATE_{new_state.name}",
transition_log
)
def _send_alert(self, alert_type: str, data: Dict[str, Any]) -> None:
"""Send alert via callback or webhook."""
if self._alert_callback:
try:
self._alert_callback(alert_type, data)
except Exception as e:
logger.error(f"Alert callback failed: {e}")
# TODO: Integrate with TickDB webhook or Slack/email notification system
# Example webhook integration:
# if self._api_key:
# self._send_webhook_alert(alert_type, data)
def _get_metrics_snapshot(self) -> Dict[str, Any]:
"""Capture current metrics for logging and alerts."""
return {
"consecutive_loss_count": self.metrics.consecutive_loss_count,
"consecutive_profit_count": self.metrics.consecutive_profit_count,
"daily_pnl": self.metrics.daily_pnl,
"current_equity": self.metrics.current_equity,
"peak_equity": self.metrics.peak_equity,
"state_duration_seconds": (datetime.now() - self._last_state_change).total_seconds()
}
def get_state(self) -> CircuitState:
"""Return current circuit breaker state (thread-safe)."""
with self._lock:
return self.state
def get_position_size_multiplier(self) -> float:
"""
Return the position size multiplier based on current state.
Used by the strategy to adjust position sizing dynamically.
"""
with self._lock:
if self.state == CircuitState.RECOVERY:
return self.config.recovery_size_factor
elif self.state in (CircuitState.CIRCUIT_OPEN, CircuitState.MANUAL_OVERRIDE):
return 0.0
return 1.0
def reset_daily(self) -> None:
"""
Reset daily metrics. Should be called at the start of each trading day.
"""
with self._lock:
logger.info("Daily metrics reset")
self.metrics.daily_pnl = 0.0
self.metrics.peak_equity = self.metrics.current_equity
self.metrics.consecutive_loss_count = 0
self.metrics.consecutive_profit_count = 0
self.metrics.last_reset_time = datetime.now()
Integration Example: Strategy Wrapper
from dataclasses import dataclass
@dataclass
class TradeSignal:
"""A trading signal generated by a strategy."""
symbol: str
direction: int # 1 = long, -1 = short
size: float
confidence: float
signal_id: str
@dataclass
class TradeResult:
"""Result of an executed trade."""
trade_id: str
signal: TradeSignal
pnl: float
executed_size: float
execution_price: float
class CircuitProtectedStrategy:
"""
Wrapper that integrates a circuit breaker with any trading strategy.
The wrapper intercepts signals from the base strategy and only
passes them through when the circuit breaker permits trading.
"""
def __init__(
self,
base_strategy, # Your existing strategy object
circuit_breaker: CircuitBreaker,
initial_equity: float = 100000.0
):
self.strategy = base_strategy
self.circuit = circuit_breaker
self.equity = initial_equity
self._trade_counter = 0
# Inject equity tracker into circuit breaker
self.circuit.metrics.current_equity = initial_equity
self.circuit.metrics.peak_equity = initial_equity
def generate_and_filter_signals(self, market_data) -> list[TradeSignal]:
"""
Generate signals from the base strategy and filter through circuit breaker.
Args:
market_data: Current market data for signal generation
Returns:
List of TradeSignal objects that passed the circuit breaker filter
"""
state = self.circuit.get_state()
# Log current state for monitoring
logger.info(
f"Circuit state: {state.name}, "
f"position multiplier: {self.circuit.get_position_size_multiplier():.2f}"
)
# Block signal generation in restricted states
if state in (CircuitState.CIRCUIT_OPEN, CircuitState.MANUAL_OVERRIDE):
logger.warning(
f"Signals blocked: circuit in {state.name} state"
)
return []
# Generate signals from base strategy
raw_signals = self.strategy.generate_signals(market_data)
# Apply position size adjustment
size_multiplier = self.circuit.get_position_size_multiplier()
filtered_signals = []
for signal in raw_signals:
adjusted_signal = TradeSignal(
symbol=signal.symbol,
direction=signal.direction,
size=signal.size * size_multiplier,
confidence=signal.confidence,
signal_id=signal.signal_id
)
filtered_signals.append(adjusted_signal)
return filtered_signals
def record_result(self, result: TradeResult) -> None:
"""
Record a trade result with the circuit breaker.
Args:
result: TradeResult from executed trade
"""
# Update equity
self.equity += result.pnl
# Record with circuit breaker
trading_permitted = self.circuit.record_trade(result.pnl, result.trade_id)
logger.info(
f"Trade {result.trade_id} recorded: PnL={result.pnl:+.2f}, "
f"equity={self.equity:.2f}, "
f"trading_permitted={trading_permitted}"
)
# Update circuit breaker equity
self.circuit.metrics.current_equity = self.equity
def check_health(self) -> bool:
"""
Perform periodic health check (drawdown evaluation).
Should be called at regular intervals (e.g., every minute)
by a background monitoring task.
"""
return self.circuit.check_drawdown()
Alert Callback Implementation
import requests
import os
def slack_alert_callback(alert_type: str, data: dict) -> None:
"""
Send alerts to Slack via webhook.
Configure SLACK_WEBHOOK_URL in your environment.
"""
webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
if not webhook_url:
logger.warning("SLACK_WEBHOOK_URL not configured; skipping Slack alert")
return
# Color code by severity
color_map = {
"STATE_CIRCUIT_OPEN": "#FF0000", # Red
"STATE_MANUAL_OVERRIDE": "#FFA500", # Orange
"STATE_WATCH": "#FFFF00", # Yellow
"MANUAL_OVERRIDE": "#FFA500", # Orange
}
payload = {
"attachments": [{
"color": color_map.get(alert_type, "#808080"),
"title": f"Circuit Breaker Alert: {alert_type}",
"fields": [
{"title": k, "value": str(v), "short": False}
for k, v in data.items()
if k != "metrics_snapshot"
],
"footer": "TickDB Circuit Breaker",
"ts": data.get("timestamp", "")
}]
}
try:
response = requests.post(
webhook_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=(3.05, 10) # Timeout: 3.05s connect, 10s read
)
response.raise_for_status()
logger.info(f"Slack alert sent: {alert_type}")
except requests.exceptions.Timeout:
logger.error("Slack webhook timeout")
except requests.exceptions.RequestException as e:
logger.error(f"Slack webhook failed: {e}")
# Example usage
config = CircuitBreakerConfig(
consecutive_loss_threshold=5,
severe_loss_threshold=10,
daily_drawdown_threshold=0.08, # 8% daily drawdown
cooldown_seconds=1800, # 30-minute cooldown
recovery_trade_count=3,
recovery_size_factor=0.25 # 25% size during recovery
)
circuit_breaker = CircuitBreaker(
config=config,
alert_callback=slack_alert_callback
)
Configuration by Asset Class
Different asset classes have different volatility profiles and trading costs. The following table provides recommended starting configurations:
| Asset class | Consecutive loss threshold | Severe loss threshold | Daily drawdown | Cooldown |
|---|---|---|---|---|
| US Equities (intraday) | 5 | 10 | 8% | 30 min |
| Crypto (24/7) | 4 | 8 | 12% | 15 min |
| Futures | 3 | 6 | 6% | 60 min |
| Forex | 6 | 12 | 5% | 45 min |
These are starting points. Backtest each configuration against your specific strategy and adjust based on observed loss distributions during both normal operation and stress periods.
Deployment Checklist
Before deploying the circuit breaker to production:
- Backtest against historical drawdown periods — Stress test with March 2020, August 2015, COVID crash data.
- Set appropriate thresholds — Thresholds too tight generate false positives; too loose defeats the purpose.
- Configure alerts — Ensure alerts reach someone who can act, not just a dashboard.
- Test manual override flow — Simulate the override button being pressed at 3 AM.
- Verify cooldown logic — Ensure cooldown is long enough to observe regime change.
- Set recovery size factor — Never resume at 100% immediately; use 25–50% until confidence is restored.
- Monitor state transitions — Log every transition to support post-incident analysis.
What This Protects Against—and What It Does Not
The circuit breaker protects against:
- Regime shifts that invalidate strategy assumptions
- Data feed anomalies causing phantom signals
- Cascade losses during high-volatility events
- Human error in leaving an automated strategy unattended
The circuit breaker does not protect against:
- Single-trade losses exceeding the drawdown threshold (use per-trade stop-losses)
- Latent bugs in the strategy logic itself
- Exchange outages or fill failures
- Correlated positions that all exit simultaneously (liquidity crunch)
The circuit breaker is a last line of defense, not a substitute for sound strategy design.
Next Steps
If you're building systematic strategies, integrate this state machine framework into your execution layer. Start with conservative thresholds, backtest rigorously, and tighten only when you have statistical evidence.
If you want to stress-test your strategy against historical crash data, TickDB provides 10+ years of cleaned US equity OHLCV data via the /v1/market/kline endpoint, suitable for regime-aware strategy development and historical drawdown analysis.
If you need real-time monitoring alongside your circuit breaker, the TickDB WebSocket API supports sub-second market data delivery, which you can pipe into your strategy's signal generation pipeline.
If you're using AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to access integrated market data references directly in your development environment.
This article does not constitute investment advice. Markets involve risk; automated trading systems can incur substantial losses. Always validate circuit breaker configurations with thorough backtesting before live deployment.