Add control residual options table with liquidity-only manual close.

Fleet status exposes enriched residuals; manual close skips the premium recovery gate while auto mid-close remains unchanged.

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

backend/app/api/fleet.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import hmac
 import logging
@@ -233,6 +234,12 @@ async def fleet_status(_tok: Annotated[str, Depends(require_fleet_token)]) -> di
         except Exception:
             latest_funds = 0.0
 
+    residuals: list[dict] = []
+    try:
+        residuals = get_engine().matcher.list_residual_options_enriched()
+    except Exception:
+        residuals = []
+
     return {
         "ok": True,
         "mode": settings.mode,
@@ -240,6 +247,7 @@ async def fleet_status(_tok: Annotated[str, Depends(require_fleet_token)]) -> di
         "exchange": exchange_name,
         "sim": settings.is_sim,
         "latest_funds": latest_funds,
+        "residuals": residuals,
         "market_connected": bool(snap.connected) if snap else False,
         "pair": snap.pair.to_dict() if snap and snap.pair else None,
         "index_px": (
@@ -315,6 +323,31 @@ async def fleet_pause(_tok: Annotated[str, Depends(require_fleet_token)]) -> dic
     return await get_engine().pause()
 
 
+class ResidualCloseBody(BaseModel):
+    group_id: str = Field(min_length=1, max_length=128)
+
+
+@router.post("/residual/close")
+async def fleet_residual_close(
+    body: ResidualCloseBody,
+    _tok: Annotated[str, Depends(require_fleet_token)],
+) -> dict:
+    """中控手动平单条残留：只验流动性，不验权利金回收比例。"""
+    matcher = get_engine().matcher
+    result = await asyncio.to_thread(matcher.close_residual_manual, body.group_id)
+    if not result.ok:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=result.detail or "平残留失败",
+        )
+    return {
+        "ok": True,
+        "detail": result.detail,
+        "data": result.data,
+        "liquidity_wait": result.liquidity_wait,
+    }
+
+
 @router.post("/issue-login")
 async def fleet_issue_login(_tok: Annotated[str, Depends(require_fleet_token)]) -> dict:
     username, _ = get_credentials()

backend/app/live/binance_executor.py

@@ -1205,21 +1205,26 @@ class BinanceLiveExecutor(Matcher):
             }
         return row
 
-    def try_close_one_residual(self, row: dict) -> dict | None:
-        """LIVE-BN：权利金达标后按最新买一 IOC 限价卖出归档期权（不扫市价）。"""
+    def try_close_one_residual(
+        self, row: dict, *, skip_premium_ratio: bool = False
+    ) -> dict | None:
+        """LIVE-BN：流动性通过后按最新买一 IOC 限价卖；自动路径另要求权利金比例。"""
         err = self._guard_live()
         if err:
-            logger.warning("residual premium close blocked: %s", err)
+            logger.warning("residual close blocked: %s", err)
             return None
         synced = self._sync_residual_contracts_with_exchange(row)
         if synced is None:
             return None
         row = synced
-        skip, close_bid, _oq = self._evaluate_residual_premium_close(row)
+        require_ratio = not skip_premium_ratio
+        skip, close_bid, _oq = self._evaluate_residual_premium_close(
+            row, require_premium_ratio=require_ratio
+        )
         if skip or close_bid is None:
             if skip:
                 logger.debug(
-                    "residual premium close skip %s: %s",
+                    "residual close skip %s: %s",
                     row.get("group_id"),
                     skip,
                 )
@@ -1232,17 +1237,19 @@ class BinanceLiveExecutor(Matcher):
             opt_contracts = float(contracts_for_eth(opt_qty, ct) or 0)
         if opt_contracts <= 0:
             logger.warning(
-                "residual premium close skip %s: bad contracts", row.get("group_id")
+                "residual close skip %s: bad contracts", row.get("group_id")
             )
             return None
         oq2 = self._quote_held_option(option_inst_id)
         if oq2 is None or oq2.bid is None:
             return None
         bid_px = float(oq2.bid)
