feat: autonomous trading engine — full implementation
- Database migration 018 with 13 tables for trading engine state - Trading engine service (services/trading/) with 12 pure computation modules: position sizer, stop-loss manager, reserve pool, circuit breaker, risk tier controller, correlation matrix, tax lots, trading window, gradual entry, notifications, micro-trading, backtester - Core TradingEngine with pre-trade evaluation pipeline and integration wiring - FastAPI HTTP service with 14 endpoints (health, config, decisions, metrics, backtest) - Performance tracker with Sharpe ratio, drawdown, profit factor computation - 194 Python tests (165 property-based + 29 integration) - Frontend: 13 TanStack Query hooks, 7 dashboard panels, tabbed Trading Engine page - Helm chart entry, network policy, nginx proxy, ingress for trading-engine - Shared infrastructure: enums, Redis keys, TradingConfig in AppConfig
This commit is contained in:
Celes Renata
@@ -0,0 +1,711 @@
|
||||
"""Property-based tests for the Stop-Loss Manager.
|
||||
|
||||
Feature: autonomous-trading-engine
|
||||
|
||||
Tests properties 9, 10, 11, 15, and 25 from the design specification,
|
||||
covering initial stop/take-profit computation, price crossing triggers,
|
||||
trailing stop activation, high-severity event tightening, and proactive
|
||||
heat-based stop tightening.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from hypothesis import given, settings, assume
|
||||
from hypothesis import strategies as st
|
||||
|
||||
from services.trading.models import (
|
||||
OpenPosition,
|
||||
RiskTierConfig,
|
||||
StopLevels,
|
||||
)
|
||||
from services.trading.stop_loss_manager import StopLossManager
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hypothesis strategies
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _risk_tier_config_strategy() -> st.SearchStrategy[RiskTierConfig]:
|
||||
"""Generate random RiskTierConfig objects with valid parameter ranges."""
|
||||
return st.builds(
|
||||
RiskTierConfig,
|
||||
name=st.sampled_from(["conservative", "moderate", "aggressive"]),
|
||||
min_confidence=st.floats(min_value=0.10, max_value=0.95, allow_nan=False, allow_infinity=False),
|
||||
max_position_pct=st.floats(min_value=0.02, max_value=0.50, allow_nan=False, allow_infinity=False),
|
||||
stop_loss_atr_multiplier=st.floats(min_value=0.5, max_value=5.0, allow_nan=False, allow_infinity=False),
|
||||
reward_risk_ratio=st.floats(min_value=0.5, max_value=5.0, allow_nan=False, allow_infinity=False),
|
||||
max_sector_pct=st.floats(min_value=0.05, max_value=0.60, allow_nan=False, allow_infinity=False),
|
||||
max_portfolio_heat=st.floats(min_value=0.05, max_value=0.50, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
|
||||
|
||||
def _open_position_strategy(
|
||||
ticker: st.SearchStrategy[str] | None = None,
|
||||
entry_price: st.SearchStrategy[float] | None = None,
|
||||
signal_confidence: st.SearchStrategy[float] | None = None,
|
||||
) -> st.SearchStrategy[OpenPosition]:
|
||||
"""Generate random OpenPosition objects."""
|
||||
return st.builds(
|
||||
OpenPosition,
|
||||
ticker=ticker if ticker is not None else st.from_regex(r"[A-Z]{3,5}", fullmatch=True),
|
||||
quantity=st.integers(min_value=1, max_value=100),
|
||||
entry_price=entry_price if entry_price is not None else st.floats(
|
||||
min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False,
|
||||
),
|
||||
current_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
unrealized_pnl=st.floats(min_value=-1000.0, max_value=1000.0, allow_nan=False, allow_infinity=False),
|
||||
market_value=st.floats(min_value=10.0, max_value=5000.0, allow_nan=False, allow_infinity=False),
|
||||
sector=st.sampled_from(["Technology", "Healthcare", "Energy", "Financials", "Consumer"]),
|
||||
stop_loss_price=st.floats(min_value=1.0, max_value=400.0, allow_nan=False, allow_infinity=False),
|
||||
take_profit_price=st.floats(min_value=10.0, max_value=1000.0, allow_nan=False, allow_infinity=False),
|
||||
signal_confidence=signal_confidence if signal_confidence is not None else st.floats(
|
||||
min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False,
|
||||
),
|
||||
is_micro_trade=st.just(False),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property 9: Stop-loss and take-profit initial computation
|
||||
# **Validates: Requirements 4.1, 4.2**
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProperty9InitialComputation:
|
||||
"""Property 9: Stop-loss and take-profit initial computation.
|
||||
|
||||
**Validates: Requirements 4.1, 4.2**
|
||||
"""
|
||||
|
||||
manager = StopLossManager()
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.1, max_value=20.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_stop_loss_equals_entry_minus_atr_times_multiplier(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Stop-loss = entry_price - (ATR * stop_loss_atr_multiplier)."""
|
||||
levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
expected_stop = entry_price - (atr * risk_tier.stop_loss_atr_multiplier)
|
||||
assert abs(levels.stop_loss_price - expected_stop) < 1e-9, (
|
||||
f"stop_loss {levels.stop_loss_price} != expected {expected_stop}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.1, max_value=20.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_stop_loss_always_below_entry(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Stop-loss is always below entry price (for positive ATR and multiplier)."""
|
||||
assume(atr * risk_tier.stop_loss_atr_multiplier > 0)
|
||||
levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
assert levels.stop_loss_price < entry_price, (
|
||||
f"stop_loss {levels.stop_loss_price} >= entry {entry_price}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.1, max_value=20.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_take_profit_equals_entry_plus_stop_distance_times_ratio(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Take-profit = entry_price + (stop_distance * reward_risk_ratio)."""
|
||||
levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
stop_distance = atr * risk_tier.stop_loss_atr_multiplier
|
||||
expected_tp = entry_price + (stop_distance * risk_tier.reward_risk_ratio)
|
||||
assert abs(levels.take_profit_price - expected_tp) < 1e-9, (
|
||||
f"take_profit {levels.take_profit_price} != expected {expected_tp}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.1, max_value=20.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_take_profit_always_above_entry(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Take-profit is always above entry price."""
|
||||
assume(atr * risk_tier.stop_loss_atr_multiplier * risk_tier.reward_risk_ratio > 0)
|
||||
levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
assert levels.take_profit_price > entry_price, (
|
||||
f"take_profit {levels.take_profit_price} <= entry {entry_price}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=5.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.1, max_value=20.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_trailing_stop_initially_inactive(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Trailing stop is not active on initial computation."""
|
||||
levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
assert levels.trailing_stop_active is False
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property 10: Price crossing triggers immediate sell
|
||||
# **Validates: Requirements 4.4, 4.5**
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProperty10PriceCrossingTriggers:
|
||||
"""Property 10: Price crossing triggers immediate sell.
|
||||
|
||||
**Validates: Requirements 4.4, 4.5**
|
||||
"""
|
||||
|
||||
manager = StopLossManager()
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
stop_distance_pct=st.floats(min_value=0.02, max_value=0.20, allow_nan=False, allow_infinity=False),
|
||||
tp_distance_pct=st.floats(min_value=0.02, max_value=0.40, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_sell_triggered_when_price_at_or_below_stop_loss(
|
||||
self, entry_price: float, stop_distance_pct: float, tp_distance_pct: float,
|
||||
) -> None:
|
||||
"""Sell triggered when current price <= stop_loss."""
|
||||
stop_loss = entry_price * (1 - stop_distance_pct)
|
||||
take_profit = entry_price * (1 + tp_distance_pct)
|
||||
assume(stop_loss > 0)
|
||||
assume(take_profit > stop_loss)
|
||||
|
||||
ticker = "TEST"
|
||||
position = OpenPosition(
|
||||
ticker=ticker, quantity=10, entry_price=entry_price,
|
||||
current_price=stop_loss, unrealized_pnl=0.0,
|
||||
market_value=entry_price * 10, sector="Technology",
|
||||
stop_loss_price=stop_loss, take_profit_price=take_profit,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
levels = StopLevels(
|
||||
stop_loss_price=stop_loss,
|
||||
take_profit_price=take_profit,
|
||||
trailing_stop_active=False,
|
||||
atr_value=1.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
)
|
||||
|
||||
# Price at stop_loss
|
||||
triggers = self.manager.check_price_crossings(
|
||||
positions=[position],
|
||||
prices={ticker: stop_loss},
|
||||
stop_levels={ticker: levels},
|
||||
)
|
||||
assert len(triggers) == 1
|
||||
assert triggers[0].trigger_type == "stop_loss"
|
||||
assert triggers[0].ticker == ticker
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
stop_distance_pct=st.floats(min_value=0.02, max_value=0.20, allow_nan=False, allow_infinity=False),
|
||||
tp_distance_pct=st.floats(min_value=0.02, max_value=0.40, allow_nan=False, allow_infinity=False),
|
||||
below_pct=st.floats(min_value=0.001, max_value=0.10, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_sell_triggered_when_price_below_stop_loss(
|
||||
self, entry_price: float, stop_distance_pct: float,
|
||||
tp_distance_pct: float, below_pct: float,
|
||||
) -> None:
|
||||
"""Sell triggered when current price is below stop_loss."""
|
||||
stop_loss = entry_price * (1 - stop_distance_pct)
|
||||
take_profit = entry_price * (1 + tp_distance_pct)
|
||||
price_below = stop_loss * (1 - below_pct)
|
||||
assume(stop_loss > 0)
|
||||
assume(price_below > 0)
|
||||
assume(take_profit > stop_loss)
|
||||
|
||||
ticker = "TEST"
|
||||
position = OpenPosition(
|
||||
ticker=ticker, quantity=10, entry_price=entry_price,
|
||||
current_price=price_below, unrealized_pnl=0.0,
|
||||
market_value=entry_price * 10, sector="Technology",
|
||||
stop_loss_price=stop_loss, take_profit_price=take_profit,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
levels = StopLevels(
|
||||
stop_loss_price=stop_loss,
|
||||
take_profit_price=take_profit,
|
||||
trailing_stop_active=False,
|
||||
atr_value=1.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
)
|
||||
|
||||
triggers = self.manager.check_price_crossings(
|
||||
positions=[position],
|
||||
prices={ticker: price_below},
|
||||
stop_levels={ticker: levels},
|
||||
)
|
||||
assert len(triggers) == 1
|
||||
assert triggers[0].trigger_type == "stop_loss"
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
stop_distance_pct=st.floats(min_value=0.02, max_value=0.20, allow_nan=False, allow_infinity=False),
|
||||
tp_distance_pct=st.floats(min_value=0.02, max_value=0.40, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_sell_triggered_when_price_at_or_above_take_profit(
|
||||
self, entry_price: float, stop_distance_pct: float, tp_distance_pct: float,
|
||||
) -> None:
|
||||
"""Sell triggered when current price >= take_profit."""
|
||||
stop_loss = entry_price * (1 - stop_distance_pct)
|
||||
take_profit = entry_price * (1 + tp_distance_pct)
|
||||
assume(stop_loss > 0)
|
||||
assume(take_profit > stop_loss)
|
||||
|
||||
ticker = "TEST"
|
||||
position = OpenPosition(
|
||||
ticker=ticker, quantity=10, entry_price=entry_price,
|
||||
current_price=take_profit, unrealized_pnl=0.0,
|
||||
market_value=entry_price * 10, sector="Technology",
|
||||
stop_loss_price=stop_loss, take_profit_price=take_profit,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
levels = StopLevels(
|
||||
stop_loss_price=stop_loss,
|
||||
take_profit_price=take_profit,
|
||||
trailing_stop_active=False,
|
||||
atr_value=1.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
)
|
||||
|
||||
triggers = self.manager.check_price_crossings(
|
||||
positions=[position],
|
||||
prices={ticker: take_profit},
|
||||
stop_levels={ticker: levels},
|
||||
)
|
||||
assert len(triggers) == 1
|
||||
assert triggers[0].trigger_type == "take_profit"
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
stop_distance_pct=st.floats(min_value=0.05, max_value=0.20, allow_nan=False, allow_infinity=False),
|
||||
tp_distance_pct=st.floats(min_value=0.05, max_value=0.40, allow_nan=False, allow_infinity=False),
|
||||
between_pct=st.floats(min_value=0.01, max_value=0.99, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_no_trigger_when_price_between_levels(
|
||||
self, entry_price: float, stop_distance_pct: float,
|
||||
tp_distance_pct: float, between_pct: float,
|
||||
) -> None:
|
||||
"""No trigger when price is strictly between stop_loss and take_profit."""
|
||||
stop_loss = entry_price * (1 - stop_distance_pct)
|
||||
take_profit = entry_price * (1 + tp_distance_pct)
|
||||
assume(stop_loss > 0)
|
||||
assume(take_profit > stop_loss + 0.02)
|
||||
|
||||
# Price strictly between stop_loss and take_profit
|
||||
price_between = stop_loss + (take_profit - stop_loss) * between_pct
|
||||
# Ensure strictly between (not at boundaries)
|
||||
assume(price_between > stop_loss)
|
||||
assume(price_between < take_profit)
|
||||
|
||||
ticker = "TEST"
|
||||
position = OpenPosition(
|
||||
ticker=ticker, quantity=10, entry_price=entry_price,
|
||||
current_price=price_between, unrealized_pnl=0.0,
|
||||
market_value=entry_price * 10, sector="Technology",
|
||||
stop_loss_price=stop_loss, take_profit_price=take_profit,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
levels = StopLevels(
|
||||
stop_loss_price=stop_loss,
|
||||
take_profit_price=take_profit,
|
||||
trailing_stop_active=False,
|
||||
atr_value=1.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
)
|
||||
|
||||
triggers = self.manager.check_price_crossings(
|
||||
positions=[position],
|
||||
prices={ticker: price_between},
|
||||
stop_levels={ticker: levels},
|
||||
)
|
||||
assert len(triggers) == 0
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property 11: Trailing stop activation at 50% of take-profit distance
|
||||
# **Validates: Requirements 4.6**
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProperty11TrailingStopActivation:
|
||||
"""Property 11: Trailing stop activation at 50% of take-profit distance.
|
||||
|
||||
**Validates: Requirements 4.6**
|
||||
"""
|
||||
|
||||
manager = StopLossManager()
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.5, max_value=10.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
move_fraction=st.floats(min_value=0.51, max_value=0.99, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_trailing_stop_activates_when_move_exceeds_50_pct(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
move_fraction: float,
|
||||
) -> None:
|
||||
"""Trailing stop activates when favorable move > 50% of TP distance."""
|
||||
# Compute initial levels to get the TP distance
|
||||
initial_levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
tp_distance = initial_levels.take_profit_price - entry_price
|
||||
assume(tp_distance > 0.01)
|
||||
|
||||
# Current price moved favorably by more than 50% of TP distance
|
||||
current_price = entry_price + (tp_distance * move_fraction)
|
||||
|
||||
position = OpenPosition(
|
||||
ticker="TEST", quantity=10, entry_price=entry_price,
|
||||
current_price=current_price, unrealized_pnl=0.0,
|
||||
market_value=current_price * 10, sector="Technology",
|
||||
stop_loss_price=initial_levels.stop_loss_price,
|
||||
take_profit_price=initial_levels.take_profit_price,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
|
||||
result = self.manager.re_evaluate_levels(
|
||||
position=position,
|
||||
current_price=current_price,
|
||||
atr=atr,
|
||||
risk_tier=risk_tier,
|
||||
last_levels=initial_levels,
|
||||
)
|
||||
|
||||
# re_evaluate_levels returns None when no material change, or StopLevels
|
||||
# Since trailing stop activation IS a material change, we expect a result
|
||||
assert result is not None, "Expected re_evaluate to return updated levels"
|
||||
assert result.trailing_stop_active is True
|
||||
# When trailing stop is active, stop should be at least at entry (breakeven)
|
||||
assert result.stop_loss_price >= entry_price - 1e-9, (
|
||||
f"Trailing stop {result.stop_loss_price} should be >= entry {entry_price}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.5, max_value=10.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
move_fraction=st.floats(min_value=0.0, max_value=0.49, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_trailing_stop_does_not_activate_when_move_at_or_below_50_pct(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
move_fraction: float,
|
||||
) -> None:
|
||||
"""Trailing stop does NOT activate when favorable move <= 50% of TP distance."""
|
||||
initial_levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
tp_distance = initial_levels.take_profit_price - entry_price
|
||||
assume(tp_distance > 0.01)
|
||||
|
||||
# Current price moved favorably by <= 50% of TP distance
|
||||
current_price = entry_price + (tp_distance * move_fraction)
|
||||
|
||||
position = OpenPosition(
|
||||
ticker="TEST", quantity=10, entry_price=entry_price,
|
||||
current_price=current_price, unrealized_pnl=0.0,
|
||||
market_value=current_price * 10, sector="Technology",
|
||||
stop_loss_price=initial_levels.stop_loss_price,
|
||||
take_profit_price=initial_levels.take_profit_price,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
|
||||
result = self.manager.re_evaluate_levels(
|
||||
position=position,
|
||||
current_price=current_price,
|
||||
atr=atr,
|
||||
risk_tier=risk_tier,
|
||||
last_levels=initial_levels,
|
||||
)
|
||||
|
||||
# Either None (no change) or result with trailing_stop_active=False
|
||||
if result is not None:
|
||||
assert result.trailing_stop_active is False, (
|
||||
f"Trailing stop should not activate at {move_fraction*100:.1f}% move"
|
||||
)
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property 15: Stop tightening during high-severity events
|
||||
# **Validates: Requirements 7.2**
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProperty15HighSeverityEventTightening:
|
||||
"""Property 15: Stop tightening during high-severity events.
|
||||
|
||||
**Validates: Requirements 7.2**
|
||||
"""
|
||||
|
||||
manager = StopLossManager()
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.5, max_value=10.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_tightened_stop_uses_half_normal_multiplier(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""During high-severity events, stop uses 0.5x normal ATR multiplier."""
|
||||
initial_levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
|
||||
# Use a current price near entry so trailing stop doesn't activate
|
||||
current_price = entry_price + 0.01
|
||||
|
||||
position = OpenPosition(
|
||||
ticker="TEST", quantity=10, entry_price=entry_price,
|
||||
current_price=current_price, unrealized_pnl=0.0,
|
||||
market_value=current_price * 10, sector="Technology",
|
||||
stop_loss_price=initial_levels.stop_loss_price,
|
||||
take_profit_price=initial_levels.take_profit_price,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
|
||||
result = self.manager.re_evaluate_levels(
|
||||
position=position,
|
||||
current_price=current_price,
|
||||
atr=atr,
|
||||
risk_tier=risk_tier,
|
||||
last_levels=initial_levels,
|
||||
high_severity_event=True,
|
||||
)
|
||||
|
||||
# High-severity event changes the multiplier, so we expect a result
|
||||
assert result is not None, "Expected updated levels during high-severity event"
|
||||
|
||||
# The tightened stop should use 0.5x the normal multiplier
|
||||
expected_tightened_multiplier = risk_tier.stop_loss_atr_multiplier * 0.5
|
||||
expected_stop = entry_price - (atr * expected_tightened_multiplier)
|
||||
assert abs(result.stop_loss_price - expected_stop) < 1e-9, (
|
||||
f"Tightened stop {result.stop_loss_price} != expected {expected_stop} "
|
||||
f"(0.5x multiplier = {expected_tightened_multiplier})"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
entry_price=st.floats(min_value=50.0, max_value=500.0, allow_nan=False, allow_infinity=False),
|
||||
atr=st.floats(min_value=0.5, max_value=10.0, allow_nan=False, allow_infinity=False),
|
||||
risk_tier=_risk_tier_config_strategy(),
|
||||
)
|
||||
def test_tightened_stop_closer_to_entry_than_normal(
|
||||
self, entry_price: float, atr: float, risk_tier: RiskTierConfig,
|
||||
) -> None:
|
||||
"""Tightened stop is closer to entry price (higher) than normal stop."""
|
||||
initial_levels = self.manager.compute_initial_levels(
|
||||
entry_price=entry_price, atr=atr, risk_tier=risk_tier,
|
||||
)
|
||||
|
||||
current_price = entry_price + 0.01
|
||||
|
||||
position = OpenPosition(
|
||||
ticker="TEST", quantity=10, entry_price=entry_price,
|
||||
current_price=current_price, unrealized_pnl=0.0,
|
||||
market_value=current_price * 10, sector="Technology",
|
||||
stop_loss_price=initial_levels.stop_loss_price,
|
||||
take_profit_price=initial_levels.take_profit_price,
|
||||
signal_confidence=0.7,
|
||||
)
|
||||
|
||||
result = self.manager.re_evaluate_levels(
|
||||
position=position,
|
||||
current_price=current_price,
|
||||
atr=atr,
|
||||
risk_tier=risk_tier,
|
||||
last_levels=initial_levels,
|
||||
high_severity_event=True,
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
# Tightened stop should be closer to entry (higher value) than normal stop
|
||||
normal_stop = entry_price - (atr * risk_tier.stop_loss_atr_multiplier)
|
||||
assert result.stop_loss_price >= normal_stop - 1e-9, (
|
||||
f"Tightened stop {result.stop_loss_price} should be >= normal stop {normal_stop}"
|
||||
)
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Property 25: Proactive stop tightening at 80% heat threshold
|
||||
# **Validates: Requirements 13.3**
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProperty25ProactiveHeatTightening:
|
||||
"""Property 25: Proactive stop tightening at 80% heat threshold.
|
||||
|
||||
**Validates: Requirements 13.3**
|
||||
"""
|
||||
|
||||
manager = StopLossManager()
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
num_positions=st.integers(min_value=2, max_value=5),
|
||||
max_heat=st.floats(min_value=0.10, max_value=0.30, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_lowest_confidence_positions_tightened_first(
|
||||
self, num_positions: int, max_heat: float,
|
||||
) -> None:
|
||||
"""Lowest-confidence positions get stops tightened first when heat > 80% of max."""
|
||||
# Create positions with distinct confidence levels
|
||||
positions: list[OpenPosition] = []
|
||||
stop_levels_dict: dict[str, StopLevels] = {}
|
||||
|
||||
for i in range(num_positions):
|
||||
ticker = f"T{i}"
|
||||
entry_price = 100.0
|
||||
confidence = 0.3 + (i * 0.15) # ascending confidence: 0.3, 0.45, 0.6, ...
|
||||
stop_loss = 90.0 # 10% below entry
|
||||
take_profit = 115.0
|
||||
|
||||
positions.append(OpenPosition(
|
||||
ticker=ticker, quantity=10, entry_price=entry_price,
|
||||
current_price=entry_price, unrealized_pnl=0.0,
|
||||
market_value=entry_price * 10, sector="Technology",
|
||||
stop_loss_price=stop_loss, take_profit_price=take_profit,
|
||||
signal_confidence=confidence,
|
||||
))
|
||||
stop_levels_dict[ticker] = StopLevels(
|
||||
stop_loss_price=stop_loss,
|
||||
take_profit_price=take_profit,
|
||||
trailing_stop_active=False,
|
||||
atr_value=5.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
)
|
||||
|
||||
# Set heat above 80% of max to trigger tightening
|
||||
portfolio_heat = max_heat * 0.85
|
||||
active_pool = 10000.0
|
||||
|
||||
updated = self.manager.tighten_for_heat(
|
||||
positions=positions,
|
||||
stop_levels=stop_levels_dict,
|
||||
portfolio_heat=portfolio_heat,
|
||||
max_heat=max_heat,
|
||||
active_pool=active_pool,
|
||||
)
|
||||
|
||||
if updated:
|
||||
# Verify that tightened positions are ordered by confidence (lowest first)
|
||||
tightened_tickers = list(updated.keys())
|
||||
tightened_confidences = [
|
||||
next(p.signal_confidence for p in positions if p.ticker == t)
|
||||
for t in tightened_tickers
|
||||
]
|
||||
|
||||
# The first tightened position should have the lowest confidence
|
||||
min_confidence_in_portfolio = min(p.signal_confidence for p in positions)
|
||||
if tightened_tickers:
|
||||
first_tightened_confidence = next(
|
||||
p.signal_confidence for p in positions if p.ticker == tightened_tickers[0]
|
||||
)
|
||||
assert first_tightened_confidence == min_confidence_in_portfolio, (
|
||||
f"First tightened position confidence {first_tightened_confidence} "
|
||||
f"!= min confidence {min_confidence_in_portfolio}"
|
||||
)
|
||||
|
||||
# All tightened stops should be >= original stops (moved closer to entry)
|
||||
for ticker, new_levels in updated.items():
|
||||
original_stop = stop_levels_dict[ticker].stop_loss_price
|
||||
assert new_levels.stop_loss_price >= original_stop - 1e-9, (
|
||||
f"{ticker}: tightened stop {new_levels.stop_loss_price} "
|
||||
f"< original {original_stop}"
|
||||
)
|
||||
|
||||
@settings(max_examples=100)
|
||||
@given(
|
||||
max_heat=st.floats(min_value=0.10, max_value=0.30, allow_nan=False, allow_infinity=False),
|
||||
heat_fraction=st.floats(min_value=0.0, max_value=0.79, allow_nan=False, allow_infinity=False),
|
||||
)
|
||||
def test_no_tightening_when_heat_below_80_pct_threshold(
|
||||
self, max_heat: float, heat_fraction: float,
|
||||
) -> None:
|
||||
"""No tightening when portfolio heat <= 80% of max."""
|
||||
portfolio_heat = max_heat * heat_fraction
|
||||
|
||||
positions = [
|
||||
OpenPosition(
|
||||
ticker="T0", quantity=10, entry_price=100.0,
|
||||
current_price=100.0, unrealized_pnl=0.0,
|
||||
market_value=1000.0, sector="Technology",
|
||||
stop_loss_price=90.0, take_profit_price=115.0,
|
||||
signal_confidence=0.5,
|
||||
),
|
||||
]
|
||||
stop_levels_dict = {
|
||||
"T0": StopLevels(
|
||||
stop_loss_price=90.0,
|
||||
take_profit_price=115.0,
|
||||
trailing_stop_active=False,
|
||||
atr_value=5.0,
|
||||
atr_multiplier=2.0,
|
||||
reward_risk_ratio=1.5,
|
||||
last_updated=datetime.utcnow(),
|
||||
),
|
||||
}
|
||||
|
||||
updated = self.manager.tighten_for_heat(
|
||||
positions=positions,
|
||||
stop_levels=stop_levels_dict,
|
||||
portfolio_heat=portfolio_heat,
|
||||
max_heat=max_heat,
|
||||
active_pool=10000.0,
|
||||
)
|
||||
|
||||
assert updated == {}, (
|
||||
f"Expected no tightening at heat {portfolio_heat} "
|
||||
f"(80% threshold = {max_heat * 0.8}), got {len(updated)} updates"
|
||||
)
|
||||
Reference in New Issue
Block a user