Celes Renata
c85c0068a2
- Replace all datetime.utcnow() with datetime.now(tz=timezone.utc) across 8 files - Fix 12 failing tests to match current implementation behavior - Fix pytest_plugins in non-top-level conftest (moved to root conftest.py) - Auto-fix 189 lint issues (import sorting, unused imports) - Add CI/CD pipeline infrastructure (ARC, ArgoCD, Kargo manifests) - Add values-beta.yaml and values-paper.yaml for staged deployments - Update GitHub Actions workflow to use self-hosted-gremlin runners - Add integration-test job to CI pipeline Result: 1596 passed, 0 failed, 0 warnings
422 lines
17 KiB
Python
422 lines
17 KiB
Python
"""Property-based tests for the Circuit Breaker.
|
|
|
|
Feature: autonomous-trading-engine
|
|
|
|
Tests properties 13 and 14 from the design specification,
|
|
covering circuit breaker activation triggers (daily loss, single position,
|
|
volatility) and cooldown expiry behaviour.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from hypothesis import assume, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.trading.circuit_breaker import CircuitBreaker
|
|
from services.trading.models import CircuitBreakerState
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hypothesis strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _circuit_breaker_strategy() -> st.SearchStrategy[CircuitBreaker]:
|
|
"""Generate CircuitBreaker instances with random but valid thresholds."""
|
|
return st.builds(
|
|
CircuitBreaker,
|
|
daily_loss_pct=st.floats(min_value=0.01, max_value=0.20, allow_nan=False, allow_infinity=False),
|
|
single_position_loss_pct=st.floats(min_value=0.05, max_value=0.50, allow_nan=False, allow_infinity=False),
|
|
ticker_cooldown_hours=st.integers(min_value=1, max_value=168),
|
|
volatility_pause_hours=st.integers(min_value=1, max_value=24),
|
|
stop_loss_hits_threshold=st.integers(min_value=2, max_value=10),
|
|
stop_loss_window_minutes=st.integers(min_value=5, max_value=120),
|
|
)
|
|
|
|
|
|
def _aware_datetime_strategy(
|
|
min_dt: datetime | None = None,
|
|
max_dt: datetime | None = None,
|
|
) -> st.SearchStrategy[datetime]:
|
|
"""Generate timezone-aware UTC datetimes."""
|
|
_min = min_dt or datetime(2024, 1, 1, tzinfo=timezone.utc)
|
|
_max = max_dt or datetime(2025, 12, 31, tzinfo=timezone.utc)
|
|
return st.datetimes(min_value=_min.replace(tzinfo=None), max_value=_max.replace(tzinfo=None)).map(
|
|
lambda dt: dt.replace(tzinfo=timezone.utc)
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 13: Circuit breaker activation
|
|
# **Validates: Requirements 6.1, 6.2, 6.3**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty13CircuitBreakerActivation:
|
|
"""Property 13: Circuit breaker activation.
|
|
|
|
**Validates: Requirements 6.1, 6.2, 6.3**
|
|
"""
|
|
|
|
# -- Daily loss trigger ------------------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
portfolio_value=st.floats(min_value=100.0, max_value=100000.0, allow_nan=False, allow_infinity=False),
|
|
excess_pct=st.floats(min_value=0.001, max_value=0.50, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_daily_loss_triggers_when_loss_exceeds_threshold(
|
|
self, cb: CircuitBreaker, portfolio_value: float, excess_pct: float,
|
|
) -> None:
|
|
"""Daily loss circuit breaker triggers when abs(daily_pnl)/portfolio_value > daily_loss_pct."""
|
|
# Construct a daily_pnl that exceeds the threshold
|
|
loss_ratio = cb.daily_loss_pct + excess_pct
|
|
daily_pnl = -(loss_ratio * portfolio_value)
|
|
|
|
result = cb.check_daily_loss(daily_pnl=daily_pnl, portfolio_value=portfolio_value)
|
|
assert result is True, (
|
|
f"Expected daily_loss trigger: loss_ratio={loss_ratio:.4f} > threshold={cb.daily_loss_pct:.4f}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
portfolio_value=st.floats(min_value=100.0, max_value=100000.0, allow_nan=False, allow_infinity=False),
|
|
below_fraction=st.floats(min_value=0.0, max_value=0.99, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_daily_loss_does_not_trigger_below_threshold(
|
|
self, cb: CircuitBreaker, portfolio_value: float, below_fraction: float,
|
|
) -> None:
|
|
"""Daily loss circuit breaker does NOT trigger when loss ratio <= threshold."""
|
|
# Loss ratio strictly below threshold
|
|
loss_ratio = cb.daily_loss_pct * below_fraction
|
|
daily_pnl = -(loss_ratio * portfolio_value)
|
|
|
|
result = cb.check_daily_loss(daily_pnl=daily_pnl, portfolio_value=portfolio_value)
|
|
assert result is False, (
|
|
f"Should not trigger: loss_ratio={loss_ratio:.4f} <= threshold={cb.daily_loss_pct:.4f}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
portfolio_value=st.floats(min_value=100.0, max_value=100000.0, allow_nan=False, allow_infinity=False),
|
|
profit=st.floats(min_value=0.01, max_value=10000.0, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_daily_loss_does_not_trigger_on_positive_pnl(
|
|
self, cb: CircuitBreaker, portfolio_value: float, profit: float,
|
|
) -> None:
|
|
"""Daily loss circuit breaker never triggers on positive P&L."""
|
|
result = cb.check_daily_loss(daily_pnl=profit, portfolio_value=portfolio_value)
|
|
assert result is False, "Should not trigger on positive daily P&L"
|
|
|
|
# -- Single position trigger -------------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
excess=st.floats(min_value=0.001, max_value=0.50, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_single_position_triggers_when_loss_exceeds_threshold(
|
|
self, cb: CircuitBreaker, excess: float,
|
|
) -> None:
|
|
"""Single position circuit breaker triggers when position_loss_pct > threshold."""
|
|
position_loss_pct = cb.single_position_loss_pct + excess
|
|
|
|
result = cb.check_single_position(position_loss_pct=position_loss_pct)
|
|
assert result is True, (
|
|
f"Expected single_position trigger: loss={position_loss_pct:.4f} > threshold={cb.single_position_loss_pct:.4f}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
below_fraction=st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_single_position_does_not_trigger_at_or_below_threshold(
|
|
self, cb: CircuitBreaker, below_fraction: float,
|
|
) -> None:
|
|
"""Single position circuit breaker does NOT trigger when loss <= threshold."""
|
|
position_loss_pct = cb.single_position_loss_pct * below_fraction
|
|
|
|
result = cb.check_single_position(position_loss_pct=position_loss_pct)
|
|
assert result is False, (
|
|
f"Should not trigger: loss={position_loss_pct:.4f} <= threshold={cb.single_position_loss_pct:.4f}"
|
|
)
|
|
|
|
# -- Volatility trigger ------------------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
base_time=_aware_datetime_strategy(),
|
|
)
|
|
def test_volatility_triggers_when_enough_stops_within_window(
|
|
self, cb: CircuitBreaker, base_time: datetime,
|
|
) -> None:
|
|
"""Volatility circuit breaker triggers when >= threshold stop-losses fire within the window."""
|
|
# Generate exactly threshold hits within the window
|
|
window_minutes = cb.stop_loss_window_minutes
|
|
n_hits = cb.stop_loss_hits_threshold
|
|
|
|
# Space hits evenly within the window
|
|
interval = timedelta(minutes=window_minutes / n_hits)
|
|
stop_loss_hits = [base_time + interval * i for i in range(n_hits)]
|
|
|
|
result = cb.check_volatility(stop_loss_hits=stop_loss_hits)
|
|
assert result is True, (
|
|
f"Expected volatility trigger: {n_hits} hits within {window_minutes}min window"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
base_time=_aware_datetime_strategy(),
|
|
)
|
|
def test_volatility_does_not_trigger_when_too_few_stops(
|
|
self, cb: CircuitBreaker, base_time: datetime,
|
|
) -> None:
|
|
"""Volatility circuit breaker does NOT trigger with fewer than threshold hits."""
|
|
n_hits = cb.stop_loss_hits_threshold - 1
|
|
assume(n_hits >= 0)
|
|
|
|
# Even if all within the window, not enough hits
|
|
stop_loss_hits = [base_time + timedelta(minutes=i) for i in range(n_hits)]
|
|
|
|
result = cb.check_volatility(stop_loss_hits=stop_loss_hits)
|
|
assert result is False, (
|
|
f"Should not trigger: only {n_hits} hits < threshold {cb.stop_loss_hits_threshold}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
base_time=_aware_datetime_strategy(),
|
|
)
|
|
def test_volatility_does_not_trigger_when_stops_spread_outside_window(
|
|
self, cb: CircuitBreaker, base_time: datetime,
|
|
) -> None:
|
|
"""Volatility circuit breaker does NOT trigger when hits are spread beyond the window."""
|
|
n_hits = cb.stop_loss_hits_threshold
|
|
window_minutes = cb.stop_loss_window_minutes
|
|
|
|
# Space hits so that no contiguous sub-sequence of threshold size fits in the window
|
|
# Each hit is window_minutes + 1 apart, so any threshold-sized group spans
|
|
# (threshold - 1) * (window_minutes + 1) minutes > window_minutes
|
|
gap = timedelta(minutes=window_minutes + 1)
|
|
stop_loss_hits = [base_time + gap * i for i in range(n_hits)]
|
|
|
|
result = cb.check_volatility(stop_loss_hits=stop_loss_hits)
|
|
assert result is False, (
|
|
f"Should not trigger: hits spread {window_minutes + 1}min apart, "
|
|
f"window is {window_minutes}min"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
)
|
|
def test_volatility_does_not_trigger_on_empty_hits(
|
|
self, cb: CircuitBreaker,
|
|
) -> None:
|
|
"""Volatility circuit breaker does NOT trigger with no stop-loss hits."""
|
|
result = cb.check_volatility(stop_loss_hits=[])
|
|
assert result is False, "Should not trigger on empty stop-loss hits list"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 14: Circuit breaker cooldown expiry
|
|
# **Validates: Requirements 6.5**
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProperty14CircuitBreakerCooldownExpiry:
|
|
"""Property 14: Circuit breaker cooldown expiry.
|
|
|
|
**Validates: Requirements 6.5**
|
|
"""
|
|
|
|
# -- is_active cooldown expiry -----------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
extra_seconds=st.integers(min_value=1, max_value=86400),
|
|
)
|
|
def test_is_active_returns_false_after_cooldown_expires(
|
|
self, cb: CircuitBreaker, triggered_at: datetime, extra_seconds: int,
|
|
) -> None:
|
|
"""is_active returns False when current time > cooldown_expires."""
|
|
cooldown_expires = triggered_at + timedelta(hours=cb.volatility_pause_hours)
|
|
now = cooldown_expires + timedelta(seconds=extra_seconds)
|
|
|
|
state = CircuitBreakerState(
|
|
active=True,
|
|
trigger_type="daily_loss",
|
|
triggered_at=triggered_at,
|
|
cooldown_expires=cooldown_expires,
|
|
)
|
|
|
|
result = cb.is_active(state=state, now=now)
|
|
assert result is False, (
|
|
f"Expected inactive: now={now} > cooldown_expires={cooldown_expires}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
before_fraction=st.floats(min_value=0.0, max_value=0.99, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_is_active_returns_true_before_cooldown_expires(
|
|
self, cb: CircuitBreaker, triggered_at: datetime, before_fraction: float,
|
|
) -> None:
|
|
"""is_active returns True when current time < cooldown_expires."""
|
|
cooldown_duration = timedelta(hours=cb.volatility_pause_hours)
|
|
cooldown_expires = triggered_at + cooldown_duration
|
|
# now is some fraction of the way through the cooldown (before expiry)
|
|
now = triggered_at + cooldown_duration * before_fraction
|
|
|
|
assume(now < cooldown_expires)
|
|
|
|
state = CircuitBreakerState(
|
|
active=True,
|
|
trigger_type="volatility",
|
|
triggered_at=triggered_at,
|
|
cooldown_expires=cooldown_expires,
|
|
)
|
|
|
|
result = cb.is_active(state=state, now=now)
|
|
assert result is True, (
|
|
f"Expected active: now={now} < cooldown_expires={cooldown_expires}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
)
|
|
def test_is_active_returns_false_when_state_not_active(
|
|
self, cb: CircuitBreaker, triggered_at: datetime,
|
|
) -> None:
|
|
"""is_active returns False when state.active is False regardless of time."""
|
|
state = CircuitBreakerState(
|
|
active=False,
|
|
trigger_type=None,
|
|
triggered_at=triggered_at,
|
|
cooldown_expires=triggered_at + timedelta(hours=24),
|
|
)
|
|
|
|
result = cb.is_active(state=state, now=triggered_at)
|
|
assert result is False, "Expected inactive when state.active is False"
|
|
|
|
# -- is_ticker_cooled_down expiry --------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
extra_seconds=st.integers(min_value=1, max_value=86400),
|
|
)
|
|
def test_ticker_cooldown_returns_false_after_expiry(
|
|
self, cb: CircuitBreaker, triggered_at: datetime, extra_seconds: int,
|
|
) -> None:
|
|
"""is_ticker_cooled_down returns False when cooldown has expired."""
|
|
cooldown_expires = triggered_at + timedelta(hours=cb.ticker_cooldown_hours)
|
|
now = cooldown_expires + timedelta(seconds=extra_seconds)
|
|
|
|
ticker_cooldowns = {"AAPL": cooldown_expires}
|
|
|
|
result = cb.is_ticker_cooled_down(
|
|
ticker="AAPL",
|
|
ticker_cooldowns=ticker_cooldowns,
|
|
now=now,
|
|
)
|
|
assert result is False, (
|
|
f"Expected ticker not cooled down: now={now} > expiry={cooldown_expires}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
before_fraction=st.floats(min_value=0.0, max_value=0.99, allow_nan=False, allow_infinity=False),
|
|
)
|
|
def test_ticker_cooldown_returns_true_during_cooldown(
|
|
self, cb: CircuitBreaker, triggered_at: datetime, before_fraction: float,
|
|
) -> None:
|
|
"""is_ticker_cooled_down returns True when still within cooldown period."""
|
|
cooldown_duration = timedelta(hours=cb.ticker_cooldown_hours)
|
|
cooldown_expires = triggered_at + cooldown_duration
|
|
now = triggered_at + cooldown_duration * before_fraction
|
|
|
|
assume(now < cooldown_expires)
|
|
|
|
ticker_cooldowns = {"TSLA": cooldown_expires}
|
|
|
|
result = cb.is_ticker_cooled_down(
|
|
ticker="TSLA",
|
|
ticker_cooldowns=ticker_cooldowns,
|
|
now=now,
|
|
)
|
|
assert result is True, (
|
|
f"Expected ticker cooled down: now={now} < expiry={cooldown_expires}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
now=_aware_datetime_strategy(),
|
|
)
|
|
def test_ticker_cooldown_returns_false_for_unknown_ticker(
|
|
self, cb: CircuitBreaker, now: datetime,
|
|
) -> None:
|
|
"""is_ticker_cooled_down returns False for a ticker with no cooldown entry."""
|
|
result = cb.is_ticker_cooled_down(
|
|
ticker="UNKNOWN",
|
|
ticker_cooldowns={},
|
|
now=now,
|
|
)
|
|
assert result is False, "Expected no cooldown for unknown ticker"
|
|
|
|
# -- compute_cooldown_expiry -------------------------------------------
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
)
|
|
def test_cooldown_expiry_single_position_uses_ticker_cooldown_hours(
|
|
self, cb: CircuitBreaker, triggered_at: datetime,
|
|
) -> None:
|
|
"""compute_cooldown_expiry for single_position uses ticker_cooldown_hours."""
|
|
expiry = cb.compute_cooldown_expiry(
|
|
trigger_type="single_position",
|
|
triggered_at=triggered_at,
|
|
)
|
|
expected = triggered_at + timedelta(hours=cb.ticker_cooldown_hours)
|
|
assert expiry == expected, (
|
|
f"single_position expiry {expiry} != expected {expected}"
|
|
)
|
|
|
|
@settings(max_examples=100)
|
|
@given(
|
|
cb=_circuit_breaker_strategy(),
|
|
triggered_at=_aware_datetime_strategy(),
|
|
trigger_type=st.sampled_from(["daily_loss", "volatility"]),
|
|
)
|
|
def test_cooldown_expiry_daily_loss_and_volatility_use_pause_hours(
|
|
self, cb: CircuitBreaker, triggered_at: datetime, trigger_type: str,
|
|
) -> None:
|
|
"""compute_cooldown_expiry for daily_loss/volatility uses volatility_pause_hours."""
|
|
expiry = cb.compute_cooldown_expiry(
|
|
trigger_type=trigger_type,
|
|
triggered_at=triggered_at,
|
|
)
|
|
expected = triggered_at + timedelta(hours=cb.volatility_pause_hours)
|
|
assert expiry == expected, (
|
|
f"{trigger_type} expiry {expiry} != expected {expected}"
|
|
)
|