Upload Quasar_axrvi_ranker.py

Quasar_axrvi_ranker.py

@@ -260,7 +260,7 @@ class ShreveConfig:
 # SECTION 1 — SHARED CONSTANTS
 # ══════════════════════════════════════════════════════════════════════════════════════
 
-DERIV_API_KEY  = os.environ.get("DERIV_API_KEY", "1KJKxIJKR8LCyKB")
+DERIV_API_KEY  = os.environ.get("DERIV_API_KEY", "mXZY9NxhIqIJyrM")
 DERIV_WS_URL   = "wss://ws.binaryws.com/websockets/v3?app_id=1089"
 
 # Deriv API symbol → AXRVI internal symbol
@@ -385,9 +385,10 @@ class TradeDirection(Enum):
 
 
 class PositionState(Enum):
-    OPEN    = "open"
-    CLOSING = "closing"
-    CLOSED  = "closed"
+    PENDING = "pending"   # buy sent to broker; awaiting buy confirmation
+    OPEN    = "open"      # broker confirmed; contract live
+    CLOSING = "closing"   # sell sent to broker; awaiting terminal event
+    CLOSED  = "closed"    # broker reported terminal state (won/lost/sold/expired)
 
 
 @dataclass
@@ -403,39 +404,145 @@ class PriceTick:
 
 @dataclass
 class Trade:
+    """
+    Broker-backed trade record.  All authoritative state comes from Deriv events.
+
+    Lifecycle:
+        PENDING  — buy sent; no contract yet
+        OPEN     — broker confirmed; contract_id bound
+        CLOSING  — sell/early-exit sent; awaiting terminal event
+        CLOSED   — broker reported final outcome (won / lost / sold / expired)
+
+    Fields marked [BROKER] must NOT be set locally;
+    they are written only from incoming Deriv WebSocket messages.
+    """
+    # ── Identity ─────────────────────────────────────────────────────────────
     trade_id:       str
-    asset:          str
+    asset:          str           # internal AXRVI symbol (e.g. "V75")
     direction:      TradeDirection
-    entry_price:    float
-    entry_time:     float
     quantity:       float
-    state:          PositionState = PositionState.OPEN
+    entry_time:     float         # wall-clock time of buy send (for monitoring only)
+
+    # ── Broker primary key [BROKER] ───────────────────────────────────────────
+    contract_id:    Optional[str]   = None   # Deriv contract_id from buy response
+    transaction_id: Optional[str]   = None   # Deriv transaction_id
+
+    # ── Broker contract details [BROKER] ─────────────────────────────────────
+    shortcode:      Optional[str]   = None   # Deriv shortcode
+    broker_symbol:  Optional[str]   = None   # Deriv API symbol (e.g. "R_75")
+    status:         Optional[str]   = None   # open | won | lost | sold | expired
+
+    # ── Price fields [BROKER] ─────────────────────────────────────────────────
+    buy_price:      Optional[float] = None   # price paid for the contract
+    sell_price:     Optional[float] = None   # price received on close/sell
+    entry_tick:     Optional[float] = None   # spot price at contract open tick
+    current_spot:   Optional[float] = None   # latest spot (updated from poc stream)
+    profit:         Optional[float] = None   # broker-confirmed net P&L
+
+    # ── Internal tracking (non-authoritative) ────────────────────────────────
+    # entry_price is kept for optional-stopping log-return calc until broker
+    # provides the authoritative entry_tick.  Updated from entry_tick on confirm.
+    entry_price:    float           = 0.0
     exit_price:     Optional[float] = None
     exit_time:      Optional[float] = None
-    unrealized_pnl: float = 0.0
-    realized_pnl:   float = 0.0
+    state:          PositionState   = PositionState.PENDING
+
+    # These fields are retained as read-only mirrors of broker data and must
+    # NEVER be used as authoritative execution truth.
+    unrealized_pnl: float = 0.0    # live spot estimate only — NOT authoritative
+    realized_pnl:   float = 0.0    # set from broker profit field on close
     fees:           float = 0.0
