315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Unit tests for v3 stop-defined portfolio heat and risk tier auto-adjustment.
|
||
|
||
Validates: Requirements 15.1–15.5, 18.1–18.6
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
import pytest
|
||
|
||
from services.risk.engine import (
|
||
TierMetrics,
|
||
check_heat_capacity,
|
||
compute_available_heat_capacity,
|
||
compute_portfolio_heat,
|
||
evaluate_tier_adjustment,
|
||
)
|
||
|
||
# ===========================================================================
|
||
# Heat computation tests (Requirements 15.1, 15.2)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestComputePortfolioHeat:
|
||
"""Test stop-defined portfolio heat calculation."""
|
||
|
||
def test_heat_three_positions_known_stops(self):
|
||
"""3 positions with known stops → expected heat = 710.
|
||
|
||
risk_dollars = position_value × stop_distance_pct
|
||
AAPL: 10000 × 0.03 = 300
|
||
MSFT: 5000 × 0.05 = 250
|
||
GOOG: 8000 × 0.02 = 160
|
||
Total heat = 710
|
||
"""
|
||
positions = [
|
||
{"ticker": "AAPL", "position_value": 10000},
|
||
{"ticker": "MSFT", "position_value": 5000},
|
||
{"ticker": "GOOG", "position_value": 8000},
|
||
]
|
||
stop_distances = {"AAPL": 0.03, "MSFT": 0.05, "GOOG": 0.02}
|
||
|
||
heat = compute_portfolio_heat(positions, stop_distances)
|
||
assert heat == pytest.approx(710.0)
|
||
|
||
def test_heat_empty_positions(self):
|
||
"""No positions → zero heat."""
|
||
heat = compute_portfolio_heat([], {})
|
||
assert heat == 0.0
|
||
|
||
def test_heat_missing_stop_distance_defaults_to_zero(self):
|
||
"""Position with no stop distance entry contributes zero risk."""
|
||
positions = [{"ticker": "AAPL", "position_value": 10000}]
|
||
stop_distances = {} # no entry for AAPL
|
||
|
||
heat = compute_portfolio_heat(positions, stop_distances)
|
||
assert heat == 0.0
|
||
|
||
def test_heat_single_position(self):
|
||
"""Single position → risk = value × stop."""
|
||
positions = [{"ticker": "TSLA", "position_value": 20000}]
|
||
stop_distances = {"TSLA": 0.04}
|
||
|
||
heat = compute_portfolio_heat(positions, stop_distances)
|
||
assert heat == pytest.approx(800.0)
|
||
|
||
|
||
# ===========================================================================
|
||
# Heat capacity / rejection tests (Requirements 15.3, 15.4, 15.5)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestCheckHeatCapacity:
|
||
"""Test heat capacity check — rejection when exceeding limit."""
|
||
|
||
def test_heat_at_limit_rejects_new_entry(self):
|
||
"""current_heat=4500, new_risk=600, max_heat=5000 → rejected (False).
|
||
|
||
4500 + 600 = 5100 > 5000 → False
|
||
"""
|
||
result = check_heat_capacity(
|
||
current_heat=4500,
|
||
new_risk_dollars=600,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=100000,
|
||
)
|
||
assert result is False
|
||
|
||
def test_heat_within_limit_allows_entry(self):
|
||
"""current_heat=4000, new_risk=500, max_heat=5000 → allowed (True).
|
||
|
||
4000 + 500 = 4500 <= 5000 → True
|
||
"""
|
||
result = check_heat_capacity(
|
||
current_heat=4000,
|
||
new_risk_dollars=500,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=100000,
|
||
)
|
||
assert result is True
|
||
|
||
def test_heat_exactly_at_limit_allows_entry(self):
|
||
"""current_heat=4500, new_risk=500, max_heat=5000 → allowed (True).
|
||
|
||
4500 + 500 = 5000 <= 5000 → True (at boundary)
|
||
"""
|
||
result = check_heat_capacity(
|
||
current_heat=4500,
|
||
new_risk_dollars=500,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=100000,
|
||
)
|
||
assert result is True
|
||
|
||
def test_heat_zero_portfolio_rejects(self):
|
||
"""Zero portfolio value → max_heat = 0 → any new risk rejected."""
|
||
result = check_heat_capacity(
|
||
current_heat=0,
|
||
new_risk_dollars=100,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=0,
|
||
)
|
||
assert result is False
|
||
|
||
|
||
class TestComputeAvailableHeatCapacity:
|
||
"""Test available heat capacity computation."""
|
||
|
||
def test_available_capacity_normal(self):
|
||
"""max_heat=5000, current=3000 → available=2000."""
|
||
available = compute_available_heat_capacity(
|
||
current_heat=3000,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=100000,
|
||
)
|
||
assert available == pytest.approx(2000.0)
|
||
|
||
def test_available_capacity_fully_used(self):
|
||
"""current >= max → available = 0."""
|
||
available = compute_available_heat_capacity(
|
||
current_heat=5500,
|
||
max_heat_pct=0.05,
|
||
portfolio_value=100000,
|
||
)
|
||
assert available == 0.0
|
||
|
||
|
||
# ===========================================================================
|
||
# Tier downgrade tests (Requirements 18.2, 18.3, 18.5)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestTierDowngrade:
|
||
"""Test that a single bad metric triggers downgrade."""
|
||
|
||
def _good_metrics(self, **overrides) -> TierMetrics:
|
||
"""Create metrics that pass all upgrade conditions, then override."""
|
||
defaults = {
|
||
"profit_factor_30d": 1.5,
|
||
"max_drawdown_30d": 0.03,
|
||
"calibration_error": 0.08,
|
||
"realized_sharpe_30d": 1.5,
|
||
"n_trades_30d": 25,
|
||
"reserve_pool_pct": 0.25,
|
||
}
|
||
defaults.update(overrides)
|
||
return TierMetrics(**defaults)
|
||
|
||
def test_downgrade_low_profit_factor(self):
|
||
"""profit_factor=0.9 (< 1.0) → downgrade."""
|
||
metrics = self._good_metrics(profit_factor_30d=0.9)
|
||
assert evaluate_tier_adjustment(metrics) == "downgrade"
|
||
|
||
def test_downgrade_high_drawdown(self):
|
||
"""max_drawdown=0.15 (> 0.12) → downgrade."""
|
||
metrics = self._good_metrics(max_drawdown_30d=0.15)
|
||
assert evaluate_tier_adjustment(metrics) == "downgrade"
|
||
|
||
def test_downgrade_high_calibration_error(self):
|
||
"""calibration_error=0.25 (> 0.20) → downgrade."""
|
||
metrics = self._good_metrics(calibration_error=0.25)
|
||
assert evaluate_tier_adjustment(metrics) == "downgrade"
|
||
|
||
def test_downgrade_negative_sharpe(self):
|
||
"""sharpe=-0.5 (< 0) → downgrade."""
|
||
metrics = self._good_metrics(realized_sharpe_30d=-0.5)
|
||
assert evaluate_tier_adjustment(metrics) == "downgrade"
|
||
|
||
|
||
# ===========================================================================
|
||
# Tier upgrade tests (Requirements 18.4)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestTierUpgrade:
|
||
"""Test that all metrics must be good for upgrade."""
|
||
|
||
def test_upgrade_all_good(self):
|
||
"""All upgrade conditions met → upgrade.
|
||
|
||
profit_factor=1.5, drawdown=0.03, cal_error=0.08,
|
||
sharpe=1.5, n_trades=25, reserve=0.25
|
||
"""
|
||
metrics = TierMetrics(
|
||
profit_factor_30d=1.5,
|
||
max_drawdown_30d=0.03,
|
||
calibration_error=0.08,
|
||
realized_sharpe_30d=1.5,
|
||
n_trades_30d=25,
|
||
reserve_pool_pct=0.25,
|
||
)
|
||
assert evaluate_tier_adjustment(metrics) == "upgrade"
|
||
|
||
def test_hold_insufficient_trades(self):
|
||
"""All upgrade conditions met EXCEPT n_trades=15 (< 20) → hold."""
|
||
metrics = TierMetrics(
|
||
profit_factor_30d=1.5,
|
||
max_drawdown_30d=0.03,
|
||
calibration_error=0.08,
|
||
realized_sharpe_30d=1.5,
|
||
n_trades_30d=15,
|
||
reserve_pool_pct=0.25,
|
||
)
|
||
assert evaluate_tier_adjustment(metrics) == "hold"
|
||
|
||
def test_hold_insufficient_reserve(self):
|
||
"""All good except reserve=0.15 (< 0.20) → hold."""
|
||
metrics = TierMetrics(
|
||
profit_factor_30d=1.5,
|
||
max_drawdown_30d=0.03,
|
||
calibration_error=0.08,
|
||
realized_sharpe_30d=1.5,
|
||
n_trades_30d=25,
|
||
reserve_pool_pct=0.15,
|
||
)
|
||
assert evaluate_tier_adjustment(metrics) == "hold"
|
||
|
||
def test_hold_drawdown_too_high_for_upgrade(self):
|
||
"""drawdown=0.06 (> 0.05 for upgrade) but < 0.12 (no downgrade) → hold."""
|
||
metrics = TierMetrics(
|
||
profit_factor_30d=1.5,
|
||
max_drawdown_30d=0.06,
|
||
calibration_error=0.08,
|
||
realized_sharpe_30d=1.5,
|
||
n_trades_30d=25,
|
||
reserve_pool_pct=0.25,
|
||
)
|
||
assert evaluate_tier_adjustment(metrics) == "hold"
|
||
|
||
|
||
# ===========================================================================
|
||
# 7-day cooldown enforcement (Requirements 18.5, 18.6)
|
||
# ===========================================================================
|
||
|
||
|
||
class TestTierCooldown:
|
||
"""Test 7-day cooldown enforcement logic at the caller level.
|
||
|
||
The evaluate_tier_adjustment function is pure — it doesn't track state.
|
||
The 7-day cooldown is enforced by the caller. Here we verify the pure
|
||
logic that a caller would use: compare last_upgrade_time to now and
|
||
only allow upgrade if >= 7 days have passed.
|
||
"""
|
||
|
||
def test_cooldown_blocks_upgrade_within_7_days(self):
|
||
"""Upgrade blocked when last upgrade was < 7 days ago."""
|
||
last_upgrade = datetime(2024, 1, 10, tzinfo=timezone.utc)
|
||
now = datetime(2024, 1, 15, tzinfo=timezone.utc) # 5 days later
|
||
cooldown_days = 7
|
||
|
||
days_since_last = (now - last_upgrade).days
|
||
can_upgrade = days_since_last >= cooldown_days
|
||
|
||
assert can_upgrade is False
|
||
|
||
def test_cooldown_allows_upgrade_after_7_days(self):
|
||
"""Upgrade allowed when last upgrade was >= 7 days ago."""
|
||
last_upgrade = datetime(2024, 1, 10, tzinfo=timezone.utc)
|
||
now = datetime(2024, 1, 17, tzinfo=timezone.utc) # 7 days later
|
||
cooldown_days = 7
|
||
|
||
days_since_last = (now - last_upgrade).days
|
||
can_upgrade = days_since_last >= cooldown_days
|
||
|
||
assert can_upgrade is True
|
||
|
||
def test_cooldown_allows_upgrade_well_past_7_days(self):
|
||
"""Upgrade allowed when last upgrade was well past cooldown."""
|
||
last_upgrade = datetime(2024, 1, 1, tzinfo=timezone.utc)
|
||
now = datetime(2024, 2, 1, tzinfo=timezone.utc) # 31 days later
|
||
cooldown_days = 7
|
||
|
||
days_since_last = (now - last_upgrade).days
|
||
can_upgrade = days_since_last >= cooldown_days
|
||
|
||
assert can_upgrade is True
|
||
|
||
def test_downgrade_ignores_cooldown(self):
|
||
"""Downgrade is applied immediately regardless of cooldown.
|
||
|
||
Even if an upgrade happened yesterday, downgrade still fires.
|
||
"""
|
||
# The evaluate function doesn't have cooldown logic — it always
|
||
# returns "downgrade" when conditions are met, regardless of timing.
|
||
metrics = TierMetrics(
|
||
profit_factor_30d=0.8, # triggers downgrade
|
||
max_drawdown_30d=0.03,
|
||
calibration_error=0.08,
|
||
realized_sharpe_30d=1.5,
|
||
n_trades_30d=25,
|
||
reserve_pool_pct=0.25,
|
||
)
|
||
# Downgrade fires regardless of when last upgrade occurred
|
||
assert evaluate_tier_adjustment(metrics) == "downgrade"
|