""" APEX V16 — Trade opener. SINGLE POINT for opening a trade. Replaces V15's 4 duplicated paths (paper, live success, duplicate-recovered, orphan-recovered) — that duplication was V15-BUG-1, the root cause of subtle drift between paths (e.g., one path saved rsi_start, others saved different fields). Responsibilities: - Take an EntryDecision (Brain output) + current tech snapshot - Optionally place real broker order (or simulate for paper) - Build immutable TradeEntry + initial TradeRuntime - Return result to orchestrator NOT responsibilities: - Risk checks (caller already validated via risk_manager) - Sizing (caller already computed via sizing.compute_contracts) - State persistence (caller saves via state_store) - Logging the trade_opened event (caller logs after success) Failure modes: - Returns TradeOpenResult.success=False with structured error. - NEVER raises on broker errors (those are wrapped). - DOES raise on programmer errors (missing required fields). """ from __future__ import annotations import asyncio import logging from dataclasses import dataclass from typing import Optional from broker.broker_base import BrokerBase, OrderResult from core.contracts import ( BrainName, Direction, EntryDecision, TradeEntry, TradeRuntime, utc_now, ) log = logging.getLogger("trade_opener") # How long to wait before polling the broker for orphan positions after # a place_market_bracket exception/failure. V15 used 2s — gives the # broker eventual-consistency window to surface the position. POST_ORDER_SAFETY_DELAY_SECONDS: float = 2.0 # ============================================================ # RESULT TYPE # ============================================================ @dataclass(frozen=True) class TradeOpenResult: """ Result of open_trade(). Either trade is set (success), or error/error_kind populated (failure). """ success: bool entry: Optional[TradeEntry] = None runtime: Optional[TradeRuntime] = None error: str = "" error_kind: str = "" # "broker_failure" | "config_error" | "duplicate" | "unknown" # ============================================================ # OPENER # ============================================================ class TradeOpener: """ Builds TradeEntry + TradeRuntime from a Brain EntryDecision. Optionally places broker order (live mode) or skips it (paper mode). Single-instance: typically held by orchestrator, instantiated once with broker + logger references. """ def __init__(self, broker: BrokerBase, is_paper: bool, logger=None) -> None: """ Args: broker: BrokerBase instance (used only in live mode) is_paper: True = simulation only, no broker call logger: LoggerBundle (optional but recommended) """ self.broker = broker self.is_paper = is_paper self.logger = logger # ============================================================ # PUBLIC API # ============================================================ async def open_trade( self, symbol: str, brain_name: str, decision: EntryDecision, contracts: int, tech_now: dict, ) -> TradeOpenResult: """ Open a trade based on Brain decision. Args: symbol: e.g. "MES", "6B" brain_name: BrainName.TF.value or BrainName.MR.value decision: EntryDecision from Brain (direction, prices, confidence) contracts: from sizing.compute_contracts (already validated) tech_now: current tech indicators (used for entry snapshot) Returns: TradeOpenResult.success=True with entry+runtime on success, success=False with structured error on failure. """ # --- Validate inputs (programmer errors, raise) --- if contracts <= 0: raise ValueError(f"contracts must be > 0, got {contracts}") if decision.direction.upper() not in ("BUY", "SELL"): raise ValueError(f"invalid direction: {decision.direction!r}") if decision.entry_price <= 0 or decision.sl_price <= 0 or decision.tp_price <= 0: raise ValueError( f"prices must be > 0: entry={decision.entry_price}, " f"sl={decision.sl_price}, tp={decision.tp_price}" ) # --- Branch: paper vs live --- if self.is_paper: return self._open_paper(symbol, brain_name, decision, contracts, tech_now) else: return await self._open_live(symbol, brain_name, decision, contracts, tech_now) # ============================================================ # PAPER PATH # ============================================================ def _open_paper( self, symbol: str, brain_name: str, decision: EntryDecision, contracts: int, tech_now: dict, ) -> TradeOpenResult: """ Paper: no broker call, just build TradeEntry as if it filled immediately at decision.entry_price. """ entry = self._build_trade_entry( symbol=symbol, brain_name=brain_name, decision=decision, contracts=contracts, tech_now=tech_now, actual_entry_price=decision.entry_price, actual_sl_price=decision.sl_price, actual_tp_price=decision.tp_price, entry_order_id=None, stop_order_id=None, target_order_id=None, is_paper=True, is_orphan_recovered=False, ) runtime = TradeRuntime() return TradeOpenResult(success=True, entry=entry, runtime=runtime) # ============================================================ # LIVE PATH # ============================================================ async def _open_live( self, symbol: str, brain_name: str, decision: EntryDecision, contracts: int, tech_now: dict, ) -> TradeOpenResult: """ Live: place market bracket via broker, then build TradeEntry from the actual filled prices and order IDs. Pre-check: if a position already exists at broker for this symbol, return duplicate error (orchestrator decides what to do). """ # --- Pre-check: existing position at broker? --- try: existing = await self.broker.positions_get(symbol) except Exception as e: return TradeOpenResult( success=False, error=f"positions_get failed: {e}", error_kind="broker_failure", ) if existing: # V15 behavior: register the existing position as an "orphan # recovered" trade. V16 returns this case explicitly so the # orchestrator decides (may want to skip, recover, or warn). pos = existing[0] return TradeOpenResult( success=False, error=( f"position already exists at broker: " f"{pos.direction} {pos.contracts}ct @ {pos.avg_price}" ), error_kind="duplicate", ) # --- Place market bracket --- try: order_result: OrderResult = await self.broker.place_market_bracket( symbol=symbol, direction=decision.direction, contracts=contracts, sl_price=decision.sl_price, tp_price=decision.tp_price, ) except Exception as e: return TradeOpenResult( success=False, error=f"place_market_bracket exception: {e}", error_kind="broker_failure", ) if not order_result.success: return TradeOpenResult( success=False, error=order_result.error or "place_market_bracket failed (no detail)", error_kind="broker_failure", ) # --- Build TradeEntry from actual broker-side prices --- actual_entry = order_result.entry_price or decision.entry_price actual_sl = order_result.sl_price or decision.sl_price actual_tp = order_result.tp_price or decision.tp_price entry = self._build_trade_entry( symbol=symbol, brain_name=brain_name, decision=decision, contracts=contracts, tech_now=tech_now, actual_entry_price=actual_entry, actual_sl_price=actual_sl, actual_tp_price=actual_tp, entry_order_id=order_result.entry_id, stop_order_id=order_result.stop_id, target_order_id=order_result.target_id, is_paper=False, is_orphan_recovered=False, ) runtime = TradeRuntime() return TradeOpenResult(success=True, entry=entry, runtime=runtime) # ============================================================ # SHARED BUILDER (single source for TradeEntry construction) # ============================================================ @staticmethod def _build_trade_entry( *, symbol: str, brain_name: str, decision: EntryDecision, contracts: int, tech_now: dict, actual_entry_price: float, actual_sl_price: float, actual_tp_price: float, entry_order_id: Optional[str], stop_order_id: Optional[str], target_order_id: Optional[str], is_paper: bool, is_orphan_recovered: bool, orphan_entry_price_diff: float = 0.0, ) -> TradeEntry: """ SINGLE point of TradeEntry construction. All paths funnel here. Snapshot every relevant tech indicator from tech_now. This eliminates V15-BUG-5 (rsi_start saved by some paths but not propagated to Brain) — V16's TradeEntry carries every snapshot field that any Brain might need. """ return TradeEntry( # Identity symbol=symbol, brain_name=brain_name, direction=decision.direction.upper(), contracts=contracts, # Prices (broker-confirmed) entry_price=actual_entry_price, sl_price=actual_sl_price, tp_price=actual_tp_price, # Timing opened_at=utc_now(), # Snapshot at entry (the data V15 was missing for Brain) rsi_m5_at_entry=float(tech_now.get("rsi", 50.0)), rsi_h1_at_entry=float(tech_now.get("rsi_h1", 50.0)), rsi_h4_at_entry=float(tech_now.get("rsi_h4", 50.0)), atr_ratio_at_entry=float(tech_now.get("atr_ratio", 1.0)), market_structure_at_entry=str(tech_now.get("market_structure", "RANGING")), regime_at_entry=str(tech_now.get("regime", "RANGING")), h1_compat_at_entry=float(tech_now.get("h1_compatibility", 1.0)), confidence_at_entry=int(decision.confidence), # Order IDs entry_order_id=entry_order_id, stop_order_id=stop_order_id, target_order_id=target_order_id, # Flags is_paper=is_paper, is_orphan_recovered=is_orphan_recovered, orphan_entry_price_diff=orphan_entry_price_diff, ) # ============================================================ # POST-ORDER SAFETY CHECK (C2b — V16 improvement on V15 parity) # ============================================================ async def post_order_safety_check( self, symbol: str, brain_name: str, decision: EntryDecision, contracts: int, tech_now: dict, error_context: str = "", ) -> TradeOpenResult: """ Called after place_market_bracket has raised or returned failure. Determines whether the position actually opened broker-side despite the failure (SDK timeout / WS race / partial response) and brings it under V16 management if so. V15 parity: _post_order_safety_check (riga 1828-1900) — sleep, positions_get, register orphan with stop_id=None. V16 IMPROVEMENT: when a position is found, attempt SL/TP re-attach via place_stop_order + place_limit_order BEFORE registering. If both succeed, the orphan trade is registered with valid order IDs and is_orphan_recovered=False (not naked). If either re-attach fails, fall back to V15 behavior (is_orphan_recovered=True, stop_order_id=None) and emit a critical-level log so the operator flattens manually or the watchdog catches it. Returns: - success=True with entry+runtime if a position was found and registered (with or without bracket re-attach). - success=False, error_kind="not_found" if no orphan exists. - success=False, error_kind="broker_failure" if positions_get itself raised (degraded mode — caller can retry next tick). """ await asyncio.sleep(POST_ORDER_SAFETY_DELAY_SECONDS) try: positions = await self.broker.positions_get(symbol) except Exception as e: return TradeOpenResult( success=False, error=f"positions_get failed during safety check: {e}", error_kind="broker_failure", ) if not positions: self._log_safety("safety_check_no_position", symbol=symbol, error_context=error_context[:200]) return TradeOpenResult( success=False, error="no orphan position found", error_kind="not_found", ) pos = positions[0] actual_entry = float(pos.avg_price) intended = float(decision.entry_price) diff = actual_entry - intended protective_side = "SELL" if str(pos.direction).upper() == "BUY" else "BUY" sl_id, tp_id, reattach_error = await self._reattach_protection( symbol=symbol, protective_side=protective_side, contracts=int(pos.contracts), sl_price=decision.sl_price, tp_price=decision.tp_price, ) bracket_attached = (sl_id is not None) and (tp_id is not None) is_orphan = not bracket_attached entry = self._build_trade_entry( symbol=symbol, brain_name=brain_name, decision=decision, contracts=int(pos.contracts), tech_now=tech_now, actual_entry_price=actual_entry, actual_sl_price=decision.sl_price, actual_tp_price=decision.tp_price, entry_order_id=None, # SDK lost it stop_order_id=sl_id, # populated only if re-attach succeeded target_order_id=tp_id, is_paper=False, is_orphan_recovered=is_orphan, orphan_entry_price_diff=diff, ) runtime = TradeRuntime() # Initialize current_sl_price so MOVE_SL has a baseline even on orphan path runtime.current_sl_price = decision.sl_price if bracket_attached: self._log_safety( "orphan_recovered_with_bracket_reattached", level="warning", symbol=symbol, actual_entry_price=actual_entry, intended_entry_price=intended, entry_price_diff=diff, contracts=int(pos.contracts), stop_id=sl_id, target_id=tp_id, error_context=error_context[:200], ) else: self._log_safety( "orphan_recovered_naked", level="critical", symbol=symbol, actual_entry_price=actual_entry, intended_entry_price=intended, entry_price_diff=diff, contracts=int(pos.contracts), reattach_error=reattach_error, error_context=error_context[:200], ) return TradeOpenResult(success=True, entry=entry, runtime=runtime) async def _reattach_protection( self, symbol: str, protective_side: str, contracts: int, sl_price: float, tp_price: float, ) -> tuple[Optional[str], Optional[str], str]: """ Attempt to attach SL + TP to an orphan position. Returns (sl_id_or_None, tp_id_or_None, error_summary). If SL fails, TP is still attempted so a naked orphan keeps at least the TP if achievable — V15 behavior in place_market_bracket riga 1213-1216. """ sl_id: Optional[str] = None tp_id: Optional[str] = None errors: list[str] = [] try: sl_res = await self.broker.place_stop_order( symbol, protective_side, contracts, sl_price, ) if sl_res.success: sl_id = sl_res.stop_id or sl_res.entry_id else: errors.append(f"sl: {sl_res.error}") except Exception as e: errors.append(f"sl raised: {e}") try: tp_res = await self.broker.place_limit_order( symbol, protective_side, contracts, tp_price, ) if tp_res.success: tp_id = tp_res.target_id or tp_res.entry_id else: errors.append(f"tp: {tp_res.error}") except Exception as e: errors.append(f"tp raised: {e}") return sl_id, tp_id, "; ".join(errors) def _log_safety(self, event: str, *, level: str = "info", **fields) -> None: if self.logger is not None: try: self.logger.brain_log.write(event, **fields) except Exception: pass try: if level == "critical": self.logger.system.critical("[safety] %s %s", event, fields) elif level == "warning": self.logger.system.warning("[safety] %s %s", event, fields) else: self.logger.system.info("[safety] %s %s", event, fields) except Exception: pass else: getattr(log, level if level != "critical" else "critical")( "[safety] %s %s", event, fields, )