""" Smoke tests for brain/brain_tf.py. Uses FakeAIClient to avoid hitting Anthropic; canned AIResponse drives each post-validation and entry/exit branch deterministically. Run: cd ~/apex_v16 python -m tests.test_brain_tf """ from __future__ import annotations import asyncio import json import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from analysis.price_action import extract_pa_signals, score_pa from analysis.tech_snapshot import TechSnapshot from brain.ai_client import AIResponse from brain.brain_tf import BrainTF, TF_ASSET_PROFILES, GENERIC_PROFILE from core.config import RuntimeConfig, RunMode, AccountKind from core.contracts import ( BrainContext, BrainDecision, Direction, EntryDecision, MarketStructure, Regime, TradeAction, TradeEntry, TradeRuntime, utc_now, ) # ============================================================ # 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: """ Mimics AIClient.ask_for_decision: returns a pre-set AIResponse. Tracks call count + last prompt for assertions. """ 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: """In-memory JSONL sink for asserting structured rejection events.""" def __init__(self) -> None: self.events: list[dict] = [] def write(self, event: str, **fields) -> None: self.events.append({"event": event, **fields}) class FakeLoggerBundle: """Just enough of LoggerBundle for BrainTF._reject to find brain_log.""" def __init__(self) -> None: self.brain_log = FakeJsonlLogger() # Also expose a system attribute in case fall-back path runs import logging as _logging self.system = _logging.getLogger("test.brain_tf.fake") def make_ai_response(payload: dict) -> AIResponse: """Wrap a dict as a successful AIResponse (JSON text).""" return AIResponse(text=json.dumps(payload), attempts=1) def _to_snap(d: dict) -> TechSnapshot: """ Build a TechSnapshot from a legacy test-dict so existing test factories can stay dict-mutable. Defaults fill any missing field so per-test fixtures only specify what's relevant. """ 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) def make_tech_buy_valid(symbol: str = "MES") -> dict: """Tech dict with valid TF BUY entry conditions for MES.""" return { "symbol": symbol, "price": 5800.00, "open": 5798.00, # green candle (no capitulation) "rsi": 45.0, # in [42, 58] "rsi_prev": 43.0, # bouncing "rsi_h1": 52.0, "rsi_h4": 55.0, "atr_ratio": 1.10, # > 0.8 "atr_m5_points": 8.0, # ~32 ticks at 0.25 tick_size "candle_strength": 1.0, "vwap": 5798.0, "vwap_deviation_pct": -0.03, "market_structure": MarketStructure.BULLISH_EXPANSION.value, "regime": Regime.TRENDING.value, "vol_regime": "NORMAL", "trend_maturity": 3, "candle_time": 1714291800.0, "bias": "BULLISH", "allowed_direction": "BUY", "h1_compatibility": 1.0, "h1_reason": "trend", "swing_data": {"swing_found": False}, } def make_entry(direction: str = "BUY", symbol: str = "MES", structure: str = MarketStructure.BULLISH_EXPANSION.value, rsi_h4: float = 55.0, h1_compat: float = 1.0, confidence: int = 75) -> TradeEntry: return TradeEntry( symbol=symbol, brain_name="TF", direction=direction, contracts=1, entry_price=5800.0 if symbol.startswith("M") else 1.0850, sl_price=5790.0 if direction == "BUY" else 5810.0, tp_price=5820.0 if direction == "BUY" else 5780.0, opened_at=datetime(2026, 4, 28, 14, 0, tzinfo=timezone.utc), rsi_m5_at_entry=45.0, rsi_h1_at_entry=52.0, rsi_h4_at_entry=rsi_h4, atr_ratio_at_entry=1.10, market_structure_at_entry=structure, regime_at_entry=Regime.TRENDING.value, h1_compat_at_entry=h1_compat, confidence_at_entry=confidence, ) def _ok(label: str) -> None: print(f" ok {label}") # ============================================================ # ENTRY: PRE-VALIDATION (no AI call) # ============================================================ async def test_pre_val_rsi_outside_zone(): """RSI outside [42,58] -> reject without calling AI.""" cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["rsi"] = 65.0 # too high for TF result = await _eval(brain,"MES", _to_snap(tech)) assert result is None assert len(ai.calls) == 0, "AI should NOT be called when pre-val fails" _ok("pre-val: RSI outside TF zone -> None, no AI") async def test_pre_val_atr_below_floor(): cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["atr_ratio"] = 0.5 # below 0.8 floor result = await _eval(brain,"MES", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 _ok("pre-val: ATR ratio < 0.8 -> None, no AI") async def test_pre_val_capitulation_candle(): cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["candle_strength"] = 2.0 tech["price"] = 5790.0 tech["open"] = 5800.0 # bearish candle for BUY -> capitulation result = await _eval(brain,"MES", _to_snap(tech)) assert result is None assert len(ai.calls) == 0 _ok("pre-val: capitulation candle against BUY -> None, no AI") # ── PRE-VAL: MACD accelerating against (V18 12-mag) ────────── # Incidente MYM/MES BUY: macd_decel=False + hist<0 → SL. # Gate identico a brain_mr.evaluate_entry. async def test_pre_val_macd_accelerating_against_buy_tf(): """BUY + macd_decel=False + hist<0 -> MACD_ACCELERATING_AGAINST.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log) tech = make_tech_buy_valid() tech["macd_decelerating"] = False tech["macd_hist_last"] = -2.32 # bearish, accelerating against BUY result = await _eval(brain, "MES", _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_decel"] is False assert events[0]["macd_hist"] == -2.32 _ok("pre-val TF: BUY + MACD bearish accelerating -> MACD_ACCELERATING_AGAINST") async def test_pre_val_macd_accelerating_against_sell_tf(): """SELL + macd_decel=False + hist>0 -> MACD_ACCELERATING_AGAINST.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log) # Build a valid TF SELL setup, then add accelerating-against MACD. tech = make_tech_buy_valid() tech["bias"] = "BEARISH" tech["allowed_direction"] = "SELL" tech["market_structure"] = MarketStructure.BEARISH_EXPANSION.value tech["price"] = 5800.0 tech["open"] = 5802.0 # red candle, no capitulation vs SELL tech["macd_decelerating"] = False tech["macd_hist_last"] = 1.85 # bullish, accelerating against SELL result = await _eval(brain, "MES", _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"] == 1.85 _ok("pre-val TF: SELL + MACD bullish accelerating -> MACD_ACCELERATING_AGAINST") async def test_pre_val_macd_accelerating_in_favor_passes_tf(): """BUY + macd_decel=False + hist>0 -> gate passes (MACD favors direction).""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log) tech = make_tech_buy_valid() tech["macd_decelerating"] = False tech["macd_hist_last"] = 0.85 # bullish, accelerating WITH BUY await _eval(brain, "MES", _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, f"gate should pass when MACD favors direction: {rejected}" _ok("pre-val TF: BUY + MACD bullish accelerating -> gate passes") async def test_pre_val_macd_decelerating_passes_tf(): """BUY + macd_decel=True -> gate passes regardless of hist sign.""" cfg = make_config() ai = FakeAIClient() log = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log) tech = make_tech_buy_valid() tech["macd_decelerating"] = True tech["macd_hist_last"] = -1.20 # bearish but decelerating -> OK await _eval(brain, "MES", _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 TF: macd_decel=True -> gate passes (any sign)") # ============================================================ # ENTRY: POST-VALIDATION watchdog # ============================================================ async def test_post_val_confidence_below_floor(): """AI approves with confidence 65% -> watchdog rejects (floor 70).""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 65, # too low "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.78, # in range MES TF (0.36, 1.20) post-calibrazione V16-29-apr "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) brain = BrainTF(cfg, ai) result = await _eval(brain,"MES", _to_snap(make_tech_buy_valid())) assert result is None _ok("post-val: confidence 65% < 70% floor -> None") async def test_post_val_c95_h4_weak(): """confidence 95% but H4 RSI 35 against BUY -> watchdog rejects.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "OTTIMO", "step_2_timing_pullback": "OTTIMO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 95, # very high "direction": "BUY", "risk_multiplier": 1.2, "sl_atr_multiplier": 0.78, # in range MES TF (0.36, 1.20) post-calibrazione V16-29-apr "rr_multiplier": 0.67, "key_risk": "h4 weak", "reason": "deep discount", })) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["rsi_h4"] = 35.0 # against BUY (need >=45) result = await _eval(brain,"MES", _to_snap(tech)) assert result is None _ok("post-val: C95% + H4 RSI 35 vs BUY -> None") async def test_post_val_rr_out_of_range(): """ TP variante γ: AI emits rr_multiplier outside advisory prompt-range [0.17, 0.67] -> POST_VAL_RR_RANGE rejection (calibration signal). Hard clamp [0.10, 0.80] in tp_resolver is defense-in-depth and not triggered for this rejection — the brain rejects first. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 78, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.78, "rr_multiplier": 0.75, # > 0.67 prompt-range upper bound "key_risk": "ok", "reason": "ok", })) logger = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=logger) result = await _eval(brain, "MES", _to_snap(make_tech_buy_valid())) assert result is None events = [e for e in logger.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["rule"] == "POST_VAL_RR_RANGE" _ok("post-val: rr_multiplier 0.75 outside [0.17, 0.67] -> POST_VAL_RR_RANGE") async def test_post_val_rr_missing(): """rr_multiplier <= 0 (missing/zero) -> POST_VAL_RR_MISSING.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 78, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.78, # rr_multiplier omitted -> defaults to 0.0 "key_risk": "ok", "reason": "ok", })) logger = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=logger) result = await _eval(brain, "MES", _to_snap(make_tech_buy_valid())) assert result is None events = [e for e in logger.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["rule"] == "POST_VAL_RR_MISSING" _ok("post-val: rr_multiplier missing -> POST_VAL_RR_MISSING") async def test_post_val_sl_out_of_profile_range(): """ BUG 9 fix: AI returns sl_atr_multiplier outside profile.sl_range. Post-fix: rejected via POST_VAL_SL_RANGE. """ cfg = make_config() # MES sl_range=(0.36, 1.20) post-V16-cal. sl 0.20 is OUT (below 0.36). ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 78, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.20, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) logger = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=logger) result = await _eval(brain, "MES", _to_snap(make_tech_buy_valid())) assert result is None events = [e for e in logger.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["rule"] == "POST_VAL_SL_RANGE" _ok("post-val: sl_atr 0.20 < profile sl_range min 0.36 -> POST_VAL_SL_RANGE") # ============================================================ # ENTRY: HAPPY PATH # ============================================================ async def test_entry_happy_path(): """ Valid setup -> EntryDecision with computed SL prices. TP variante γ: brain emits tp_price=0.0 sentinel + rr_multiplier; orchestrator finalizes tp_price post-sizing via tp_resolver. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 78, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.78, # in range MES TF (0.36, 1.20) post-calibrazione V16-29-apr "rr_multiplier": 0.50, "key_risk": "trend stale", "reason": "deep discount confluence", })) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() # Bump ATR to a value that does NOT trigger MIN_SL_TICKS clamp. tech["atr_m5_points"] = 30.0 result = await _eval(brain,"MES", _to_snap(tech)) assert isinstance(result, EntryDecision) assert result.direction == "BUY" assert result.confidence == 78 assert result.entry_price == 5800.00 assert result.sl_price < result.entry_price # BUY: SL below # tp_price is sentinel 0.0 from brain; finalized post-sizing assert result.tp_price == 0.0 assert result.rr_multiplier == 0.50 assert result.metadata["brain"] == "TF" assert result.metadata["rr_multiplier_ai"] == 0.50 assert 8 <= result.metadata["sl_ticks_used"] <= 40, \ f"MES sl_ticks {result.metadata['sl_ticks_used']} out of [8,40]" assert result.metadata["profile_origin"] == "US500" _ok("happy: EntryDecision with SL prices + rr_multiplier (TP resolved post-sizing)") class _SLTech: """Minimal tech stub for _compute_sl_only direct unit tests.""" def __init__(self, price: float, atr_m5_points: float) -> None: self.price = price self.atr_m5_points = atr_m5_points def test_sl_floored_mym_min_sl_ticks(): """MYM BUY: AI's sl_atr produces sl_ticks_pre=5 < MIN_SL=10 → floor a 10.""" # MYM: tick_size=1.0, MIN_SL_TICKS=10, profile sl_range=(0.16, 0.55). # sl_atr=0.16, atr=30 → sl_distance_pre = 30*0.16*1.10 = 5.28 → 5 ticks. pc = BrainTF._compute_sl_only( symbol="MYM", direction="BUY", tech=_SLTech(price=44000.0, atr_m5_points=30.0), sl_atr=0.16, ) assert pc["sl_ticks_pre_clamp"] == 5 assert pc["sl_ticks_post_clamp"] == 10 # floored to MIN_SL assert pc["sl_floored"] is True assert pc["sl_min_ticks"] == 10 # SL price moved by 10 ticks (MYM tick_size=1.0 → 10 points) assert pc["sl_price"] == 44000.0 - 10.0 _ok("sl_floored: MYM AI proposes 5 ticks → floored to MIN 10") def test_sl_floored_false_when_above_min(): """6E sl_ticks_pre=15 ≥ MIN_SL=10 → no floor, sl_floored=False.""" # 6E: tick_size=0.00005, MIN_SL_TICKS=10, profile sl_range=(3.0, 5.0). # sl_atr=4.5, atr=0.00015 → 0.00015*4.5*1.10 = 0.000742 → 0.000742/0.00005 = 14.85 → 15. pc = BrainTF._compute_sl_only( symbol="6E", direction="BUY", tech=_SLTech(price=1.0850, atr_m5_points=0.00015), sl_atr=4.5, ) assert pc["sl_ticks_pre_clamp"] == 15 assert pc["sl_ticks_post_clamp"] == 15 # invariato assert pc["sl_floored"] is False assert pc["sl_min_ticks"] == 10 _ok("sl_floored: 6E sl=15 ≥ MIN 10 → unchanged, sl_floored=False") def test_sl_floored_false_when_symbol_has_no_min_config(monkeypatch): """Symbol non listato in MIN_SL_TICKS → sl_floored=False (no floor applied).""" # Rimuovi temporaneamente MYM da MIN_SL_TICKS per simulare "no config". import core.config_futures as cfg_fut patched = dict(cfg_fut.MIN_SL_TICKS) patched.pop("MYM", None) monkeypatch.setattr(cfg_fut, "MIN_SL_TICKS", patched) pc = BrainTF._compute_sl_only( symbol="MYM", direction="BUY", tech=_SLTech(price=44000.0, atr_m5_points=30.0), sl_atr=0.16, ) # Stesso sl_ticks_pre del test 1 (5), ma senza floor: sl_ticks_post=5. assert pc["sl_ticks_pre_clamp"] == 5 assert pc["sl_ticks_post_clamp"] == 5 assert pc["sl_floored"] is False assert pc["sl_min_ticks"] == 0 _ok("sl_floored: no MIN_SL_TICKS config → no floor, sl_floored=False") async def test_rejection_logging_pullback_zone(): """Pre-val rejection emits structured event with rsi/rsi_low/rsi_high.""" cfg = make_config() ai = FakeAIClient() log_bundle = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log_bundle) tech = make_tech_buy_valid() tech["rsi"] = 65.0 await _eval(brain,"MES", _to_snap(tech)) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 ev = events[0] assert ev["rule"] == "PULLBACK_ZONE" assert ev["rsi_m5"] == 65.0 assert ev["rsi_low"] == 42.0 assert ev["rsi_high"] == 58.0 assert ev["brain"] == "TF" _ok("logging: PULLBACK_ZONE event with rsi fields") async def test_rejection_logs_radar_tech_fields(): """V18 dashboard radar: entry_rejected carries rsi_m5/rsi_h4/h1_compat/ macd_*/pattern/atr_ratio/bias/h1_struct_*/volume_weak/divergence/bouncing_rsi.""" cfg = make_config() ai = FakeAIClient() log_bundle = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log_bundle) tech = make_tech_buy_valid() tech["rsi"] = 65.0 # PULLBACK_ZONE tech["rsi_h4"] = 58.0 tech["h1_compatibility"] = 0.85 tech["macd_hist_last"] = 0.0042 tech["macd_decelerating"] = True tech["hammer"] = True # pattern label tech["atr_ratio"] = 1.20 tech["bias"] = "BULLISH" tech["h1_struct_bull"] = True tech["divergence"] = "BULLISH" tech["rsi_prev"] = 60.0 # bouncing_rsi=True (65 > 60) await _eval(brain, "MES", _to_snap(tech)) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] ev = events[0] assert ev["rule"] == "PULLBACK_ZONE" assert ev["rsi_m5"] == 65.0 assert ev["rsi_h4"] == 58.0 assert ev["h1_compat"] == 0.85 assert ev["macd_hist"] == 0.0042 assert ev["macd_decel"] is True assert ev["pattern"] == "HAMMER" assert ev["atr_ratio"] == 1.20 assert ev["bias"] == "BULLISH" assert ev["h1_struct_bull"] is True assert ev["h1_struct_bear"] is False assert ev["volume_weak"] is False assert ev["divergence"] == "BULLISH" assert ev["bouncing_rsi"] is True _ok("logging: entry_rejected carries full RADAR tech fields (V18)") async def test_rejection_logging_post_val(): """Post-val rejection includes ai_confidence + ai_sl_atr + ai_rr_multiplier.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 65, # below floor "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": 0.78, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) log_bundle = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log_bundle) await _eval(brain,"MES", _to_snap(make_tech_buy_valid())) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 ev = events[0] assert ev["rule"] == "POST_VAL_CONF" assert ev["ai_confidence"] == 65 assert ev["ai_sl_atr"] == 0.78 assert ev["ai_rr_multiplier"] == 0.50 _ok("logging: POST_VAL_CONF event with ai_* diagnostics") 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({ "step_1_qualita_sconto": "DEBOLE", "step_2_timing_pullback": "PREMATURO", "step_3_contesto_macro": "NEUTRO", "approved": False, "confidence": 60, "direction": "BUY", "risk_multiplier": 0.5, "sl_atr_multiplier": 0.78, "rr_multiplier": 0.30, "key_risk": "weak setup", "reason": "discount too shallow", "rejection_details": { "rule_failed": "STEP1_DEBOLE", "detail": "RSI 45 ai margini, sconto poco profondo", }, })) log_bundle = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log_bundle) await _eval(brain, "MES", _to_snap(make_tech_buy_valid())) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 ev = events[0] assert ev["rule"] == "AI_REJECTED" assert ev["ai_rule_failed"] == "STEP1_DEBOLE" assert ev["ai_rejection_detail"].startswith("RSI 45 ai margini") _ok("logging: AI_REJECTED event carries ai_rule_failed + ai_rejection_detail") async def test_ai_rejection_missing_details_safe(): """AI returns approved=false WITHOUT rejection_details -> fallback UNSPECIFIED (no crash).""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "DEBOLE", "step_2_timing_pullback": "PREMATURO", "step_3_contesto_macro": "NEUTRO", "approved": False, "confidence": 60, "direction": "BUY", "risk_multiplier": 0.5, "sl_atr_multiplier": 0.78, "rr_multiplier": 0.30, "key_risk": "n/a", "reason": "no", # rejection_details key intentionally missing })) log_bundle = FakeLoggerBundle() brain = BrainTF(cfg, ai, logger=log_bundle) await _eval(brain, "MES", _to_snap(make_tech_buy_valid())) events = [e for e in log_bundle.brain_log.events if e["event"] == "entry_rejected"] assert len(events) == 1 assert events[0]["ai_rule_failed"] == "UNSPECIFIED" _ok("logging: AI rejection without rejection_details -> UNSPECIFIED fallback") 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 TF_ASSET_PROFILES) per esercitare il path GENERIC fallback. """ cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "step_1_qualita_sconto": "BUONO", "step_2_timing_pullback": "BUONO", "step_3_contesto_macro": "FAVOREVOLE", "approved": True, "confidence": 75, "direction": "BUY", "risk_multiplier": 1.0, "sl_atr_multiplier": GENERIC_PROFILE["sl_range"][0] + 0.1, "rr_multiplier": 0.50, "key_risk": "ok", "reason": "ok", })) brain = BrainTF(cfg, ai) fake_symbol = "ZZZ_TEST_NO_PROFILE" # Sanity: the fake symbol is genuinely absent from the profile dict assert fake_symbol not in TF_ASSET_PROFILES tech = make_tech_buy_valid(fake_symbol) # Realistic small-tick fake quote (PRICE_COMPUTATION_ERROR is OK — # the warning fires before _compute_sl_only is reached). tech["price"] = 0.6500 tech["open"] = 0.6498 tech["atr_m5_points"] = 0.0010 # First call: should warn 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 # Second call: no duplicate warn (set already has it) _ = await _eval(brain, fake_symbol, _to_snap(tech)) _ok(f"generic profile: {fake_symbol} (no profile) tracked + warned once") # ============================================================ # PROMPT: anti-revenge re-entry post-SL # ============================================================ def _render_tf_prompt(consec_sl: int) -> str: """Build a TF entry prompt with given consecutive_sl_count.""" d = make_tech_buy_valid() d["consecutive_sl_count"] = consec_sl tech = _to_snap(d) signals = extract_pa_signals(tech) score = score_pa(signals, "BUY") profile = TF_ASSET_PROFILES["MES"] return BrainTF._build_entry_prompt( symbol="MES", direction="BUY", bias=tech.bias, regime=tech.regime, rsi=tech.rsi, rsi_prev=tech.rsi_prev, atr_ratio=tech.atr_ratio, candle_strength=tech.candle_strength, tech=tech, signals=signals, score=score, profile=profile, ) def test_prompt_consecutive_sl_warning(): prompt = _render_tf_prompt(consec_sl=2) # V17: warning anti-revenge è inline in sezione 4 "CONTESTO STRUTTURALE" # (V16 aveva un blocco ANTI-REVENGE dedicato). assert "2 SL recenti" in prompt assert "la struttura ti ha già stoppato" in prompt assert "RSI in zona ottimale (42-46" in prompt assert "SL consecutivi sessione: 2" in prompt assert "re-entry post-SL: SÌ" in prompt _ok("prompt: consec_sl=2 -> inline anti-revenge warning + RSI ottimale rule") def test_prompt_no_sl_no_warning(): prompt = _render_tf_prompt(consec_sl=0) # V17: per consec_sl=0 non viene emesso testo di warning (non più un # placeholder "Nessun SL recente"). Solo la riga di stato. assert "SL consecutivi sessione: 0" in prompt assert "re-entry post-SL: NO" in prompt assert "la struttura ti ha già stoppato" not in prompt _ok("prompt: consec_sl=0 -> no inline anti-revenge warning") # ============================================================ # EXIT: manage_exit # ============================================================ 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=20.0, last_exit_eval_time=0.0), tech_now=_to_snap(tech if tech is not None else make_tech_buy_valid()), ) async def test_exit_time_stop_negative_pnl(): """P&L<=0 + minutes >= threshold -> EXIT without AI call.""" cfg = make_config() ai = FakeAIClient() # AI must NOT be called brain = BrainTF(cfg, ai) ctx = make_ctx( runtime=TradeRuntime(minutes_open=300.0, net_profit_usd=-5.0, last_exit_eval_time=0.0), ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert "Time Stop TF" in decision.reason assert len(ai.calls) == 0 _ok("exit: time stop hit (P&L<=0) -> EXIT, no AI") async def test_exit_same_candle_dedup(): """tech.candle_time <= runtime.last_exit_eval_time -> HOLD without AI.""" cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["candle_time"] = 1000.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=20.0, last_exit_eval_time=1000.0), tech=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_struct_flip_detected(): """LONG with new BEARISH struct (after grace) -> flag in metadata.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({"exit_now": True, "reason": "struct flip"})) brain = BrainTF(cfg, ai) entry = make_entry(structure=MarketStructure.BULLISH_EXPANSION.value) tech = make_tech_buy_valid() tech["market_structure"] = MarketStructure.BEARISH_EXPANSION.value tech["candle_time"] = 2000.0 ctx = make_ctx( entry=entry, runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=10.0, last_exit_eval_time=1500.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert decision.metadata.get("struct_flipped") is True assert decision.metadata.get("evaluated_candle_time") == 2000.0 _ok("exit: struct flip detected -> metadata flag set, AI called") async def test_exit_ai_says_hold(): """AI returns exit_now=false -> HOLD.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({"exit_now": False, "reason": "no inversion"})) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["candle_time"] = 3000.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=20.0, net_profit_usd=15.0, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.HOLD.value assert "no inversion" in decision.reason assert decision.metadata["evaluated_candle_time"] == 3000.0 _ok("exit: AI says HOLD -> HOLD with metadata candle_time") async def test_exit_ai_says_exit(): """AI returns exit_now=true -> EXIT.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({ "exit_now": True, "reason": "engulfing+H4 crollo", })) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["candle_time"] = 4000.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=30.0, net_profit_usd=8.0, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert "engulfing" in decision.reason _ok("exit: AI says EXIT -> EXIT") async def test_exit_ai_error_holds(): """AI returns None text -> HOLD default (conservative).""" cfg = make_config() ai = FakeAIClient(canned=AIResponse(text=None, error_kind="overload", attempts=3)) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["candle_time"] = 5000.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=5.0, last_exit_eval_time=0.0), tech=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") async def test_exit_auto_partial_50_fires_with_set_be(): """progress >= 50% + !partial_done -> PARTIAL_50 + set_be_after_partial.""" cfg = make_config() ai = FakeAIClient() # AI must NOT be called brain = BrainTF(cfg, ai) entry = make_entry() tech = make_tech_buy_valid() tech["tick_size"] = 0.25 # MES ctx = make_ctx( entry=entry, runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=15.0, progress_pct=55.0, partial_done=False, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) 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 assert decision.metadata["progress_pct"] == 55.0 assert len(ai.calls) == 0 _ok("exit: progress 55% -> PARTIAL_50 + set_be (no AI call)") async def test_exit_auto_partial_50_threshold_exact(): """progress == 50.0 (boundary) -> trigger fires (>=, not >).""" cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["tick_size"] = 0.25 ctx = make_ctx( runtime=TradeRuntime(minutes_open=8.0, net_profit_usd=10.0, progress_pct=50.0, partial_done=False, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.PARTIAL_50.value assert decision.metadata["trigger"] == "auto_partial_50" _ok("exit: progress 50.0 (boundary) -> PARTIAL_50") async def test_exit_auto_partial_50_skipped_if_partial_done(): """One-shot guard: partial_done=True -> trigger NOT fired (falls through to AI).""" cfg = make_config() # AI canned to HOLD so we can verify the path falls through past partial ai = FakeAIClient(canned=make_ai_response( {"exit_now": False, "reason": "hold partial done"}, )) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["tick_size"] = 0.25 tech["candle_time"] = 9000.0 # bypass same-candle dedup ctx = make_ctx( runtime=TradeRuntime(minutes_open=15.0, net_profit_usd=25.0, progress_pct=75.0, partial_done=True, # already partialed last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.HOLD.value assert len(ai.calls) == 1, "AI should be consulted because partial path skipped" _ok("exit: partial_done=True -> no re-partial (falls through)") async def test_exit_auto_partial_50_below_threshold_no_fire(): """progress < 50% -> trigger NOT fired (falls through to AI / dedup).""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response( {"exit_now": False, "reason": "low progress hold"}, )) brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["tick_size"] = 0.25 tech["candle_time"] = 9100.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=8.0, net_profit_usd=4.0, progress_pct=49.9, partial_done=False, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.HOLD.value assert len(ai.calls) == 1 _ok("exit: progress 49.9% -> below threshold, AI consulted") async def test_exit_auto_partial_50_fires_before_dedup(): """Same-candle dedup must NOT block the auto-partial (deterministic action).""" cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) tech = make_tech_buy_valid() tech["tick_size"] = 0.25 tech["candle_time"] = 7000.0 ctx = make_ctx( runtime=TradeRuntime(minutes_open=12.0, net_profit_usd=18.0, progress_pct=62.0, partial_done=False, last_exit_eval_time=7000.0), # same candle already eval'd tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.PARTIAL_50.value assert decision.metadata["trigger"] == "auto_partial_50" assert len(ai.calls) == 0 _ok("exit: partial trigger fires even on already-evaluated candle") async def test_exit_auto_partial_50_be_price_tick_aligned(): """be_price aligned to instrument tick grid (defensive against off-grid entry_price).""" cfg = make_config() ai = FakeAIClient() brain = BrainTF(cfg, ai) # Force an off-grid entry to verify rounding kicks in. entry = make_entry() entry = TradeEntry( symbol=entry.symbol, brain_name=entry.brain_name, direction=entry.direction, contracts=entry.contracts, entry_price=5800.13, # not on 0.25 grid sl_price=entry.sl_price, tp_price=entry.tp_price, opened_at=entry.opened_at, rsi_m5_at_entry=entry.rsi_m5_at_entry, rsi_h1_at_entry=entry.rsi_h1_at_entry, rsi_h4_at_entry=entry.rsi_h4_at_entry, atr_ratio_at_entry=entry.atr_ratio_at_entry, market_structure_at_entry=entry.market_structure_at_entry, regime_at_entry=entry.regime_at_entry, h1_compat_at_entry=entry.h1_compat_at_entry, confidence_at_entry=entry.confidence_at_entry, ) tech = make_tech_buy_valid() tech["tick_size"] = 0.25 ctx = make_ctx( entry=entry, runtime=TradeRuntime(minutes_open=10.0, net_profit_usd=12.0, progress_pct=51.0, partial_done=False, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.PARTIAL_50.value # 5800.13 rounded to nearest 0.25 -> 5800.25 assert decision.metadata["be_price"] == 5800.25 _ok("exit: be_price tick-aligned to 0.25 grid") async def test_exit_short_symmetry(): """SHORT trade with new BULLISH struct -> flip detected.""" cfg = make_config() ai = FakeAIClient(canned=make_ai_response({"exit_now": True, "reason": "flip"})) brain = BrainTF(cfg, ai) entry = make_entry(direction="SELL", structure=MarketStructure.BEARISH_EXPANSION.value) tech = make_tech_buy_valid() tech["market_structure"] = MarketStructure.BULLISH_EXPANSION.value tech["candle_time"] = 6000.0 ctx = make_ctx( entry=entry, runtime=TradeRuntime(minutes_open=20.0, net_profit_usd=8.0, last_exit_eval_time=0.0), tech=tech, ) decision = await brain.manage_exit(ctx) assert decision.action == TradeAction.EXIT.value assert decision.metadata["struct_flipped"] is True _ok("exit: SHORT symmetry — new BULLISH flip detected") # ============================================================ # RUN # ============================================================ async def main_async() -> int: print("test_brain_tf.py") # Pre-val await test_pre_val_rsi_outside_zone() await test_pre_val_atr_below_floor() await test_pre_val_capitulation_candle() # PRE-VAL: MACD gate (V18 12-mag) await test_pre_val_macd_accelerating_against_buy_tf() await test_pre_val_macd_accelerating_against_sell_tf() await test_pre_val_macd_accelerating_in_favor_passes_tf() await test_pre_val_macd_decelerating_passes_tf() # Post-val await test_post_val_confidence_below_floor() await test_post_val_c95_h4_weak() await test_post_val_rr_out_of_range() await test_post_val_rr_missing() await test_post_val_sl_out_of_profile_range() # Happy await test_entry_happy_path() # Structured logging await test_rejection_logging_pullback_zone() await test_rejection_logs_radar_tech_fields() await test_rejection_logging_post_val() await test_ai_rejection_with_details_logged() await test_ai_rejection_missing_details_safe() # Generic profile await test_generic_profile_warning() # Prompt anti-revenge test_prompt_consecutive_sl_warning() test_prompt_no_sl_no_warning() # SL MIN_SL_TICKS floor (V18 12-mag) — sync, not async; monkeypatch # test is pytest-only (uses fixture), excluded from standalone runner. test_sl_floored_mym_min_sl_ticks() test_sl_floored_false_when_above_min() # Exit await test_exit_time_stop_negative_pnl() await test_exit_same_candle_dedup() await test_exit_struct_flip_detected() await test_exit_ai_says_hold() await test_exit_ai_says_exit() await test_exit_ai_error_holds() await test_exit_short_symmetry() # Auto partial 50 + BE (V18) await test_exit_auto_partial_50_fires_with_set_be() await test_exit_auto_partial_50_threshold_exact() await test_exit_auto_partial_50_skipped_if_partial_done() await test_exit_auto_partial_50_below_threshold_no_fire() await test_exit_auto_partial_50_fires_before_dedup() await test_exit_auto_partial_50_be_price_tick_aligned() print("ALL TESTS PASSED") return 0 def main() -> int: return asyncio.run(main_async()) if __name__ == "__main__": sys.exit(main())