""" Smoke tests for brain/brain_mr.py. Mirrors the brain_tf test structure: FakeAIClient + FakeLoggerBundle to avoid network calls and assert structured rejection events. Run: cd ~/apex_v16 python -m tests.test_brain_mr """ from __future__ import annotations import asyncio import json import logging import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from analysis.tech_snapshot import TechSnapshot from brain.ai_client import AIResponse from brain import brain_mr as bmr_mod from brain.brain_mr import ( BrainMR, MR_ASSET_PROFILES, GENERIC_PROFILE, GRACE_INDICES_EXTRA_MIN, ) class _SessionPatch: """Context manager to pin _classify_session for deterministic time-stop tests.""" def __init__(self, session: str) -> None: self.session = session self._orig = None def __enter__(self): self._orig = bmr_mod._classify_session bmr_mod._classify_session = lambda _h: self.session return self def __exit__(self, *exc): bmr_mod._classify_session = self._orig from core.config import RuntimeConfig, RunMode, AccountKind from core.contracts import ( BrainContext, BrainDecision, Direction, EntryDecision, MarketStructure, Regime, TradeAction, TradeEntry, TradeRuntime, ) # ============================================================ # ADAPTER — evaluate_entry now returns EntryEvalResult; legacy tests # assert on the inner decision. Keeps the diff minimal. # ============================================================ async def _eval(brain, symbol, snap, *args, **kwargs): # Auto-derive bias_data from snap so legacy tests don't need to # construct a BiasData explicitly. Production code passes it from # BiasResolver in orchestrator._scan_one. if "bias_data" not in kwargs: from analysis.bias import BiasData kwargs["bias_data"] = BiasData( bias=snap.bias, allowed_direction=snap.allowed_direction, h1_compatibility=getattr(snap, "h1_compatibility", 1.0), h1_reason=getattr(snap, "h1_reason", "test"), ambiguous=False, ) result = await brain.evaluate_entry(symbol, snap, *args, **kwargs) return result.decision # ============================================================ # FAKES # ============================================================ class FakeAIClient: def __init__(self, canned: AIResponse | None = None) -> None: self.canned = canned or AIResponse(text=None, error_kind="unknown") self.calls: list[tuple[str, dict]] = [] async def ask_for_decision( self, prompt: str, max_tokens: int = 600, where=None, ) -> AIResponse: self.calls.append(("ask_for_decision", { "prompt_preview": prompt[:80], "max_tokens": max_tokens, "where": where, })) return self.canned class FakeJsonlLogger: def __init__(self) -> None: self.events: list[dict] = [] def write(self, event: str, **fields) -> None: self.events.append({"event": event, **fields}) class FakeLoggerBundle: def __init__(self) -> None: self.brain_log = FakeJsonlLogger() self.system = logging.getLogger("test.brain_mr.fake") def make_ai_response(payload: dict) -> AIResponse: return AIResponse(text=json.dumps(payload), attempts=1) def _to_snap(d: dict) -> TechSnapshot: """Build a TechSnapshot from a legacy test-dict (mirrors test_brain_tf).""" return TechSnapshot( symbol=d["symbol"], price=float(d.get("price", 0.0)), open=float(d.get("open", d.get("price", 0.0))), candle_time=int(d.get("candle_time", 0) or 0), is_candle_closed=bool(d.get("is_candle_closed", True)), candle_age_seconds=float(d.get("candle_age_seconds", 10.0)), rsi=float(d.get("rsi", 50.0)), rsi_prev=float(d.get("rsi_prev", d.get("rsi", 50.0))), rsi_h1=float(d.get("rsi_h1", 50.0)), rsi_h4=float(d.get("rsi_h4", 50.0)), atr_m5_points=float(d.get("atr_m5_points", 0.0)), atr_ratio=float(d.get("atr_ratio", 1.0)), vol_regime=d.get("vol_regime", "NORMAL"), vol_spike=bool(d.get("vol_spike", False)), market_structure=d.get("market_structure", "RANGING"), h1_struct_bull=bool(d.get("h1_struct_bull", False)), h1_struct_bear=bool(d.get("h1_struct_bear", False)), trend_maturity=int(d.get("trend_maturity", 0)), regime=d.get("regime", "RANGING"), regime_reason=d.get("regime_reason", ""), regime_near_trending=list(d.get("regime_near_trending", [])), deviation_pct=float(d.get("deviation_pct", 0.0)), divergence=d.get("divergence", "NONE"), macd_decelerating=bool(d.get("macd_decelerating", False)), macd_hist_last=float(d.get("macd_hist_last", 0.0)), candle_strength=float(d.get("candle_strength", 1.0)), hammer=bool(d.get("hammer", False)), shooting_star=bool(d.get("shooting_star", False)), bull_engulfing=bool(d.get("bull_engulfing", False)), bear_engulfing=bool(d.get("bear_engulfing", False)), doji=bool(d.get("doji", False)), doji_type=d.get("doji_type"), piercing=bool(d.get("piercing", False)), dark_cloud=bool(d.get("dark_cloud", False)), morning_star=bool(d.get("morning_star", False)), evening_star=bool(d.get("evening_star", False)), volume_weak=bool(d.get("volume_weak", False)), buy_absorption=bool(d.get("buy_absorption", False)), sell_absorption=bool(d.get("sell_absorption", False)), vwap=float(d.get("vwap", d.get("price", 0.0))), vwap_deviation_pct=float(d.get("vwap_deviation_pct", 0.0)), bias=d.get("bias", "NEUTRO"), allowed_direction=d.get("allowed_direction", "NONE"), h1_compatibility=float(d.get("h1_compatibility", 1.0)), h1_reason=d.get("h1_reason", ""), swing_data=dict(d.get("swing_data", {})), consecutive_sl_count=int(d.get("consecutive_sl_count", 0)), tick_size=float(d.get("tick_size", 0.0)), tick_value=float(d.get("tick_value", 0.0)), ) def make_config() -> RuntimeConfig: return RuntimeConfig(mode=RunMode.PAPER, account=AccountKind.INELIGIBLE) # ── Tech generator: BUY at oversold extreme on 6E (FX) ─────────── def make_tech_buy_extreme(symbol: str = "6E") -> dict: """Tech with valid MR BUY conditions: RSI oversold, ATR ok, FX (no index restriction).""" return { "symbol": symbol, "price": 1.0850, "open": 1.0852, "rsi": 28.0, # < 32 oversold "rsi_prev": 30.0, "rsi_h1": 38.0, "rsi_h4": 45.0, "atr_ratio": 1.30, # <= 1.8 cap "atr_m5_points": 0.0040, # 40 ticks at 0.00005 tick "candle_strength": 1.0, "vwap": 1.0855, "vwap_deviation_pct": -0.05, "market_structure": MarketStructure.RANGING.value, "regime": Regime.RANGING.value, "vol_regime": "NORMAL", "trend_maturity": 2, "candle_time": 1714291800.0, "bias": "NEUTRO", "allowed_direction": "BOTH", "h1_compatibility": 0.8, "h1_reason": "ranging", "swing_data": {"swing_found": False}, "divergence": "BULLISH", "macd_decelerating": True, "deviation_pct": -0.45, "consecutive_sl_count": 0, # PA: bullish reversal pattern present "hammer": True, } def make_tech_buy_index(symbol: str = "MNQ", protrend: bool = True) -> dict: """MNQ BUY oversold; protrend=True -> BULLISH structure (allowed).""" base = { "symbol": symbol, "price": 18000.0, "open": 18002.0, "rsi": 28.0, "rsi_prev": 30.0, "rsi_h1": 38.0, "rsi_h4": 48.0, "atr_ratio": 1.30, "atr_m5_points": 50.0, "candle_strength": 1.0, "vwap": 18005.0, "vwap_deviation_pct": -0.05, "vol_regime": "NORMAL", "trend_maturity": 4, "candle_time": 1714291800.0, "bias": "BULLISH", "allowed_direction": "BOTH", "h1_compatibility": 1.0, "h1_reason": "trending", "swing_data": {"swing_found": False}, "divergence": "BULLISH", "macd_decelerating": True, "deviation_pct": -0.45, "consecutive_sl_count": 0, "hammer": True, } if protrend: base["market_structure"] = MarketStructure.BULLISH_EXPANSION.value base["regime"] = Regime.TRENDING.value else: base["market_structure"] = MarketStructure.BEARISH_EXPANSION.value base["regime"] = Regime.TRENDING.value return base def make_entry( direction: str = "BUY", symbol: str = "6E", structure: str = MarketStructure.RANGING.value, regime: str = Regime.RANGING.value, rsi_m5: float = 28.0, rsi_h4: float = 45.0, h1_compat: float = 0.8, confidence: int = 70, entry_price: float = 1.0850, sl_price: float = 1.0830, tp_price: float = 1.0880, ) -> TradeEntry: return TradeEntry( symbol=symbol, brain_name="MR", direction=direction, contracts=1, entry_price=entry_price, sl_price=sl_price, tp_price=tp_price, opened_at=datetime(2026, 4, 28, 14, 0, tzinfo=timezone.utc), rsi_m5_at_entry=rsi_m5, rsi_h1_at_entry=38.0, rsi_h4_at_entry=rsi_h4, atr_ratio_at_entry=1.30, market_structure_at_entry=structure, regime_at_entry=regime, h1_compat_at_entry=h1_compat, confidence_at_entry=confidence, ) def make_ctx( entry: TradeEntry | None = None, runtime: TradeRuntime | None = None, tech: dict | None = None, ) -> BrainContext: return BrainContext( entry=entry or make_entry(), runtime=runtime or TradeRuntime(minutes_open=10.0, net_profit_usd=5.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech if tech is not None else make_tech_buy_extreme()), ) def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # ENTRY: PRE-VALIDATION # ============================================================ async def test_pre_val_rsi_not_extreme_buy(): cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["rsi"] = 38.0 # not oversold result = await _eval(brain,"6E", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "RSI_NOT_EXTREME" assert events[0]["side"] == "oversold" _ok("pre-val: RSI not oversold for BUY -> None, no AI") async def test_pre_val_rsi_not_extreme_sell(): cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["rsi"] = 60.0 tech["allowed_direction"] = "SELL" result = await _eval(brain,"6E", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 _ok("pre-val: RSI not overbought for SELL -> None, no AI") async def test_pre_val_indices_not_pro_trend(): """MNQ BUY oversold but H1 structure BEARISH -> reject.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_index(protrend=False) # BEARISH structure result = await _eval(brain,"MNQ", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "INDICES_NOT_PRO_TREND" assert events[0]["market_structure"] == MarketStructure.BEARISH_EXPANSION.value _ok("pre-val: index BUY in BEARISH struct -> reject INDICES_NOT_PRO_TREND") async def test_pre_val_indices_ranging(): """MES BUY oversold but RANGING regime -> reject.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_index("MES", protrend=True) tech["regime"] = Regime.RANGING.value result = await _eval(brain,"MES", _to_snap(tech)) assert result is None events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "INDICES_RANGING" _ok("pre-val: index in RANGING regime -> reject INDICES_RANGING") async def test_pre_val_atr_too_high(): cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["atr_ratio"] = 2.0 # > 1.8 cap result = await _eval(brain,"6E", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "ATR_TOO_HIGH" assert events[0]["atr_ratio"] == 2.0 _ok("pre-val: ATR ratio > 1.8 cap -> None, no AI") # ── PRE-VAL 5: MACD accelerating against (V18 12-mag) ────── # Incidente 6B BUY: MACD accelerando bearish contro → SL -$375. async def test_pre_val_macd_accelerating_against_buy(): """BUY + macd_decel=False + hist<0 -> MACD_ACCELERATING_AGAINST.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["macd_decelerating"] = False tech["macd_hist_last"] = -0.0021 # bearish, accelerating against BUY result = await _eval(brain, "6E", _to_snap(tech)) assert result is None assert len(ai.calls) == 0, "AI must NOT be called when gate rejects" events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "MACD_ACCELERATING_AGAINST" assert events[0]["macd_decelerating"] is False assert events[0]["macd_hist_last"] == -0.0021 _ok("pre-val: BUY + MACD bearish accelerating -> MACD_ACCELERATING_AGAINST") async def test_pre_val_macd_accelerating_against_sell(): """SELL + macd_decel=False + hist>0 -> MACD_ACCELERATING_AGAINST.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() # Symmetric SELL setup: RSI overbought + bullish reversal pattern. tech["rsi"] = 72.0 tech["rsi_prev"] = 70.0 tech["divergence"] = "BEARISH" tech["shooting_star"] = True tech["hammer"] = False tech["allowed_direction"] = "BOTH" tech["deviation_pct"] = 0.45 tech["macd_decelerating"] = False tech["macd_hist_last"] = 0.0034 # bullish, accelerating against SELL result = await _eval(brain, "6E", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "MACD_ACCELERATING_AGAINST" assert events[0]["macd_hist_last"] == 0.0034 _ok("pre-val: SELL + MACD bullish accelerating -> MACD_ACCELERATING_AGAINST") async def test_pre_val_macd_accelerating_in_favor_passes(): """BUY + macd_decel=False + hist>0 -> gate passes (MACD accelera a favore).""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["macd_decelerating"] = False tech["macd_hist_last"] = 0.0018 # bullish, accelerating WITH BUY await _eval(brain, "6E", _to_snap(tech)) # The gate must NOT have rejected. Downstream may reject for other # reasons (AI stub returns no text -> AI_NONE), but never with our rule. rejected = [ e for e in log.brain_log.events if e["event"] == "entry_rejected" and e.get("rule") == "MACD_ACCELERATING_AGAINST" ] assert not rejected, f"gate should pass when MACD favors direction: {rejected}" _ok("pre-val: BUY + MACD bullish accelerating -> gate passes") async def test_pre_val_macd_decelerating_passes_either_sign(): """BUY + macd_decel=True -> gate passes regardless of hist sign.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["macd_decelerating"] = True tech["macd_hist_last"] = -0.0025 # bearish but decelerating -> OK await _eval(brain, "6E", _to_snap(tech)) rejected = [ e for e in log.brain_log.events if e["event"] == "entry_rejected" and e.get("rule") == "MACD_ACCELERATING_AGAINST" ] assert not rejected, "decelerating momentum must not trigger the gate" _ok("pre-val: macd_decel=True -> gate passes (any sign)") # ============================================================ # ENTRY: POST-VALIDATION # ============================================================ async def test_post_val_conf_below_with_pattern(): """Pattern confirmed -> floor 60. AI returns conf=55 -> reject.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 55, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 4.0, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) result = await _eval(brain,"6E", _to_snap(make_tech_buy_extreme())) assert result is None events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "POST_VAL_CONF" _ok("post-val: conf 55 < 60 floor with pattern -> reject") async def test_post_val_conf_below_without_pattern(): """No reversal pattern -> floor 75. AI returns conf=70 -> reject.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 70, # below 75 floor "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 4.0, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["hammer"] = False # no reversal pattern result = await _eval(brain,"6E", _to_snap(tech)) assert result is None events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert events[0]["rule"] == "POST_VAL_CONF_NO_PATTERN" assert events[0]["pattern_confirmed"] is False assert events[0]["ai_confidence"] == 70 _ok("post-val: conf 70 < 75 floor without pattern -> reject") async def test_post_val_c95_h4_weak(): cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 95, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 3.0, "rr_multiplier": 0.67, "key_risk": "ok", "reason": "ok", })) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["rsi_h4"] = 35.0 # against BUY result = await _eval(brain,"6E", _to_snap(tech)) assert result is None _ok("post-val: C95% + H4 RSI 35 vs BUY -> reject") async def test_post_val_rr_out_of_range(): """ TP variante γ parity with brain_tf: AI emits rr_multiplier outside advisory prompt-range [0.17, 0.67] -> POST_VAL_RR_RANGE rejection. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 75, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 4.0, "rr_multiplier": 0.05, # below 0.17 prompt-range lower bound "key_risk": "ok", "reason": "ok", })) log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) result = await _eval(brain, "6E", _to_snap(make_tech_buy_extreme())) assert result is None events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["rule"] == "POST_VAL_RR_RANGE" _ok("post-val MR: rr_multiplier 0.05 outside [0.17, 0.67] -> POST_VAL_RR_RANGE") async def test_post_val_sl_out_of_profile_range(): """ BUG 9 fix (parity with brain_tf): AI returns sl_atr_multiplier above profile.sl_range -> POST_VAL_SL_RANGE. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 75, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 8.0, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) result = await _eval(brain, "6E", _to_snap(make_tech_buy_extreme())) assert result is None events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["rule"] == "POST_VAL_SL_RANGE" _ok("post-val MR: sl_atr 8.0 > profile sl_range max 5.0 -> POST_VAL_SL_RANGE") # ============================================================ # ENTRY: structured logging # ============================================================ async def test_rejection_logging_rsi_not_extreme(): cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) tech = make_tech_buy_extreme() tech["rsi"] = 40.0 # < 50 -> direction BUY, but >= 32 -> not oversold await _eval(brain,"6E", _to_snap(tech)) events = [e for e in log.brain_log.events if e["event"] == "entry_rejected"] ev = events[0] assert ev["rule"] == "RSI_NOT_EXTREME" assert ev["rsi_m5"] == 40.0 assert ev["rsi_threshold"] == 32.0 assert ev["side"] == "oversold" assert ev["brain"] == "MR" _ok("logging: RSI_NOT_EXTREME with full fields") async def test_ai_rejection_with_details_logged(): """AI returns approved=false + rejection_details -> entry_rejected event carries ai_rule_failed/ai_rejection_detail.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": False, "confidence": 60, "direction": "BUY", "risk_multiplier": 0.5, "sl_atr_multiplier": 3.5, "rr_multiplier": 0.30, "key_risk": "weak", "reason": "no extreme + no pattern", "rejection_details": { "rule_failed": "NO_PATTERN_LOW_CONF", "detail": "RSI 28 borderline, no reversal candle confluence", }, })) log_bundle = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log_bundle) await _eval(brain, "6E", _to_snap(make_tech_buy_extreme())) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] ev = next(e for e in events if e["rule"] == "AI_REJECTED") assert ev["ai_rule_failed"] == "NO_PATTERN_LOW_CONF" assert "borderline" in ev["ai_rejection_detail"] _ok("logging: MR AI_REJECTED carries ai_rule_failed + ai_rejection_detail") async def test_ai_rejection_missing_details_safe(): cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": False, "confidence": 60, "direction": "BUY", "risk_multiplier": 0.5, "sl_atr_multiplier": 3.5, "rr_multiplier": 0.30, "key_risk": "n/a", "reason": "no", # rejection_details intentionally missing })) log_bundle = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log_bundle) await _eval(brain, "6E", _to_snap(make_tech_buy_extreme())) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] ev = next(e for e in events if e["rule"] == "AI_REJECTED") assert ev["ai_rule_failed"] == "UNSPECIFIED" _ok("logging: MR AI rejection without rejection_details -> UNSPECIFIED") # ============================================================ # ENTRY: HAPPY PATH # ============================================================ async def test_entry_happy_with_pattern(): """ Valid MR setup with reversal pattern -> EntryDecision with SL prices. TP variante γ: tp_price=0.0 sentinel + rr_multiplier; orchestrator finalizes tp_price post-sizing. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 65, # >= 60 with pattern "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 3.5, "rr_multiplier": 0.50, "key_risk": "h4 weak", "reason": "deep oversold + hammer", })) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["atr_m5_points"] = 0.0050 result = await _eval(brain,"6E", _to_snap(tech)) assert isinstance(result, EntryDecision) assert result.direction == "BUY" assert result.confidence == 65 assert result.sl_price < result.entry_price assert result.tp_price == 0.0 # sentinel assert result.rr_multiplier == 0.50 assert result.metadata["brain"] == "MR" assert result.metadata["rr_multiplier_ai"] == 0.50 assert result.metadata["pattern_confirmed"] is True assert result.metadata["profile_origin"] == "EURUSD" _ok("happy: MR setup with pattern + conf 65 -> EntryDecision (TP resolved post-sizing)") async def test_entry_happy_without_pattern_high_conf(): cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 80, # >= 75 without pattern "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 3.5, "rr_multiplier": 0.50, "key_risk": "no pattern", "reason": "deep oversold no pattern", })) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["hammer"] = False # no reversal pattern tech["atr_m5_points"] = 0.0050 result = await _eval(brain,"6E", _to_snap(tech)) assert isinstance(result, EntryDecision) assert result.metadata["pattern_confirmed"] is False _ok("happy: no pattern + conf 80 -> EntryDecision") async def test_generic_profile_warning(): """ Symbol senza profile dedicato -> usa GENERIC_PROFILE + emette warning una sola volta. Dopo Fix A V16, tutti gli asset attivi hanno profile dedicato; usiamo un symbol fittizio (non in MR_ASSET_PROFILES) per esercitare il path GENERIC fallback. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "approved": True, "confidence": 80, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": GENERIC_PROFILE["sl_range"][0] + 0.5, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) log = FakeLoggerBundle() brain = BrainMR(cfg, ai, logger=log) fake_symbol = "ZZZ_TEST_NO_PROFILE" assert fake_symbol not in MR_ASSET_PROFILES tech = make_tech_buy_extreme() tech["symbol"] = fake_symbol # the warning fires before _compute_sl_only is reached. tech["price"] = 75.50 tech["open"] = 75.55 tech["atr_m5_points"] = 0.30 tech["hammer"] = False assert fake_symbol not in brain._warned_generic_profile _ = await _eval(brain, fake_symbol, _to_snap(tech)) assert fake_symbol in brain._warned_generic_profile _ok(f"generic profile: {fake_symbol} (no profile) tracked + warned once") # ============================================================ # EXIT: manage_exit # ============================================================ async def test_exit_time_stop_deep_loss(): """P&L < -50 + minutes >= deep_threshold -> EXIT, no AI.""" cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) ctx = make_ctx( runtime=TradeRuntime(minutes_open=200.0, net_profit_usd=-80.0, last_exit_eval_time=0.0), ) with _SessionPatch("LONDON"): # 6E LONDON: deep=70 decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert "Time Stop MR" in decision.reason assert decision.metadata["trigger"] == "time_stop_deep" assert len(ai.calls) == 0 _ok("exit: time stop deep loss -> EXIT, no AI") async def test_exit_time_stop_breakeven_failed(): cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) ctx = make_ctx( runtime=TradeRuntime(minutes_open=200.0, net_profit_usd=-5.0, last_exit_eval_time=0.0), ) with _SessionPatch("LONDON"): # 6E LONDON: breakeven=115 decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert "BE failed" in decision.reason assert decision.metadata["trigger"] == "time_stop_be_failed" assert len(ai.calls) == 0 _ok("exit: time stop BE failed -> EXIT, no AI") async def test_exit_grace_period_indices_active(): """MNQ in break-even, RSI toward mean, within grace -> NOT exit. Pinned to LONDON_NY (deep=65, breakeven=110, grace_limit=80). minutes_open=70: 70>=65 enters grace block, 70<80 grace ACTIVE, no exit. Falls through to AI which says HOLD. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({"exit_now": False, "reason": "wait"})) brain = BrainMR(cfg, ai) entry = make_entry(symbol="MNQ", structure=MarketStructure.BULLISH_EXPANSION.value, regime=Regime.TRENDING.value, rsi_m5=28.0, entry_price=18000.0, sl_price=17970.0, tp_price=18050.0) tech = make_tech_buy_index("MNQ", protrend=True) tech["candle_time"] = 1000.0 tech["price"] = 18000.0 # no progress -> no trailing/partial tech["rsi"] = 35.0 # moved up from entry 28 -> toward mean ctx = BrainContext( entry=entry, runtime=TradeRuntime(minutes_open=70.0, net_profit_usd=-5.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) with _SessionPatch("LONDON_NY"): decision = await brain.manage_exit(ctx) assert decision.action != TradeAction.EXIT.value, \ f"unexpected EXIT: {decision.reason} meta={decision.metadata}" _ok("exit: grace period indices active (BE + toward mean) -> not EXIT") async def test_exit_grace_period_indices_expired(): """Grace period expired (minutes >= grace_limit) -> EXIT grace_expired. Pinned to LONDON_NY (deep=65, grace_limit=80). minutes_open=85 >=80 -> grace_expired branch fires. """ cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) entry = make_entry(symbol="MNQ", structure=MarketStructure.BULLISH_EXPANSION.value, regime=Regime.TRENDING.value, rsi_m5=28.0, entry_price=18000.0, sl_price=17970.0, tp_price=18050.0) tech = make_tech_buy_index("MNQ", protrend=True) tech["rsi"] = 35.0 ctx = BrainContext( entry=entry, runtime=TradeRuntime(minutes_open=85.0, net_profit_usd=-5.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) with _SessionPatch("LONDON_NY"): decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert decision.metadata["trigger"] == "grace_expired" assert len(ai.calls) == 0 _ok("exit: grace period expired -> EXIT grace_expired") async def test_exit_trailing_emergency(): """progress > 30, TP wide enough -> MOVE_SL trailing.""" cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) # entry 1.0850, tp 1.0900 -> tp_distance 0.005. atr 0.001 -> trailing_dist=0.0015. # tp_distance (0.005) > trailing_dist*2.5 (0.00375) -> trailing fires. # progress: price=1.0875 -> (1.0875-1.0850)/(1.0900-1.0850) = 50% entry = make_entry(entry_price=1.0850, sl_price=1.0820, tp_price=1.0900) tech = make_tech_buy_extreme() tech["price"] = 1.0875 tech["atr_m5_points"] = 0.001 ctx = BrainContext( entry=entry, runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=20.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.MOVE_SL.value assert decision.metadata["sl_target"] == "trailing" assert decision.metadata["trailing_atr_mult"] == 1.5 assert decision.move_sl_to is not None # For BUY trailing SL = price - 1.5*atr = 1.0875 - 0.0015 = 1.0860 assert abs(decision.move_sl_to - 1.0860) < 1e-6 assert len(ai.calls) == 0 _ok("exit: trailing emergency progress 50% -> MOVE_SL trailing") async def test_exit_rsi50_partial_with_set_be(): """BUY: RSI crossed 48, P&L>0, progress>15, !rsi50_partial_done -> PARTIAL_50 + set_be.""" cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) entry = make_entry(entry_price=1.0850, sl_price=1.0820, tp_price=1.0900) tech = make_tech_buy_extreme() tech["rsi"] = 50.0 # >= 48 -> reached_mean tech["price"] = 1.0860 # progress 20% (above 15 threshold) ctx = BrainContext( entry=entry, runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=8.0, last_exit_eval_time=0.0, rsi50_partial_done=False), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.PARTIAL_50.value assert decision.metadata["trigger"] == "rsi50_cross" assert decision.metadata["set_be_after_partial"] is True assert decision.metadata["be_price"] == entry.entry_price assert decision.metadata["rsi50_partial_done"] is True assert len(ai.calls) == 0 _ok("exit: RSI50 cross + P&L>0 -> PARTIAL_50 + set_be metadata") async def test_exit_auto_partial_50(): """progress >= 50, !partial_done -> PARTIAL_50 auto + set_be_after_partial. V18: soglia abbassata 65→50 e aggiunto set_be per parità con Brain TF. """ cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) entry = make_entry(entry_price=1.0850, sl_price=1.0820, tp_price=1.0900) tech = make_tech_buy_extreme() tech["price"] = 1.0885 # progress (1.0885-1.0850)/(1.0900-1.0850) = 70% tech["rsi"] = 47.0 # NOT yet reached_mean (<48), so RSI50 partial doesn't fire tech["tick_size"] = 0.00005 # 6E ctx = BrainContext( entry=entry, runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=15.0, last_exit_eval_time=0.0, partial_done=False), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) # Trailing fires first if conditions met. Check that progress > 30 -> trailing. # tp_distance = 0.005, atr default tech 0.0040 -> trailing_dist=0.006. # 0.005 > 0.006*2.5 = 0.015 ? NO -> trailing skipped. # So auto_partial_50 fires. assert decision.action == TradeAction.PARTIAL_50.value assert decision.metadata["trigger"] == "auto_partial_50" assert decision.metadata["set_be_after_partial"] is True assert decision.metadata["be_price"] == entry.entry_price _ok("exit: progress >= 50 -> auto PARTIAL_50 + set_be (be_price tick-aligned)") async def test_exit_same_candle_dedup(): cfg = make_config() ai = FakeAIClient() brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["candle_time"] = 1000.0 tech["price"] = 1.0851 # tiny progress, no trailing ctx = BrainContext( entry=make_entry(), runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=2.0, last_exit_eval_time=1000.0), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.HOLD.value assert "stessa candela" in decision.reason assert len(ai.calls) == 0 _ok("exit: same-candle dedup -> HOLD, no AI") async def test_exit_ai_says_exit(): cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "exit_now": True, "reason": "regime flipped" })) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["price"] = 1.0851 tech["candle_time"] = 5000.0 ctx = BrainContext( entry=make_entry(), runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=2.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert "regime flipped" in decision.reason assert decision.metadata["evaluated_candle_time"] == 5000.0 _ok("exit: AI exit_now=True -> EXIT") async def test_exit_ai_says_partial(): cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "exit_now": False, "partial_close": True, "reason": "near vwap target" })) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["price"] = 1.0851 tech["candle_time"] = 6000.0 ctx = BrainContext( entry=make_entry(), runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=2.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.PARTIAL_50.value assert decision.metadata["trigger"] == "ai_partial" _ok("exit: AI partial_close=True -> PARTIAL_50") async def test_exit_ai_error_holds(): cfg = make_config() ai = FakeAIClient(canned=AIResponse(text=None, error_kind="overload", attempts=3)) brain = BrainMR(cfg, ai) tech = make_tech_buy_extreme() tech["price"] = 1.0851 tech["candle_time"] = 7000.0 ctx = BrainContext( entry=make_entry(), runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=2.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.HOLD.value assert "AI error" in decision.reason _ok("exit: AI error -> HOLD default") # ============================================================ # RUN # ============================================================ async def main_async() -> int: print("test_brain_mr.py") # Pre-val (5) await test_pre_val_rsi_not_extreme_buy() await test_pre_val_rsi_not_extreme_sell() await test_pre_val_indices_not_pro_trend() await test_pre_val_indices_ranging() await test_pre_val_atr_too_high() # Pre-val 5: MACD gate (V18) await test_pre_val_macd_accelerating_against_buy() await test_pre_val_macd_accelerating_against_sell() await test_pre_val_macd_accelerating_in_favor_passes() await test_pre_val_macd_decelerating_passes_either_sign() # Post-val await test_post_val_conf_below_with_pattern() await test_post_val_conf_below_without_pattern() await test_post_val_c95_h4_weak() await test_post_val_rr_out_of_range() await test_post_val_sl_out_of_profile_range() # Structured logging await test_rejection_logging_rsi_not_extreme() await test_ai_rejection_with_details_logged() await test_ai_rejection_missing_details_safe() # Happy (3) await test_entry_happy_with_pattern() await test_entry_happy_without_pattern_high_conf() await test_generic_profile_warning() # Exit (9) await test_exit_time_stop_deep_loss() await test_exit_time_stop_breakeven_failed() await test_exit_grace_period_indices_active() await test_exit_grace_period_indices_expired() await test_exit_trailing_emergency() await test_exit_rsi50_partial_with_set_be() await test_exit_auto_partial_50() await test_exit_same_candle_dedup() await test_exit_ai_says_exit() await test_exit_ai_says_partial() await test_exit_ai_error_holds() print("ALL TESTS PASSED") return 0 def main() -> int: return asyncio.run(main_async()) if __name__ == "__main__": sys.exit(main())