Pin market watch to held option strike while a position is open.

Stop falling back to ATM quotes for unrealized/close PnL after ATM drifts or restart.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/strategy/session.py

@@ -9,6 +9,7 @@ from typing import Any
 
 from ..config import Settings, get_settings
 from ..exchange import get_exchange, set_exchange, build_exchange
+from ..exchange.option_ids import pair_from_option_inst
 from ..exchange.protocol import ExchangeMarket
 from ..exchange.types import MarketSnapshot, OptionPair
 from .selection import (
@@ -36,6 +37,22 @@ def _has_open_position() -> bool:
         return False
 
 
+def _held_option_inst_id() -> str | None:
+    """活跃持仓期权合约；无仓返回 None。"""
+    try:
+        from ..models.db import get_db
+
+        row = get_db().fetchone(
+            "SELECT status, option_inst_id FROM positions WHERE id=1"
+        )
+        if not row or row["status"] != "open":
+            return None
+        inst = str(row["option_inst_id"] or "").strip()
+        return inst or None
+    except Exception:
+        return None
+
+
 def _as_bool_setting(raw: str | None, default: bool) -> bool:
     if raw is None or raw == "":
         return default
@@ -107,23 +124,40 @@ class StrategySession:
     def pair(self) -> OptionPair | None:
         return self._pair
 
+    def _watch_ids(self, pair: OptionPair | None = None) -> list[str]:
+        """永续 + 监控对 + 持仓腿（有仓时绝不能 drop 持仓盘口）。"""
+        s = self.settings
+        p = pair if pair is not None else self._pair
+        ids: list[str] = [s.perp_inst_id]
+        if p is not None:
+            ids.extend([p.call_inst_id, p.put_inst_id])
+        held = _held_option_inst_id()
+        if held:
+            ids.append(held)
+        # 去重保序
+        out: list[str] = []
+        seen: set[str] = set()
+        for i in ids:
+            if i and i not in seen:
+                seen.add(i)
+                out.append(i)
+        return out
+
     async def start(self) -> None:
         if self._started:
             return
         self._started = True
         await self.ex.start()
         try:
-            await asyncio.to_thread(self.align_instruments)
+            # 有持仓时必须钉在持仓行权价，禁止重启后漂到新 ATM
+            if _has_open_position():
+                await asyncio.to_thread(self.align_to_held_position)
+            else:
+                await asyncio.to_thread(self.align_instruments)
         except Exception as e:
             # eapi 418/429 时允许先起会话，后续 refresh 再对齐
             logger.warning("initial ATM align failed (will retry): %s", e)
-        await self.ex.resubscribe(
-            [
-                self.settings.perp_inst_id,
-                self._pair.call_inst_id if self._pair else "",
-                self._pair.put_inst_id if self._pair else "",
-            ]
-        )
+        await self.ex.resubscribe(self._watch_ids())
         self._refresh_task = asyncio.create_task(self._refresh_loop(), name="strategy-align")
 
     async def stop(self) -> None:
@@ -143,8 +177,7 @@ class StrategySession:
         self.ex.set_pair(pair)
         if idx is not None:
             self.ex.set_index_px(idx)
-        ids = [s.perp_inst_id, pair.call_inst_id, pair.put_inst_id]
-        self.ex.warm_and_subscribe(ids)
+        self.ex.warm_and_subscribe(self._watch_ids(pair))
         logger.info(
             "aligned pair exchange=%s expiry=%s strike=%s mark=%.2f hours=%.1f",
             getattr(self.ex, "name", "?"),
@@ -155,7 +188,33 @@ class StrategySession:
         )
         return pair
 
+    def align_to_held_position(self) -> OptionPair | None:
+        """有活跃仓时：监控对锁定为持仓合约的到期/行权价。"""
+        held = _held_option_inst_id()
+        if not held:
+            return None
+        pair = pair_from_option_inst(held)
+        if pair is None:
+            logger.warning("cannot rebuild pair from held option %s", held)
+            return None
+        mark = self._mark_for_atm() or float(pair.strike)
+        idx = None
+        try:
+            idx = self.ex.fetch_index(self.settings.index_inst_id)
+        except Exception:
+            pass
+        logger.info(
+            "pin watch to held option %s strike=%.0f expiry=%s",
+            held,
+            pair.strike,
+            pair.expiry_ymd,
+        )
+        return self._apply_pair(pair, mark=float(mark), idx=idx)
+
     def align_instruments(self) -> OptionPair | None:
+        # 重启/刷新时若仍有仓，绝不切到新 ATM
+        if _has_open_position():
+            return self.align_to_held_position()
         s = self.settings
         idx = self.ex.fetch_index(s.index_inst_id)
         mark = self.ex.fetch_mark(s.perp_inst_id) or idx
@@ -271,13 +330,7 @@ class StrategySession:
                 or pair.put_inst_id != old.put_inst_id
             )
         ):
-            await self.ex.resubscribe(
-                [
-                    self.settings.perp_inst_id,
-                    pair.call_inst_id,
-                    pair.put_inst_id,
-                ]
-            )
+            await self.ex.resubscribe(self._watch_ids(pair))
         return pair
 
     async def pick_for_open_async(self) -> OpenPick | None:
@@ -288,13 +341,7 @@ class StrategySession:
             or pick.pair.call_inst_id != old.call_inst_id
             or pick.pair.put_inst_id != old.put_inst_id
         ):
-            await self.ex.resubscribe(
-                [
-                    self.settings.perp_inst_id,
-                    pick.pair.call_inst_id,
-                    pick.pair.put_inst_id,
-                ]
-            )
+            await self.ex.resubscribe(self._watch_ids(pick.pair))
         return pick
 
     def _mark_for_atm(self) -> float | None:
@@ -324,6 +371,13 @@ class StrategySession:
 
     async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None:
         if _has_open_position():
+            # 持仓期间：钉住持仓行权价（禁止漂到新 ATM）
+            held = _held_option_inst_id()
+            if held and (
+                self._pair is None
+                or held not in (self._pair.call_inst_id, self._pair.put_inst_id)
+            ):
+                return await asyncio.to_thread(self.align_to_held_position)
             return self._pair
         if force or self.atm_needs_realign():
             logger.info(