""" Smoke tests for trading/risk_manager.py. Covers all 13 RiskRule rejection codes + lifecycle hooks (halt, daily P&L update, register_sl_hit, register_tp_hit). Calendar boundary tests for FORCE_FLAT (21:08 vs 21:07) and LAST_FRIDAY (15:00 vs 14:59), plus 4-Friday and 5-Friday months. Run: cd ~/apex_v16 python -m tests.test_risk_manager """ from __future__ import annotations import sys from datetime import datetime, timedelta, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from core.config import RuntimeConfig, RunMode, AccountKind from core.contracts import EntryDecision, RiskRule, utc_now from persistence.state_store import ( ActiveTrade, BiasCache, BrainCounters, CooldownState, DailyCounters, SessionState, ) from trading.risk_manager import ( CORRELATION_GROUPS, RiskAudit, RiskCheckResult, RiskManager, ) from trading.sizing import SizingAudit, SizingDecision def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # FIXTURES # ============================================================ def make_config( *, enable_correlation: bool = False, daily_loss_hard_stop: float = -1500.0, daily_loss_soft_stop: float = -1000.0, daily_profit_target: float = 300.0, max_daily_trades: int = 10, max_open_trades_total: int = 2, max_risk_vs_daily_budget: float = 0.33, force_flat_utc_hour: int = 21, force_flat_utc_minute: int = 8, last_friday_of_month_cutoff_utc_hour: int = 15, ) -> RuntimeConfig: cfg = RuntimeConfig(mode=RunMode.PAPER, account=AccountKind.INELIGIBLE) cfg.enable_correlation = enable_correlation cfg.daily_loss_hard_stop = daily_loss_hard_stop cfg.daily_loss_soft_stop = daily_loss_soft_stop cfg.daily_profit_target = daily_profit_target cfg.max_daily_trades = max_daily_trades cfg.max_open_trades_total = max_open_trades_total cfg.max_risk_vs_daily_budget = max_risk_vs_daily_budget cfg.force_flat_utc_hour = force_flat_utc_hour cfg.force_flat_utc_minute = force_flat_utc_minute cfg.last_friday_of_month_cutoff_utc_hour = last_friday_of_month_cutoff_utc_hour return cfg def make_state() -> SessionState: return SessionState() def make_entry(direction: str = "BUY") -> EntryDecision: return EntryDecision( direction=direction, entry_price=5800.00, sl_price=5797.50 if direction == "BUY" else 5802.50, tp_price=5805.00 if direction == "BUY" else 5795.00, rr_multiplier=0.50, confidence=75, rationale="test", ) def make_sizing( *, skip: bool = False, reason: str = "ok", contracts: int = 2, real_risk_usd: float = 100.0, sl_ticks: int = 30, ) -> SizingDecision: audit = SizingAudit( symbol="MES", risk_multiplier=1.0, target_float=2.0, sl_distance_points=2.5, tick_size=0.25, tick_value=1.25, sl_usd_per_contract=37.5, risk_usd_target=100.0, daily_budget_cap_usd=495.0, max_contracts_used=8, clamp_active=False, inverted_quote=False, ) return SizingDecision( contracts=contracts, skip=skip, reason=reason, sl_ticks=sl_ticks, real_risk_usd=real_risk_usd, audit=audit, ) def wednesday_14_utc() -> datetime: """A safe Wednesday at 14:00 UTC (no weekly cutoffs).""" return datetime(2026, 4, 29, 14, 0, 0, tzinfo=timezone.utc) def make_active_trade() -> ActiveTrade: """Tiny ActiveTrade fixture for occupancy/correlation tests.""" from core.contracts import TradeEntry, TradeRuntime entry = TradeEntry( symbol="MES", brain_name="TF", direction="BUY", contracts=1, entry_price=5800.0, sl_price=5797.5, tp_price=5805.0, opened_at=utc_now(), rsi_m5_at_entry=55.0, rsi_h1_at_entry=58.0, rsi_h4_at_entry=60.0, atr_ratio_at_entry=1.0, market_structure_at_entry="BULLISH_EXPANSION", regime_at_entry="TRENDING", h1_compat_at_entry=1.0, confidence_at_entry=75, ) return ActiveTrade(entry=entry, runtime=TradeRuntime()) # ============================================================ # 1. Happy path # ============================================================ def test_check_entry_all_clear_approved(): cfg = make_config() state = make_state() rm = RiskManager(cfg, state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.approved is True assert res.rule == RiskRule.OK.value assert res.audit.daily_pnl == 0.0 _ok("check_entry: all clear -> approved=True, rule=OK") # ============================================================ # 2. SIZING_SKIP propagation # ============================================================ def test_sizing_skip_propagates(): rm = RiskManager(make_config(), make_state()) sizing = make_sizing(skip=True, reason="size_too_small") res = rm.check_entry( entry=make_entry(), sizing=sizing, symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.approved is False assert res.rule == RiskRule.SIZING_SKIP.value assert "size_too_small" in res.reason _ok("check_entry: SizingDecision.skip=True -> SIZING_SKIP") # ============================================================ # 3. HALTED # ============================================================ def test_halted_blocks(): state = make_state() state.halted = True state.halt_reason = "manual halt for test" rm = RiskManager(make_config(), state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.HALTED.value assert "manual halt" in res.reason _ok("check_entry: state.halted -> HALTED") # ============================================================ # 4. DAILY_LOSS_HARD_STOP_HIT triggers auto-halt # ============================================================ def test_daily_loss_hard_stop_triggers_auto_halt(): cfg = make_config(daily_loss_hard_stop=-1500.0) state = make_state() state.daily.daily_pnl = -1500.0 rm = RiskManager(cfg, state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.DAILY_LOSS_HARD_STOP_HIT.value assert state.halted is True, "hard stop must auto-halt" assert state.daily.daily_loss_hard_stop_hit is True, "explicit V16 flag must set" _ok("DAILY_LOSS_HARD_STOP_HIT -> auto-halt + state.daily.daily_loss_hard_stop_hit=True") # ============================================================ # 5. DAILY_LOSS_SOFT_STOP_HIT blocks but does NOT halt # ============================================================ def test_daily_loss_soft_stop_blocks_no_halt(): cfg = make_config(daily_loss_soft_stop=-1000.0) state = make_state() state.daily.daily_pnl = -1000.0 rm = RiskManager(cfg, state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.DAILY_LOSS_SOFT_STOP_HIT.value assert state.halted is False, "soft stop must NOT halt" _ok("DAILY_LOSS_SOFT_STOP_HIT -> block, halted stays False (V15 parity)") def test_daily_loss_soft_stop_recovers_passes(): """One penny above soft -> ok again (recovery scenario).""" cfg = make_config(daily_loss_soft_stop=-1000.0) state = make_state() state.daily.daily_pnl = -999.99 rm = RiskManager(cfg, state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.approved is True _ok("daily_pnl=-999.99 (one cent above soft) -> approved (recovery)") # ============================================================ # 6. DAILY_PROFIT_TARGET_REACHED, no halt, profit_target_hit set # ============================================================ def test_daily_profit_target_blocks_no_halt(): cfg = make_config(daily_profit_target=300.0) state = make_state() state.daily.daily_pnl = 300.0 rm = RiskManager(cfg, state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.DAILY_PROFIT_TARGET_REACHED.value assert state.halted is False assert state.daily.profit_target_hit is True _ok("DAILY_PROFIT_TARGET_REACHED -> block, NO halt, flag set") # ============================================================ # 7. MAX_OPEN_TRADES_REACHED (V15 parity) # ============================================================ def test_max_open_trades_reached(): cfg = make_config(max_open_trades_total=2) rm = RiskManager(cfg, make_state()) actives = {"6E": make_active_trade(), "GC": make_active_trade()} res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades=actives, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.MAX_OPEN_TRADES_REACHED.value assert "2/2" in res.reason _ok("MAX_OPEN_TRADES_REACHED: 2 active vs cap=2 -> block") # ============================================================ # 8. MAX_DAILY_TRADES_REACHED # ============================================================ def test_max_daily_trades_reached(): state = make_state() state.daily.executed_count = 10 rm = RiskManager(make_config(max_daily_trades=10), state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.MAX_DAILY_TRADES_REACHED.value _ok("MAX_DAILY_TRADES_REACHED: executed_count=10/10 -> block") # ============================================================ # 9-10. FORCE_FLAT boundary 21:08 vs 21:07 # ============================================================ def test_force_flat_at_2108_blocks(): rm = RiskManager(make_config(), make_state()) # Wed 21:08:00 UTC now = datetime(2026, 4, 29, 21, 8, 0, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.rule == RiskRule.FORCE_FLAT_TIME_REACHED.value _ok("FORCE_FLAT boundary: 21:08:00 UTC -> block") def test_force_flat_at_2107_passes(): rm = RiskManager(make_config(), make_state()) now = datetime(2026, 4, 29, 21, 7, 59, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.approved is True _ok("FORCE_FLAT boundary: 21:07:59 UTC -> approved") # ============================================================ # 11-12. LAST_FRIDAY boundary 15:00 vs 14:59 # ============================================================ def test_last_friday_15_00_blocks(): rm = RiskManager(make_config(), make_state()) # 24 Apr 2026 is the last Friday of April (next Fri 1 May -> month 5) now = datetime(2026, 4, 24, 15, 0, 0, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.rule == RiskRule.LAST_FRIDAY_CUTOFF.value _ok("LAST_FRIDAY: 24-Apr 15:00 -> block") def test_last_friday_14_59_passes(): rm = RiskManager(make_config(), make_state()) now = datetime(2026, 4, 24, 14, 59, 59, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.approved is True _ok("LAST_FRIDAY: 24-Apr 14:59:59 -> approved (boundary)") # ============================================================ # 13. 4-Friday month: only the last Friday triggers # ============================================================ def test_4_friday_month_only_last_triggers(): """ Feb 2026 has 4 Fridays: 6, 13, 20, 27. Only 27/02 is the last; the others must NOT trigger LAST_FRIDAY. """ rm = RiskManager(make_config(), make_state()) for day in (6, 13, 20): now = datetime(2026, 2, day, 16, 0, 0, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.approved is True, f"Friday {day}/02 must NOT trigger" last = datetime(2026, 2, 27, 16, 0, 0, tzinfo=timezone.utc) res_last = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=last, ) assert res_last.rule == RiskRule.LAST_FRIDAY_CUTOFF.value _ok("4-Friday month (Feb 2026): only 27/02 triggers LAST_FRIDAY_CUTOFF") # ============================================================ # 14. 5-Friday month: only the last Friday triggers # ============================================================ def test_5_friday_month_only_last_triggers(): """ Jan 2026 has 5 Fridays: 2, 9, 16, 23, 30. Only 30/01 is the last. """ rm = RiskManager(make_config(), make_state()) for day in (2, 9, 16, 23): now = datetime(2026, 1, day, 16, 0, 0, tzinfo=timezone.utc) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.approved is True, f"Friday {day}/01 must NOT trigger" last = datetime(2026, 1, 30, 16, 0, 0, tzinfo=timezone.utc) res_last = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=last, ) assert res_last.rule == RiskRule.LAST_FRIDAY_CUTOFF.value _ok("5-Friday month (Jan 2026): only 30/01 triggers LAST_FRIDAY_CUTOFF") # ============================================================ # 15-16. COOLDOWN_ACTIVE / expiry # ============================================================ def test_cooldown_active_blocks(): state = make_state() now = wednesday_14_utc() state.cooldown.cooldown_until["MES"] = (now + timedelta(minutes=30)).isoformat() rm = RiskManager(make_config(), state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.rule == RiskRule.COOLDOWN_ACTIVE.value _ok("COOLDOWN_ACTIVE: cooldown_until in future -> block") def test_cooldown_expired_passes_and_clears(): state = make_state() now = wednesday_14_utc() state.cooldown.cooldown_until["MES"] = (now - timedelta(seconds=1)).isoformat() rm = RiskManager(make_config(), state) res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MES", active_trades={}, now_utc=now, ) assert res.approved is True assert "MES" not in state.cooldown.cooldown_until, "expired entry should be cleared" _ok("COOLDOWN expired -> approved + entry cleared") # ============================================================ # 17-18. CORRELATION_BLOCKED (V15 EQUITY_INDICES) + disabled passes # ============================================================ def test_correlation_blocked_equity_indices(): rm = RiskManager(make_config(enable_correlation=True), make_state()) actives = {"MES": make_active_trade()} res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MNQ", active_trades=actives, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.CORRELATION_BLOCKED.value assert res.audit.correlation_blocker == "MES" _ok("CORRELATION: MES open + try MNQ (EQUITY_INDICES) -> block") def test_correlation_disabled_passes(): rm = RiskManager(make_config(enable_correlation=False), make_state()) actives = {"MES": make_active_trade()} res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="MNQ", active_trades=actives, now_utc=wednesday_14_utc(), ) assert res.approved is True _ok("CORRELATION disabled: MES+MNQ both ok") # ============================================================ # 19. CORRELATION FX_MAJORS bundle (V15 parity) # ============================================================ def test_correlation_fx_majors_bundle(): """ V15 bundles 6E/6B/6A/6J/6C into FX_MAJORS. Opening 6E and 6J together must be blocked when correlation is enabled (dollar-index correlation). """ rm = RiskManager(make_config(enable_correlation=True), make_state()) actives = {"6E": make_active_trade()} res = rm.check_entry( entry=make_entry(), sizing=make_sizing(), symbol="6J", active_trades=actives, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.CORRELATION_BLOCKED.value assert res.audit.correlation_blocker == "6E" _ok("CORRELATION: V15 FX_MAJORS bundle -> 6E open blocks 6J") # ============================================================ # 20. MAX_CONTRACTS_EXCEEDED (sanity-check) # ============================================================ def test_max_contracts_exceeded(): """Sizing produced more contracts than the per-symbol cap (regression guard).""" rm = RiskManager(make_config(), make_state()) # MES cap from config_futures is small (8 typically); use 99 to force trip sizing = make_sizing(contracts=99) res = rm.check_entry( entry=make_entry(), sizing=sizing, symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.MAX_CONTRACTS_EXCEEDED.value _ok("MAX_CONTRACTS_EXCEEDED: contracts=99 > MES cap (sanity guard)") # ============================================================ # 21. MAX_RISK_VS_DAILY_BUDGET_EXCEEDED # ============================================================ def test_max_risk_vs_budget_exceeded(): """ daily_pnl=-1400, hard=-1500 -> remaining=$100. Cap=33% -> $33 allowed. sizing.real_risk_usd=50 -> exceeds. """ # Soft must be below current daily_pnl so we don't trip soft-stop first; # this test isolates the budget-cap gate. cfg = make_config( daily_loss_hard_stop=-1500.0, daily_loss_soft_stop=-1500.0, max_risk_vs_daily_budget=0.33, ) state = make_state() state.daily.daily_pnl = -1400.0 rm = RiskManager(cfg, state) sizing = make_sizing(real_risk_usd=50.0) res = rm.check_entry( entry=make_entry(), sizing=sizing, symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) assert res.rule == RiskRule.MAX_RISK_VS_DAILY_BUDGET_EXCEEDED.value _ok("MAX_RISK_VS_DAILY_BUDGET_EXCEEDED: $50 risk on $33 cap -> block") # ============================================================ # 22. RiskAudit populated on rejection # ============================================================ def test_audit_populated_on_rejection(): state = make_state() state.daily.daily_pnl = -800.0 state.daily.executed_count = 4 state.cooldown.cooldown_until["MES"] = ( wednesday_14_utc() + timedelta(minutes=30) ).isoformat() rm = RiskManager(make_config(), state) sizing = make_sizing(real_risk_usd=120.0) res = rm.check_entry( entry=make_entry(), sizing=sizing, symbol="MES", active_trades={}, now_utc=wednesday_14_utc(), ) a: RiskAudit = res.audit assert a.daily_pnl == -800.0 assert a.daily_loss_hard_stop == -1500.0 assert a.daily_loss_soft_stop == -1000.0 assert a.daily_profit_target == 300.0 assert a.daily_remaining_budget == 700.0 # |-1500 - (-800)| assert a.pending_risk_usd == 120.0 assert a.risk_vs_budget_pct == 120.0 / 700.0 assert a.open_trades_count == 0 assert a.cooldown_until is not None assert a.now_utc.startswith("2026-04-29T14:00") # audit_dict serializable d = res.audit_dict() assert d["daily_pnl"] == -800.0 and d["pending_risk_usd"] == 120.0 _ok("RiskAudit fully populated; audit_dict() serializes") # ============================================================ # 23-24. register_sl_hit cooldown durations # ============================================================ def test_register_sl_hit_first_30min_cooldown(): state = make_state() rm = RiskManager(make_config(), state) now = wednesday_14_utc() rm.register_sl_hit("MES", now_utc=now) cu = state.cooldown.cooldown_until["MES"] expiry = datetime.fromisoformat(cu) assert expiry == now + timedelta(minutes=30) assert state.metadata["consecutive_sl_count"]["MES"] == 1 _ok("register_sl_hit: first SL -> 30min cooldown, consecutive=1") def test_register_sl_hit_consecutive_2h_cooldown(): state = make_state() state.metadata.setdefault("consecutive_sl_count", {})["MES"] = 2 rm = RiskManager(make_config(), state) now = wednesday_14_utc() rm.register_sl_hit("MES", now_utc=now) cu = state.cooldown.cooldown_until["MES"] expiry = datetime.fromisoformat(cu) assert expiry == now + timedelta(hours=2) assert state.metadata["consecutive_sl_count"]["MES"] == 3 _ok("register_sl_hit: prior consecutive=2 -> 2h cooldown (circuit breaker)") # ============================================================ # 25-26. register_tp_hit + update_daily_pnl routing # ============================================================ def test_register_tp_hit_resets_consecutive_sl(): state = make_state() state.metadata.setdefault("consecutive_sl_count", {})["MES"] = 3 rm = RiskManager(make_config(), state) rm.register_tp_hit("MES") assert state.metadata["consecutive_sl_count"]["MES"] == 0 assert "MES" in state.cooldown.last_trade_close_at _ok("register_tp_hit: consecutive_sl reset to 0") def test_update_daily_pnl_routes_to_brain_counters(): state = make_state() rm = RiskManager(make_config(), state) rm.update_daily_pnl(50.0, is_win=True, brain="TF") rm.update_daily_pnl(-30.0, is_win=False, brain="MR") assert state.daily.daily_pnl == 20.0 assert state.session_pnl == 20.0 assert state.brain.tf_wins == 1 and state.brain.mr_losses == 1 _ok("update_daily_pnl: routes win/loss to brain.{tf,mr}_{wins,losses}") # ============================================================ # MAIN # ============================================================ def main() -> int: print("test_risk_manager.py") test_check_entry_all_clear_approved() test_sizing_skip_propagates() test_halted_blocks() test_daily_loss_hard_stop_triggers_auto_halt() test_daily_loss_soft_stop_blocks_no_halt() test_daily_loss_soft_stop_recovers_passes() test_daily_profit_target_blocks_no_halt() test_max_open_trades_reached() test_max_daily_trades_reached() test_force_flat_at_2108_blocks() test_force_flat_at_2107_passes() test_last_friday_15_00_blocks() test_last_friday_14_59_passes() test_4_friday_month_only_last_triggers() test_5_friday_month_only_last_triggers() test_cooldown_active_blocks() test_cooldown_expired_passes_and_clears() test_correlation_blocked_equity_indices() test_correlation_disabled_passes() test_correlation_fx_majors_bundle() test_max_contracts_exceeded() test_max_risk_vs_budget_exceeded() test_audit_populated_on_rejection() test_register_sl_hit_first_30min_cooldown() test_register_sl_hit_consecutive_2h_cooldown() test_register_tp_hit_resets_consecutive_sl() test_update_daily_pnl_routes_to_brain_counters() print("ALL 27 TESTS PASSED") return 0 if __name__ == "__main__": sys.exit(main())