Revise strategy: TTM+ATM+leverage option pick, % exit, perp leverage/margin.

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

backend/app/market/gateway.py

@@ -4,18 +4,24 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from dataclasses import dataclass
 from typing import Any
 
 from ..config import Settings, get_settings
 from .book_cache import BookCache
-from .instruments import next_session_expiry_ymd, select_option_pair
+from .instruments import (
+    hours_until_expiry,
+    list_eligible_expiry_ymds,
+    option_leverage,
+    select_option_pair,
+)
 from .okx_rest import OkxRestClient
 from .okx_ws import OkxPublicWs
 from .types import MarketSnapshot, OptionPair
 
 logger = logging.getLogger(__name__)
 
-# 现价偏离当前行权价超过该点数则重选 ATM（ETH 期权常见步进 5）
+# 展示用：现价偏离当前行权超过该点数则重选 ATM（空仓）
 _ATM_DRIFT_POINTS = 5.0
 
 
@@ -29,6 +35,40 @@ def _has_open_position() -> bool:
         return False
 
 
+def _strategy_floats() -> tuple[float, float]:
+    """(min_option_hours, min_option_leverage)"""
+    s = get_settings()
+    try:
+        from ..models.db import get_db
+
+        db = get_db()
+        hours = float(
+            db.get_setting("min_option_hours", str(s.min_option_hours))
+            or s.min_option_hours
+        )
+        lev = float(
+            db.get_setting("min_option_leverage", str(s.min_option_leverage))
+            or s.min_option_leverage
+        )
+        return hours, lev
+    except Exception:
+        return s.min_option_hours, s.min_option_leverage
+
+
+@dataclass(slots=True)
+class OpenPick:
+    pair: OptionPair
+    option_side: str
+    perp_side: str
+    bias: str
+    call_ask: float
+    put_ask: float
+    option_ask: float
+    option_leverage: float
+    hours_left: float
+    underlying_px: float
+
+
 class MarketGateway:
     def __init__(self, settings: Settings | None = None) -> None:
         self.settings = settings or get_settings()
@@ -64,25 +104,13 @@ class MarketGateway:
         await self.ws.stop()
         self.rest.close()
 
-    def align_instruments(self) -> OptionPair | None:
-        """同步：拉期权列表，选次日到期 ATM Call/Put，REST 预热盘口，切换 WS 订阅。"""
+    def _apply_pair(self, pair: OptionPair, *, mark: float, idx: float | None) -> OptionPair:
         s = self.settings
-        idx = self.rest.fetch_index_ticker(s.index_inst_id)
-        mark = self.rest.fetch_mark_price(s.perp_inst_id) or idx
-        if mark is None or mark <= 0:
-            raise RuntimeError("无法获取 ETH 标记/指数价格，无法选 ATM")
-
-        instruments = self.rest.fetch_option_instruments(s.option_inst_family)
-        ymd = next_session_expiry_ymd()
-        pair = select_option_pair(instruments, mark_px=float(mark), expiry_ymd=ymd)
-        if pair is None:
-            raise RuntimeError(f"未找到到期 {ymd} 的 ATM Call/Put 合约 pair (family={s.option_inst_family})")
-
         self._pair = pair
         self.cache.set_pair(pair)
-        self.cache.set_index_px(idx)
+        if idx is not None:
+            self.cache.set_index_px(idx)
 
-        # REST 预热：永续 + Call + Put
         for inst in (s.perp_inst_id, pair.call_inst_id, pair.put_inst_id):
             bids, asks, ts = self.rest.fetch_books(inst, sz=5)
             self.cache.upsert_book(inst, bids=bids, asks=asks, ts_ms=ts)
@@ -94,15 +122,102 @@ class MarketGateway:
         self.cache.drop_except(keep)
         self.ws.set_instruments([s.perp_inst_id, pair.call_inst_id, pair.put_inst_id])
         logger.info(
-            "aligned pair expiry=%s strike=%s call=%s put=%s mark=%.2f",
+            "aligned pair expiry=%s strike=%s call=%s put=%s mark=%.2f hours=%.1f",
             pair.expiry_ymd,
             pair.strike,
             pair.call_inst_id,
             pair.put_inst_id,
             mark,
+            hours_until_expiry(pair.expiry_ymd),
         )
         return pair
 
