""" APEX V16 — Brain Trend Following. Migrated from V15's Braintf.py. The TRADING RULES are unchanged: - Entry only on pullbacks in zona sconto RSI 42-58 - ATR ratio >= 0.8 floor (block low-volatility / congested markets) - Capitulation candle veto (candle_strength > 1.8 against direction) - AI Chain-of-Thought entry prompt with 3 steps (qualità sconto, timing pullback, contesto macro) - Post-validation watchdog (RSI re-check, conf>=70, RR floor, C>=90% + H4 weak guardia) - Time-stop matrix per (asset, session) - Same-candle dedup in manage_exit - Struct-flip detection (LONG with new BEARISH, SHORT with new BULLISH) - Deep-discount H4 paradox guard - AI exit prompt biased toward HOLD WHAT CHANGED (orchestration only, V16-BUG-prevention): - Async-first (uses AIClient.ask_for_decision) - Returns typed EntryDecision / BrainDecision instead of dicts - Reads ctx.entry / ctx.runtime instead of init_data dict - Does NOT mutate ctx.runtime (orchestrator owns runtime state) - PA pattern reading goes through analysis.price_action (single source) - TF_ASSET_PROFILES at module top, calibrated on V15 CFD assets; futures asset symbols mapped by equivalence (MES <- US500, etc.) """ from __future__ import annotations import datetime as _dt import json import logging from typing import Optional from analysis.price_action import ( PASignals, PAScore, extract_pa_signals, score_pa, ) from analysis.tech_snapshot import tech_log_fields from brain.ai_client import AIClient, extract_json_from_response from brain.brain_base import BrainBase from core import config_futures as cfg_fut from core.contracts import ( BrainContext, BrainDecision, BrainName, Direction, EntryDecision, EntryEvalResult, TradeAction, ) from trading.tp_resolver import TP_SUGGESTED_CONSERVATIVE_MARGIN _log = logging.getLogger("brain.tf") # ============================================================ # TF_ASSET_PROFILES — sl_range / sl_min_points # # V16-29-apr: rivalidato su pattern di clamp osservati in produzione. # I profili V15 (CFD US500/US100/XAUUSD) erano calibrati su scale ATR # diverse dai futures micro: MES/MNQ/MYM/MGC/MCL/6C avevano sl_range # fuori dai bound MIN_SL_TICKS/MAX_SL_TICKS → clamp permanente. Allineati # qui agli ATR reali futures. Post BUG C: tp_range rimosso (TP è scelta # AI come prezzo assoluto, non multiplier ATR). # # Mapping V15 -> V16 futures: # US500 -> MES EURUSD -> 6E USDJPY -> 6J # US100 -> MNQ GBPUSD -> 6B USDCAD -> 6C # US30 -> MYM AUDUSD -> 6A XAUUSD -> MGC # # MCL (oil) and 6A (AUD) had no equivalent in V15. They use # GENERIC_PROFILE and emit a one-time warning the first time they # are evaluated, so the calibration round can spot them in the logs. # ============================================================ TF_ASSET_PROFILES: dict = { # ── EQUITY INDEX MICROS ──────────────────────────────────────── "MES": { # ≡ V15 US500 "sl_range": (0.36, 1.20), # V16-29-apr: era (0.055, 0.100), CLAMP_UP @ MIN=8 "sl_min_points": 35, "v15_origin": "US500", "note": "S&P500 micro: V16 calibrato su ATR_M5 ~6 (futures), sl_ticks 10-30", }, "MNQ": { # ≡ V15 US100 "sl_range": (0.075, 0.300), # V16-29-apr: era (0.055, 0.100), CLAMP_UP marginale "sl_min_points": 70, "v15_origin": "US100", "note": "Nasdaq micro: V16 calibrato su ATR_M5 ~30 (futures), sl_ticks 10-36", }, "MYM": { # ≡ V15 US30 "sl_range": (0.16, 0.55), # V16-29-apr: era (0.070, 0.110), CLAMP_UP severo "sl_min_points": 150, "v15_origin": "US30", "note": "Dow micro: V16 calibrato su ATR_M5 ~80 (futures), sl_ticks 14-44", }, # ── METAL MICRO ──────────────────────────────────────────────── "MGC": { # ≡ V15 XAUUSD "sl_range": (1.0, 2.9), # V16-29-apr: era (3.5, 6.0), CLAMP_DOWN @ MAX=60 "sl_min_points": 8.0, "v15_origin": "XAUUSD", "note": "Gold micro: V16 calibrato su ATR_M5 ~2.5, sl_ticks 27-58", }, # ── FX FUTURES ───────────────────────────────────────────────── "6E": { # ≡ V15 EURUSD "sl_range": (3.0, 5.0), "v15_origin": "EURUSD", "note": "EUR/USD: SL 9-15p (V15 calibration)", }, "6B": { # ≡ V15 GBPUSD "sl_range": (2.8, 4.8), "v15_origin": "GBPUSD", "note": "GBP/USD: SL 15-26p (V15 calibration)", }, "6J": { # ≡ V15 USDJPY (inverted_quote=True at broker layer) "sl_range": (0.45, 0.72), "v15_origin": "USDJPY", "note": "JPY: SL 25-40p (V15 calibration)", }, "6C": { # ≡ V15 USDCAD (inverted_quote=True at broker layer) "sl_range": (1.5, 2.7), # V16-29-apr: era (3.0, 4.8), CLAMP_DOWN @ MAX=50 "v15_origin": "USDCAD", "note": "CAD: V16 calibrato (ATR_TYPICAL 0.0008 alto vs altri FX)", }, "6A": { # AUD/USD - new in V16, derived from 6E (FX major analog) "sl_range": (3.0, 5.0), "v15_origin": "AUDUSD (new in V16)", "note": "AUD/USD: derived from 6E analog, refine in CALIBRAZIONE 5-9 may", }, "MCL": { # WTI Crude Oil micro - new in V16, energy commodity "sl_range": (1.2, 1.8), # V16-29-apr: era (1.5, 2.0), vicino MAX=80 "v15_origin": "WTI (new in V16)", "note": "Crude Oil micro: ATR_TYPICAL 0.40, sl_ticks 53-79", }, } # Generic fallback for symbols with no V15 equivalent. GENERIC_PROFILE: dict = { "sl_range": (3.5, 5.0), "v15_origin": "GENERIC", "note": "no V15 equivalent — generic conservative profile, recalibrate in V16", } # ============================================================ # TF_TIME_STOP — per (asset, session) in minutes. # Migrated from V15: TF needs more time than MR (trend needs breath). # Default fallback 180 min. # ============================================================ TF_TIME_STOP: dict = { # Forex futures (mapped from V15 .r symbols) "6E": {"LONDON": 120, "LONDON_NY": 120, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 200, "ASIA_LONDON_GAP": 200}, "6B": {"LONDON": 120, "LONDON_NY": 120, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 200, "ASIA_LONDON_GAP": 200}, "6C": {"LONDON": 150, "LONDON_NY": 120, "NY_CLOSE": 120, "NY_LATE": 180, "ASIA_EARLY": 210, "ASIA_LONDON_GAP": 210}, "6A": {"LONDON": 150, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 150, "ASIA_EARLY": 120, "ASIA_LONDON_GAP": 150}, "6J": {"LONDON": 150, "LONDON_NY": 120, "NY_CLOSE": 150, "NY_LATE": 150, "ASIA_EARLY": 120, "ASIA_LONDON_GAP": 150}, # Volatile / metals "MGC": {"LONDON": 150, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 210, "ASIA_LONDON_GAP": 210}, # Equity index micros (USA session relevant) "MNQ": {"LONDON": 180, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 240, "ASIA_LONDON_GAP": 240}, "MYM": {"LONDON": 180, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 240, "ASIA_LONDON_GAP": 240}, "MES": {"LONDON": 180, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 240, "ASIA_LONDON_GAP": 240}, # Energy: no V15 equivalent, conservative defaults aligned with metals "MCL": {"LONDON": 150, "LONDON_NY": 150, "NY_CLOSE": 150, "NY_LATE": 180, "ASIA_EARLY": 210, "ASIA_LONDON_GAP": 210}, } TF_TIME_STOP_DEFAULT_MIN = 180 # Pre-validation thresholds (V15) RSI_TF_LOW = 42.0 RSI_TF_HIGH = 58.0 ATR_RATIO_FLOOR = 0.8 CAPITULATION_STRENGTH = 1.8 MIN_CONFIDENCE = 70 # Same-candle exit grace (V15-04-28: avoid closing on M5-entry candle) STRUCT_FLIP_GRACE_MIN = 1 # Auto-partial 50% + SL→BE quando il trade è a metà strada verso il TP. # V18: secure half the position e protegge il residuo a BE prima che un # ritracciamento vanifichi il progress. One-shot via runtime.partial_done. TF_AUTO_PARTIAL_PROGRESS_PCT = 50.0 # ============================================================ # SESSION CLASSIFIER # ============================================================ def _classify_session(utc_hour: int) -> str: if 7 <= utc_hour < 12: return "LONDON" if 12 <= utc_hour < 17: return "LONDON_NY" if 17 <= utc_hour < 20: return "NY_CLOSE" if 20 <= utc_hour <= 23: return "NY_LATE" if 0 <= utc_hour < 3: return "ASIA_EARLY" return "ASIA_LONDON_GAP" # ============================================================ # BRAIN TF # ============================================================ class BrainTF(BrainBase): """ Trend Following Brain. Async (uses AIClient). The orchestrator is responsible for: - candle-gating (calling evaluate_entry only on M5 just-closed) - populating tech with bias_data keys (bias / allowed_direction / regime / h1_compatibility / swing_data) and the technical fields (price / open / rsi / rsi_prev / rsi_h1 / rsi_h4 / atr_ratio / atr_m5_points / vwap / vwap_deviation_pct / market_structure / vol_regime / trend_maturity / candle_strength / candle_time) - applying the BrainDecision (EXIT/HOLD) and updating TradeRuntime.last_exit_eval_time from BrainDecision.metadata before persisting state. """ name = BrainName.TF.value def __init__(self, config, ai_client: AIClient, logger=None) -> None: super().__init__(config) self.ai = ai_client self.logger = logger # Track which symbols already produced a "no V15 calibration" warn self._warned_generic_profile: set[str] = set() # ============================================================ # PROFILE LOOKUP # ============================================================ def _profile_for(self, symbol: str) -> dict: prof = TF_ASSET_PROFILES.get(symbol) if prof is not None: return prof if symbol not in self._warned_generic_profile: self._warned_generic_profile.add(symbol) msg = (f"[BrainTF] {symbol}: no V15-calibrated profile — " f"using GENERIC. Recalibrate in V16 (5-9 may).") _log.warning(msg) if self.logger is not None: # LoggerBundle exposes .system (rotating logger) self.logger.system.warning(msg) return GENERIC_PROFILE # ============================================================ # ENTRY # ============================================================ async def evaluate_entry( self, symbol: str, tech, *, bias_data, last_entry_eval_time: float = 0.0, ) -> EntryEvalResult: """ Evaluate a TF entry. `tech` is a TechSnapshot. Returns EntryEvalResult: - decision = EntryDecision (open) | None (no setup / rejected) - evaluated_candle_time = candle_time (float) when AI was called and either responded or returned a permanent error; None when dedup-skipped, pre-val-rejected, or AI failed transiently. - dedup_skipped = True when same-candle dedup short-circuited the AI call (no AI cost, no cache update needed). Flow: pre-val (RSI zone, ATR floor, capitulation) -> SAME-CANDLE DEDUP gate -> AI prompt with Chain-of-Thought -> post-val watchdog (RSI/conf/RR/C90+H4) -> price computation -> decision. """ # Source A (BiasResolver: algo + AI override). tech.bias / # tech.allowed_direction are Source B (algo-only) and intentionally # bypassed here — see brain_base.evaluate_entry docstring. bias = bias_data.bias allowed = bias_data.allowed_direction regime = tech.regime direction = "BUY" if allowed in ("BUY", "BOTH") else "SELL" rsi = float(tech.rsi) rsi_prev = float(tech.rsi_prev) atr_ratio = float(tech.atr_ratio) candle_strength = float(tech.candle_strength) candle_time = float(getattr(tech, "candle_time", 0) or 0) # ── CANDLE-STATE GATE (before pre-val) ──────────────────── # Skip if candle isn't a settled, fresh M5 close. Logged as # scan_skip by the orchestrator (not entry_rejected: this is # state-of-data, not a trading decision). if not bool(getattr(tech, "is_candle_closed", False)): return EntryEvalResult( decision=None, reject_reason="CANDLE_NOT_CLOSED", ) age = float(getattr(tech, "candle_age_seconds", 0.0) or 0.0) if age < float(self.config.candle_close_delay_seconds): return EntryEvalResult( decision=None, reject_reason="CANDLE_STABILIZING", ) if age > float(self.config.candle_max_age_seconds): return EntryEvalResult( decision=None, reject_reason="CANDLE_TOO_OLD", ) # ── PRE-VAL 1: pullback zone ────────────────────────────── if not (RSI_TF_LOW <= rsi <= RSI_TF_HIGH): return EntryEvalResult(decision=self._reject( symbol, direction, rule="PULLBACK_ZONE", detail=(f"RSI {rsi:.1f} outside TF zone " f"[{RSI_TF_LOW},{RSI_TF_HIGH}]"), tech=tech, rsi_prev=rsi_prev, rsi_low=RSI_TF_LOW, rsi_high=RSI_TF_HIGH, )) # ── PRE-VAL 2: volatility floor ─────────────────────────── if atr_ratio < ATR_RATIO_FLOOR: return EntryEvalResult(decision=self._reject( symbol, direction, rule="VOLATILITA_BASSA", detail=f"ATR ratio {atr_ratio:.2f}x < floor {ATR_RATIO_FLOOR}", tech=tech, atr_floor=ATR_RATIO_FLOOR, )) # ── PRE-VAL 3: capitulation candle ──────────────────────── if candle_strength > CAPITULATION_STRENGTH: price = tech.price opn = tech.open common = dict( candle_strength=candle_strength, strength_floor=CAPITULATION_STRENGTH, price=price, open=opn, ) if direction == "BUY" and (opn is None or price is None or price < opn): return EntryEvalResult(decision=self._reject( symbol, direction, rule="CAPITULATION_CANDLE", detail=f"Bearish capitulation candle {candle_strength:.2f}x vs BUY", tech=tech, **common, )) if direction == "SELL" and (opn is None or price is None or price > opn): return EntryEvalResult(decision=self._reject( symbol, direction, rule="CAPITULATION_CANDLE", detail=f"Bullish capitulation candle {candle_strength:.2f}x vs SELL", tech=tech, **common, )) # ── PRE-VAL 4: RSI H4 estremo (V18 tightening 09-mag) ── # Soglie abbassate da 70/30 a 65/35: già a H4=65 il setup è # compromesso per TF BUY (simmetrico H4=35 per TF SELL). # Pre-AI: setup statisticamente perdente, niente token cost. rsi_h4 = float(tech.rsi_h4) if direction == "BUY" and rsi_h4 > 65.0: return EntryEvalResult(decision=self._reject( symbol, direction, rule="RSI_H4_OVERBOUGHT_TF_BUY", detail=f"RSI H4 {rsi_h4:.1f} > 65 — TF BUY rejected", tech=tech, )) if direction == "SELL" and rsi_h4 < 35.0: return EntryEvalResult(decision=self._reject( symbol, direction, rule="RSI_H4_OVERSOLD_TF_SELL", detail=f"RSI H4 {rsi_h4:.1f} < 35 — TF SELL rejected", tech=tech, )) # ── PRE-VAL 5: H1 compatibility floor (V18 09-mag) ── # h1_compat < 0.70 = setup against-trend / mismatched H1. # Gate deterministico pre-AI, no token cost. h1_compat = float(tech.h1_compatibility) if h1_compat < 0.70: return EntryEvalResult(decision=self._reject( symbol, direction, rule="H1_COMPAT_TOO_LOW", detail=f"H1 compat {h1_compat:.2f} < 0.70 — {direction} rejected", tech=tech, )) # ── PRE-VAL 6: MACD accelerating against (V18 12-mag) ── # Incidente 12-mag: TF approvava MYM/MES BUY con macd_decel=False # e macd_hist<0 (momentum bearish in accelerazione contro la # direzione). Gate identico a brain_mr.evaluate_entry — anche TF # vuole momentum a favore o in rallentamento, non in accelerazione # opposta. Pre-AI, no token cost. macd_decel = bool(tech.macd_decelerating) macd_hist = float(tech.macd_hist_last) macd_against = ( (direction == "BUY" and not macd_decel and macd_hist < 0) or (direction == "SELL" and not macd_decel and macd_hist > 0) ) if macd_against: return EntryEvalResult(decision=self._reject( symbol, direction, rule="MACD_ACCELERATING_AGAINST", detail=(f"MACD accelerando contro {direction} " f"(hist={macd_hist:.6f}) — TF rejected"), tech=tech, )) # ── SAME-CANDLE DEDUP (gate just before AI call) ────────── # If the AI was already invoked on this M5 candle for this # symbol, short-circuit. Pre-val above keeps running every iter # so the orchestrator still emits scan_skip / entry_rejected # for those — only the AI call is suppressed. if candle_time and candle_time <= last_entry_eval_time: return EntryEvalResult( decision=None, dedup_skipped=True, ) # ── BUILD PROMPT + AI ───────────────────────────────────── profile = self._profile_for(symbol) signals = extract_pa_signals(tech) score = score_pa(signals, direction) prompt = self._build_entry_prompt( symbol=symbol, direction=direction, bias=bias, regime=regime, rsi=rsi, rsi_prev=rsi_prev, atr_ratio=atr_ratio, candle_strength=candle_strength, tech=tech, signals=signals, score=score, profile=profile, ) resp = await self.ai.ask_for_decision(prompt) # Cache-update policy per error_kind (V16 same-candle dedup): # text present (any parse outcome) -> update (AI consumed) # text None + kind in {credit,invalid} -> update (permanent) # text None + kind unknown (transient) -> NOT updated (retry) TRANSIENT_KINDS = {"unknown", "overload"} if resp.text is not None: ai_eval_ts: Optional[float] = candle_time if candle_time else None elif resp.error_kind in TRANSIENT_KINDS: ai_eval_ts = None # retry next iter else: ai_eval_ts = candle_time if candle_time else None # permanent if resp.text is None: return EntryEvalResult( decision=self._reject( symbol, direction, rule="API_ERROR", detail=f"AI unavailable: {resp.error_kind}", tech=tech, error_kind=resp.error_kind, attempts=resp.attempts, ), evaluated_candle_time=ai_eval_ts, ) result = self._parse_ai_json(resp.text) if not result or "approved" not in result: return EntryEvalResult( decision=self._reject( symbol, direction, rule="API_PARSE_ERROR", detail="AI returned malformed JSON", tech=tech, response_preview=(resp.text or "")[:200], ), evaluated_candle_time=ai_eval_ts, ) # ── POST-VALIDATION WATCHDOG ───────────────────────────── rsi_h4 = float(tech.rsi_h4) rejection = self._post_validate( result=result, symbol=symbol, direction=direction, rsi=rsi, rsi_h4=rsi_h4, profile=profile, entry_price=float(tech.price), ) if rejection is not None: extra_fields: dict = {} if rejection[0] == "TP_BELOW_MIN_PROFIT": extra_fields["tp_source"] = "rejected_tp_too_small" extra_fields["tp_price_suggested_ai"] = float( result.get("tp_price_suggested", 0.0) or 0.0 ) return EntryEvalResult( decision=self._reject( symbol, direction, rule=rejection[0], detail=rejection[1], tech=tech, ai_confidence=int(result.get("confidence", 0)), ai_sl_atr=float(result.get("sl_atr_multiplier", 0.0)), ai_rr_multiplier=float(result.get("rr_multiplier", 0.0)), **extra_fields, ), evaluated_candle_time=ai_eval_ts, ) if not result.get("approved"): rd = result.get("rejection_details") or {} ai_rule = str(rd.get("rule_failed", "UNSPECIFIED"))[:40] if isinstance(rd, dict) else "UNSPECIFIED" ai_detail = str(rd.get("detail", ""))[:200] if isinstance(rd, dict) else "" return EntryEvalResult( decision=self._reject( symbol, direction, rule="AI_REJECTED", detail=str(result.get("reason", "AI did not approve"))[:200], tech=tech, ai_confidence=int(result.get("confidence", 0)), ai_rule_failed=ai_rule, ai_rejection_detail=ai_detail, ), evaluated_candle_time=ai_eval_ts, ) # ── SL PRICE COMPUTATION (TP variante γ: resolved post-sizing) ─ # Brain only emits SL price + rr_multiplier. Orchestrator calls # tp_resolver.resolve_tp_price(...) post-sizing to finalize tp_price. try: pc = self._compute_sl_only( symbol=symbol, direction=direction, tech=tech, sl_atr=float(result["sl_atr_multiplier"]), ) except (KeyError, ValueError) as e: return EntryEvalResult( decision=self._reject( symbol, direction, rule="PRICE_COMPUTATION_ERROR", detail=f"cannot compute prices: {e}", tech=tech, atr_m5_points=getattr(tech, "atr_m5_points", None), ), evaluated_candle_time=ai_eval_ts, ) confidence = int(result.get("confidence", 0)) rationale = str(result.get("reason", "TF setup approved"))[:120] rr_multiplier_ai = float(result.get("rr_multiplier", 0.0)) decision = EntryDecision( direction=direction, entry_price=pc["entry_price"], sl_price=pc["sl_price"], tp_price=0.0, # sentinel: orchestrator finalizes post-sizing rr_multiplier=rr_multiplier_ai, confidence=confidence, rationale=rationale, metadata={ "brain": "TF", "sl_atr_multiplier": pc["sl_atr"], "rr_multiplier_ai": rr_multiplier_ai, "tp_price_suggested_ai": float(result.get("tp_price_suggested", 0.0) or 0.0), "tp_rationale_ai": str(result.get("tp_rationale", ""))[:200], "risk_multiplier": float(result.get("risk_multiplier", 1.0)), "key_risk": str(result.get("key_risk", ""))[:120], "step_1": str(result.get("step_1_qualita_sconto", ""))[:200], "step_2": str(result.get("step_2_timing_pullback", ""))[:200], "step_3": str(result.get("step_3_contesto_macro", ""))[:200], "sl_ticks_used": pc["sl_ticks_post_clamp"], "clamp_active": pc["clamp_active"], "sl_floored": pc["sl_floored"], "sl_min_ticks": pc["sl_min_ticks"], # A/B test passivo SL (v18-dev): consumato dall'orchestrator # per popolare TradeRuntime.sl_price_ai_raw. "sl_price_ai_raw": pc["sl_price_ai_raw"], "sl_ticks_pre": pc["sl_ticks_pre"], "sl_floored_or_capped": pc["sl_floored_or_capped"], "profile_origin": profile.get("v15_origin", "?"), }, ) return EntryEvalResult( decision=decision, evaluated_candle_time=ai_eval_ts, ) # ============================================================ # POST-VALIDATION (watchdog on AI output) # ============================================================ @staticmethod def _post_validate( result: dict, symbol: str, direction: str, rsi: float, rsi_h4: float, profile: dict, entry_price: float, ) -> Optional[tuple[str, str]]: """ Returns (rule, detail) if AI output should be rejected; None if OK. AI is unreliable. We re-check the same hard rules even if AI approved (V15 lesson learned). """ if not result.get("approved"): return None # nothing to police; AI already rejected # 1. RSI zone paranoid recheck if direction == "BUY" and not (RSI_TF_LOW <= rsi <= RSI_TF_HIGH): return ("POST_VAL_RSI", f"AI approved with RSI {rsi:.1f} outside TF BUY zone") if direction == "SELL" and not (RSI_TF_LOW <= rsi <= RSI_TF_HIGH): return ("POST_VAL_RSI", f"AI approved with RSI {rsi:.1f} outside TF SELL zone") # 2. Confidence floor confidence = int(result.get("confidence", 0)) if confidence < MIN_CONFIDENCE: return ("POST_VAL_CONF", f"confidence {confidence}% < {MIN_CONFIDENCE}% floor") # 3. (rimosso V16-29-apr) RR floor pre-clamp. # Filosofia APEX: SL=risk, TP=rr_multiplier × $rischio reale, # RR=conseguenza diretta della scelta AI. sl_mult = float(result.get("sl_atr_multiplier", 0.0)) rr_mult = float(result.get("rr_multiplier", 0.0)) # 4. Guardia C>=90% + H4 weak (V14 fix) if confidence >= 90: dir_ok = ( (direction == "BUY" and rsi_h4 >= 45) or (direction == "SELL" and rsi_h4 <= 55) ) if not dir_ok: return ("POST_VAL_C90_H4", f"C{confidence}% but H4 RSI {rsi_h4:.0f} against {direction}") # 5. SL multiplier range — enforcement, not advisory. # V15 only listed sl_range in the prompt; the AI was free to drift. # V16 incident 29 apr: MGC TF SELL came back with sl_atr way out of # profile range. Enforce code-side. sl_low, sl_high = profile.get("sl_range", (0.0, float("inf"))) if not (sl_low <= sl_mult <= sl_high): return ("POST_VAL_SL_RANGE", f"sl_atr_multiplier {sl_mult:.3f} out of profile sl_range " f"[{sl_low:.3f},{sl_high:.3f}] for {symbol}") # 6. rr_multiplier in advisory prompt-range [0.17, 0.67]. # Hard clamp [0.10, 0.80] is applied later by tp_resolver as # defense-in-depth; this advisory rejection catches AI drift # outside the documented confidence→rr mapping (calibration signal). if rr_mult <= 0: return ("POST_VAL_RR_MISSING", f"rr_multiplier {rr_mult} not positive") if not (0.17 <= rr_mult <= 0.67): return ("POST_VAL_RR_RANGE", f"rr_multiplier {rr_mult:.3f} outside advisory " f"prompt-range [0.17, 0.67]") # 7. TP_BELOW_MIN_PROFIT (V17). Applies only to AI-suggested TP # path: if tp_price_suggested is provided and direction-coherent, # compute the per-contract net after the 15% conservative margin # and reject when below the configured threshold. This is a HARD # rejection (no floor lift) — a tight AI-suggested TP is itself # signal that the setup is marginal, prefer to skip. The # rr-fallback path (handled in tp_resolver via MIN_TP_TICKS) is # NOT subject to this rule. tp_suggested = float(result.get("tp_price_suggested", 0.0) or 0.0) if tp_suggested > 0: coherent = ( (direction == "BUY" and tp_suggested > entry_price) or (direction == "SELL" and tp_suggested < entry_price) ) if coherent: spec = cfg_fut.ASSETS_MAP.get(symbol, {}) tick_size = float(spec.get("tick_size", 0.0)) tick_value = float(spec.get("tick_value", 0.0)) if tick_size > 0 and tick_value > 0: scale = 1.0 - TP_SUGGESTED_CONSERVATIVE_MARGIN if direction == "BUY": tp_dist_post = (tp_suggested - entry_price) * scale else: tp_dist_post = (entry_price - tp_suggested) * scale tp_ticks = max(0, int(round(tp_dist_post / tick_size))) tp_gross = tp_ticks * tick_value tp_net = tp_gross - cfg_fut.COMMISSION_PER_CONTRACT_USD if tp_net < cfg_fut.TP_MIN_NET_PROFIT_USD: return ( "TP_BELOW_MIN_PROFIT", f"AI tp_suggested={tp_suggested} → " f"{tp_ticks} ticks post-15%-margin → " f"gross/ct ${tp_gross:.2f} − " f"commission ${cfg_fut.COMMISSION_PER_CONTRACT_USD:.2f} = " f"net/ct ${tp_net:.2f} < " f"${cfg_fut.TP_MIN_NET_PROFIT_USD:.2f} threshold", ) return None # ============================================================ # SL PRICE COMPUTATION (TP variante γ: TP resolved post-sizing) # ============================================================ @staticmethod def _compute_sl_only( symbol: str, direction: str, tech, sl_atr: float, ) -> dict: """ Compute SL price + tick diagnostics from sl_atr_multiplier. TP variante γ: TP is finalized post-sizing by orchestrator via tp_resolver.resolve_tp_price; this Brain step only fixes SL. Logic: sl_distance_pre = atr_m5_points * sl_atr * SL_SAFETY_MULT sl_ticks_pre = round(sl_distance_pre / tick_size) sl_ticks_post = clamp(sl_ticks_pre, [MIN_SL_TICKS, MAX_SL_TICKS]) sl_distance_post = sl_ticks_post * tick_size sl_price = entry ∓ sl_distance_post (BUY: -, SELL: +) Returns dict with entry_price, sl_price, sl_atr, sl ticks/distances, clamp_active. NO TP fields (resolved later). Raises: KeyError if symbol unknown. ValueError if atr_m5_points missing or non-positive. """ spec = cfg_fut.ASSETS_MAP[symbol] tick_size = float(spec["tick_size"]) digits = int(spec["digits"]) entry_price = float(tech.price) atr_points = float(getattr(tech, "atr_m5_points", 0.0) or 0.0) if atr_points <= 0: raise ValueError("atr_m5_points missing or non-positive in tech") sl_distance_pre = atr_points * sl_atr * cfg_fut.SL_SAFETY_MULT sl_ticks_pre = round(sl_distance_pre / tick_size) sl_min = cfg_fut.MIN_SL_TICKS.get(symbol, 0) sl_max = cfg_fut.MAX_SL_TICKS.get(symbol, 10_000) sl_ticks_post = max(sl_min, min(sl_ticks_pre, sl_max)) sl_distance_post = sl_ticks_post * tick_size if direction == "BUY": sl_price = entry_price - sl_distance_post sl_price_ai_raw = entry_price - (sl_ticks_pre * tick_size) else: sl_price = entry_price + sl_distance_post sl_price_ai_raw = entry_price + (sl_ticks_pre * tick_size) # V18 12-mag — `sl_floored` distingue il caso "AI propone SL più # stretto del MIN_SL_TICKS configurato" dal clamp generico. # Incidente MYM: SL troppo vicino veniva toccato dal rumore di # mercato; loggiamo questo specifico bump per analisi calibrazione. sl_floored = sl_min > 0 and sl_ticks_pre < sl_min return { "entry_price": round(entry_price, digits), "sl_price": round(sl_price, digits), "atr_m5_points": atr_points, "tick_size": tick_size, "sl_atr": sl_atr, "sl_distance_pre_clamp": round(sl_distance_pre, digits + 2), "sl_distance_post_clamp": round(sl_distance_post, digits + 2), "sl_ticks_pre_clamp": sl_ticks_pre, "sl_ticks_post_clamp": sl_ticks_post, "sl_min_ticks": sl_min, "sl_max_ticks": sl_max, "clamp_active": sl_ticks_pre != sl_ticks_post, "sl_floored": sl_floored, # A/B test passivo SL (v18-dev): SL grezzo proposto dall'AI prima # del clamp MIN/MAX. Loggato solo, mai inviato al broker. "sl_ticks_pre": sl_ticks_pre, "sl_price_ai_raw": round(sl_price_ai_raw, digits), "sl_floored_or_capped": sl_ticks_post != sl_ticks_pre, } # ============================================================ # ENTRY PROMPT BUILDER # ============================================================ @staticmethod def _build_entry_prompt( *, symbol: str, direction: str, bias: str, regime: str, rsi: float, rsi_prev: float, atr_ratio: float, candle_strength: float, tech, signals: PASignals, score: PAScore, profile: dict, ) -> str: """ V17 narrative entry prompt. Espone TUTTI i campi del TechSnapshot al modello, organizzati in sei sezioni narrative: 1. Contesto mercato (sessione/liquidità, regime, bias, news) 2. Storia del prezzo (narrativa naturale) 3. Segnali esaurimento pullback (checklist tecnica) 4. Contesto strutturale (swing, EMA50 H1, anti-revenge) 5. Profilo rischio (range SL meccanico, candle metadata) 6. Domanda esplicita + Chain of Thought 3-step Invarianti rispetto a V16: - Chain of Thought 3-step (qualità sconto / timing / macro) - Confidence floor MIN_CONFIDENCE (=70) - SL profile range mapping (sl_atr_multiplier ∈ profile.sl_range) V17 transitional bridge: - JSON output mantiene il contratto V16 (`step_1/2/3`, `risk_multiplier`, `rr_multiplier` ∈ [0.17, 0.67]) — pipeline downstream (evaluate_entry / _post_validate / tp_resolver) NON cambia in questo commit. - Aggiunto `tp_price_suggested` (livello TP assoluto basato su VWAP/swing/EMA50) accanto a `rr_multiplier`. Per ora viene loggato; downstream lo consumerà nei commit V17 successivi (margine conservativo 15% applicato in pipeline). """ # ── SL profile ───────────────────────────────────────────── sl_low, sl_high = profile["sl_range"] sl_mid = round((sl_low + sl_high) / 2, 3) sl_floor = profile.get("sl_min_points", 0) # ── Tech extracts ────────────────────────────────────────── rsi_h1 = float(tech.rsi_h1) rsi_h4 = float(tech.rsi_h4) struct = tech.market_structure regime_reason = tech.regime_reason h1_compat = float(tech.h1_compatibility) h1_reason = tech.h1_reason consec_sl = int(tech.consecutive_sl_count) vwap = tech.vwap vwap_dev = float(tech.vwap_deviation_pct) deviation_pct = float(tech.deviation_pct) vol_regime = tech.vol_regime vol_spike = bool(tech.vol_spike) divergence = tech.divergence macd_decel = bool(tech.macd_decelerating) tech_allowed = tech.allowed_direction rsi_bouncing = ( (direction == "BUY" and rsi > rsi_prev) or (direction == "SELL" and rsi < rsi_prev) ) rsi_motion = ( "in risalita" if rsi > rsi_prev else "ancora in discesa" if rsi < rsi_prev else "fermo" ) # ── Session + liquidity (UTC buckets) ────────────────────── ct = float(getattr(tech, "candle_time", 0) or 0) if ct > 0: now_dt = _dt.datetime.fromtimestamp(ct, tz=_dt.timezone.utc) else: now_dt = _dt.datetime.now(_dt.timezone.utc) now_utc_str = now_dt.strftime("%H:%M UTC") utc_hr = now_dt.hour if 0 <= utc_hr < 7: session_name, liquidity = "ASIA", "bassa" elif 7 <= utc_hr < 12: session_name, liquidity = "EUROPA", "media" elif 12 <= utc_hr < 16: session_name, liquidity = "OVERLAP EU-USA", "alta" elif 16 <= utc_hr < 21: session_name, liquidity = "USA", "alta" else: session_name, liquidity = "POST-USA", "bassa" # Brain runs only when news_filter doesn't block (gate is in # orchestrator BEFORE evaluate_entry). So when this prompt is # built, no HIGH-impact event is currently imminent. news_context = ( "nessun evento HIGH a calendario imminente " "(filtro news in orchestrator già passato)" ) # ── Narrative qualifiers ─────────────────────────────────── if "BULLISH" in struct: trend_word = "rialzista" elif "BEARISH" in struct: trend_word = "ribassista" else: trend_word = "laterale" if vwap_dev > 0: vwap_side = "sopra" elif vwap_dev < 0: vwap_side = "sotto" else: vwap_side = "su" if atr_ratio >= 1.2: atr_quality = "pulito" elif atr_ratio >= 0.8: atr_quality = "normale" else: atr_quality = "debole" macd_state = "decelerando" if macd_decel else "accelerando" # ── Active candle patterns ───────────────────────────────── pattern_flags: list[str] = [] if tech.hammer: pattern_flags.append("hammer") if tech.shooting_star: pattern_flags.append("shooting_star") if tech.bull_engulfing: pattern_flags.append("bull_engulfing") if tech.bear_engulfing: pattern_flags.append("bear_engulfing") if tech.piercing: pattern_flags.append("piercing") if tech.dark_cloud: pattern_flags.append("dark_cloud") if tech.morning_star: pattern_flags.append("morning_star") if tech.evening_star: pattern_flags.append("evening_star") if tech.doji: pattern_flags.append(f"doji({tech.doji_type or 'std'})") pattern_str = ", ".join(pattern_flags) if pattern_flags else "NESSUNO" doji_type_str = tech.doji_type or "NESSUNO" # ── PA / absorption alignment with trade direction ───────── pa_pro_dir = ( (direction == "BUY" and score.dominant == "BULLISH") or (direction == "SELL" and score.dominant == "BEARISH") ) if pa_pro_dir: pa_align = "favor" elif score.dominant in ("BULLISH", "BEARISH"): pa_align = "adverse" else: pa_align = "neutro" abs_pro_dir = ( (direction == "BUY" and tech.buy_absorption) or (direction == "SELL" and tech.sell_absorption) ) # ── Swing structural (lazy) ──────────────────────────────── swing = tech.swing_data or {} sw_found = swing.get("swing_found", False) sw_type = swing.get("swing_type", "N/A") if sw_found else "NON TROVATO" sw_price_str = ( f"{swing.get('swing_price', 0):.5f}" if sw_found else "N/A" ) sw_bars = swing.get("swing_index", 0) # ── H1 compat label ──────────────────────────────────────── if h1_compat >= 0.8: compat_label = "pro-trend" elif h1_compat >= 0.5: compat_label = "neutro" else: compat_label = "contro-trend" # ── Entry digits per asset ───────────────────────────────── entry_for_tp = float(tech.price) spec = cfg_fut.ASSETS_MAP.get(symbol, {}) digits_for_tp = int(spec.get("digits", 5)) # ── TP time-budget (V17 calibrazione: vincolo raggiungibilità) ── # Stima quanti tick il prezzo può percorrere realisticamente prima # del time-stop dell'asset/sessione. L'AI userà max_reachable_ticks # per evitare TP statisticamente fuori portata nel tempo disponibile. ts_session = _classify_session(utc_hr) minutes_to_timestop = TF_TIME_STOP.get(symbol, {}).get( ts_session, TF_TIME_STOP_DEFAULT_MIN, ) tick_size = float(spec.get("tick_size", 0.0) or 0.0) atr_m5_points = float(getattr(tech, "atr_m5_points", 0.0) or 0.0) atr_m5_ticks = (atr_m5_points / tick_size) if tick_size > 0 else 0.0 n_candles_to_ts = minutes_to_timestop / 5.0 max_reachable_ticks = int(round(atr_m5_ticks * n_candles_to_ts * 0.6)) # ── 6J inverted-quote note (carried over from V16) ───────── if symbol == "6J": usdjpy_entry = (1.0 / entry_for_tp) if entry_for_tp > 0 else 0.0 inverted_note = ( f"\n⚠️ 6J INVERTED QUOTE: 6J quote {entry_for_tp:.7f} ≡ " f"USDJPY {usdjpy_entry:.3f}. Movimento opposto: 6J↑ = yen " f"forte = USDJPY↓. BUY 6J ≡ SELL USDJPY (yen rafforza)." ) else: inverted_note = "" # ── Asset character (orienta la scelta TP) ───────────────── asset_character = { "MES": "S&P 500 micro: momentum forte, breakout trade-able, target swing high.", "MNQ": "Nasdaq micro: ad alta volatilità, candele ampie, target livello tondo.", "MYM": "Dow micro: meno volatile di MNQ, swing più ordinato.", "MGC": "Gold micro: spike-prone, range giornaliero ampio. Target conservativo.", "MCL": "Crude Oil micro: direzionale con inversioni rapide. Target su VWAP/swing.", "6E": "EUR/USD futures: mean-reverting, range giornaliero definito.", "6B": "GBP/USD futures: spike e poi range. Target su swing recenti.", "6A": "AUD/USD futures: risk-on proxy, scalper-friendly.", "6J": "JPY futures: INVERTED quote (6J↑ = USDJPY↓), tick scale separata.", "6C": "CAD futures: INVERTED quote, oil-correlated.", }.get(symbol, f"{symbol}: profile non dedicato, target conservativo.") favor_str = ", ".join(score.favor_direction) or "nessuno" adverse_str = ", ".join(score.adverse_direction) or "nessuno" candle_age = float(getattr(tech, "candle_age_seconds", 0.0) or 0.0) return f"""\ Sei un Senior Quant Trader specializzato in DEEP DISCOUNT Trend Following. Il tuo unico lavoro: entrare SOLO sui pullback quando il prezzo è in sconto all'interno di un trend confermato. Restituisci SOLO JSON valido — zero testo fuori dal JSON. ═══ 1. CONTESTO MERCATO OGGI ═══ Sessione : {session_name} ({now_utc_str}), liquidità {liquidity} Regime : {regime} — {regime_reason} Bias H4 : {bias}, direzione consentita: {tech_allowed} RSI H4 : {rsi_h4:.1f} Vol regime : {vol_regime}, Vol spike: {"SI" if vol_spike else "NO"} News : {news_context} ═══ 2. STORIA DEL PREZZO ═══ {symbol} è in trend {trend_word} confermato da {tech.trend_maturity} candele H1 (struttura attuale: {struct}). Il prezzo quota {entry_for_tp:.{digits_for_tp}f} e ha appena fatto un pullback — RSI M5 è passato da {rsi_prev:.1f} a {rsi:.1f} ({rsi_motion}). RSI H1: {rsi_h1:.1f}. Deviazione del prezzo da EMA50 H1: {deviation_pct:+.2f}%. VWAP intra-day: {vwap:.{digits_for_tp}f} — il prezzo è {vwap_side} del VWAP di {vwap_dev:+.2f}%. Struttura H1: {struct} — {h1_reason}. H1 compatibility con {direction}: {h1_compat:.2f} ({compat_label}). Divergenza RSI/prezzo: {divergence}. MACD: {macd_state}.{inverted_note} ═══ 3. SEGNALI DI ESAURIMENTO PULLBACK ═══ Conta i segnali presenti (checklist tecnica): • Bouncing RSI M5 : {"SI" if rsi_bouncing else "NO"} (BUY: rsi>rsi_prev = pullback finito; SELL: rsi 0 else "NO"}) {f"⚠️ {consec_sl} SL recenti: la struttura ti ha già stoppato. Nuova entry SOLO con almeno 1 pattern reversal attivo + volume_weak + RSI in zona ottimale ({'42-46' if direction == 'BUY' else '54-58'})." if consec_sl > 0 else ""} ═══ 5. PROFILO RISCHIO ═══ Profilo SL {symbol} (V15 origin: {profile.get('v15_origin','?')}): {profile.get('note','')} Range sl_atr_multiplier : {sl_low:.3f} – {sl_high:.3f} (mid {sl_mid:.3f}, ATR-meccanico) {f"⚠️ SL FLOOR ASSOLUTO: mai sotto {sl_floor} punti." if sl_floor else ""} Mappatura confidence → sl_atr_multiplier: 60-69%: sl_atr={sl_high:.3f} 70-79%: sl_atr={sl_mid:.3f} 80-84%: sl_atr={sl_mid:.3f} 85%+: sl_atr={sl_low:.3f} Candle metadata (M5 corrente): Open : {tech.open} Close : {entry_for_tp:.{digits_for_tp}f} Age post-close : {candle_age:.0f}s Asset character: {asset_character} ═══ 5b. VINCOLO TEMPORALE TP ═══ Time-stop {symbol} ({ts_session}): {minutes_to_timestop} minuti. ATR M5 corrente: {atr_m5_points:.{digits_for_tp}f} punti = {atr_m5_ticks:.0f} ticks/candela. Candele M5 disponibili prima del time-stop: ~{n_candles_to_ts:.0f}. Movimento statisticamente raggiungibile (fattore conservativo 0.6): max_reachable_ticks = {max_reachable_ticks} ⚠️ Scegli un `tp_price_suggested` raggiungibile entro {max_reachable_ticks} ticks dal prezzo {entry_for_tp:.{digits_for_tp}f}. Anche se VWAP/swing/EMA50 indicano livelli oltre questa soglia, NON li scegliere — l'orchestratore chiuderà il trade per time-stop prima che il target venga colpito. ═══ 6. DOMANDA + CHAIN OF THOUGHT ═══ Considerando il contesto di mercato (sez. 1), la storia del prezzo (sez. 2), i segnali di esaurimento pullback (sez. 3), il contesto strutturale (sez. 4) e il profilo rischio (sez. 5): → Questo pullback su {symbol} rappresenta un'opportunità di entrata {direction} con buona probabilità di successo? Se approvi, indica DUE campi TP (entrambi richiesti, transitional bridge V17): • `tp_price_suggested` (NEW V17): livello TP assoluto basato sui livelli strutturali visibili — VWAP {vwap:.{digits_for_tp}f}, swing {sw_type} {sw_price_str}, EMA50 H1 (deviazione {deviation_pct:+.2f}%). Il sistema applicherà un margine conservativo del 15% prima di piazzare l'ordine. Coerente con la direzione (BUY: > entry; SELL: < entry). • `rr_multiplier` (LEGACY V16, ancora consumato da tp_resolver): frazione del RISCHIO REALE in dollari, range [0.17, 0.67]. tp_usd_target = rr_multiplier × sl_usd_actual Mappatura confidence → rr_multiplier: 70-74%: 0.17 – 0.33 (TP molto vicino, alta probabilità hit) 75-79%: 0.33 – 0.50 (TP medio) 80-84%: 0.50 – 0.67 (TP ambizioso) 85%+ : 0.67 (max) Filosofia APEX "WR > RR": frazione conservativa, target raggiungibile intraday su Topstep no-overnight. Hard clamp [0.10, 0.80] applicato dall'orchestrator come defense-in-depth. Velocità di hit TP per la sessione corrente ({session_name}, liquidità {liquidity}): adatta sia tp_price_suggested sia rr_multiplier alla probabilità che il prezzo raggiunga il target intraday. CHAIN OF THOUGHT (3 step obbligatori — sintetizza in `reason`): STEP 1 — QUALITÀ DELLO SCONTO: RSI {rsi:.1f}: vicino a {'42' if direction == 'BUY' else '58'} = sconto OTTIMO, vicino al lato opposto della zona = sconto DEBOLE. STEP 2 — TIMING ESAURIMENTO PULLBACK: bouncing={rsi_bouncing}, volume_weak={signals.volume_weak}, PA dominant={score.dominant} (forza {score.strength:.1f}), VWAP {vwap_dev:+.2f}%, absorption pro {direction}: {abs_pro_dir}. 2+ segnali = +10%, 3+ = +15%, tutti = +20%. Absorption AVVERSA a {direction} = -10%. STEP 3 — CONTESTO MACRO + LIVELLI: H1 struct={struct}, trend_mat={tech.trend_maturity}, RSI H4={rsi_h4:.1f}, divergence={divergence}, swing={sw_type}@{sw_price_str}. trend_mat>5 = -10%, divergence contro = -15%. ═══ CRITERI APPROVAZIONE ═══ approved=true SOLO se: 1) confidence finale >= {MIN_CONFIDENCE} 2) STEP 1 non è "DEBOLE" 3) STEP 2 non è "PREMATURO" (>=1 segnale esaurimento) 4) STEP 3 non è "SFAVOREVOLE" 5) sl_atr_multiplier in [{sl_low:.3f}, {sl_high:.3f}] 6) rr_multiplier in [0.17, 0.67] (mapping confidence sopra) 7) tp_price_suggested coerente con la direzione (BUY: > entry {entry_for_tp:.{digits_for_tp}f}; SELL: < entry {entry_for_tp:.{digits_for_tp}f}) 8) tp_price_suggested entro max_reachable_ticks={max_reachable_ticks} ticks dal prezzo (BUY: ≤ entry + {max_reachable_ticks}×tick_size; SELL: ≥ entry − {max_reachable_ticks}×tick_size) ═══ REJECTION CODES (rule_failed quando approved=false) ═══ - "STEP1_DEBOLE" : sconto RSI insufficiente - "STEP2_PREMATURO" : pullback non esaurito (no bouncing/volume_weak/PA) - "STEP3_SFAVOREVOLE" : macro/H4/divergence/trend_mat contro - "CONFIDENCE_FLOOR" : conf < {MIN_CONFIDENCE} - "ABSORPTION_AGAINST" : absorption avversa alla direction - "ANTI_REVENGE" : SL recenti senza pattern + volume_weak + RSI ottimale - "TP_BEYOND_MAX_REACHABLE": tp_price_suggested oltre max_reachable_ticks (time-budget insufficiente) - "OTHER" : altri motivi (specifica in detail) Quando approved=true → rule_failed="NONE", detail="ok". ═══ OUTPUT JSON (RISPETTA L'ORDINE) ═══ {{ "step_1_qualita_sconto": "OTTIMO/BUONO/DEBOLE: ...", "step_2_timing_pullback": "OTTIMO/BUONO/PREMATURO: ...", "step_3_contesto_macro": "FAVOREVOLE/NEUTRO/SFAVOREVOLE: ...", "approved": true, "confidence": 75, "direction": "{direction}", "risk_multiplier": 1.0, "sl_atr_multiplier": {sl_mid:.3f}, "rr_multiplier": 0.50, "tp_price_suggested": 0.0, "tp_rationale": "max 25 parole — livello tecnico + frazione $rischio", "key_risk": "rischio principale del setup", "reason": "max 8 parole, no virgolette", "rejection_details": {{"rule_failed": "NONE", "detail": "ok"}} }} """ # ============================================================ # EXIT # ============================================================ async def manage_exit(self, ctx: BrainContext) -> BrainDecision: decision = await self._manage_exit_logic(ctx) self._emit_manage_decision_log(ctx, decision) return decision async def _manage_exit_logic(self, ctx: BrainContext) -> BrainDecision: """ TF exit management. Bias toward HOLD (V15 philosophy). Order: 1. Time-stop (only if P&L <= 0): EXIT. 2. Same-candle dedup: HOLD without AI call. 3. Build exit prompt (with struct-flip / deep-discount notes). 4. Call AI (deterministic). 5. Map exit_now=true -> EXIT, false -> HOLD, error -> HOLD default. Brain does NOT mutate runtime. The orchestrator must read BrainDecision.metadata["evaluated_candle_time"] (when set) and update TradeRuntime.last_exit_eval_time before persisting state. """ entry = ctx.entry rt = ctx.runtime tech = ctx.tech_now symbol = entry.symbol direction = entry.direction is_long = direction == Direction.BUY.value net_profit = rt.net_profit_usd minutes_open = rt.minutes_open # ── TIME STOP ───────────────────────────────────────────── utc_hour = _dt.datetime.now(_dt.UTC).hour session = _classify_session(utc_hour) ts_threshold = TF_TIME_STOP.get(symbol, {}).get( session, TF_TIME_STOP_DEFAULT_MIN ) if net_profit <= 0 and minutes_open >= ts_threshold: return self._exit( reason=f"Time Stop TF — {minutes_open:.0f}m/{ts_threshold}m ({session})", metadata={"trigger": "time_stop", "session": session}, ) # ── AUTO PARTIAL 50% + SL→BE ───────────────────────────── # Deterministico (no AI), one-shot via rt.partial_done. Posto # PRIMA del dedup come in brain_mr: anche su candele già valutate # vogliamo che il trigger scatti appena progress varca la soglia. progress = rt.progress_pct if progress >= TF_AUTO_PARTIAL_PROGRESS_PCT and not rt.partial_done: be_price = self._round_to_tick( entry.entry_price, float(tech.tick_size), ) return BrainDecision( action=TradeAction.PARTIAL_50.value, reason=(f"TP {progress:.0f}% — partial 50% TF + SL→BE"), metadata={ "trigger": "auto_partial_50", "set_be_after_partial": True, "be_price": be_price, "progress_pct": progress, }, ) # ── SAME-CANDLE DEDUP (idempotency, not gating) ────────── candle_time = float(getattr(tech, "candle_time", 0) or 0) if candle_time and candle_time <= rt.last_exit_eval_time: return self._hold( reason="stessa candela già analizzata", metadata={"trigger": "dedup"}, ) # ── STRUCT FLIP DETECTION (V14 + V15 grace) ────────────── struct_now = tech.market_structure struct_at_entry = entry.market_structure_at_entry struct_unchanged = (struct_now == struct_at_entry) struct_flipped = False if minutes_open >= STRUCT_FLIP_GRACE_MIN: if is_long and "BEARISH" in struct_now and "BEARISH" not in struct_at_entry: struct_flipped = True if (not is_long) and "BULLISH" in struct_now and "BULLISH" not in struct_at_entry: struct_flipped = True # ── DEEP-DISCOUNT H4 PARADOX ───────────────────────────── rsi_h4_at_entry = entry.rsi_h4_at_entry deep_discount = ( (is_long and rsi_h4_at_entry < 25) or ((not is_long) and rsi_h4_at_entry > 75) ) # ── BUILD AI PROMPT ────────────────────────────────────── signals = extract_pa_signals(tech) score = score_pa(signals, direction) prompt = self._build_exit_prompt( ctx=ctx, signals=signals, score=score, struct_unchanged=struct_unchanged, struct_flipped=struct_flipped, deep_discount=deep_discount, session=session, ) resp = await self.ai.ask_for_decision( prompt, max_tokens=400, where="manage_exit", ) if resp.text is None: return self._hold( reason=f"AI error ({resp.error_kind}) — hold default", metadata={"trigger": "ai_error", "error_kind": resp.error_kind}, ) result = self._parse_ai_json(resp.text) if not result or "exit_now" not in result: return self._hold( reason="AI malformed JSON — hold default", metadata={"trigger": "ai_malformed"}, ) meta = { "brain": "TF", "trigger": "ai_exit_eval", "session": session, "struct_flipped": struct_flipped, "deep_discount": deep_discount, "evaluated_candle_time": candle_time if candle_time else None, } if bool(result.get("exit_now")): return self._exit( reason=str(result.get("reason", "AI exit"))[:120], metadata=meta, ) # HOLD return BrainDecision( action="HOLD", reason=str(result.get("reason", "AI hold"))[:120], metadata=meta, ) # ============================================================ # EXIT PROMPT BUILDER # ============================================================ @staticmethod def _build_exit_prompt( *, ctx: BrainContext, signals: PASignals, score: PAScore, struct_unchanged: bool, struct_flipped: bool, deep_discount: bool, session: str, ) -> str: """ V17 narrative anti-hallucination exit prompt (TF). Default conservativo = HOLD (V15 philosophy preservata). Quattro sezioni narrative: 1. Contesto entry originale (snapshot al fill) 2. Situazione attuale (numeri reali aggiornati: P&L, prezzo, RSI M5/H1/H4, struttura now vs entry, ATR, MACD, divergenza, sessione) 3. Price action attuale (PA dominant, favor/adverse, volume, absorption, candle strength, VWAP deviation, pattern attivi) 4. Domanda rigida + regole anti-allucinazione Anti-hallucination guardrails (V17): - Risposta SOLO con i numeri/flag forniti — niente livelli/swing inventati, nessuna speculazione su candele non elencate - Se incerto → HOLD (default) - EXIT richiede ALMENO 2 segnali confluenti oggettivi citati nel `reason`. Single-signal exits sono rifiutati a priori. Specifici V16 preservati: - struct_flipped (con grace period) → 🚨 alert in sez. 2 - deep_discount H4 paradox → h4_rule dedicato in sez. 2 - VWAP trend-confirmed flag → labellato in sez. 3 - Paradosso struttura against-trend deliberato → nota in sez. 2 Output JSON invariato: {"exit_now": bool, "reason": str}. """ entry = ctx.entry rt = ctx.runtime tech = ctx.tech_now is_long = entry.direction == Direction.BUY.value trade_type = "LONG" if is_long else "SHORT" # ── RSI multi-TF: entry vs ora ──────────────────────────── rsi_m5_now = float(tech.rsi) rsi_h1_now = float(tech.rsi_h1) rsi_h4_now = float(tech.rsi_h4) rsi_m5_entry = entry.rsi_m5_at_entry rsi_h1_entry = entry.rsi_h1_at_entry rsi_h4_entry = entry.rsi_h4_at_entry # ── Struct note (V16 calibrated) ────────────────────────── if struct_flipped: struct_note = ( f"🚨 STRUTTURA INVERTITA CONTRO {trade_type}: " f"{entry.market_structure_at_entry} → {tech.market_structure}. " f"Il trend che supportava il TF è terminato — segnale forte di EXIT." ) elif struct_unchanged: entered_against_h1 = entry.h1_compat_at_entry < 0.5 if entered_against_h1: struct_note = ( f"⚠️ PARADOSSO STRUTTURA: entry against-trend deliberato " f"(compat={entry.h1_compat_at_entry:.2f}) approvato a " f"C{entry.confidence_at_entry}%. Struttura INVARIATA → " f"NON è motivo di exit." ) else: struct_note = ( f"Struttura H1 INVARIATA: " f"{entry.market_structure_at_entry}. Non motivo di exit." ) else: struct_note = ( f"Struttura H1 CAMBIATA: {entry.market_structure_at_entry} → " f"{tech.market_structure}. Va valutata insieme ad altri segnali." ) # ── H4 paradox / delta rule (V16 calibrated) ────────────── if deep_discount: if is_long: h4_rule = ( f"DEEP DISCOUNT (H4 entry {rsi_h4_entry:.1f}): IGNORA " f"livello assoluto < 35. EXIT solo se H4 era salito >35 " f"post-entry e ora ricaduto, o sceso sotto " f"{rsi_h4_entry - 5:.0f}." ) else: h4_rule = ( f"DEEP DISCOUNT (H4 entry {rsi_h4_entry:.1f}): IGNORA " f"livello assoluto > 65. EXIT solo se H4 era sceso <65 " f"post-entry e ora risalito, o salito sopra " f"{rsi_h4_entry + 5:.0f}." ) else: delta = rsi_h4_now - rsi_h4_entry warn = "" if is_long and delta < -15: warn = " ⚠️ Crollo H4 significativo — possibile inversione." elif (not is_long) and delta > 15: warn = " ⚠️ Rally H4 significativo — possibile inversione." h4_rule = ( f"RSI H4: entry {rsi_h4_entry:.1f} → ora {rsi_h4_now:.1f} " f"({delta:+.1f}).{warn}" ) # ── VWAP context ────────────────────────────────────────── vwap_dev = float(tech.vwap_deviation_pct) vwap_trend_confirmed = ( (is_long and vwap_dev > 0.2) or ((not is_long) and vwap_dev < -0.2) ) vwap_label = "trend confermato" if vwap_trend_confirmed else "non confermato" # ── PA / absorption alignment with trade direction ──────── pa_pro_dir = ( (is_long and score.dominant == "BULLISH") or ((not is_long) and score.dominant == "BEARISH") ) if pa_pro_dir: pa_align = "favor" elif score.dominant in ("BULLISH", "BEARISH"): pa_align = "adverse" else: pa_align = "neutro" abs_pro_dir = ( (is_long and tech.buy_absorption) or ((not is_long) and tech.sell_absorption) ) favor_str = ", ".join(score.favor_direction) or "nessuno" adverse_str = ", ".join(score.adverse_direction) or "nessuno" # ── Active candle patterns ──────────────────────────────── pattern_flags: list[str] = [] if tech.hammer: pattern_flags.append("hammer") if tech.shooting_star: pattern_flags.append("shooting_star") if tech.bull_engulfing: pattern_flags.append("bull_engulfing") if tech.bear_engulfing: pattern_flags.append("bear_engulfing") if tech.piercing: pattern_flags.append("piercing") if tech.dark_cloud: pattern_flags.append("dark_cloud") if tech.morning_star: pattern_flags.append("morning_star") if tech.evening_star: pattern_flags.append("evening_star") if tech.doji: pattern_flags.append(f"doji({tech.doji_type or 'std'})") pattern_str = ", ".join(pattern_flags) if pattern_flags else "NESSUNO" # ── SL distance % from current price ────────────────────── sl_dist_pct = 0.0 if entry.entry_price: current = float(tech.price) if is_long: sl_dist_pct = (current - entry.sl_price) / entry.entry_price * 100 else: sl_dist_pct = (entry.sl_price - current) / entry.entry_price * 100 # ── Asset digits per consistent number formatting ───────── spec = cfg_fut.ASSETS_MAP.get(entry.symbol, {}) digits = int(spec.get("digits", 5)) return f"""\ Sei un gestore di posizioni Trend Following. Valuta SOLO la full exit. Default conservativo = HOLD. Rispondi SOLO con JSON valido — zero testo fuori dal JSON. ═══ 1. CONTESTO ENTRY ORIGINALE ═══ {entry.symbol} trade {trade_type} aperto {rt.minutes_open:.0f} minuti fa. Entry {entry.entry_price:.{digits}f}, SL {entry.sl_price:.{digits}f} (distanza dal prezzo attuale {sl_dist_pct:+.2f}%), TP {entry.tp_price:.{digits}f}. Confidence originale: {entry.confidence_at_entry}%. RSI M5 entry: {rsi_m5_entry:.1f}, RSI H4 entry: {rsi_h4_entry:.1f}. Struttura H1 al fill: {entry.market_structure_at_entry}. Regime al fill: {entry.regime_at_entry}. Deep Discount: {"SÌ" if deep_discount else "NO"}. NOTA: Partial close gestito dal Router/Orchestrator. Qui valuti SOLO la FULL EXIT. ═══ 2. SITUAZIONE ATTUALE ═══ P&L netto : ${rt.net_profit_usd:.2f} Tempo aperto : {rt.minutes_open:.0f}m (sessione {session}) Prezzo attuale : {tech.price:.{digits}f} RSI multi-TF (entry → ora): M5: {rsi_m5_entry:.1f} → {rsi_m5_now:.1f} ({rsi_m5_now - rsi_m5_entry:+.1f}) H1: {rsi_h1_entry:.1f} → {rsi_h1_now:.1f} ({rsi_h1_now - rsi_h1_entry:+.1f}) H4: {rsi_h4_entry:.1f} → {rsi_h4_now:.1f} ({rsi_h4_now - rsi_h4_entry:+.1f}) {h4_rule} Struttura H1 ora : {tech.market_structure} Struttura H1 entry : {entry.market_structure_at_entry} {struct_note} Trend Maturity : {tech.trend_maturity} candele H1 ATR Ratio : {tech.atr_ratio:.2f}x Vol Regime : {tech.vol_regime} Vol Spike : {"SI" if tech.vol_spike else "NO"} MACD : {"decelerando" if tech.macd_decelerating else "accelerando"} Divergenza : {tech.divergence} ═══ 3. PRICE ACTION ATTUALE ═══ PA dominante : {score.dominant} (forza {score.strength:.1f}x) — {pa_align} a {trade_type} Bullish/Bearish : {score.bullish_score:.1f} / {score.bearish_score:.1f} FAVOR {trade_type:5} : {favor_str} ADVERSE {trade_type:5}: {adverse_str} Volume : {"DEBOLE" if signals.volume_weak else "NORMALE"} (volume_weak={signals.volume_weak}) Absorption pro {trade_type}: {"SI" if abs_pro_dir else "NO"} (BUY_ABS={tech.buy_absorption}, SELL_ABS={tech.sell_absorption}) Candle Strength : {tech.candle_strength:.2f}x (1.0 = media) Pattern attivi : {pattern_str} VWAP intra-day : {tech.vwap:.{digits}f} VWAP deviation : {vwap_dev:+.2f}% ({vwap_label}) ═══ 4. DOMANDA RIGIDA — ANTI-ALLUCINAZIONE ═══ Basandoti SOLO sui dati oggettivi sopra: il prezzo si sta muovendo CONTRO il trade {trade_type} con struttura cambiata? EXIT solo se almeno 2 segnali oggettivi confluenti confermano inversione. Se incerto → HOLD. Non ipotizzare, non speculare — usa SOLO i numeri forniti. Esempi di segnali oggettivi citabili dal contesto sopra: - struttura H1 CAMBIATA contro la direzione (sez. 2, struct_note) - RSI H4 delta significativo contro (sez. 2, h4_rule) - PA dominante ADVERSE con forza > 1.5x (sez. 3) - pattern reversal contrario attivo + candle strength > 1.8x (sez. 3) - divergenza confermata + MACD decelerando (sez. 2) - VWAP deviation contraria + vol_regime HIGH (sez. 2 + 3) - candela gigante contraria (candle_strength > 2.5x) + vol_spike (sez. 3) REGOLE OPERATIVE: • Rispondi SOLO con i dati forniti — non inventare livelli, swing, pattern non elencati, né target ipotetici • Se incerto → HOLD (default conservativo) • EXIT richiede ALMENO 2 segnali confluenti oggettivi (citali nel reason) • Single-signal exits sono RIFIUTATI a priori (es. "RSI estremo" da solo, "VWAP contro" da solo, "struct invariata" da sola → HOLD) • Paradosso struttura: se entry era against-trend deliberato (vedi nota sez. 2), struttura invariata NON è motivo di exit {"• Deep discount: ignora livello assoluto H4, applica solo h4_rule sez. 2" if deep_discount else ""} ═══ OUTPUT (SOLO JSON) ═══ {{ "exit_now": false, "reason": "max 12 parole — cita dati specifici (es. 'struct flip H1 + RSI H4 -18 + adverse PA 1.7x')" }} """ # ============================================================ # HELPERS # ============================================================ @staticmethod def _parse_ai_json(text: str) -> Optional[dict]: try: return json.loads(extract_json_from_response(text)) except (json.JSONDecodeError, TypeError): return None def _reject( self, symbol: str, direction: str, rule: str, detail: str, *, tech=None, **fields, ) -> None: """ Emit a structured "entry_rejected" event and return None. Every reject path passes through here. Fields (rule + caller- provided numerics) end up in brain_log.jsonl for post-mortem analytics — esp. CALIBRAZIONE 5-9 mag where we need to histogram skip causes and inspect ATR/RR distributions on real trades. V18 12-mag — `tech` opzionale: se passato, i campi RADAR (rsi_m5/rsi_h4/h1_compat/macd_hist/pattern/atr_ratio/bias/...) vengono auto-popolati via `tech_log_fields(tech)`. Eventuali valori in `fields` hanno precedenza sui campi auto-derivati. Args: rule: short stable code for the rejection cause (e.g. "PULLBACK_ZONE", "POST_VAL_TP_RANGE"). detail: human-readable explanation, max ~200 chars. tech: TechSnapshot per arricchimento dashboard. **fields: rule-specific numerics (rsi_threshold, sl_distance_*, ...). """ merged_fields = {**tech_log_fields(tech), **fields} if self.logger is not None: try: self.logger.brain_log.write( "entry_rejected", symbol=symbol, brain="TF", direction=direction, rule=rule, reason=detail, **merged_fields, ) except AttributeError: # Fall back to system logger if LoggerBundle missing # the brain_log attribute (e.g. in unit tests) self.logger.system.info( f"[TF reject] {symbol} {direction} {rule}: {detail} " f"fields={merged_fields}" ) else: _log.debug( "[TF reject] %s %s %s: %s fields=%s", symbol, direction, rule, detail, merged_fields, ) return None