-        skip2 = self._residual_bid_gate(row, bid=bid_px, oq=oq2)
+        skip2 = self._residual_bid_gate(
+            row, bid=bid_px, oq=oq2, require_premium_ratio=require_ratio
+        )
         if skip2:
             logger.debug(
-                "residual premium close recheck skip %s: %s",
+                "residual close recheck skip %s: %s",
                 row.get("group_id"),
                 skip2,
             )
@@ -1258,7 +1265,7 @@ class BinanceLiveExecutor(Matcher):
             )
         except Exception as e:
             logger.warning(
-                "residual premium close bid-ioc sell failed %s: %s",
+                "residual close bid-ioc sell failed %s: %s",
                 row.get("group_id"),
                 e,
             )
@@ -1276,6 +1283,7 @@ class BinanceLiveExecutor(Matcher):
         )
         fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id))
         now_ms = int(time.time() * 1000)
+        tag = "manual" if skip_premium_ratio else "mid"
         return self._book_residual_market_close(
             row,
             fill_px=of_px,
@@ -1283,11 +1291,15 @@ class BinanceLiveExecutor(Matcher):
             notional=of_px * fill_eth,
             slip=0.0,
             now_ms=now_ms,
-            note=f"LIVE-BN residual mid-close at bid IOC px={bid_px}",
+            note=f"LIVE-BN residual {tag}-close at bid IOC px={bid_px}",
             exec_mode="LIVE",
             filled_contracts=filled_c,
             remaining_contracts=remaining,
-            close_reason="residual_premium_close",
+            close_reason=(
+                "residual_manual_close"
+                if skip_premium_ratio
+                else "residual_premium_close"
+            ),
         )
 
     def _try_exchange_flatten_residual(

backend/app/live/executor.py

@@ -1250,21 +1250,26 @@ class OkxLiveExecutor(Matcher):
             }
         return row
 
-    def try_close_one_residual(self, row: dict) -> dict | None:
-        """LIVE：权利金达标后按最新买一 IOC 限价卖出归档期权（不扫市价）。"""
+    def try_close_one_residual(
+        self, row: dict, *, skip_premium_ratio: bool = False
+    ) -> dict | None:
+        """LIVE：流动性通过后按最新买一 IOC 限价卖；自动路径另要求权利金比例。"""
         err = self._guard_live()
         if err:
-            logger.warning("residual premium close blocked: %s", err)
+            logger.warning("residual close blocked: %s", err)
             return None
         synced = self._sync_residual_contracts_with_exchange(row)
         if synced is None:
             return None
         row = synced
-        skip, close_bid, _oq = self._evaluate_residual_premium_close(row)
+        require_ratio = not skip_premium_ratio
+        skip, close_bid, _oq = self._evaluate_residual_premium_close(
+            row, require_premium_ratio=require_ratio
+        )
         if skip or close_bid is None:
             if skip:
                 logger.debug(
-                    "residual premium close skip %s: %s",
+                    "residual close skip %s: %s",
                     row.get("group_id"),
                     skip,
                 )
@@ -1277,17 +1282,19 @@ class OkxLiveExecutor(Matcher):
             opt_contracts = float(contracts_for_eth(opt_qty, ct) or 0)
         if opt_contracts <= 0:
             logger.warning(
-                "residual premium close skip %s: bad contracts", row.get("group_id")
+                "residual close skip %s: bad contracts", row.get("group_id")
             )
             return None
         oq2 = self._quote_held_option(option_inst_id)
         if oq2 is None or oq2.bid is None:
             return None
         bid_px = float(oq2.bid)
-        skip2 = self._residual_bid_gate(row, bid=bid_px, oq=oq2)
+        skip2 = self._residual_bid_gate(
+            row, bid=bid_px, oq=oq2, require_premium_ratio=require_ratio
+        )
         if skip2:
             logger.debug(
-                "residual premium close recheck skip %s: %s",
+                "residual close recheck skip %s: %s",
                 row.get("group_id"),
                 skip2,
             )