+    def align_instruments(self) -> OptionPair | None:
+        """空仓展示：选剩余时长合格的最近到期 ATM（不校验期权杠杆）。"""
+        s = self.settings
+        idx = self.rest.fetch_index_ticker(s.index_inst_id)
+        mark = self.rest.fetch_mark_price(s.perp_inst_id) or idx
+        if mark is None or mark <= 0:
+            raise RuntimeError("无法获取 ETH 标记/指数价格，无法选 ATM")
+
+        min_hours, _ = _strategy_floats()
+        instruments = self.rest.fetch_option_instruments(s.option_inst_family)
+        pair = select_option_pair(
+            instruments, mark_px=float(mark), min_hours=min_hours
+        )
+        if pair is None:
+            raise RuntimeError(
+                f"未找到剩余≥{min_hours}h 的 ATM Call/Put (family={s.option_inst_family})"
+            )
+        return self._apply_pair(pair, mark=float(mark), idx=idx)
+
+    def pick_for_open(self) -> OpenPick | None:
+        """
+        开仓选约：
+        1) 剩余时长 ≥ min_hours 的到期日（由近到远）
+        2) 该到期 ATM 平值
+        3) 卖一比价定方向后校验 现价/卖一 ≥ min_option_leverage
+        """
+        s = self.settings
+        min_hours, min_lev = _strategy_floats()
+        idx = self.rest.fetch_index_ticker(s.index_inst_id)
+        mark = self.rest.fetch_mark_price(s.perp_inst_id) or idx
+        if mark is None or mark <= 0:
+            return None
+        underlying = float(mark)
+        instruments = self.rest.fetch_option_instruments(s.option_inst_family)
+        eligible = list_eligible_expiry_ymds(instruments, min_hours=min_hours)
+        if not eligible:
+            logger.info("no expiry with hours>=%.1f", min_hours)
+            return None
+
+        from ..strategy.signal import decide
+
+        for ymd in eligible:
+            pair = select_option_pair(
+                instruments, mark_px=underlying, expiry_ymd=ymd
+            )
+            if pair is None:
+                continue
+            call_bids, call_asks, _ = self.rest.fetch_books(pair.call_inst_id, sz=5)
+            put_bids, put_asks, _ = self.rest.fetch_books(pair.put_inst_id, sz=5)
+            call_ask = call_asks[0].px if call_asks else None
+            put_ask = put_asks[0].px if put_asks else None
+            sig = decide(call_ask, put_ask)
+            if sig is None:
+                continue
+            opt_ask = sig.call_ask if sig.option_side == "call" else sig.put_ask
+            lev = option_leverage(underlying, opt_ask)
+            hours_left = hours_until_expiry(ymd)
+            if lev is None or lev + 1e-9 < min_lev:
+                logger.info(
+                    "skip expiry=%s strike=%.0f side=%s lev=%s need>=%.0f hours=%.1f",
+                    ymd,
+                    pair.strike,
+                    sig.option_side,
+                    f"{lev:.1f}" if lev else "n/a",
+                    min_lev,
+                    hours_left,
+                )
+                continue
+            self._apply_pair(pair, mark=underlying, idx=idx)
+            # 写入刚拉的盘口，避免 WS 尚未推送
+            self.cache.upsert_book(pair.call_inst_id, bids=call_bids, asks=call_asks)
+            self.cache.upsert_book(pair.put_inst_id, bids=put_bids, asks=put_asks)
+            return OpenPick(
+                pair=pair,
+                option_side=sig.option_side,
+                perp_side=sig.perp_side,
+                bias=sig.bias,
+                call_ask=float(sig.call_ask),
+                put_ask=float(sig.put_ask),
+                option_ask=float(opt_ask),
+                option_leverage=float(lev),
+                hours_left=hours_left,
+                underlying_px=underlying,
+            )
+        return None
+
     async def realign_async(self) -> OptionPair | None:
         old = self._pair
         pair = await asyncio.to_thread(self.align_instruments)
@@ -122,6 +237,23 @@ class MarketGateway:
             )
         return pair
 
+    async def pick_for_open_async(self) -> OpenPick | None:
+        old = self._pair
+        pick = await asyncio.to_thread(self.pick_for_open)
+        if pick and (
+            old is None
+            or pick.pair.call_inst_id != old.call_inst_id
+            or pick.pair.put_inst_id != old.put_inst_id
+        ):
+            await self.ws.resubscribe(
+                [
+                    self.settings.perp_inst_id,
+                    pick.pair.call_inst_id,
+                    pick.pair.put_inst_id,
+                ]
+            )
+        return pick
+
     def _mark_for_atm(self) -> float | None:
         snap = self.snapshot()
         if snap.perp and snap.perp.mark_px:
@@ -133,11 +265,10 @@ class MarketGateway:
         return None
 
     def atm_needs_realign(self, mark_px: float | None = None) -> bool:
-        """到期日变了，或现价已偏离当前行权价超过阈值。"""
         if self._pair is None:
             return True
-        want = next_session_expiry_ymd()
-        if self._pair.expiry_ymd != want:
+        min_hours, _ = _strategy_floats()
+        if hours_until_expiry(self._pair.expiry_ymd) + 1e-9 < min_hours:
             return True
         mark = mark_px if mark_px is not None else self._mark_for_atm()
         if mark is None or mark <= 0:
@@ -145,14 +276,15 @@ class MarketGateway:
         return abs(float(self._pair.strike) - float(mark)) >= _ATM_DRIFT_POINTS
 
     async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None:
-        """空仓时按现价对齐 ATM。有持仓时不切换，避免盯市合约被换掉。"""
+        """空仓时按剩余时长+ATM 对齐。有持仓不切换。"""
         if _has_open_position():
             return self._pair
         if force or self.atm_needs_realign():
             logger.info(
-                "ATM realign force=%s old_strike=%s",
+                "ATM realign force=%s old_strike=%s old_exp=%s",
                 force,
                 self._pair.strike if self._pair else None,
+                self._pair.expiry_ymd if self._pair else None,
             )
             return await self.realign_async()
         return self._pair
@@ -164,7 +296,6 @@ class MarketGateway:
         return self.snapshot().to_dict()
 
     async def _refresh_loop(self) -> None:
-        """周期性刷新指数价；空仓时按到期/ATM 偏离重对齐。"""
         while True:
             await asyncio.sleep(30)
             try:
@@ -184,7 +315,6 @@ class MarketGateway:
                 logger.warning("market refresh failed: %s", e)
 
 
-# 进程级单例（FastAPI lifespan 注入）
 _gateway: MarketGateway | None = None
 
 

backend/app/market/instruments.py

@@ -1,4 +1,4 @@
-"""合约选择：次日 16:00（上海）到期 + ATM 行权价（暂定默认，待拍板可改）。"""
+"""合约选择：剩余时长过滤 + ATM 平值期权。"""
 
 from __future__ import annotations
 
@@ -42,13 +42,15 @@ def expiry_ms_from_ymd(ymd: str) -> int:
     return int(dt.timestamp() * 1000)
 
 
+def hours_until_expiry(ymd: str, now: datetime | None = None) -> float:
+    """距到期剩余小时（可为负）。"""
+    n = (now or datetime.now(tz=_SH)).astimezone(_SH)
+    left_ms = expiry_ms_from_ymd(ymd) - int(n.timestamp() * 1000)
+    return left_ms / 3_600_000.0
+
+
 def next_session_expiry_ymd(now: datetime | None = None) -> str:
-    """
-    业务约定：开仓选「次日 16:00」到期。
-    - 上海时间 >= 当日 16:00：目标到期日 = 次日
-    - 上海时间 < 当日 16:00：目标到期日 = 当日（当日 16:00 到期仍可用作盘口对齐/预热）
-    正式开仓窗从当日 16:00 起，届时「次日」即日历次日。
-    """
+    """兼容旧逻辑：次日/当日 16:00 到期键（展示/测试用）。"""
     now_sh = (now or datetime.now(tz=_SH)).astimezone(_SH)
     open_today = now_sh.replace(hour=16, minute=0, second=0, microsecond=0)
     if now_sh >= open_today:
@@ -64,30 +66,19 @@ def pick_atm_strike(strikes: list[float], mark_px: float) -> float | None:
     return min(strikes, key=lambda s: (abs(s - mark_px), s))
 
 