-    contract_id:    Optional[str] = None   # Deriv contract ID — set on buy confirmation
+
+    # ── Broker state update (called from _on_deriv_message on buy confirm) ───
+
+    def confirm_open(
+        self,
+        contract_id:    str,
+        buy_price:      float,
+        entry_tick:     float,
+        transaction_id: Optional[str] = None,
+        shortcode:      Optional[str] = None,
+        broker_symbol:  Optional[str] = None,
+    ) -> None:
+        """Transition PENDING → OPEN from broker buy confirmation."""
+        self.contract_id    = contract_id
+        self.buy_price      = buy_price
+        self.entry_price    = entry_tick   # use broker tick as authoritative entry
+        self.entry_tick     = entry_tick
+        self.current_spot   = entry_tick
+        self.transaction_id = transaction_id
+        self.shortcode      = shortcode
+        self.broker_symbol  = broker_symbol
+        self.status         = "open"
+        self.state          = PositionState.OPEN
+
+    def update_from_poc(self, poc: dict) -> None:
+        """
+        Apply a proposal_open_contract (poc) stream update.
+        Handles both live tick updates and terminal states.
+        """
+        self.current_spot = float(poc.get("current_spot", self.current_spot or 0.0))
+        self.status       = poc.get("status", self.status)
+
+        # Terminal event: populate close fields from broker data
+        if poc.get("is_expired") or poc.get("is_sold") or self.status in ("won", "lost", "sold", "expired"):
+            sell_price       = poc.get("sell_price") or poc.get("bid_price")
+            if sell_price is not None:
+                self.sell_price  = float(sell_price)
+                self.exit_price  = self.sell_price
+            self.exit_time   = time.time()
+            raw_profit       = poc.get("profit")
+            if raw_profit is None:
+                raw_profit = poc.get("bid_price", 0.0)
+            self.profit      = float(raw_profit) if raw_profit is not None else 0.0
+            self.realized_pnl = self.profit   # mirror broker profit
+            self.state       = PositionState.CLOSED
+
+    def close(self, exit_price: float, exit_time: float, fees: float) -> None:
+        """
+        Legacy compatibility close — only called when a broker terminal event
+        has been received and the broker profit is authoritative.
+        Uses broker sell_price if already set (preferred), otherwise
+        falls back to the passed exit_price for reward calc continuity.
+        """
+        if self.sell_price is not None:
+            self.exit_price = self.sell_price
+        else:
+            self.exit_price = exit_price
+        self.exit_time = exit_time
+        self.fees      = fees
+        self.state     = PositionState.CLOSED
+        # Do NOT recompute realized_pnl from local arithmetic; use broker profit
+        # if already available.  If not (edge case), use local arithmetic as
+        # last-resort fallback so callers never get None.
+        if self.profit is not None:
+            self.realized_pnl = self.profit
+        else:
+            if self.direction == TradeDirection.LONG:
+                self.realized_pnl = (self.exit_price - self.entry_price) * self.quantity - fees
+            else:
+                self.realized_pnl = (self.entry_price - self.exit_price) * self.quantity - fees
 
     def compute_unrealized_pnl(self, current_price: float) -> float:
+        """
+        NON-AUTHORITATIVE spot estimate for optional-stopping G_t calc only.
+        NEVER used as execution truth.
+        """
+        ref = self.entry_tick if self.entry_tick else self.entry_price
+        if ref <= 0:
+            return 0.0
         if self.direction == TradeDirection.LONG:
-            return (current_price - self.entry_price) * self.quantity
-        return (self.entry_price - current_price) * self.quantity
+            return (current_price - ref) * self.quantity
+        return (ref - current_price) * self.quantity
 
     def compute_unrealized_return(self, current_price: float) -> float:
+        """Non-authoritative spot return — for G_t [S8] only."""
+        ref = self.entry_tick if self.entry_tick else self.entry_price
+        if ref <= 0:
+            return 0.0
         if self.direction == TradeDirection.LONG:
