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
711 lines
30 KiB
Python
711 lines
30 KiB
Python
"""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, timezone
|
|
|
|
from hypothesis import assume, given, settings
|
|
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.now(tz=timezone.utc),
|
|
)
|
|
|
|
# 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.now(tz=timezone.utc),
|
|
)
|
|
|
|
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.now(tz=timezone.utc),
|
|
)
|
|
|
|
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.now(tz=timezone.utc),
|
|
)
|
|
|
|
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.now(tz=timezone.utc),
|
|
)
|
|
|
|
# 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.now(tz=timezone.utc),
|
|
),
|
|
}
|
|
|
|
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"
|
|
)
|