""" Brain selection — port verbatim of V15 APEX_PREDATOR_V15._choose_brain (riga 1098-1215). Decides whether a given symbol/tech/bias combination should route to: - Brain TF (trend following) - Brain MR (mean reversion) - or skip (None) Pure function (no I/O, no state). Lives in brain/ because the decision is trading-policy semantics, not market analysis. analysis.regime.determine_regime produces the regime label that is INPUT to this router. V15 calibration constants (MR_EXCLUDED, TF_ALLOWED_SYMBOLS, INDICI_FUTURES) are inlined here — they are routing policy, not market data. If more consumers emerge they can be promoted to core/config_futures.py. DA RIVALIDARE IN CALIBRAZIONE V16: - night_tf_block window (20:00-04:00 UTC) is V15-empirical - thresholds 35/65 (TRENDING_SOFT MR widened) vs 32/68 (RANGING strict MR) - pullback band 42-58 (TF entries on indices and forex) """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional # ============================================================ # ROUTING CONSTANTS (V15 config_futures.py:455-482, port verbatim) # ============================================================ MR_EXCLUDED: set[str] = { "MYM", # V14: MR -$1,314 WR50% on US30 -> only TF } TF_ALLOWED_SYMBOLS: set[str] = { # Indices (V14 strong TF history) "MES", "MNQ", "MYM", # Forex majors with V14 positive TF history "6E", "6B", "6C", # Forex minors (TF unvalidated, kept on whitelist for opportunity) "6A", "6J", # Commodities "MGC", "MCL", } INDICI_FUTURES: set[str] = {"MES", "MNQ", "MYM", "YM"} # ============================================================ # RESULT TYPE # ============================================================ @dataclass(frozen=True) class BrainSelectionResult: """ Outcome of choose_brain. chosen != None ⇒ reject_reason is None (accept) chosen is None ⇒ reject_reason valued (reject) details always carries a base snapshot (rsi, rsi_prev, regime, allowed_direction, trend_maturity, h1_struct, is_index) plus branch-specific flags. Designed for jq aggregation on brain_log.jsonl — flags are spread top-level by orchestrator._log_skip. """ chosen: Optional[str] reject_reason: Optional[str] details: dict[str, Any] = field(default_factory=dict) # ============================================================ # HELPERS # ============================================================ def _is_night_tf_block(now_utc: Optional[datetime] = None) -> bool: """ True during V15-empirical illiquid window (20:00-04:00 UTC). Trend-Following may sit hours in loss in thin markets; Mean-Reversion handled OK by RSI-extremes even when illiquid. Caller can pass now_utc for testability; default = real clock. """ now = now_utc if now_utc is not None else datetime.now(timezone.utc) h = now.hour return h >= 20 or h < 4 # ============================================================ # CHOOSE BRAIN # ============================================================ def choose_brain( *, symbol: str, tech, # analysis.tech_snapshot.TechSnapshot bias, # analysis.bias.BiasData now_utc: Optional[datetime] = None, ) -> BrainSelectionResult: """ Routes (symbol, tech, bias) to BrainSelectionResult with chosen ∈ {"TF", "MR", None}. Port verbatim of V15 _choose_brain. The shape of the matrix is intentionally preserved — this is calibrated logic, not architecture. Refactoring it semantically would invalidate V15 calibration. """ allowed = bias.allowed_direction if bias is not None else "BOTH" rsi = float(tech.rsi) rsi_prev = float(tech.rsi_prev) rsi_bouncing = rsi > rsi_prev h1_struct = tech.market_structure is_index = symbol in INDICI_FUTURES regime = tech.regime trend_maturity = int(tech.trend_maturity) base: dict[str, Any] = { "rsi": rsi, "rsi_prev": rsi_prev, "rsi_bouncing": rsi_bouncing, "regime": regime, "allowed_direction": allowed, "trend_maturity": trend_maturity, "h1_struct": h1_struct, "is_index": is_index, } def _accept(brain: str, **extra: Any) -> BrainSelectionResult: d = {**base, **extra} return BrainSelectionResult(chosen=brain, reject_reason=None, details=d) def _reject(reason: str, **extra: Any) -> BrainSelectionResult: d = {**base, **extra} return BrainSelectionResult(chosen=None, reject_reason=reason, details=d) if allowed == "NONE": return _reject("BIAS_NONE") # NB V16: regime label here is what determine_regime() emits: # "TRENDING" | "TRENDING_SOFT" | "BREAKOUT" | "RANGING" | "UNKNOWN". # Port preserves V15 semantics for these labels. # ---- BREAKOUT: skip volatility spike ---- if regime == "BREAKOUT": return _reject("REGIME_BREAKOUT") # ---- RANGING: only MR with RSI extreme; never on indices (V15 patch #20) ---- if regime == "RANGING": if is_index: return _reject("RANGING_INDEX_BLOCKED") if symbol in MR_EXCLUDED: return _reject("MR_EXCLUDED_SYMBOL", mr_excluded=True) if allowed in ("BUY", "BOTH") and rsi < 32: return _accept("MR") if allowed in ("SELL", "BOTH") and rsi > 68: return _accept("MR") return _reject( "RANGING_RSI_NOT_EXTREME", rsi_below_32=(rsi < 32), rsi_above_68=(rsi > 68), ) # ---- TRENDING / TRENDING_SOFT: TF if whitelisted else MR ---- if regime in ("TRENDING", "TRENDING_SOFT") and trend_maturity >= 3: is_soft = (regime == "TRENDING_SOFT") night_block = _is_night_tf_block(now_utc) if symbol in TF_ALLOWED_SYMBOLS: # ---- INDICI: TF only with-trend, MR pro-trend allowed ---- if is_index: trend_is_bull = "BULLISH" in h1_struct trend_is_bear = "BEARISH" in h1_struct # TF pro-trend (pullback 42-58) if ( allowed in ("BUY", "BOTH") and trend_is_bull and 42 <= rsi <= 58 and rsi_bouncing and not night_block ): return _accept("TF") if ( allowed in ("SELL", "BOTH") and trend_is_bear and 42 <= rsi <= 58 and not rsi_bouncing and not night_block ): return _accept("TF") # MR pro-trend (V15 patch #20: indices only with-trend) if symbol not in MR_EXCLUDED: if allowed in ("BUY", "BOTH") and trend_is_bull and rsi < 32: return _accept("MR") if allowed in ("SELL", "BOTH") and trend_is_bear and rsi > 68: return _accept("MR") return _reject( "INDICES_NO_PRO_TREND_SETUP", is_soft=is_soft, trend_is_bull=trend_is_bull, trend_is_bear=trend_is_bear, in_pullback_42_58=(42 <= rsi <= 58), night_tf_block=night_block, mr_excluded=(symbol in MR_EXCLUDED), rsi_below_32=(rsi < 32), rsi_above_68=(rsi > 68), ) # ---- FOREX / COMMODITIES ---- if not is_soft: # Full TRENDING -> TF on pullback 42-58 if ( allowed in ("BUY", "BOTH") and 42 <= rsi <= 58 and rsi_bouncing and not night_block ): return _accept("TF") if ( allowed in ("SELL", "BOTH") and 42 <= rsi <= 58 and not rsi_bouncing and not night_block ): return _accept("TF") # TRENDING_SOFT or pullback incomplete -> MR widened 35/65 if symbol not in MR_EXCLUDED: if allowed in ("BUY", "BOTH") and rsi < 35: return _accept("MR") if allowed in ("SELL", "BOTH") and rsi > 65: return _accept("MR") return _reject( "TRENDING_NO_TF_NO_MR", is_soft=is_soft, in_pullback_42_58=(42 <= rsi <= 58), night_tf_block=night_block, mr_excluded=(symbol in MR_EXCLUDED), rsi_below_35=(rsi < 35), rsi_above_65=(rsi > 65), ) # ---- Asset NOT in TF whitelist: only MR with extreme RSI ---- if symbol not in MR_EXCLUDED: if allowed in ("BUY", "BOTH") and rsi < 32: return _accept("MR") if allowed in ("SELL", "BOTH") and rsi > 68: return _accept("MR") return _reject( "NON_WHITELIST_NO_MR_EXTREME", mr_excluded=(symbol in MR_EXCLUDED), rsi_below_32=(rsi < 32), rsi_above_68=(rsi > 68), ) # No valid setup — distinguish maturity-shortfall (frequent, expected) # from genuinely unhandled regime labels (anomaly). if regime in ("TRENDING", "TRENDING_SOFT"): return _reject("MATURITY_TOO_LOW") return _reject("REGIME_UNHANDLED")