-def select_option_pair(
+def _complete_by_expiry(
     instruments: list[dict[str, Any]],
-    *,
-    mark_px: float,
-    expiry_ymd: str | None = None,
-    now: datetime | None = None,
-) -> OptionPair | None:
-    """
-    从 live 合约列表中选出：目标到期日 + ATM 同行权价 Call/Put。
-    行权价规则暂定 ATM（最接近标记/指数价）；待拍板后可替换。
-    """
-    ymd = expiry_ymd or next_session_expiry_ymd(now)
-    by_strike: dict[float, dict[str, str]] = {}
-
+) -> dict[str, dict[float, dict[str, str]]]:
+    """expiry_ymd -> strike -> {C|P: instId}，仅完整 Call+Put。"""
+    by_exp: dict[str, dict[float, dict[str, str]]] = {}
     for row in instruments:
         if not isinstance(row, dict):
             continue
         state = str(row.get("state") or "live").lower()
         if state and state != "live":
             continue
-
         inst_id = str(row.get("instId") or "")
         y, stk, opt = parse_option_inst_id(inst_id)
-
         if y is None or stk is None or opt is None:
             exp = safe_float(row.get("expTime"))
             if exp:
@@ -96,20 +87,77 @@ def select_option_pair(
             stk = safe_float(row.get("stk"))
             opt_raw = str(row.get("optType") or "").upper()
             opt = opt_raw if opt_raw in ("C", "P") else None
-
-        if not inst_id or y != ymd or stk is None or opt not in ("C", "P"):
+        if not inst_id or not y or stk is None or opt not in ("C", "P"):
             continue
-        by_strike.setdefault(float(stk), {})[opt] = inst_id
+        by_exp.setdefault(y, {}).setdefault(float(stk), {})[opt] = inst_id
 
-    complete = {s: v for s, v in by_strike.items() if "C" in v and "P" in v}
+    out: dict[str, dict[float, dict[str, str]]] = {}
+    for ymd, strikes in by_exp.items():
+        complete = {s: v for s, v in strikes.items() if "C" in v and "P" in v}
+        if complete:
+            out[ymd] = complete
+    return out
+
+
+def list_eligible_expiry_ymds(
+    instruments: list[dict[str, Any]],
+    *,
+    min_hours: float,
+    now: datetime | None = None,
+) -> list[str]:
+    """剩余时间 >= min_hours 的到期日，由近到远。"""
+    complete = _complete_by_expiry(instruments)
+    eligible = [
+        ymd
+        for ymd in complete
+        if hours_until_expiry(ymd, now) + 1e-9 >= float(min_hours)
+    ]
+    return sorted(eligible, key=lambda y: expiry_ms_from_ymd(y))
+
+
+def select_option_pair(
+    instruments: list[dict[str, Any]],
+    *,
+    mark_px: float,
+    expiry_ymd: str | None = None,
+    min_hours: float | None = None,
+    now: datetime | None = None,
+) -> OptionPair | None:
+    """
+    选 ATM Call/Put。
+    - 若给 expiry_ymd：在该到期日选平值。
+    - 若给 min_hours：选「剩余时长合格」中最近到期日的平值。
+    - 否则回退 next_session_expiry_ymd。
+    """
+    complete = _complete_by_expiry(instruments)
     if not complete:
         return None
 
-    atm = pick_atm_strike(list(complete.keys()), mark_px)
+    if expiry_ymd:
+        ymd = expiry_ymd
+        if ymd not in complete:
+            return None
+    elif min_hours is not None:
+        eligible = list_eligible_expiry_ymds(
+            instruments, min_hours=min_hours, now=now
+        )
+        if not eligible:
+            return None
+        ymd = eligible[0]
+    else:
+        ymd = next_session_expiry_ymd(now)
+        if ymd not in complete:
+            # 回退到最近合格到期
+            eligible = list_eligible_expiry_ymds(instruments, min_hours=0, now=now)
+            if not eligible:
+                return None
+            ymd = eligible[0]
+
+    strikes_map = complete[ymd]
+    atm = pick_atm_strike(list(strikes_map.keys()), mark_px)
     if atm is None:
         return None
-
-    legs = complete[atm]
+    legs = strikes_map[atm]
     return OptionPair(
         expiry_ymd=ymd,
         expiry_ms=expiry_ms_from_ymd(ymd),
@@ -117,3 +165,10 @@ def select_option_pair(
         call_inst_id=legs["C"],
         put_inst_id=legs["P"],
     )
+
+
+def option_leverage(underlying_px: float, premium_ask: float) -> float | None:
+    """现价 / 卖一权利金。"""
+    if underlying_px <= 0 or premium_ask is None or premium_ask <= 0:
+        return None
+    return float(underlying_px) / float(premium_ask)