Quant Trading Systems: Architecture & Infrastructure › Robustness and Testing
Quantitative FinanceAll chapters Quant Trading Systems: Architecture & Infrastructure › Robustness and TestingOne section · Sections in this chapter10 / 14 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Production trading systems face adversarial conditions: data feeds fail, networks partition, and bugs lurk in rarely-executed code paths. Building systems that fail safely requires defensive design, broad test coverage, and fail-safe mechanisms. The financial consequences of system failures can be severe, from missed trading opportunities to catastrophic losses.
Robustness is not a feature that can be added after development; it must be designed in from the beginning. Every component should be designed to fail gracefully, every external dependency should be monitored, and every code path should be tested.
Circuit Breakers and Kill Switches
Circuit breakers automatically halt or reduce trading when anomalies are detected. They serve as the last line of defense against runaway losses, protecting the firm from scenarios where automated trading goes wrong.
The concept comes from electrical engineering, where circuit breakers prevent fires by cutting power when current exceeds safe levels. In trading, circuit breakers cut trading activity when metrics exceed safe levels. The goal is the same: prevent a small problem from becoming a catastrophe.
class CircuitBreaker:
"""Monitors trading activity and triggers circuit breakers when limits are breached."""
def __init__(self):
self.trading_enabled = True
self.breach_reasons: list[str] = []
# Configurable limits
self.max_orders_per_second = 100
self.max_daily_loss = 50000
self.max_position_notional = 1000000
self.max_drawdown_pct = 0.05
# Tracking state
self.orders_this_second = 0
self.last_second = 0
self.daily_pnl = 0
self.peak_nav = 0
def check_rate_limit(self, current_time: float) -> bool:
"""Check if order rate exceeds limit."""
current_second = int(current_time)
if current_second != self.last_second:
self.orders_this_second = 0
self.last_second = current_second
self.orders_this_second += 1
if self.orders_this_second > self.max_orders_per_second:
self._trip_breaker(
f"Order rate exceeded: {self.orders_this_second}/sec"
)
return False
return True
def check_daily_loss(self, current_pnl: float) -> bool:
"""Check if daily loss limit is breached."""
if current_pnl < -self.max_daily_loss:
self._trip_breaker(f"Daily loss limit breached: ${current_pnl:.2f}")
return False
return True
def check_drawdown(self, current_nav: float) -> bool:
"""Check if drawdown exceeds limit."""
if current_nav > self.peak_nav:
self.peak_nav = current_nav
drawdown = (self.peak_nav - current_nav) / self.peak_nav
if drawdown > self.max_drawdown_pct:
self._trip_breaker(f"Drawdown exceeded: {drawdown:.2%}")
return False
return True
def check_position_size(self, position_notional: float) -> bool:
"""Check if position size exceeds limit."""
if abs(position_notional) > self.max_position_notional:
self._trip_breaker(
f"Position limit exceeded: ${position_notional:.2f}"
)
return False
return True
def _trip_breaker(self, reason: str):
"""Trip the circuit breaker and disable trading."""
self.trading_enabled = False
self.breach_reasons.append(reason)
print(f"⚠️ CIRCUIT BREAKER TRIPPED: {reason}")
def reset(self, password: str = None):
"""Reset circuit breaker (requires manual intervention in production)."""
if password == "CONFIRM_RESET": # Simple safety check
self.trading_enabled = True
self.breach_reasons = []
print("Circuit breaker reset")
else:
print("Reset requires confirmation password")
def can_trade(self) -> bool:
"""Check if trading is currently allowed."""
return self.trading_enabled# Demonstrate circuit breaker behavior
cb = CircuitBreaker()
cb.max_daily_loss = 10000
cb.max_drawdown_pct = 0.03
# Simulate trading day
scenarios = [
{"name": "Normal trading", "nav": 1000000, "daily_pnl": 5000},
{"name": "Small loss", "nav": 995000, "daily_pnl": -5000},
{"name": "Approaching limit", "nav": 992000, "daily_pnl": -8000},
{"name": "Loss limit breach", "nav": 988000, "daily_pnl": -12000},
]
log_entries = []
for scenario in scenarios:
if cb.can_trade():
cb.check_daily_loss(scenario["daily_pnl"])
cb.check_drawdown(scenario["nav"])
status = "✓ Trading enabled" if cb.can_trade() else "✗ Trading HALTED"
log_entries.append(
f"{scenario['name']:20s} | P&L: ${scenario['daily_pnl']:>8,.0f} | {status}"
)Circuit Breaker Demo ================================================== Normal trading | P&L: $ 5,000 | ✓ Trading enabled Small loss | P&L: $ -5,000 | ✓ Trading enabled Approaching limit | P&L: $ -8,000 | ✓ Trading enabled Loss limit breach | P&L: $ -12,000 | ✗ Trading HALTED Breach reasons: ['Daily loss limit breached: $-12000.00']
The circuit breaker monitors the trading activity and halts the system when the daily loss limit is breached in the final scenario. Notice that once the breaker trips, it stays tripped. Resetting requires explicit action (simulated here by the password check), so a human reviews the situation before trading resumes.
Testing Strategies
Testing validates that your system behaves correctly under normal and failure conditions. Testing is not a phase that happens after development; it's an ongoing practice that shapes how code is written and systems are designed.
Unit tests verify individual components in isolation. Each function should have tests covering normal inputs, edge cases, and error conditions. Unit tests run quickly and catch bugs early in development.
Integration tests verify that components work together correctly. Test the full flow from data ingestion through order generation. Integration tests catch issues that arise from component interactions, such as data format mismatches or timing dependencies.
Simulation tests run your system against historical data with simulated execution. Compare results to your backtests to ensure consistency. Simulation tests verify that the production system implements the same logic as the research backtester.
Chaos testing deliberately injects failures to verify graceful degradation. What happens when a data feed disconnects mid-day? When an order is rejected? When prices gap? Chaos testing reveals weaknesses in error handling that normal testing doesn't expose.
import random
class ChaosTestingFramework:
"""Framework for injecting failures to test system robustness."""
def __init__(self, failure_probability: float = 0.1):
self.failure_probability = failure_probability
self.failure_log: list[dict] = []
def maybe_fail_data_feed(self) -> tuple[bool, Optional[str]]:
"""Simulate potential data feed failure."""
if random.random() < self.failure_probability:
failure_type = random.choice(
[
"connection_lost",
"stale_data",
"garbled_message",
"delayed_data",
]
)
self.failure_log.append(
{"component": "data_feed", "failure": failure_type}
)
return True, failure_type
return False, None
def maybe_fail_order_submission(self) -> tuple[bool, Optional[str]]:
"""Simulate potential order submission failure."""
if random.random() < self.failure_probability:
failure_type = random.choice(
[
"timeout",
"rejected_invalid_price",
"rejected_insufficient_margin",
"venue_unavailable",
]
)
self.failure_log.append(
{"component": "execution", "failure": failure_type}
)
return True, failure_type
return False, None
def inject_price_anomaly(self, price: float) -> float:
"""Potentially inject anomalous price."""
if random.random() < self.failure_probability / 2:
anomaly_type = random.choice(["spike", "zero", "negative"])
if anomaly_type == "spike":
return price * random.uniform(1.5, 3.0)
elif anomaly_type == "zero":
return 0
else:
return -price
return price# Demonstrate chaos testing
random.seed(42)
chaos = ChaosTestingFramework(failure_probability=0.2)
# Simulate 20 trading operations
operations = ["data_update", "order_submit", "price_check"]
failure_count = 0
success_count = 0
failed_ops = []
for i in range(20):
op = random.choice(operations)
if op == "data_update":
failed, failure_type = chaos.maybe_fail_data_feed()
elif op == "order_submit":
failed, failure_type = chaos.maybe_fail_order_submission()
else:
price = 100.0
new_price = chaos.inject_price_anomaly(price)
failed = new_price != price
failure_type = f"price_anomaly ({new_price:.2f})" if failed else None
if failed:
failure_count += 1
failed_ops.append(
f" Op {i + 1:2d}: {op:15s} -> FAILED: {failure_type}"
)
else:
success_count += 1
obs_failure_rate = failure_count / 20Chaos Testing Simulation (20% failure rate) ============================================================ Op 3: data_update -> FAILED: connection_lost Op 5: price_check -> FAILED: price_anomaly (0.00) Op 6: data_update -> FAILED: stale_data Op 16: order_submit -> FAILED: venue_unavailable Results: 16 successful, 4 failures Observed failure rate: 20.0%
The chaos testing framework injects simulated failures, allowing us to verify how the system handles connection drops and order rejections. Running chaos tests regularly uncovers edge cases in error handling code that rarely executes under normal conditions.
Data Validation and Sanity Checks
Never trust incoming data without validation. Market data contains errors more often than you might expect. Exchange systems have bugs, network errors corrupt packets, and data vendors occasionally publish incorrect values.
The consequences of trading on bad data can be severe. An erroneous price might trigger large orders in the wrong direction, or risk calculations might fail to recognize dangerous positions. Input validation catches these issues before they affect trading decisions.
class DataValidator:
"""Validates market data for common anomalies."""
def __init__(self, symbol: str):
self.symbol = symbol
self.last_price: Optional[float] = None
self.price_history: list[float] = []
self.max_price_change_pct = 0.10 # 10% max single-tick move
def validate_price(self, price: float) -> tuple[bool, Optional[str]]:
"""Validate a price update."""
# Check for invalid values
if price <= 0:
return False, "Non-positive price"
if np.isnan(price) or np.isinf(price):
return False, "NaN or Inf price"
# Check for unreasonable change
if self.last_price is not None:
change_pct = abs(price - self.last_price) / self.last_price
if change_pct > self.max_price_change_pct:
return (
False,
f"Price change {change_pct:.1%} exceeds {self.max_price_change_pct:.1%}",
)
# Update state
self.price_history.append(price)
self.last_price = price
return True, None
def validate_quote(
self, bid: float, ask: float
) -> tuple[bool, Optional[str]]:
"""Validate a quote update."""
# Check for crossed market
if bid >= ask:
return False, f"Crossed market: bid {bid} >= ask {ask}"
# Check for unreasonable spread
spread_pct = (ask - bid) / ((ask + bid) / 2)
if spread_pct > 0.05: # 5% spread
return False, f"Wide spread: {spread_pct:.1%}"
return True, None# Demonstrate data validation
validator = DataValidator("AAPL")
test_data = [
("price", 185.50),
("price", 185.75),
("price", 0), # Invalid: zero
("price", 185.80),
("price", 220.00), # Invalid: too large a jump
("price", 186.00),
("quote", (185.50, 185.52)), # Valid quote
("quote", (185.55, 185.50)), # Invalid: crossed
]
validation_log = []
for test in test_data:
data_type = test[0]
value = test[1]
if data_type == "price":
valid, reason = validator.validate_price(value)
validation_log.append(
f"Price {value:>8.2f}: {'✓ Valid' if valid else f'✗ {reason}'}"
)
else:
bid, ask = value
valid, reason = validator.validate_quote(bid, ask)
validation_log.append(
f"Quote {bid:.2f}/{ask:.2f}: {'✓ Valid' if valid else f'✗ {reason}'}"
)Data Validation Results: ------------------------------------------------------------ Price 185.50: ✓ Valid Price 185.75: ✓ Valid Price 0.00: ✗ Non-positive price Price 185.80: ✓ Valid Price 220.00: ✗ Price change 18.4% exceeds 10.0% Price 186.00: ✓ Valid Quote 185.50/185.52: ✓ Valid Quote 185.55/185.50: ✗ Crossed market: bid 185.55 >= ask 185.5
The validator flags invalid prices and crossed markets, preventing bad data from triggering erroneous trading signals. Notice that after rejecting the zero price, the validator continues to work correctly for subsequent valid prices. This resilience is essential; a single bad data point should not permanently break the system.