@@ -1304,7 +1311,7 @@ class OkxLiveExecutor(Matcher):
             )
         except Exception as e:
             logger.warning(
-                "residual premium close bid-ioc sell failed %s: %s",
+                "residual close bid-ioc sell failed %s: %s",
                 row.get("group_id"),
                 e,
             )
@@ -1321,6 +1328,7 @@ class OkxLiveExecutor(Matcher):
             remaining = max(0.0, opt_contracts - filled_c)
         fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id))
         now_ms = int(time.time() * 1000)
+        tag = "manual" if skip_premium_ratio else "mid"
         return self._book_residual_market_close(
             row,
             fill_px=of_px,
@@ -1328,11 +1336,15 @@ class OkxLiveExecutor(Matcher):
             notional=of_px * fill_eth,
             slip=0.0,
             now_ms=now_ms,
-            note=f"LIVE residual mid-close at bid IOC px={bid_px}",
+            note=f"LIVE residual {tag}-close at bid IOC px={bid_px}",
             exec_mode="LIVE",
             filled_contracts=filled_c,
             remaining_contracts=remaining,
-            close_reason="residual_premium_close",
+            close_reason=(
+                "residual_manual_close"
+                if skip_premium_ratio
+                else "residual_premium_close"
+            ),
         )
 
     def _try_exchange_flatten_residual(

backend/app/sim/matcher.py

@@ -838,24 +838,32 @@ class Matcher:
         )
 
     def _residual_bid_gate(
-        self, row: dict[str, Any], *, bid: float, oq: Any
+        self,
+        row: dict[str, Any],
+        *,
+        bid: float,
+        oq: Any,
+        require_premium_ratio: bool = True,
     ) -> str | None:
-        """权利金比例 + 深度 + 买一/标记偏差。通过返回 None。"""
+        """深度 + 买一/标记偏差；可选权利金比例。通过返回 None。"""
         s = get_settings()
         initial_premium = float(row.get("initial_premium") or 0)
         opt_qty = float(row.get("option_qty_eth") or 0)
-        if initial_premium <= 0 or opt_qty <= 0:
-            return "invalid_initial_premium_or_qty"
+        if opt_qty <= 0:
+            return "invalid_qty"
         if bid <= 0:
             return "option_bid_unavailable"
-        current_premium = float(bid) * opt_qty
-        min_pct = self._residual_min_premium_pct()
-        threshold = initial_premium * (min_pct / 100.0)
-        if current_premium + 1e-12 < threshold:
-            return (
-                f"premium_below_threshold curr={current_premium:.4f} "
-                f"need>={threshold:.4f} ({min_pct:g}%)"
-            )
+        if require_premium_ratio:
+            if initial_premium <= 0:
+                return "invalid_initial_premium_or_qty"
+            current_premium = float(bid) * opt_qty
+            min_pct = self._residual_min_premium_pct()
+            threshold = initial_premium * (min_pct / 100.0)
+            if current_premium + 1e-12 < threshold:
+                return (
+                    f"premium_below_threshold curr={current_premium:.4f} "
+                    f"need>={threshold:.4f} ({min_pct:g}%)"
+                )
         option_inst_id = str(row.get("option_inst_id") or "")
         ct_mult = self._ct_mult(option_inst_id)
         if not bid_covers_eth(
@@ -877,12 +885,11 @@ class Matcher:
         return None
 
     def _evaluate_residual_premium_close(
-        self, row: dict[str, Any]
+        self, row: dict[str, Any], *, require_premium_ratio: bool = True
     ) -> tuple[str | None, float | None, Any]:
         """
-        残留中途平前置：权利金比例 + 买一流动性。
+        残留平前置：买一流动性；自动路径另加权利金比例。
         返回 (skip_reason, close_bid, option_quote)；skip_reason 非空则本轮不卖。
-        成交价口径：最新买一（不再抬到内在价值）。
         """
         option_inst_id = str(row.get("option_inst_id") or "")
         if not option_inst_id:
@@ -891,11 +898,73 @@ class Matcher:
         if oq is None or oq.bid is None:
             return ("option_bid_unavailable", None, None)
         close_bid = float(oq.bid)
-        skip = self._residual_bid_gate(row, bid=close_bid, oq=oq)
+        skip = self._residual_bid_gate(
+            row,
+            bid=close_bid,
+            oq=oq,
+            require_premium_ratio=require_premium_ratio,
+        )
         if skip:
             return (skip, None, None)
         return (None, close_bid, oq)
 
+    def list_residual_options_enriched(self) -> list[dict[str, Any]]:
+        """pending 残留 + 买一权利金/回收占比/流动性是否可手动平。"""
+        out: list[dict[str, Any]] = []
+        for row in self.list_residual_options(pending_only=True):
+            d = dict(row)
+            option_inst_id = str(d.get("option_inst_id") or "")
+            opt_qty = float(d.get("option_qty_eth") or 0)
+            init = float(d.get("initial_premium") or 0)
+            oq = self._quote_held_option(option_inst_id) if option_inst_id else None
+            bid = float(oq.bid) if oq is not None and oq.bid is not None else None
+            cur = float(bid) * opt_qty if bid is not None else None
+            ratio = (cur / init * 100.0) if cur is not None and init > 1e-12 else None
+            liq_detail: str | None
+            if bid is None or oq is None:
+                liq_detail = "option_bid_unavailable"
+            else:
+                liq_detail = self._residual_bid_gate(
+                    d, bid=bid, oq=oq, require_premium_ratio=False
+                )
+            d.update(
+                {
+                    "bid_px": bid,
+                    "current_premium": cur,
+                    "recovery_pct": ratio,
+                    "liquidity_ok": liq_detail is None,
+                    "liquidity_detail": liq_detail,
+                }
+            )
+            out.append(d)
+        return out
+
+    def close_residual_manual(self, group_id: str) -> CloseResult:
+        """中控手动平单条残留：只验流动性，不验权利金比例。"""
+        gid = str(group_id or "").strip()
+        if not gid:
+            return CloseResult(ok=False, detail="缺少 group_id")
+        row = self.db.fetchone(
+            "SELECT * FROM residual_options WHERE group_id=? AND status='pending'",
+            (gid,),
+        )
+        if row is None:
+            return CloseResult(ok=False, detail="无该组 pending 残留")
+        d = dict(row)
+        skip, _bid, _oq = self._evaluate_residual_premium_close(
+            d, require_premium_ratio=False
+        )
+        if skip:
+            return CloseResult(ok=False, detail=skip, liquidity_wait=True)
+        booked = self.try_close_one_residual(d, skip_premium_ratio=True)
+        if not booked:
+            return CloseResult(
+                ok=False,
+                detail="平残留失败（流动性变化或下单未成交）",
+                liquidity_wait=True,
+            )
+        return CloseResult(ok=True, detail="residual_manual_closed", data=booked)
+
     def _book_residual_market_close(
         self,
         row: dict[str, Any],
@@ -1094,25 +1163,31 @@ class Matcher:
             "fully_done": fully_done,
         }
 
-    def try_close_one_residual(self, row: dict[str, Any]) -> dict[str, Any] | None:
-        """SIM：权利金达标且流动性通过则本地吃买一平残留。"""
-        skip, close_bid, oq = self._evaluate_residual_premium_close(row)
+    def try_close_one_residual(
+        self, row: dict[str, Any], *, skip_premium_ratio: bool = False
+    ) -> dict[str, Any] | None:
+        """SIM：流动性通过则吃买一平残留；自动路径另要求权利金比例。"""
+        require_ratio = not skip_premium_ratio
+        skip, close_bid, oq = self._evaluate_residual_premium_close(
+            row, require_premium_ratio=require_ratio
+        )
         if skip or close_bid is None or oq is None:
             if skip:
                 logger.debug(
-                    "residual premium close skip %s: %s",
+                    "residual close skip %s: %s",
                     row.get("group_id"),
                     skip,
                 )
             return None
-        # 下单前再刷买一并重跑门槛
         option_inst_id = str(row.get("option_inst_id") or "")
         oq2 = self._quote_held_option(option_inst_id) or oq
         bid2 = float(oq2.bid) if oq2.bid is not None else float(close_bid)
-        skip2 = self._residual_bid_gate(row, bid=bid2, oq=oq2)
+        skip2 = self._residual_bid_gate(
+            row, bid=bid2, oq=oq2, require_premium_ratio=require_ratio
+        )
         if skip2:
             logger.debug(
-                "residual premium close recheck skip %s: %s",
+                "residual close recheck skip %s: %s",
                 row.get("group_id"),
                 skip2,
             )
@@ -1125,6 +1200,11 @@ class Matcher:
             fee_rate=self._fee_rate(),
         )
         now_ms = int(time.time() * 1000)
+        note = (
+            f"residual manual close at bid px={bid2}"
+            if skip_premium_ratio
+            else f"residual mid-close at bid px={bid2}"
+        )
         return self._book_residual_market_close(
             row,
             fill_px=of.fill_px,
@@ -1132,9 +1212,14 @@ class Matcher:
             notional=of.notional,
             slip=of.slip,
             now_ms=now_ms,
-            note=f"residual mid-close at bid px={bid2}",
+            note=note,
             filled_contracts=float(row.get("option_qty_contracts") or 0) or None,
             remaining_contracts=0.0,
+            close_reason=(
+                "residual_manual_close"
+                if skip_premium_ratio
+                else "residual_premium_close"
+            ),
         )
 
     def try_close_pending_residuals(self) -> list[dict[str, Any]]:

backend/tests/test_residual_premium_close.py

@@ -189,6 +189,57 @@ def test_residual_book_pending_guard(tmp_path, monkeypatch) -> None:
     db.close()
 
 
+def test_manual_close_skips_premium_ratio(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("MODE", "SIM")
+    db = Database(tmp_path / "manual.db")
+    db.set_setting("residual_min_premium_pct", "20")
+    _seed_residual(db, initial_premium=100.0, qty=2.0, entry_px=50.0)
+    m = Matcher(db)
+
+    # bid=5 → premium=10 < 20%，自动路径应跳过，手动可平
+    oq = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=10_000.0, mark_px=5.0)
+    monkeypatch.setattr(m, "_quote_held_option", lambda _id: oq)
+    monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0)
+    monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01)
+
+    assert m.try_close_one_residual(m.list_residual_options()[0]) is None
+    enriched = m.list_residual_options_enriched()
+    assert len(enriched) == 1
+    assert enriched[0]["liquidity_ok"] is True
+    assert enriched[0]["recovery_pct"] is not None
+    assert float(enriched[0]["recovery_pct"]) < 20.0
+
+    r = m.close_residual_manual("G-res")
+    assert r.ok is True
+    row = db.fetchone(
+        "SELECT status FROM residual_options WHERE group_id=?", ("G-res",)
+    )
+    assert row is not None and row["status"] == "settled"
+    db.close()
+
+
+def test_manual_close_liquidity_still_required(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("MODE", "SIM")
+    db = Database(tmp_path / "manual_liq.db")
+    db.set_setting("residual_min_premium_pct", "20")
+    _seed_residual(db, initial_premium=100.0, qty=2.0)
+    m = Matcher(db)
+
+    oq = SimpleNamespace(bid=5.0, ask=5.5, bid_sz=1.0, mark_px=5.0)
+    monkeypatch.setattr(m, "_quote_held_option", lambda _id: oq)
+    monkeypatch.setattr(m, "_close_spot_px", lambda _snap: 1900.0)
+    monkeypatch.setattr(m, "_ct_mult", lambda _id: 0.01)
+
+    r = m.close_residual_manual("G-res")
+    assert r.ok is False
+    assert "liquidity" in (r.detail or "")
+    row = db.fetchone(
+        "SELECT status FROM residual_options WHERE group_id=?", ("G-res",)
+    )
+    assert row is not None and row["status"] == "pending"
+    db.close()
+
+
 def test_settings_exposes_residual_min_premium_pct(tmp_path, monkeypatch) -> None:
     monkeypatch.setenv("MODE", "SIM")
     from app.api import settings as settings_api