-            return (current_price - self.entry_price) / self.entry_price
-        return (self.entry_price - current_price) / self.entry_price
-
-    def close(self, exit_price: float, exit_time: float, fees: float) -> None:
-        self.exit_price = exit_price
-        self.exit_time  = exit_time
-        self.fees       = fees
-        self.state      = PositionState.CLOSED
-        if self.direction == TradeDirection.LONG:
-            self.realized_pnl = (exit_price - self.entry_price) * self.quantity - fees
-        else:
-            self.realized_pnl = (self.entry_price - exit_price) * self.quantity - fees
+            return (current_price - ref) / ref
+        return (ref - current_price) / ref
 
     @property
     def holding_duration(self) -> float:
@@ -2548,86 +2655,256 @@ class PriceStreamer:
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 class PositionManager:
-    """Thread-safe trade lifecycle manager with optional structured logger hooks (v3)."""
+    """
+    Broker-backed trade registry.
+
+    Lifecycle contract:
+        register_pending_buy()  — buy sent to broker; trade enters PENDING
+        confirm_buy()           — broker buy confirmation received; trade enters OPEN
+        mark_closing()          — sell sent to broker; trade enters CLOSING
+        close_trade_from_broker() — broker terminal event; trade enters CLOSED
+
+    NO local PnL computation is authoritative.
+    NO state transitions happen without a broker event.
+
+    The legacy close_trade() method is retained as a thin wrapper around
+    close_trade_from_broker() for backward compatibility with callers that
+    pass an exit_price (e.g. checkpoint restore).  It does NOT simulate fills.
+    """
 
     def __init__(self, ranker_logger: Optional[object] = None):
-        self._open_trades:   Dict[str, Trade] = {}
-        self._closed_trades: List[Trade]      = []
-        self._lock           = Lock()
+        self._open_trades:    Dict[str, Trade] = {}   # contract_id NOT yet known (PENDING)
+        self._closed_trades:  List[Trade]      = []
+        self._lock            = Lock()
         self.total_realized_pnl: float = 0.0
         self.total_fees:         float = 0.0
         self.trades_opened:      int   = 0
         self.trades_closed:      int   = 0
         self.ranker_logger = ranker_logger
 
