""" APEX V16 — Brain abstract interface. Both Brain TF (Trend Following) and Brain MR (Mean Reversion) inherit from BrainBase. This enforces a SINGLE contract for the Brain<->Orchestrator boundary, fixing V15-BUG-4 (manage_exit signature differed between TF and MR). Brain responsibilities: - evaluate_entry: given current tech indicators, decide if a new trade should be opened. Returns EntryDecision or None. - manage_exit: given a BrainContext (entry snapshot + runtime + tech_now), decide HOLD / EXIT / PARTIAL_50 / MOVE_SL. Brain MUST NOT: - Place orders directly (orchestrator's job) - Mutate state files (persistence layer's job) - Talk to broker (broker layer's job) Brain MAY: - Call AI (Gemini) for decision support - Read tech_now and entry snapshot - Log to brain_log.jsonl via injected logger """ from __future__ import annotations from abc import ABC, abstractmethod from typing import Optional from core.contracts import ( BrainContext, BrainDecision, EntryDecision, TradeAction, ) class BrainBase(ABC): """ Abstract base class for all Brains. Subclasses MUST implement evaluate_entry() and manage_exit(). Subclasses MAY override _hold(), _exit() helpers if desired, but the default implementations should suffice. """ name: str = "BASE" # subclasses override: "TF" or "MR" def __init__(self, config) -> None: """ Args: config: RuntimeConfig instance (for ai_model, risk_per_trade, etc.) """ self.config = config # ============================================================ # ENTRY EVALUATION # ============================================================ @abstractmethod async def evaluate_entry( self, symbol: str, tech, *, bias_data, ) -> Optional[EntryDecision]: """ Evaluate whether to open a new trade for `symbol` given current tech. Async because implementations may call AIClient.ask_for_decision(). Args: symbol: instrument root. tech: TechSnapshot (analysis.tech_snapshot.TechSnapshot). Untyped here to avoid a circular import; subclasses consume it via attribute access (snap.rsi, snap.atr_ratio, etc.). V15-BUG-3 fix: no more dict-with-fallbacks. bias_data: BiasData from BiasResolver (Source A — algo + AI override). Subclasses MUST read allowed_direction / bias from this parameter, NOT from tech.* (Source B is algo- only and ignores AI override). Returns: EntryDecision if a setup is found, None otherwise. Implementations must: - Check Brain-specific entry rules (RSI extremes for MR, pullback patterns for TF, etc.) - Compute proposed entry/SL/TP prices - Assign confidence 0-100 - Return None if no valid setup (most calls return None) """ raise NotImplementedError # ============================================================ # EXIT MANAGEMENT (called every loop tick for each open trade) # ============================================================ @abstractmethod async def manage_exit(self, ctx: BrainContext) -> BrainDecision: """ Decide HOLD / EXIT / PARTIAL_50 / MOVE_SL for an open trade. Async because implementations may call AIClient.ask_for_decision(). Args: ctx: BrainContext with entry snapshot, runtime state, tech_now. Returns: BrainDecision with action and reason. Implementations must: - Default to HOLD when uncertain (conservative) - Use ctx.entry (immutable snapshot) for "vs entry" comparisons, NEVER fall back to ctx.tech_now for entry-side data - Respect grace periods (e.g. minutes_open < N -> HOLD) - Be deterministic where possible (logged decisions reproducible) - NOT mutate ctx.runtime — orchestrator owns runtime updates. Brain may signal updates via BrainDecision.metadata (e.g. {"evaluated_candle_time": }) for the orchestrator to apply before persisting. """ raise NotImplementedError # ============================================================ # COMMON HELPERS (subclasses can use directly) # ============================================================ def _hold(self, reason: str, metadata: Optional[dict] = None) -> BrainDecision: return BrainDecision( action=TradeAction.HOLD.value, reason=reason, metadata=metadata or {}, ) def _emit_manage_decision_log( self, ctx: BrainContext, decision: BrainDecision, ) -> None: """Observability: every manage_exit return is logged to brain_log.""" if getattr(self, "logger", None) is None: return meta = decision.metadata or {} trigger = meta.get("trigger", "unknown") try: self.logger.brain_log.write( "manage_exit_decision", symbol=ctx.entry.symbol, brain=self.name, action=decision.action, trigger=trigger, ai_called=trigger.startswith("ai_"), minutes_open=round(ctx.runtime.minutes_open, 1), net_profit_usd=round(ctx.runtime.net_profit_usd, 2), progress_pct=round(ctx.runtime.progress_pct, 2), current_price=float(getattr(ctx.tech_now, "price", 0.0) or 0.0), reason=(decision.reason or "")[:200], ) except Exception: pass def _exit(self, reason: str, metadata: Optional[dict] = None) -> BrainDecision: return BrainDecision( action=TradeAction.EXIT.value, reason=reason, metadata=metadata or {}, ) def _partial_50(self, reason: str) -> BrainDecision: return BrainDecision(action=TradeAction.PARTIAL_50.value, reason=reason) @staticmethod def _round_to_tick(price: float, tick_size: float, digits: int = 6) -> float: """ Align `price` to the instrument's tick grid. Returns `price` unchanged when `tick_size <= 0` (defensive — should not happen for live instruments). The `digits` arg trims float artifacts introduced by the multiply (V16 incident 29 apr: 6C 0.7324928... at 16 decimals on a 0.00005 tick grid silently rejected by broker). """ if tick_size <= 0: return price return round(round(price / tick_size) * tick_size, digits) def _move_sl( self, sl_price: float, reason: str, target: str = "breakeven", extra_metadata: Optional[dict] = None, ) -> BrainDecision: """ Emit a MOVE_SL decision. `sl_price` MUST already be tick-aligned by the caller (use `_round_to_tick(price, tech.tick_size)` at the site where `sl_price` is computed). Brain emits the same value the broker will receive — keeps logs and broker state in sync. Args: sl_price: new stop-loss price (entry price for breakeven, computed price for trailing). Tick-aligned. reason: human-readable rationale (logged). target: "breakeven" or "trailing" — encodes intent in metadata so callers (orchestrator, log analytics) can tell them apart without parsing the reason string. extra_metadata: any additional fields to merge into metadata (e.g. {"trailing_atr_mult": 1.5}). """ meta = {"sl_target": target} if extra_metadata: meta.update(extra_metadata) return BrainDecision( action=TradeAction.MOVE_SL.value, reason=reason, move_sl_to=sl_price, metadata=meta, ) # ============================================================ # CONVENIENCE: directional helpers used by both TF and MR # ============================================================ @staticmethod def is_long(direction: str) -> bool: return direction.upper() == "BUY" @staticmethod def is_short(direction: str) -> bool: return direction.upper() == "SELL" @staticmethod def rsi_moved_against( direction: str, rsi_at_entry: float, rsi_now: float, ) -> float: """ Returns positive value when RSI has moved AGAINST the trade. Negative or zero means RSI moved with the trade. For LONG: against = RSI dropped (rsi_at_entry - rsi_now positive) For SHORT: against = RSI rose (rsi_now - rsi_at_entry positive) """ if direction.upper() == "BUY": return rsi_at_entry - rsi_now return rsi_now - rsi_at_entry # ============================================================ # Sentinel decisions for orchestrator convenience # ============================================================ NO_ENTRY: Optional[EntryDecision] = None