"""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"