-    def open_trade(
+    # ── Phase 1: send buy ─────────────────────────────────────────────────────
+
+    def register_pending_buy(
         self,
         trade_id:    str,
         asset:       str,
         direction:   TradeDirection,
-        entry_price: float,
         quantity:    float,
+        broker_symbol: Optional[str] = None,
     ) -> Trade:
+        """
+        Create a PENDING trade immediately after the buy message is sent.
+        No price, no PnL, no fill — contract_id is unknown until broker confirms.
+        """
+        trade = Trade(
+            trade_id      = trade_id,
+            asset         = asset,
+            direction     = direction,
+            quantity      = quantity,
+            entry_time    = time.time(),
+            broker_symbol = broker_symbol,
+            state         = PositionState.PENDING,
+        )
         with self._lock:
-            trade = Trade(
-                trade_id    = trade_id,
-                asset       = asset,
-                direction   = direction,
-                entry_price = entry_price,
-                entry_time  = time.time(),
-                quantity    = quantity,
-                state       = PositionState.OPEN,
-            )
             self._open_trades[trade_id] = trade
             self.trades_opened += 1
 
+        logger.info(
+            f"⏳ [{asset}] BUY SENT | trade_id={trade_id} | "
+            f"dir={direction.value.upper()} | awaiting broker confirmation"
+        )
+        return trade
+
+    # ── Phase 2: broker buy confirmation ──────────────────────────────────────
+
+    def confirm_buy(
+        self,
+        trade_id:       str,
+        contract_id:    str,
+        buy_price:      float,
+        entry_tick:     float,
+        transaction_id: Optional[str] = None,
+        shortcode:      Optional[str] = None,
+        broker_symbol:  Optional[str] = None,
+    ) -> Optional[Trade]:
+        """
+        Called from _on_deriv_message when a 'buy' response arrives.
+        Transitions trade PENDING → OPEN and binds all broker contract details.
+        """
+        with self._lock:
+            trade = self._open_trades.get(trade_id)
+            if trade is None:
+                logger.warning(
+                    f"[PositionManager.confirm_buy] trade_id={trade_id} not found"
+                )
+                return None
+            trade.confirm_open(
+                contract_id    = contract_id,
+                buy_price      = buy_price,
+                entry_tick     = entry_tick,
+                transaction_id = transaction_id,
+                shortcode      = shortcode,
+                broker_symbol  = broker_symbol or trade.broker_symbol,
+            )
+
         if self.ranker_logger:
             self.ranker_logger.trade_open(
-                trade_id=trade_id, asset=asset,
-                direction=direction.value, price=entry_price, qty=quantity,
+                trade_id  = trade_id,
+                asset     = trade.asset,
+                direction = trade.direction.value,
+                price     = entry_tick,
+                qty       = trade.quantity,
             )
 
         logger.info(
-            f"✅ [{asset}] TRADE OPENED | ID={trade_id} | "
-            f"Dir={direction.value.upper()} | Entry={entry_price:.4f}"
+            f"✅ [{trade.asset}] TRADE OPENED | trade_id={trade_id} | "
+            f"contract_id={contract_id} | entry_tick={entry_tick:.4f} | "
+            f"buy_price={buy_price:.4f}"
         )
         return trade
 
-    def close_trade(
+    # ── Phase 3: send sell ────────────────────────────────────────────────────
+
+    def mark_closing(self, trade_id: str) -> None:
+        """
+        Mark a trade as CLOSING after a sell request is sent.
+        Actual close happens in close_trade_from_broker() when broker confirms.
+        """
+        with self._lock:
+            trade = self._open_trades.get(trade_id)
+            if trade:
+                trade.state = PositionState.CLOSING
+        logger.info(f"[{trade_id}] ⏳ SELL SENT — awaiting broker terminal event")
+
+    # ── Phase 4: broker terminal event ────────────────────────────────────────
+
+    def close_trade_from_broker(
         self,
         trade_id:   str,
-        exit_price: float,
-        fees:       float = 0.0,
+        status:     str,
+        profit:     float,
+        sell_price: Optional[float] = None,
+        exit_tick:  Optional[float] = None,
     ) -> Optional[Trade]:
+        """
+        Called from _on_deriv_message when proposal_open_contract reports
+        a terminal state (is_expired, is_sold, status in won/lost/sold/expired).
+        Authoritative close — profit comes directly from broker.
+        """
         with self._lock:
             if trade_id not in self._open_trades:
-                logger.warning(f"⚠️  Trade {trade_id} not found")
                 return None
             trade = self._open_trades.pop(trade_id)
-            trade.close(exit_price, time.time(), fees)
+
+        exit_price = sell_price or exit_tick or (trade.current_spot or 0.0)
+        trade.profit       = profit
+        trade.realized_pnl = profit          # authoritative broker P&L
+        trade.sell_price   = sell_price
+        trade.exit_price   = exit_price
+        trade.exit_time    = time.time()
+        trade.status       = status
+        trade.state        = PositionState.CLOSED
+        trade.fees         = 0.0             # fees already reflected in broker profit
+
+        with self._lock:
             self._closed_trades.append(trade)
             self.trades_closed      += 1
             self.total_realized_pnl += trade.realized_pnl
-            self.total_fees         += fees
 
         if self.ranker_logger:
             return_pct = (
-                (trade.exit_price - trade.entry_price) / trade.entry_price
-                if trade.direction == TradeDirection.LONG
-                else (trade.entry_price - trade.exit_price) / trade.entry_price
+                (exit_price - trade.entry_price) / trade.entry_price
+                if trade.direction == TradeDirection.LONG and trade.entry_price > 0
+                else (trade.entry_price - exit_price) / trade.entry_price
+                if trade.entry_price > 0
+                else 0.0
             )
             self.ranker_logger.trade_close(
-                trade_id=trade_id, asset=trade.asset,
-                pnl=trade.realized_pnl, return_pct=return_pct,
+                trade_id  = trade_id,
+                asset     = trade.asset,
+                pnl       = trade.realized_pnl,
+                return_pct = return_pct,
             )
 
-        logger.info(f"🔴 [{trade.asset}] TRADE CLOSED | PnL={trade.realized_pnl:+.4f}")
+        logger.info(
+            f"🔴 [{trade.asset}] TRADE CLOSED | trade_id={trade_id} | "
+            f"status={status} | profit={profit:+.4f} | "
+            f"contract_id={trade.contract_id}"
+        )
+        return trade
+
+    # ── Legacy compatibility wrapper ───────────────────────────────────────────
+
+    def open_trade(
+        self,
+        trade_id:    str,
+        asset:       str,
+        direction:   TradeDirection,
+        entry_price: float,
+        quantity:    float,
+    ) -> Trade:
+        """
+        Backward-compatibility shim — delegates to register_pending_buy().
+        entry_price is stored as a non-authoritative hint; it will be
+        overwritten by the broker's entry_tick on confirm_buy().
+        """
+        trade = self.register_pending_buy(trade_id, asset, direction, quantity)
+        trade.entry_price = entry_price   # non-authoritative hint
         return trade
 
+    def close_trade(
+        self,
+        trade_id:   str,
+        exit_price: float,
+        fees:       float = 0.0,
+    ) -> Optional[Trade]:
+        """
+        Backward-compatibility shim — accepts an exit_price for callers that
+        previously supplied it.  Delegates to close_trade_from_broker() with
+        profit derived from broker data if available, otherwise uses the passed
+        exit_price only for the reward calculator (log-return path).
+        Does NOT simulate fills or invent authoritative P&L.
+        """
+        with self._lock:
+            trade = self._open_trades.get(trade_id)
+            if trade is None:
+                logger.warning(f"⚠️  Trade {trade_id} not found")
+                return None
+
+        # Use broker profit if already set; otherwise treat exit_price as
+        # best-available approximation (e.g. during optional-stopping exit
+        # before broker sends terminal event).
+        profit = (
+            trade.profit
+            if trade.profit is not None
+            else (
+                (exit_price - trade.entry_price) * trade.quantity - fees
+                if trade.direction == TradeDirection.LONG
+                else (trade.entry_price - exit_price) * trade.quantity - fees
+            )
+        )
+        return self.close_trade_from_broker(
+            trade_id   = trade_id,
+            status     = trade.status or "sold",
+            profit     = profit,
+            sell_price = exit_price,
+        )
+
+    # ── Queries ────────────────────────────────────────────────────────────────
+
     def get_open_trades(self) -> List[Trade]:
+        """Return all trades that are PENDING, OPEN, or CLOSING (i.e. not CLOSED)."""
         with self._lock:
             return list(self._open_trades.values())
 
+    def get_confirmed_open_trades(self) -> List[Trade]:
+        """Return only broker-confirmed OPEN or CLOSING trades (not PENDING)."""
+        with self._lock:
+            return [
+                t for t in self._open_trades.values()
+                if t.state in (PositionState.OPEN, PositionState.CLOSING)
+            ]
+
     def get_open_trade_by_asset(self, asset: str) -> Optional[Trade]:
         with self._lock:
             for t in self._open_trades.values():
@@ -2635,7 +2912,15 @@ class PositionManager:
                     return t
         return None
 
+    def get_open_trade_by_contract(self, contract_id: str) -> Optional[Trade]:
+        with self._lock:
+            for t in self._open_trades.values():
+                if t.contract_id == str(contract_id):
+                    return t
+        return None
+
     def get_unrealized_pnl(self, price_map: Dict[str, float]) -> float:
+        """Non-authoritative spot estimate only."""
         with self._lock:
             return sum(
                 t.compute_unrealized_pnl(price_map.get(t.asset, t.entry_price))
@@ -2987,6 +3272,35 @@ class DerivWebSocketClient:
             logger.error(f"❌ Subscription error for {symbol}: {e}")
             return False
 
+    async def subscribe_to_poc(self, contract_id: str) -> bool:
+        """
+        Subscribe to proposal_open_contract stream for a live contract.
+        Delivers real-time status updates and terminal events (won/lost/sold/expired).
+        Must be called after buy confirmation so we receive all lifecycle events.
+        """
+        try:
+            await self.ws.send(json.dumps({
+                "proposal_open_contract": 1,
+                "contract_id":            int(contract_id),
+                "subscribe":              1,
+                "req_id":                 self._next_msg_id(),
+            }))
+            logger.info(f"🔔 Subscribed to poc stream | contract_id={contract_id}")
+            return True
+        except Exception as e:
+            logger.error(f"❌ poc subscription error | contract_id={contract_id}: {e}")
+            return False
+
+    async def forget_contract(self, subscription_id: str) -> None:
+        """Unsubscribe from a poc stream after contract closes to avoid leaking subscriptions."""
+        try:
+            await self.ws.send(json.dumps({
+                "forget": subscription_id,
+                "req_id": self._next_msg_id(),
+            }))
+        except Exception:
+            pass   # best-effort cleanup
+
     async def send_message(self, msg: dict) -> bool:
         try:
             msg["req_id"] = self._next_msg_id()
@@ -3365,36 +3679,267 @@ class QuasarAXRVIBridge:
     # ── Deriv message handling ─────────────────────────────────────────────────────────
 
     def _on_deriv_message(self, msg: dict) -> None:
-        """Synchronous handler for Deriv WebSocket messages."""
+        """
+        Synchronous handler for Deriv WebSocket messages.
+
+        Handles:
+          tick                    — price update
+          buy                     — contract open confirmation (PENDING → OPEN)
+          proposal_open_contract  — live contract update; terminal event → CLOSED
+          balance                 — account balance sync to portfolio risk manager
+          error                   — broker rejection log
+        """
         try:
             if "tick" in msg:
                 self._on_price_tick(msg["tick"])
+
             elif "buy" in msg:
-                contract_id = msg["buy"].get("contract_id")
-                req_id      = msg.get("req_id")
-                trade_id    = self._pending_req_to_trade.pop(req_id, None)
-                if trade_id and contract_id:
-                    # Bind the Deriv contract_id to the internal Trade object so
-                    # _close_position can send a sell for early exit
+                self._on_buy_confirmation(msg)
+
+            elif "proposal_open_contract" in msg:
+                self._on_poc_update(msg["proposal_open_contract"], msg)
+
+            elif "balance" in msg:
+                bal = msg["balance"]
+                new_equity = float(bal.get("balance", 0.0))
+                self.portfolio_risk_mgr.update_equity(new_equity)
+                logger.debug(
+                    f"[Deriv] Balance update | equity={new_equity:.2f}"
+                )
+
+            elif "error" in msg:
+                err     = msg["error"]
+                code    = err.get("code", "UNKNOWN")
+                message = err.get("message", str(err))
+                req_id  = msg.get("req_id")
+                # Clean up any pending trade associated with this req_id
+                trade_id = self._pending_req_to_trade.pop(req_id, None)
+                if trade_id:
+                    # Remove the pending trade stub — broker rejected the buy
                     with self.position_mgr._lock:
-                        trade = self.position_mgr._open_trades.get(trade_id)
-                        if trade:
-                            trade.contract_id = str(contract_id)
-                    logger.info(
-                        f"✅ Trade confirmed | trade_id={trade_id} | "
-                        f"contract_id={contract_id}"
-                    )
+                        rejected = self.position_mgr._open_trades.pop(trade_id, None)
+                    if rejected:
+                        logger.error(
+                            f"❌ [{rejected.asset}] BROKER REJECTED buy | "
+                            f"trade_id={trade_id} | code={code} | {message}"
+                        )
+                        # Legacy-compatible execution failure log
+                        logger.error(
+                            f"[{rejected.asset}] EXECUTION FAILED | "
+                            f"reason=broker_rejected | code={code} | {message}"
+                        )
+                    else:
+                        logger.error(
+                            f"❌ Deriv error (req_id={req_id}): [{code}] {message}"
+                        )
                 else:
-                    logger.info(
-                        f"✅ Buy confirmation received | contract_id={contract_id} | "
-                        f"req_id={req_id} (trade_id lookup: {trade_id})"
+                    logger.error(
+                        f"❌ Deriv error (req_id={req_id}): [{code}] {message}"
                     )
-            elif "error" in msg:
-                logger.error(f"⚠️  Deriv error: {msg['error']}")
                 if self.ranker_logger:
                     self.ranker_logger.connection_event("Deriv", "error", str(msg["error"]))
+
         except Exception as e:
             logger.error(f"❌ Deriv message handler error: {e}")
+            traceback.print_exc()
+
+    def _on_buy_confirmation(self, msg: dict) -> None:
+        """
+        Handle broker buy confirmation:  PENDING → OPEN.
+        Bind contract_id, entry_tick, buy_price to the Trade object.
+        Then subscribe to the proposal_open_contract stream for live updates.
+        """
+        buy_data    = msg["buy"]
+        req_id      = msg.get("req_id")
+        trade_id    = self._pending_req_to_trade.pop(req_id, None)
+        contract_id = str(buy_data.get("contract_id", ""))
+
+        if not contract_id:
+            logger.error(
+                f"❌ Buy confirmation missing contract_id | req_id={req_id}"
+            )
+            return
+
+        # Map contract_id → trade_id for poc routing
+        if trade_id:
+            self._contract_to_trade[contract_id] = trade_id
+
+        buy_price  = float(buy_data.get("buy_price",   0.0))
+        start_tick = float(buy_data.get("start_time",  time.time()))
+        # Deriv may provide longcode/shortcode/transaction_id directly in buy
+        shortcode  = buy_data.get("shortcode")
+        tx_id      = str(buy_data.get("transaction_id", ""))
+        # Use spot price at confirmation as entry_tick (broker-authoritative)
+        entry_tick = float(
+            buy_data.get("spot", 0.0)
+            or buy_data.get("entry_spot", 0.0)
+            or 0.0
+        )
+        # If Deriv didn't include spot in buy, use current streamer price
+        if entry_tick == 0.0 and trade_id:
+            pending = self.position_mgr._open_trades.get(trade_id)
+            if pending:
+                s = self.price_streamers.get(pending.asset)
+                entry_tick = s.latest_mid if s else 0.0
+        broker_symbol = buy_data.get("symbol", "")
+
+        if trade_id:
+            trade = self.position_mgr.confirm_buy(
+                trade_id       = trade_id,
+                contract_id    = contract_id,
+                buy_price      = buy_price,
+                entry_tick     = entry_tick,
+                transaction_id = tx_id,
+                shortcode      = shortcode,
+                broker_symbol  = broker_symbol,
+            )
+            if trade:
+                # Legacy-compatible log: "TRADE OPENED"
+                logger.info(
+                    f"[{trade.asset}] TRADE OPENED | trade_id={trade_id} | "
+                    f"contract_id={contract_id} | buy_price={buy_price:.4f}"
+                )
+                # Subscribe to poc stream for live status + terminal event
+                if self.ws_client and self.ws_client.connected:
+                    asyncio.get_event_loop().create_task(
+                        self.ws_client.subscribe_to_poc(contract_id)
+                    )
+        else:
+            logger.warning(
+                f"[Deriv] Buy confirmation — no trade_id for req_id={req_id} | "
+                f"contract_id={contract_id} (late or orphaned confirmation)"
+            )
+
+    def _on_poc_update(self, poc: dict, raw_msg: dict) -> None:
+        """
+        Handle a proposal_open_contract update.
+
+        Live tick: update current_spot on the Trade object.
+        Terminal:  close the trade with broker-authoritative profit data.
+        """
+        contract_id = str(poc.get("contract_id", ""))
+        trade_id    = self._contract_to_trade.get(contract_id)
+
+        if not trade_id:
+            # Not a trade we're tracking — ignore
+            return
+
+        with self.position_mgr._lock:
+            trade = self.position_mgr._open_trades.get(trade_id)
+
+        if trade is None:
+            # Already closed (race condition or duplicate terminal) — ignore
+            return
+
+        # ── Update current_spot on live ticks ────────────────────────────────
+        current_spot = poc.get("current_spot")
+        if current_spot:
+            trade.current_spot = float(current_spot)
+            # Update streamer-level price context for feature engine
+            streamer = self.price_streamers.get(trade.asset)
+            if streamer:
+                trade.unrealized_pnl = trade.compute_unrealized_pnl(float(current_spot))
+
+        # ── Terminal state check ──────────────────────────────────────────────
+        is_terminal = (
+            poc.get("is_expired", False)
+            or poc.get("is_sold",    False)
+            or poc.get("status")    in ("won", "lost", "sold", "expired")
+        )
+
+        if not is_terminal:
+            return   # live update only — nothing to close yet
+
+        # ── Extract broker-authoritative close data ───────────────────────────
+        status     = poc.get("status", "expired")
+        raw_profit = poc.get("profit")
+        if raw_profit is None:
+            raw_profit = poc.get("bid_price", 0.0)
+        profit     = float(raw_profit) if raw_profit is not None else 0.0
+        sell_price = float(poc.get("sell_price") or poc.get("bid_price") or 0.0)
+        exit_tick  = float(poc.get("exit_tick") or poc.get("current_spot") or 0.0)
+
+        logger.info(
+            f"[{trade.asset}] CONTRACT TERMINAL | contract_id={contract_id} | "
+            f"status={status} | profit={profit:+.4f} | sell_price={sell_price:.4f}"
+        )
+
+        closed_trade = self.position_mgr.close_trade_from_broker(
+            trade_id   = trade_id,
+            status     = status,
+            profit     = profit,
+            sell_price = sell_price if sell_price > 0 else None,
+            exit_tick  = exit_tick  if exit_tick  > 0 else None,
+        )
+
+        if closed_trade is None:
+            return
+
+        # ── Replay, bandit, and portfolio accounting ──────────────────────────
+        fees   = 0.0   # already reflected in broker profit
+        reward = self._reward_from_broker(closed_trade)
+
+        self.portfolio_risk_mgr.register_close(trade_id, closed_trade.realized_pnl)
+        self._close_pending_episode(trade_id, reward)
+        self._trade_tick_counts.pop(trade_id, None)
+
+        self.stats["trades_closed"] += 1
+        self.stats["total_pnl"]     += closed_trade.realized_pnl
+
+        # Legacy-compatible log: "TRADE CLOSED"
+        logger.info(
+            f"💰 [{closed_trade.asset}] TRADE CLOSED | "
+            f"reward={reward:+.6f} | profit={profit:+.4f} | "
+            f"portfolio_dd={self.portfolio_risk_mgr._current_drawdown():.2%}"
+        )
+
+        # Clean up contract mapping
+        self._contract_to_trade.pop(contract_id, None)
+
+        # Subscription cleanup (best-effort)
+        sub_id = poc.get("id") or poc.get("subscription", {}).get("id")
+        if sub_id and self.ws_client and self.ws_client.connected:
+            asyncio.get_event_loop().create_task(
+                self.ws_client.forget_contract(sub_id)
+            )
+
+        # REFILL TRIGGER — if open count drops below floor after terminal event
+        open_count = len(self.position_mgr.get_open_trades())
+        if open_count < 2 and self.running:
+            logger.warning(
+                f"[poc_terminal] ⚠️  REFILL TRIGGER — "
+                f"open_count={open_count} < 2 after contract terminal. "
+                f"Scheduling immediate rank_and_gate() to restore minimum."
+            )
+            asyncio.get_event_loop().create_task(self._safe_rank_and_gate())
+
+    async def _safe_rank_and_gate(self) -> None:
+        """Wrapper that silently catches errors from refill-triggered rank_and_gate()."""
+        try:
+            await self.rank_and_gate()
+        except Exception as e:
+            logger.error(f"[refill] rank_and_gate error: {e}")
+
+    def _reward_from_broker(self, trade: "Trade") -> float:
+        """
+        Compute replay reward from broker-authoritative profit.
+        Uses log-return form [S4] if entry_tick and exit data are available;
+        falls back to raw profit / stake ratio otherwise.
+        """
+        stake = self.trade_config.amount
+        entry = trade.entry_tick if trade.entry_tick and trade.entry_tick > 0 else trade.entry_price
+        exit_ = trade.sell_price or trade.exit_price or 0.0
+        fees  = self.trade_config.commission_rate
+        slip  = self.trade_config.slippage_bps / 10_000.0
+
+        if entry > 0 and exit_ > 0:
+            log_ret = math.log(exit_ / entry)
+            if trade.direction == TradeDirection.SHORT:
+                log_ret = -log_ret
+            return float(log_ret - fees - slip)
+
+        # Fallback: normalise broker profit by stake
+        return float(trade.profit / stake) if stake > 0 and trade.profit is not None else 0.0
 
     def _on_price_tick(self, tick_data: dict) -> None:
         try: