From f48ea5bbcca4a67a7feb5949418a16a7bc24f7e9 Mon Sep 17 00:00:00 2001 From: dekun Date: Sun, 26 Jul 2026 22:28:03 +0800 Subject: [PATCH] =?UTF-8?q?Audit=20fixes:=20LIVE=20symbols/fills/expiry/pe?= =?UTF-8?q?nding,=20security=20harden,=20add=20=E6=9B=B4=E6=96=B0=E8=AF=B4?= =?UTF-8?q?=E6=98=8E.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- README.md | 1 + backend/app/api/auth.py | 8 +- backend/app/api/auth_routes.py | 87 +- backend/app/api/settings.py | 7 + backend/app/config.py | 4 + backend/app/credentials.py | 10 +- backend/app/env_store.py | 8 +- backend/app/live/binance_executor.py | 133 +- backend/app/live/binance_trade.py | 7 +- backend/app/live/executor.py | 2020 +++++++++++++------------- backend/app/live/live_pnl.py | 21 +- backend/app/live/okx_trade.py | 14 +- backend/app/live/symbols.py | 26 + backend/app/main.py | 20 +- backend/tests/test_runtime_mode.py | 7 +- backend/tests/test_symbols.py | 33 + docs/更新说明.md | 51 + frontend/src/pages/Settings.tsx | 1 + 18 files changed, 1389 insertions(+), 1069 deletions(-) create mode 100644 backend/app/live/symbols.py create mode 100644 backend/tests/test_symbols.py create mode 100644 docs/更新说明.md diff --git a/README.md b/README.md index b5d3502..b52891f 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ - [商业化与授权方案](docs/商业化与授权方案.md) - [策略说明](docs/策略说明.md) - [实盘策略说明](docs/实盘策略说明.md) +- [更新说明](docs/更新说明.md)（每次发版追加） ## 访问（测试机） diff --git a/backend/app/api/auth.py b/backend/app/api/auth.py index b683671..1ebe621 100644 --- a/backend/app/api/auth.py +++ b/backend/app/api/auth.py @@ -42,7 +42,11 @@ def _b64url_decode(s: str) -> bytes: def issue_token(username: str, settings: Settings) -> tuple[str, int]: exp = int(time.time()) + int(settings.auth_token_ttl_sec) - payload = {"u": username, "exp": exp} + payload = { + "u": username, + "exp": exp, + "v": int(settings.auth_token_version), + } raw = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8")) sig = hmac.new( settings.auth_secret.encode("utf-8"), @@ -70,6 +74,8 @@ def verify_token(token: str, settings: Settings) -> str: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token") from e if int(payload.get("exp") or 0) < int(time.time()): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token expired") + if int(payload.get("v") or 0) != int(settings.auth_token_version): + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token revoked") username = str(payload.get("u") or "") if not username: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token") diff --git a/backend/app/api/auth_routes.py b/backend/app/api/auth_routes.py index a757566..1d6b096 100644 --- a/backend/app/api/auth_routes.py +++ b/backend/app/api/auth_routes.py @@ -1,28 +1,71 @@ from __future__ import annotations +import hmac +import time +from collections import defaultdict from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel, Field from ..config import Settings, get_settings -from ..credentials import get_credentials, update_credentials +from ..credentials import get_credentials, update_credentials, upsert_env_file from .auth import LoginRequest, LoginResponse, issue_token, require_user router = APIRouter(prefix="/api/auth", tags=["auth"]) +_login_hits: dict[str, list[float]] = defaultdict(list) + class ChangeCredentialsRequest(BaseModel): current_password: str = Field(min_length=1) new_username: str = Field(min_length=1, max_length=64) - new_password: str = Field(min_length=4, max_length=128) + new_password: str = Field(min_length=8, max_length=128) + + +def _client_ip(request: Request) -> str: + xff = request.headers.get("x-forwarded-for") or "" + if xff.strip(): + return xff.split(",")[0].strip() + if request.client: + return request.client.host or "unknown" + return "unknown" + + +def _rate_limit_login(ip: str, settings: Settings) -> None: + now = time.time() + window = float(settings.login_window_sec) + max_n = int(settings.login_max_attempts) + hits = [t for t in _login_hits[ip] if now - t < window] + _login_hits[ip] = hits + if len(hits) >= max_n: + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail=f"登录过于频繁，请 {int(window)} 秒后再试", + ) @router.post("/login", response_model=LoginResponse) -async def login(body: LoginRequest, settings: Annotated[Settings, Depends(get_settings)]) -> LoginResponse: +async def login( + body: LoginRequest, + request: Request, + settings: Annotated[Settings, Depends(get_settings)], +) -> LoginResponse: + ip = _client_ip(request) + _rate_limit_login(ip, settings) user, pwd = get_credentials() - if body.username != user or body.password != pwd: - raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误") + user_ok = hmac.compare_digest( + body.username.encode("utf-8"), user.encode("utf-8") + ) + pwd_ok = hmac.compare_digest( + body.password.encode("utf-8"), pwd.encode("utf-8") + ) + if not (user_ok and pwd_ok): + _login_hits[ip].append(time.time()) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误" + ) + _login_hits.pop(ip, None) token, ttl = issue_token(body.username, settings) return LoginResponse( token=token, @@ -34,7 +77,10 @@ async def login(body: LoginRequest, settings: Annotated[Settings, Depends(get_se @router.get("/me") -async def me(username: Annotated[str, Depends(require_user)], settings: Annotated[Settings, Depends(get_settings)]) -> dict: +async def me( + username: Annotated[str, Depends(require_user)], + settings: Annotated[Settings, Depends(get_settings)], +) -> dict: return { "username": username, "env_name": settings.env_name, @@ -50,17 +96,30 @@ async def change_credentials( settings: Annotated[Settings, Depends(get_settings)], ) -> LoginResponse: _cur_user, cur_pwd = get_credentials() - if body.current_password != cur_pwd: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前密码不正确") + if not hmac.compare_digest( + body.current_password.encode("utf-8"), cur_pwd.encode("utf-8") + ): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="当前密码不正确" + ) try: - update_credentials(new_username=body.new_username, new_password=body.new_password) + update_credentials( + new_username=body.new_username, new_password=body.new_password + ) + # 作废旧 token + new_ver = int(settings.auth_token_version) + 1 + upsert_env_file("AUTH_TOKEN_VERSION", str(new_ver)) + get_settings.cache_clear() except ValueError as e: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e - token, ttl = issue_token(body.new_username.strip(), settings) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail=str(e) + ) from e + settings2 = get_settings() + token, ttl = issue_token(body.new_username.strip(), settings2) return LoginResponse( token=token, username=body.new_username.strip(), expires_in=ttl, - env_name=settings.env_name, - mode=settings.mode, + env_name=settings2.env_name, + mode=settings2.mode, ) diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 64d9696..5471821 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -215,6 +215,7 @@ async def put_strategy_settings( class RuntimeSettingsBody(BaseModel): mode: Literal["SIM", "LIVE"] | None = None confirm_live: bool | None = False + confirm_live_phrase: str | None = None okx_api_key: str | None = None okx_api_secret: str | None = None okx_api_passphrase: str | None = None @@ -272,6 +273,12 @@ async def put_runtime_settings( status_code=400, detail="切换到 LIVE 须二次确认（confirm_live=true）", ) + phrase = (body.confirm_live_phrase or "").strip() + if phrase != "LIVE": + raise HTTPException( + status_code=400, + detail="切换到 LIVE 须在 confirm_live_phrase 传入 LIVE", + ) updates: dict[str, str] = {} if body.okx_api_key is not None and body.okx_api_key.strip(): diff --git a/backend/app/config.py b/backend/app/config.py index 4904eeb..79c8d47 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -25,6 +25,10 @@ class Settings(BaseSettings): auth_password: str = "admin123" auth_secret: str = "change-me-eth-hedge-sim-secret" auth_token_ttl_sec: int = 60 * 60 * 24 * 7 + auth_token_version: int = 1 # 改密时递增，作废旧 token + disable_api_docs: bool = True # 生产默认关闭 /docs + login_max_attempts: int = 8 + login_window_sec: int = 300 okx_api_key: str = "" okx_api_secret: str = "" diff --git a/backend/app/credentials.py b/backend/app/credentials.py index 4437b58..bd1450c 100644 --- a/backend/app/credentials.py +++ b/backend/app/credentials.py @@ -38,11 +38,15 @@ def get_credentials() -> tuple[str, str]: def upsert_env_file(key: str, value: str) -> Path | None: """写入第一个已存在的 .env；都不存在则写仓库根 .env。""" + if "\n" in value or "\r" in value: + raise ValueError(f"{key} 值不能包含换行") paths = _env_paths() target = next((p for p in paths if p.is_file()), paths[0]) target.parent.mkdir(parents=True, exist_ok=True) text = target.read_text(encoding="utf-8") if target.is_file() else "" - line = f"{key}={value}" + # 简单引号，避免空格/特殊字符破坏解析 + safe = value.replace("\\", "\\\\").replace('"', '\\"') + line = f'{key}="{safe}"' pattern = re.compile(rf"(?m)^{re.escape(key)}=.*$") if pattern.search(text): text = pattern.sub(line, text) @@ -51,6 +55,10 @@ def upsert_env_file(key: str, value: str) -> Path | None: text += "\n" text += line + "\n" target.write_text(text, encoding="utf-8") + try: + target.chmod(0o600) + except Exception: + pass return target diff --git a/backend/app/env_store.py b/backend/app/env_store.py index 3c8a1b2..f29f4a3 100644 --- a/backend/app/env_store.py +++ b/backend/app/env_store.py @@ -22,14 +22,12 @@ def upsert_env_keys(updates: dict[str, str]) -> Path | None: return target -def mask_secret(raw: str | None, *, keep: int = 4) -> str | None: - """脱敏：****末尾；过短则全部打码。""" +def mask_secret(raw: str | None, *, keep: int = 0) -> str | None: + """脱敏：仅返回是否已配置（不再泄露末尾字符）。""" s = (raw or "").strip() if not s: return None - if len(s) <= keep: - return "*" * len(s) - return "*" * max(4, len(s) - keep) + s[-keep:] + return "********" def okx_keys_configured(s=None) -> bool: diff --git a/backend/app/live/binance_executor.py b/backend/app/live/binance_executor.py index 1e5621a..3640a6a 100644 --- a/backend/app/live/binance_executor.py +++ b/backend/app/live/binance_executor.py @@ -7,11 +7,12 @@ import time from ..config import get_settings from ..env_store import live_ready -from ..sim.liquidity import contracts_for_eth +from ..sim.liquidity import contracts_for_eth, eth_from_contracts from ..sim.matcher import CloseResult, Matcher, OpenResult from ..sim.pricing import option_expiry_settle, option_intrinsic from ..strategy.session import get_session from .binance_trade import BinanceTradeClient +from .symbols import live_settings, resolve_perp_inst_id logger = logging.getLogger(__name__) @@ -38,9 +39,11 @@ class BinanceLiveExecutor(Matcher): return base from .live_pnl import enrich_live_unrealized - s = get_settings() gid = base.get("group_id") open_at = None + perp_inst = resolve_perp_inst_id( + self.db, group_id=str(gid) if gid else None + ) if gid: g = self.db.fetchone( "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?", @@ -48,11 +51,8 @@ class BinanceLiveExecutor(Matcher): ) if g: open_at = int(g["open_at_ms"] or 0) or None - perp_inst = str(g["perp_inst_id"] or s.perp_inst_id) - else: - perp_inst = s.perp_inst_id - else: - perp_inst = s.perp_inst_id + if g["perp_inst_id"]: + perp_inst = str(g["perp_inst_id"]) try: client = self._client() except Exception: @@ -83,7 +83,7 @@ class BinanceLiveExecutor(Matcher): if err: return OpenResult(ok=False, detail=err) - s = get_settings() + s = live_settings() if self.has_open_position(): st = self.position_status() return OpenResult( @@ -92,6 +92,7 @@ class BinanceLiveExecutor(Matcher): ) client = self._client() + perp_inst = resolve_perp_inst_id(self.db) perp_qty = self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth) opt_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth) ct_mult = self._ct_mult(option_inst_id) @@ -107,6 +108,12 @@ class BinanceLiveExecutor(Matcher): logger.exception("binance live open option failed") return OpenResult(ok=False, detail=f"币安开期权失败: {e}") + filled_opt_contracts = float(opt_fill.sz) if opt_fill.sz and opt_fill.sz > 0 else float( + int(round(opt_contracts)) + ) + opt_contracts = filled_opt_contracts + opt_qty = eth_from_contracts(opt_contracts, ct_mult) + # 永续市价失败（多为保证金不足）→ 必须回滚期权 try: if perp_side == "long": @@ -114,7 +121,7 @@ class BinanceLiveExecutor(Matcher): else: side, pos_side = "SELL", "SHORT" perp_fill_live = client.place_perp_market( - symbol=s.perp_inst_id, + symbol=perp_inst, side=side, qty_eth=perp_qty, position_side=pos_side, @@ -159,6 +166,12 @@ class BinanceLiveExecutor(Matcher): pf_px = float(perp_fill_live.avg_px) of_fee = float(opt_fill.fee) pf_fee = float(perp_fill_live.fee) + filled_perp_qty = ( + float(perp_fill_live.sz) + if perp_fill_live.sz and perp_fill_live.sz > 0 + else perp_qty + ) + perp_qty = filled_perp_qty initial_premium = of_px * opt_qty of_notional = of_px * opt_qty pf_notional = pf_px * perp_qty @@ -194,7 +207,7 @@ class BinanceLiveExecutor(Matcher): option_side, perp_side, option_inst_id, - s.perp_inst_id, + perp_inst, strike, expiry_ymd, entry_index_px, @@ -235,7 +248,7 @@ class BinanceLiveExecutor(Matcher): "perp", "open", perp_side, - s.perp_inst_id, + perp_inst, perp_qty, None, pf_px, @@ -303,7 +316,7 @@ class BinanceLiveExecutor(Matcher): detail: str, ) -> None: """期权已成交、永续未开且回滚失败 → 落 half_open，禁止新开，待 repair。""" - s = get_settings() + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) initial_premium = of_px * opt_qty self.ledger.apply_cash( -(of_px * opt_qty + of_fee), @@ -331,7 +344,7 @@ class BinanceLiveExecutor(Matcher): option_side, perp_side, option_inst_id, - s.perp_inst_id, + perp_inst, strike, expiry_ymd, entry_index_px, @@ -475,7 +488,7 @@ class BinanceLiveExecutor(Matcher): if err: return CloseResult(ok=False, detail=err) - s = get_settings() + s = live_settings() pos = self.current_position() st = str(pos.get("status") or "") if st == "half_open": @@ -490,6 +503,7 @@ class BinanceLiveExecutor(Matcher): opt_qty = float(pos["option_qty_eth"]) perp_qty = float(pos["perp_qty_eth"]) opt_contracts = float(pos["option_qty_contracts"] or 0) + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) client = self._client() is_expiry = reason == "expiry" fee_rate = self._fee_rate() @@ -527,34 +541,45 @@ class BinanceLiveExecutor(Matcher): of_fee = float(prev["fee"] or 0) of_notional = float(prev["notional"] or (of_px * opt_qty)) of_slip = float(prev["slip"] or 0) - elif is_expiry: - if intrinsic is None: - return CloseResult(ok=False, detail="到期结算失败：缺行权价或标的价") - of = option_expiry_settle( - intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate - ) - of_px, of_fee, of_slip, of_notional = of.fill_px, of.fee, of.slip, of.notional else: + # 含到期：优先交易所真实平期权；失败且无内在价值时可本地结算 try: opt_live = client.place_option_market( symbol=option_inst_id, side="SELL", - quantity=opt_contracts, + quantity=max(1.0, opt_contracts), reduce_only=True, ) of_px = float(opt_live.avg_px) of_fee = float(opt_live.fee) + filled_c = float(opt_live.sz) if opt_live.sz and opt_live.sz > 0 else opt_contracts + opt_contracts = filled_c + opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id)) of_notional = of_px * opt_qty except Exception as e: - if not bypass_liquidity: + if is_expiry and intrinsic is not None: + of = option_expiry_settle( + intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate + ) + of_px, of_fee, of_slip, of_notional = ( + of.fill_px, + of.fee, + of.slip, + of.notional, + ) + logger.warning( + "expiry option exchange close failed, local settle: %s", e + ) + elif not bypass_liquidity: return CloseResult( ok=False, detail=f"币安平期权失败: {e}", liquidity_wait=True, ) - return CloseResult(ok=False, detail=f"币安平期权失败: {e}") + else: + return CloseResult(ok=False, detail=f"币安平期权失败: {e}") - # 期权已平：立刻落 pending，避免永续失败后重试再卖期权 + # 期权已平（或到期本地结算）：立刻落 pending，避免永续失败后重试再卖期权 self._mark_option_closed_perp_pending( group_id=group_id, option_inst_id=option_inst_id, @@ -574,7 +599,7 @@ class BinanceLiveExecutor(Matcher): else: side, pos_side = "BUY", "SHORT" perp_live = client.place_perp_market( - symbol=s.perp_inst_id, + symbol=perp_inst, side=side, qty_eth=perp_qty, position_side=pos_side, @@ -681,7 +706,8 @@ class BinanceLiveExecutor(Matcher): option_fill_already_written: bool, skip_option_cash: bool, ) -> CloseResult: - s = get_settings() + s = live_settings() + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) perp_side = str(pos["perp_side"]) perp_qty = float(pos["perp_qty_eth"]) opt_entry = float(pos["option_entry_px"]) @@ -749,7 +775,7 @@ class BinanceLiveExecutor(Matcher): "perp", "close", "flat", - s.perp_inst_id, + perp_inst, perp_qty, None, pf_px, @@ -793,7 +819,7 @@ class BinanceLiveExecutor(Matcher): client=self._client(), exchange="binance", group_id=group_id, - perp_inst_id=str((g2["perp_inst_id"] if g2 else None) or s.perp_inst_id), + perp_inst_id=str((g2["perp_inst_id"] if g2 else None) or resolve_perp_inst_id(self.db, group_id=group_id)), open_at_ms=int(g2["open_at_ms"]) if g2 and g2["open_at_ms"] else None, local_net=float(net) if net is not None else None, ) @@ -816,10 +842,7 @@ class BinanceLiveExecutor(Matcher): err = self._guard_live() if err: return CloseResult(ok=False, detail=err) - if require_deep_otm and not self.option_is_deep_otm(): - return CloseResult(ok=False, detail="期权非远虚，应走双腿全平") - s = get_settings() pos = self.current_position() st = str(pos.get("status") or "") if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"): @@ -829,17 +852,57 @@ class BinanceLiveExecutor(Matcher): return self.close_group(reason=reason, bypass_liquidity=True) group_id = str(pos["group_id"]) + option_inst_id = str(pos["option_inst_id"]) + option_side = str(pos["option_side"]) + opt_contracts = float(pos["option_qty_contracts"] or 0) + opt_qty = float(pos["option_qty_eth"]) perp_side = str(pos["perp_side"]) perp_qty = float(pos["perp_qty_eth"]) perp_entry = float(pos["perp_entry_px"]) + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) client = self._client() + + # 优先尝试交易所平期权；成功则走双腿全平 + try: + opt_live = client.place_option_market( + symbol=option_inst_id, + side="SELL", + quantity=max(1.0, opt_contracts), + reduce_only=True, + ) + of_px = float(opt_live.avg_px) + of_fee = float(opt_live.fee) + filled_c = float(opt_live.sz) if opt_live.sz and opt_live.sz > 0 else opt_contracts + opt_contracts = filled_c + opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id)) + of_notional = of_px * opt_qty + self._mark_option_closed_perp_pending( + group_id=group_id, + option_inst_id=option_inst_id, + opt_qty=opt_qty, + opt_contracts=opt_contracts, + of_px=of_px, + of_fee=of_fee, + of_notional=of_notional, + of_slip=0.0, + reason=reason, + ) + return self.close_group(reason=reason, bypass_liquidity=True) + except Exception as e: + logger.warning("abandon: option exchange sell failed: %s", e) + if require_deep_otm and not self.option_is_deep_otm(): + return CloseResult( + ok=False, + detail=f"期权平单失败且非远虚，应走双腿全平: {e}", + ) + try: if perp_side == "long": side, pos_side = "SELL", "LONG" else: side, pos_side = "BUY", "SHORT" perp_live = client.place_perp_market( - symbol=s.perp_inst_id, + symbol=perp_inst, side=side, qty_eth=perp_qty, position_side=pos_side, @@ -862,8 +925,6 @@ class BinanceLiveExecutor(Matcher): note=f"LIVE-BN close perp abandon option {reason}", ) - option_inst_id = str(pos["option_inst_id"]) - option_side = str(pos["option_side"]) strike = self._group_strike(group_id, option_inst_id) g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None @@ -893,7 +954,7 @@ class BinanceLiveExecutor(Matcher): "perp", "close", "flat", - s.perp_inst_id, + perp_inst, perp_qty, None, pf_px, diff --git a/backend/app/live/binance_trade.py b/backend/app/live/binance_trade.py index ed4755e..8530e92 100644 --- a/backend/app/live/binance_trade.py +++ b/backend/app/live/binance_trade.py @@ -219,12 +219,17 @@ class BinanceTradeClient: sz = safe_float(q.get("executedQty")) or safe_float(q.get("quantity")) st = str(q.get("status") or "").upper() data = q - if avg and avg > 0 and st in ("FILLED", "PARTIALLY_FILLED"): + if avg and avg > 0 and st == "FILLED": break if st in ("CANCELED", "REJECTED", "EXPIRED"): raise RuntimeError(f"币安期权订单失败 status={st} {q}") + if st == "PARTIALLY_FILLED": + continue if not avg or avg <= 0: raise RuntimeError(f"币安期权无成交均价: {data}") + st_final = str(data.get("status") or "").upper() + if st_final and st_final != "FILLED": + raise RuntimeError(f"币安期权未完全成交 status={st_final} {data}") from .money import abs_fee_usdt fee = abs(safe_float(data.get("fee")) or 0.0) diff --git a/backend/app/live/executor.py b/backend/app/live/executor.py index 0dfb4fd..b1371c2 100644 --- a/backend/app/live/executor.py +++ b/backend/app/live/executor.py @@ -1,993 +1,1027 @@ -"""实盘执行：OKX 真下单 + 本地账本/持仓记录（与 Matcher 同结构）。""" - -from __future__ import annotations - -import logging -import time - -from ..config import get_settings -from ..env_store import live_ready -from ..exchange.runtime import load_runtime_settings -from ..models.db import get_db -from ..sim.liquidity import contracts_for_eth -from ..sim.matcher import CloseResult, Matcher, OpenResult -from ..sim.pricing import option_expiry_settle, option_intrinsic -from ..strategy.session import get_session -from .okx_trade import OkxTradeClient - -logger = logging.getLogger(__name__) - - -class OkxLiveExecutor(Matcher): - """开平仓走 OKX 私有接口；浮盈/残留逻辑复用 Matcher。""" - - def __init__(self, db=None) -> None: - super().__init__(db) - self._trade: OkxTradeClient | None = None - - def _client(self) -> OkxTradeClient: - if self._trade is None: - self._trade = OkxTradeClient() - return self._trade - - def _guard_live(self) -> str | None: - ok, reason = live_ready() - if not ok: - return reason - return None - - def unrealized(self) -> dict: - base = super().unrealized() - if not base.get("has_position"): - return base - from .live_pnl import enrich_live_unrealized - - s = get_settings() - gid = base.get("group_id") - open_at = None - if gid: - g = self.db.fetchone( - "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?", - (gid,), - ) - if g: - open_at = int(g["open_at_ms"] or 0) or None - perp_inst = str(g["perp_inst_id"] or s.perp_inst_id) - else: - perp_inst = s.perp_inst_id - else: - perp_inst = s.perp_inst_id - try: - client = self._client() - except Exception: - return base - return enrich_live_unrealized( - base=base, - db=self.db, - client=client, - exchange="okx", - perp_inst_id=perp_inst, - perp_side=str(base.get("perp_side") or ""), - open_at_ms=open_at, - ) - - def open_group( - self, - *, - group_id: str, - bias: str, - option_side: str, - perp_side: str, - option_inst_id: str, - entry_index_px: float, - strike: float | None = None, - expiry_ymd: str | None = None, - ) -> OpenResult: - err = self._guard_live() - if err: - return OpenResult(ok=False, detail=err) - - s = get_settings() - if self.has_open_position(): - st = self.position_status() - return OpenResult( - ok=False, - detail=f"已有持仓/半仓状态({st})，请先修复或平仓", - ) - - client = self._client() - perp_qty = self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth) - opt_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth) - ct_mult = self._ct_mult(option_inst_id) - opt_contracts = contracts_for_eth(opt_qty, ct_mult) - - # 期权：买入，张数 = contracts - try: - opt_fill = client.place_market( - inst_id=option_inst_id, - side="buy", - sz=str(int(round(opt_contracts))), - td_mode="cash", # OKX 期权常见 cash；若账户不同可再扩展 - ) - except Exception as e: - logger.exception("live open option failed") - return OpenResult(ok=False, detail=f"实盘开期权失败: {e}") - - # 永续市价：按产品假设，失败原因实质为保证金不足 → 必须回滚期权 - try: - ct_val = client.get_ct_val(s.perp_inst_id, inst_type="SWAP") - perp_sz = max(1, int(round(perp_qty / ct_val))) - if perp_side == "long": - side, pos_side = "buy", "long" - else: - side, pos_side = "sell", "short" - perp_fill_live = client.place_market( - inst_id=s.perp_inst_id, - side=side, - sz=str(perp_sz), - td_mode="cross", - pos_side=pos_side, - ) - except Exception as e: - logger.exception("live open perp failed (likely margin); rollback option") - try: - client.place_market( - inst_id=option_inst_id, - side="sell", - sz=str(int(round(opt_contracts))), - td_mode="cash", - reduce_only=True, - ) - except Exception as e2: - logger.exception("live option rollback failed: %s", e2) - self._persist_half_open( - group_id=group_id, - bias=bias, - option_side=option_side, - perp_side=perp_side, - option_inst_id=option_inst_id, - entry_index_px=entry_index_px, - strike=strike, - expiry_ymd=expiry_ymd, - opt_qty=opt_qty, - opt_contracts=opt_contracts, - of_px=float(opt_fill.avg_px), - of_fee=float(opt_fill.fee), - detail=f"保证金开永续失败且期权回滚失败: {e} / {e2}", - ) - return OpenResult( - ok=False, - group_id=group_id, - detail=f"永续开仓失败(保证金)且期权回滚失败，已标记 half_open: {e} / {e2}", - ) - return OpenResult( - ok=False, - detail=f"永续开仓失败(多为保证金不足)，已回滚期权: {e}", - ) - - of_px = float(opt_fill.avg_px) - pf_px = float(perp_fill_live.avg_px) - of_fee = float(opt_fill.fee) - pf_fee = float(perp_fill_live.fee) - initial_premium = of_px * opt_qty - of_notional = of_px * opt_qty - pf_notional = pf_px * perp_qty - - # LIVE：交易所已成交，本地账本允许透支镜像，禁止因账本拒记导致「交易所有仓、DB 空」 - self.ledger.apply_cash( - -(of_notional + of_fee), - kind="open_option", - group_id=group_id, - note=f"LIVE open option {group_id}", - allow_negative=True, - ) - self.ledger.apply_cash( - -pf_fee, - kind="open_perp_fee", - group_id=group_id, - note=f"LIVE open perp {group_id}", - allow_negative=True, - ) - - now = int(time.time() * 1000) - with self.db._lock: - self.db._conn.execute( - """INSERT INTO groups( - group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, - strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, - exec_mode - ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "open", - bias, - option_side, - perp_side, - option_inst_id, - s.perp_inst_id, - strike, - expiry_ymd, - entry_index_px, - initial_premium, - now, - of_fee + pf_fee, - 0.0, - "LIVE", - ), - ) - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "open", - "long", - option_inst_id, - opt_qty, - opt_contracts, - of_px, - of_px, - of_fee, - 0.0, - of_notional, - now, - "LIVE", - ), - ) - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "perp", - "open", - perp_side, - s.perp_inst_id, - perp_qty, - None, - pf_px, - pf_px, - pf_fee, - 0.0, - pf_notional, - now + 1, - "LIVE", - ), - ) - self.db._conn.execute( - """UPDATE positions SET - group_id=?, perp_side=?, perp_qty_eth=?, perp_entry_px=?, - option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, - option_entry_px=?, entry_index_px=?, initial_premium=?, status=? - WHERE id=1""", - ( - group_id, - perp_side, - perp_qty, - pf_px, - option_inst_id, - option_side, - opt_qty, - opt_contracts, - of_px, - entry_index_px, - initial_premium, - "open", - ), - ) - self.db._conn.commit() - - return OpenResult( - ok=True, - group_id=group_id, - detail="opened_live", - data={ - "group_id": group_id, - "exec_mode": "LIVE", - "option_ord": opt_fill.ord_id, - "perp_ord": perp_fill_live.ord_id, - "initial_premium": initial_premium, - "fees": of_fee + pf_fee, - }, - ) - - def _persist_half_open( - self, - *, - group_id: str, - bias: str, - option_side: str, - perp_side: str, - option_inst_id: str, - entry_index_px: float, - strike: float | None, - expiry_ymd: str | None, - opt_qty: float, - opt_contracts: float, - of_px: float, - of_fee: float, - detail: str, - ) -> None: - """期权已成交、永续未开且回滚失败 → 落 half_open，禁止新开，待 repair。""" - s = get_settings() - initial_premium = of_px * opt_qty - self.ledger.apply_cash( - -(of_px * opt_qty + of_fee), - kind="open_option", - group_id=group_id, - note=f"LIVE half_open option {group_id}", - allow_negative=True, - ) - now = int(time.time() * 1000) - with self.db._lock: - existing = self.db._conn.execute( - "SELECT group_id FROM groups WHERE group_id=?", (group_id,) - ).fetchone() - if existing is None: - self.db._conn.execute( - """INSERT INTO groups( - group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, - strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, - exec_mode, note - ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "half_open", - bias, - option_side, - perp_side, - option_inst_id, - s.perp_inst_id, - strike, - expiry_ymd, - entry_index_px, - initial_premium, - now, - of_fee, - 0.0, - "LIVE", - detail[:200], - ), - ) - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "open", - "long", - option_inst_id, - opt_qty, - opt_contracts, - of_px, - of_px, - of_fee, - 0.0, - of_px * opt_qty, - now, - "LIVE", - ), - ) - self.db._conn.execute( - """UPDATE positions SET - group_id=?, perp_side=?, perp_qty_eth=0, perp_entry_px=NULL, - option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, - option_entry_px=?, entry_index_px=?, initial_premium=?, status='half_open' - WHERE id=1""", - ( - group_id, - perp_side, - option_inst_id, - option_side, - opt_qty, - opt_contracts, - of_px, - entry_index_px, - initial_premium, - ), - ) - self.db._conn.commit() - - def repair_half_open(self) -> CloseResult: - """卖出 half_open 残留期权，清本地状态。""" - err = self._guard_live() - if err: - return CloseResult(ok=False, detail=err) - pos = self.current_position() - if pos.get("status") != "half_open": - return CloseResult(ok=False, detail="非 half_open 状态") - group_id = str(pos.get("group_id") or "") - option_inst_id = str(pos.get("option_inst_id") or "") - opt_contracts = float(pos.get("option_qty_contracts") or 0) - opt_qty = float(pos.get("option_qty_eth") or 0) - if not option_inst_id or opt_contracts <= 0: - return CloseResult(ok=False, detail="half_open 缺期权合约信息") - client = self._client() - try: - opt_live = client.place_market( - inst_id=option_inst_id, - side="sell", - sz=str(int(round(opt_contracts))), - td_mode="cash", - reduce_only=True, - ) - except Exception as e: - return CloseResult(ok=False, detail=f"half_open 平期权失败: {e}") - of_px = float(opt_live.avg_px) - of_fee = float(opt_live.fee) - of_notional = of_px * opt_qty - opt_entry = float(pos.get("option_entry_px") or of_px) - self.ledger.apply_cash( - of_notional - of_fee, - kind="close_option", - group_id=group_id or None, - note="LIVE repair half_open", - allow_negative=True, - ) - now = int(time.time() * 1000) - with self.db._lock: - if group_id: - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "close", - "flat", - option_inst_id, - opt_qty, - opt_contracts, - of_px, - of_px, - of_fee, - 0.0, - of_notional, - now, - "LIVE", - ), - ) - opt_pnl = (of_px - opt_entry) * opt_qty - of_fee - self.db._conn.execute( - """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, note=? - WHERE group_id=?""", - ( - "closed", - now, - "half_open_repair", - float(opt_pnl), - "repaired half_open", - group_id, - ), - ) - self.db._conn.execute( - """UPDATE positions SET - group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, - option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, - option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' - WHERE id=1""" - ) - self.db._conn.commit() - return CloseResult( - ok=True, - detail="half_open_repaired", - data={"group_id": group_id, "exec_mode": "LIVE"}, - ) - - def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult: - err = self._guard_live() - if err: - return CloseResult(ok=False, detail=err) - - s = get_settings() - pos = self.current_position() - st = str(pos.get("status") or "") - if st == "half_open": - return self.repair_half_open() - if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"): - return CloseResult(ok=False, detail="无持仓可平") - - group_id = str(pos["group_id"]) - option_inst_id = str(pos["option_inst_id"]) - option_side = str(pos["option_side"]) - perp_side = str(pos["perp_side"]) - opt_qty = float(pos["option_qty_eth"]) - perp_qty = float(pos["perp_qty_eth"]) - opt_contracts = float(pos["option_qty_contracts"] or 0) - client = self._client() - is_expiry = reason == "expiry" - fee_rate = self._fee_rate() - pending_perp_only = st == "option_closed_perp_pending" - - sess = get_session() - snap = sess.snapshot() - strike = self._group_strike(group_id, option_inst_id) - spot = self._close_spot_px(snap) - intrinsic = None - if strike is not None and spot is not None: - intrinsic = option_intrinsic( - option_side=option_side, strike=float(strike), spot=float(spot) - ) - - of_px = 0.0 - of_fee = 0.0 - of_slip = 0.0 - of_notional = 0.0 - - if pending_perp_only: - # 期权已在上次成交并入账；只读上次平期权 fill - prev = self.db.fetchone( - """SELECT fill_px, fee, notional, slip FROM fills - WHERE group_id=? AND leg='option' AND action='close' - ORDER BY id DESC LIMIT 1""", - (group_id,), - ) - if prev is None: - return CloseResult( - ok=False, - detail="option_closed_perp_pending 缺期权平仓记录，请人工核对", - ) - of_px = float(prev["fill_px"]) - of_fee = float(prev["fee"] or 0) - of_notional = float(prev["notional"] or (of_px * opt_qty)) - of_slip = float(prev["slip"] or 0) - elif is_expiry: - if intrinsic is None: - return CloseResult(ok=False, detail="到期结算失败：缺行权价或标的价") - of = option_expiry_settle( - intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate - ) - of_px, of_fee, of_slip, of_notional = of.fill_px, of.fee, of.slip, of.notional - else: - try: - opt_live = client.place_market( - inst_id=option_inst_id, - side="sell", - sz=str(int(round(opt_contracts))), - td_mode="cash", - reduce_only=True, - ) - of_px = float(opt_live.avg_px) - of_fee = float(opt_live.fee) - of_notional = of_px * opt_qty - except Exception as e: - if not bypass_liquidity: - return CloseResult( - ok=False, - detail=f"实盘平期权失败: {e}", - liquidity_wait=True, - ) - return CloseResult(ok=False, detail=f"实盘平期权失败: {e}") - - # 期权已平：立刻落 pending，避免永续失败后重试再卖期权 - self._mark_option_closed_perp_pending( - group_id=group_id, - option_inst_id=option_inst_id, - opt_qty=opt_qty, - opt_contracts=opt_contracts, - of_px=of_px, - of_fee=of_fee, - of_notional=of_notional, - of_slip=of_slip, - reason=reason, - ) - pending_perp_only = True - - try: - ct_val = client.get_ct_val(s.perp_inst_id, inst_type="SWAP") - perp_sz = max(1, int(round(perp_qty / ct_val))) - if perp_side == "long": - side, pos_side = "sell", "long" - else: - side, pos_side = "buy", "short" - perp_live = client.place_market( - inst_id=s.perp_inst_id, - side=side, - sz=str(perp_sz), - td_mode="cross", - pos_side=pos_side, - reduce_only=True, - ) - pf_px = float(perp_live.avg_px) - pf_fee = float(perp_live.fee) - except Exception as e: - return CloseResult( - ok=False, - detail=f"期权已平，永续待平(option_closed_perp_pending): {e}", - ) - - return self._finalize_dual_close( - pos=pos, - group_id=group_id, - option_inst_id=option_inst_id, - opt_qty=opt_qty, - opt_contracts=opt_contracts, - of_px=of_px, - of_fee=of_fee, - of_slip=of_slip, - of_notional=of_notional, - pf_px=pf_px, - pf_fee=pf_fee, - reason=reason, - option_fill_already_written=( - st == "option_closed_perp_pending" - or (pending_perp_only and not is_expiry) - ), - skip_option_cash=( - st == "option_closed_perp_pending" - or (pending_perp_only and not is_expiry) - ), - ) - - def _mark_option_closed_perp_pending( - self, - *, - group_id: str, - option_inst_id: str, - opt_qty: float, - opt_contracts: float, - of_px: float, - of_fee: float, - of_notional: float, - of_slip: float, - reason: str, - ) -> None: - self.ledger.apply_cash( - of_notional - of_fee, - kind="close_option", - group_id=group_id, - note=f"LIVE close option pending perp {reason}", - allow_negative=True, - ) - now = int(time.time() * 1000) - with self.db._lock: - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "close", - "flat", - option_inst_id, - opt_qty, - opt_contracts, - of_px, - of_px, - of_fee, - of_slip, - of_notional, - now, - "LIVE", - ), - ) - self.db._conn.execute( - "UPDATE positions SET status='option_closed_perp_pending' WHERE id=1" - ) - self.db._conn.execute( - "UPDATE groups SET fees=COALESCE(fees,0)+?, note=? WHERE group_id=?", - (of_fee, f"option_closed_perp_pending:{reason}", group_id), - ) - self.db._conn.commit() - - def _finalize_dual_close( - self, - *, - pos: dict, - group_id: str, - option_inst_id: str, - opt_qty: float, - opt_contracts: float, - of_px: float, - of_fee: float, - of_slip: float, - of_notional: float, - pf_px: float, - pf_fee: float, - reason: str, - option_fill_already_written: bool, - skip_option_cash: bool, - ) -> CloseResult: - s = get_settings() - perp_side = str(pos["perp_side"]) - perp_qty = float(pos["perp_qty_eth"]) - opt_entry = float(pos["option_entry_px"]) - perp_entry = float(pos["perp_entry_px"] or pf_px) - opt_pnl = (of_px - opt_entry) * opt_qty - if perp_side == "long": - perp_pnl = (pf_px - perp_entry) * perp_qty - else: - perp_pnl = (perp_entry - pf_px) * perp_qty - - if not skip_option_cash: - self.ledger.apply_cash( - of_notional - of_fee, - kind="close_option", - group_id=group_id, - note=f"LIVE close option {reason}", - allow_negative=True, - ) - self.ledger.apply_cash( - perp_pnl - pf_fee, - kind="close_perp", - group_id=group_id, - note=f"LIVE close perp {reason}", - allow_negative=True, - ) - - now = int(time.time() * 1000) - g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) - base_fees = float((g["fees"] if g else 0) or 0) - fees = base_fees + (0.0 if skip_option_cash else of_fee) + pf_fee - slip = float((g["slip_cost"] if g else 0) or 0) + ( - 0.0 if option_fill_already_written else of_slip - ) - from ..sim.pnl import summarize_fills_pnl - - with self.db._lock: - if not option_fill_already_written: - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "option", - "close", - "flat", - option_inst_id, - opt_qty, - opt_contracts, - of_px, - of_px, - of_fee, - of_slip, - of_notional, - now, - "LIVE", - ), - ) - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "perp", - "close", - "flat", - s.perp_inst_id, - perp_qty, - None, - pf_px, - pf_px, - pf_fee, - 0.0, - pf_px * perp_qty, - now + 1, - "LIVE", - ), - ) - fills = self.db._conn.execute( - "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,) - ).fetchall() - summary = summarize_fills_pnl(list(fills)) - net = summary.get("net_pnl") - if net is None: - net = opt_pnl + perp_pnl - of_fee - pf_fee - self.db._conn.execute( - """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, - fees=?, slip_cost=? WHERE group_id=?""", - ("closed", now, reason, float(net), fees, slip, group_id), - ) - self.db._conn.execute( - """UPDATE positions SET - group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, - option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, - option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' - WHERE id=1""" - ) - self.db._conn.commit() - - from .live_pnl import reconcile_closed_group_pnl - - g2 = self.db.fetchone( - "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?", - (group_id,), - ) - net = reconcile_closed_group_pnl( - db=self.db, - client=self._client(), - exchange="okx", - group_id=group_id, - perp_inst_id=str((g2["perp_inst_id"] if g2 else None) or s.perp_inst_id), - open_at_ms=int(g2["open_at_ms"]) if g2 and g2["open_at_ms"] else None, - local_net=float(net) if net is not None else None, - ) - - return CloseResult( - ok=True, - detail="closed_live", - data={ - "group_id": group_id, - "reason": reason, - "net_pnl": net, - "exec_mode": "LIVE", - "pnl_source": "live_exchange", - }, - ) - - def close_perp_abandon_option( - self, *, reason: str = "target_perp_only", require_deep_otm: bool = True - ) -> CloseResult: - err = self._guard_live() - if err: - return CloseResult(ok=False, detail=err) - if require_deep_otm and not self.option_is_deep_otm(): - return CloseResult(ok=False, detail="期权非远虚，应走双腿全平") - - s = get_settings() - pos = self.current_position() - st = str(pos.get("status") or "") - if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"): - return CloseResult(ok=False, detail="无持仓可平") - # 若期权已平只剩永续，走 close_group 续平即可 - if st == "option_closed_perp_pending": - return self.close_group(reason=reason, bypass_liquidity=True) - - group_id = str(pos["group_id"]) - perp_side = str(pos["perp_side"]) - perp_qty = float(pos["perp_qty_eth"]) - perp_entry = float(pos["perp_entry_px"]) - client = self._client() - try: - ct_val = client.get_ct_val(s.perp_inst_id, inst_type="SWAP") - perp_sz = max(1, int(round(perp_qty / ct_val))) - if perp_side == "long": - side, pos_side = "sell", "long" - else: - side, pos_side = "buy", "short" - perp_live = client.place_market( - inst_id=s.perp_inst_id, - side=side, - sz=str(perp_sz), - td_mode="cross", - pos_side=pos_side, - reduce_only=True, - ) - except Exception as e: - return CloseResult(ok=False, detail=f"实盘平永续失败: {e}") - - pf_px = float(perp_live.avg_px) - pf_fee = float(perp_live.fee) - if perp_side == "long": - perp_pnl = (pf_px - perp_entry) * perp_qty - else: - perp_pnl = (perp_entry - pf_px) * perp_qty - - self.ledger.apply_cash( - perp_pnl - pf_fee, - kind="close_perp", - group_id=group_id, - note=f"LIVE close perp abandon option {reason}", - ) - - # 复用父类归档写入：临时改 fill 路径太重，直接调用父类会再平一次本地假价。 - # 因此把实盘价写入后走父类结构——这里内联父类 abandon 的 DB 段。 - option_inst_id = str(pos["option_inst_id"]) - option_side = str(pos["option_side"]) - strike = self._group_strike(group_id, option_inst_id) - g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) - expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None - expiry_ms = None - if expiry_ymd: - try: - from ..exchange.expiry import expiry_ms_from_ymd - - expiry_ms = int(expiry_ms_from_ymd(expiry_ymd)) - except Exception: - expiry_ms = None - - now = int(time.time() * 1000) - open_fees = float((g["fees"] if g else 0) or 0) - fees = open_fees + pf_fee - slip = float((g["slip_cost"] if g else 0) or 0) - interim_net = perp_pnl - open_fees - pf_fee - spot = self._close_spot_px(get_session().snapshot()) - - with self.db._lock: - self.db._conn.execute( - """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, - base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - "perp", - "close", - "flat", - s.perp_inst_id, - perp_qty, - None, - pf_px, - pf_px, - pf_fee, - 0.0, - pf_px * perp_qty, - now, - "LIVE", - ), - ) - self.db._conn.execute( - """INSERT INTO residual_options( - group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts, - option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px, - initial_premium, status, created_at_ms, note - ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - ( - group_id, - option_inst_id, - option_side, - float(pos["option_qty_eth"]), - float(pos["option_qty_contracts"] or 0), - float(pos["option_entry_px"]), - float(strike) if strike is not None else None, - expiry_ymd, - expiry_ms, - float(pos["entry_index_px"] or 0), - float(pos["initial_premium"] or 0), - "pending", - now, - f"LIVE abandoned after {reason}; spot={spot}", - ), - ) - self.db._conn.execute( - """UPDATE groups SET status=?, close_reason=?, realized_pnl=?, - fees=?, slip_cost=?, note=?, exec_mode=? WHERE group_id=?""", - ( - "option_residual", - reason, - interim_net, - fees, - slip, - "LIVE perp_closed; option residual until expiry", - "LIVE", - group_id, - ), - ) - self.db._conn.execute( - """UPDATE positions SET - group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, - option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, - option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' - WHERE id=1""" - ) - self.db._conn.commit() - - return CloseResult( - ok=True, - detail="perp_closed_option_residual_live", - data={"group_id": group_id, "reason": reason, "mode": "target_perp_only", "exec_mode": "LIVE"}, - ) - - -def get_executor(db=None) -> Matcher: - """按 MODE + 交易所返回执行器。""" - from ..models.db import get_db - - database = db or get_db() - s = get_settings() - if s.is_sim: - return Matcher(database) - ex = load_runtime_settings().exchange - if ex == "binance": - from .binance_executor import BinanceLiveExecutor - - return BinanceLiveExecutor(database) - return OkxLiveExecutor(database) +"""实盘执行：OKX 真下单 + 本地账本/持仓记录（与 Matcher 同结构）。""" + +from __future__ import annotations + +import logging +import time + +from ..config import get_settings +from ..env_store import live_ready +from ..exchange.runtime import load_runtime_settings +from ..models.db import get_db +from ..sim.liquidity import contracts_for_eth, eth_from_contracts +from ..sim.matcher import CloseResult, Matcher, OpenResult +from ..sim.pricing import option_expiry_settle, option_intrinsic +from ..strategy.session import get_session +from .okx_trade import OkxTradeClient +from .symbols import live_settings, resolve_perp_inst_id + +logger = logging.getLogger(__name__) + + +class OkxLiveExecutor(Matcher): + """开平仓走 OKX 私有接口；浮盈/残留逻辑复用 Matcher。""" + + def __init__(self, db=None) -> None: + super().__init__(db) + self._trade: OkxTradeClient | None = None + + def _client(self) -> OkxTradeClient: + if self._trade is None: + self._trade = OkxTradeClient() + return self._trade + + def _guard_live(self) -> str | None: + ok, reason = live_ready() + if not ok: + return reason + return None + + def unrealized(self) -> dict: + base = super().unrealized() + if not base.get("has_position"): + return base + from .live_pnl import enrich_live_unrealized + + gid = base.get("group_id") + open_at = None + perp_inst = resolve_perp_inst_id( + self.db, group_id=str(gid) if gid else None + ) + if gid: + g = self.db.fetchone( + "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?", + (gid,), + ) + if g: + open_at = int(g["open_at_ms"] or 0) or None + if g["perp_inst_id"]: + perp_inst = str(g["perp_inst_id"]) + try: + client = self._client() + except Exception: + return base + return enrich_live_unrealized( + base=base, + db=self.db, + client=client, + exchange="okx", + perp_inst_id=perp_inst, + perp_side=str(base.get("perp_side") or ""), + open_at_ms=open_at, + ) + + def open_group( + self, + *, + group_id: str, + bias: str, + option_side: str, + perp_side: str, + option_inst_id: str, + entry_index_px: float, + strike: float | None = None, + expiry_ymd: str | None = None, + ) -> OpenResult: + err = self._guard_live() + if err: + return OpenResult(ok=False, detail=err) + + s = live_settings() + if self.has_open_position(): + st = self.position_status() + return OpenResult( + ok=False, + detail=f"已有持仓/半仓状态({st})，请先修复或平仓", + ) + + client = self._client() + perp_inst = resolve_perp_inst_id(self.db) + perp_qty = self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth) + opt_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth) + ct_mult = self._ct_mult(option_inst_id) + opt_contracts = contracts_for_eth(opt_qty, ct_mult) + + # 期权：买入，张数 = contracts + try: + opt_fill = client.place_market( + inst_id=option_inst_id, + side="buy", + sz=str(int(round(opt_contracts))), + td_mode="cash", # OKX 期权常见 cash；若账户不同可再扩展 + ) + except Exception as e: + logger.exception("live open option failed") + return OpenResult(ok=False, detail=f"实盘开期权失败: {e}") + + # 以交易所实际成交张数回写名义 + filled_opt_contracts = float(opt_fill.sz) if opt_fill.sz and opt_fill.sz > 0 else float( + int(round(opt_contracts)) + ) + opt_contracts = filled_opt_contracts + opt_qty = eth_from_contracts(opt_contracts, ct_mult) + + # 永续市价：按产品假设，失败原因实质为保证金不足 → 必须回滚期权 + try: + ct_val = client.get_ct_val(perp_inst, inst_type="SWAP") + perp_sz = max(1, int(round(perp_qty / ct_val))) + if perp_side == "long": + side, pos_side = "buy", "long" + else: + side, pos_side = "sell", "short" + perp_fill_live = client.place_market( + inst_id=perp_inst, + side=side, + sz=str(perp_sz), + td_mode="cross", + pos_side=pos_side, + ) + except Exception as e: + logger.exception("live open perp failed (likely margin); rollback option") + try: + client.place_market( + inst_id=option_inst_id, + side="sell", + sz=str(int(round(opt_contracts))), + td_mode="cash", + reduce_only=True, + ) + except Exception as e2: + logger.exception("live option rollback failed: %s", e2) + self._persist_half_open( + group_id=group_id, + bias=bias, + option_side=option_side, + perp_side=perp_side, + option_inst_id=option_inst_id, + entry_index_px=entry_index_px, + strike=strike, + expiry_ymd=expiry_ymd, + opt_qty=opt_qty, + opt_contracts=opt_contracts, + of_px=float(opt_fill.avg_px), + of_fee=float(opt_fill.fee), + detail=f"保证金开永续失败且期权回滚失败: {e} / {e2}", + ) + return OpenResult( + ok=False, + group_id=group_id, + detail=f"永续开仓失败(保证金)且期权回滚失败，已标记 half_open: {e} / {e2}", + ) + return OpenResult( + ok=False, + detail=f"永续开仓失败(多为保证金不足)，已回滚期权: {e}", + ) + + of_px = float(opt_fill.avg_px) + pf_px = float(perp_fill_live.avg_px) + of_fee = float(opt_fill.fee) + pf_fee = float(perp_fill_live.fee) + filled_perp_sz = float(perp_fill_live.sz) if perp_fill_live.sz and perp_fill_live.sz > 0 else float(perp_sz) + perp_qty = filled_perp_sz * float(ct_val) + initial_premium = of_px * opt_qty + of_notional = of_px * opt_qty + pf_notional = pf_px * perp_qty + + # LIVE：交易所已成交，本地账本允许透支镜像，禁止因账本拒记导致「交易所有仓、DB 空」 + self.ledger.apply_cash( + -(of_notional + of_fee), + kind="open_option", + group_id=group_id, + note=f"LIVE open option {group_id}", + allow_negative=True, + ) + self.ledger.apply_cash( + -pf_fee, + kind="open_perp_fee", + group_id=group_id, + note=f"LIVE open perp {group_id}", + allow_negative=True, + ) + + now = int(time.time() * 1000) + with self.db._lock: + self.db._conn.execute( + """INSERT INTO groups( + group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, + strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, + exec_mode + ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "open", + bias, + option_side, + perp_side, + option_inst_id, + perp_inst, + strike, + expiry_ymd, + entry_index_px, + initial_premium, + now, + of_fee + pf_fee, + 0.0, + "LIVE", + ), + ) + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "option", + "open", + "long", + option_inst_id, + opt_qty, + opt_contracts, + of_px, + of_px, + of_fee, + 0.0, + of_notional, + now, + "LIVE", + ), + ) + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "perp", + "open", + perp_side, + perp_inst, + perp_qty, + None, + pf_px, + pf_px, + pf_fee, + 0.0, + pf_notional, + now + 1, + "LIVE", + ), + ) + self.db._conn.execute( + """UPDATE positions SET + group_id=?, perp_side=?, perp_qty_eth=?, perp_entry_px=?, + option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, + option_entry_px=?, entry_index_px=?, initial_premium=?, status=? + WHERE id=1""", + ( + group_id, + perp_side, + perp_qty, + pf_px, + option_inst_id, + option_side, + opt_qty, + opt_contracts, + of_px, + entry_index_px, + initial_premium, + "open", + ), + ) + self.db._conn.commit() + + return OpenResult( + ok=True, + group_id=group_id, + detail="opened_live", + data={ + "group_id": group_id, + "exec_mode": "LIVE", + "option_ord": opt_fill.ord_id, + "perp_ord": perp_fill_live.ord_id, + "initial_premium": initial_premium, + "fees": of_fee + pf_fee, + }, + ) + + def _persist_half_open( + self, + *, + group_id: str, + bias: str, + option_side: str, + perp_side: str, + option_inst_id: str, + entry_index_px: float, + strike: float | None, + expiry_ymd: str | None, + opt_qty: float, + opt_contracts: float, + of_px: float, + of_fee: float, + detail: str, + ) -> None: + """期权已成交、永续未开且回滚失败 → 落 half_open，禁止新开，待 repair。""" + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) + initial_premium = of_px * opt_qty + self.ledger.apply_cash( + -(of_px * opt_qty + of_fee), + kind="open_option", + group_id=group_id, + note=f"LIVE half_open option {group_id}", + allow_negative=True, + ) + now = int(time.time() * 1000) + with self.db._lock: + existing = self.db._conn.execute( + "SELECT group_id FROM groups WHERE group_id=?", (group_id,) + ).fetchone() + if existing is None: + self.db._conn.execute( + """INSERT INTO groups( + group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id, + strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost, + exec_mode, note + ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "half_open", + bias, + option_side, + perp_side, + option_inst_id, + perp_inst, + strike, + expiry_ymd, + entry_index_px, + initial_premium, + now, + of_fee, + 0.0, + "LIVE", + detail[:200], + ), + ) + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "option", + "open", + "long", + option_inst_id, + opt_qty, + opt_contracts, + of_px, + of_px, + of_fee, + 0.0, + of_px * opt_qty, + now, + "LIVE", + ), + ) + self.db._conn.execute( + """UPDATE positions SET + group_id=?, perp_side=?, perp_qty_eth=0, perp_entry_px=NULL, + option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?, + option_entry_px=?, entry_index_px=?, initial_premium=?, status='half_open' + WHERE id=1""", + ( + group_id, + perp_side, + option_inst_id, + option_side, + opt_qty, + opt_contracts, + of_px, + entry_index_px, + initial_premium, + ), + ) + self.db._conn.commit() + + def repair_half_open(self) -> CloseResult: + """卖出 half_open 残留期权，清本地状态。""" + err = self._guard_live() + if err: + return CloseResult(ok=False, detail=err) + pos = self.current_position() + if pos.get("status") != "half_open": + return CloseResult(ok=False, detail="非 half_open 状态") + group_id = str(pos.get("group_id") or "") + option_inst_id = str(pos.get("option_inst_id") or "") + opt_contracts = float(pos.get("option_qty_contracts") or 0) + opt_qty = float(pos.get("option_qty_eth") or 0) + if not option_inst_id or opt_contracts <= 0: + return CloseResult(ok=False, detail="half_open 缺期权合约信息") + client = self._client() + try: + opt_live = client.place_market( + inst_id=option_inst_id, + side="sell", + sz=str(int(round(opt_contracts))), + td_mode="cash", + reduce_only=True, + ) + except Exception as e: + return CloseResult(ok=False, detail=f"half_open 平期权失败: {e}") + of_px = float(opt_live.avg_px) + of_fee = float(opt_live.fee) + of_notional = of_px * opt_qty + opt_entry = float(pos.get("option_entry_px") or of_px) + self.ledger.apply_cash( + of_notional - of_fee, + kind="close_option", + group_id=group_id or None, + note="LIVE repair half_open", + allow_negative=True, + ) + now = int(time.time() * 1000) + with self.db._lock: + if group_id: + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "option", + "close", + "flat", + option_inst_id, + opt_qty, + opt_contracts, + of_px, + of_px, + of_fee, + 0.0, + of_notional, + now, + "LIVE", + ), + ) + opt_pnl = (of_px - opt_entry) * opt_qty - of_fee + self.db._conn.execute( + """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, note=? + WHERE group_id=?""", + ( + "closed", + now, + "half_open_repair", + float(opt_pnl), + "repaired half_open", + group_id, + ), + ) + self.db._conn.execute( + """UPDATE positions SET + group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, + option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, + option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' + WHERE id=1""" + ) + self.db._conn.commit() + return CloseResult( + ok=True, + detail="half_open_repaired", + data={"group_id": group_id, "exec_mode": "LIVE"}, + ) + + def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult: + err = self._guard_live() + if err: + return CloseResult(ok=False, detail=err) + + s = live_settings() + pos = self.current_position() + st = str(pos.get("status") or "") + if st == "half_open": + return self.repair_half_open() + if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"): + return CloseResult(ok=False, detail="无持仓可平") + + group_id = str(pos["group_id"]) + option_inst_id = str(pos["option_inst_id"]) + option_side = str(pos["option_side"]) + perp_side = str(pos["perp_side"]) + opt_qty = float(pos["option_qty_eth"]) + perp_qty = float(pos["perp_qty_eth"]) + opt_contracts = float(pos["option_qty_contracts"] or 0) + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) + client = self._client() + is_expiry = reason == "expiry" + fee_rate = self._fee_rate() + pending_perp_only = st == "option_closed_perp_pending" + + sess = get_session() + snap = sess.snapshot() + strike = self._group_strike(group_id, option_inst_id) + spot = self._close_spot_px(snap) + intrinsic = None + if strike is not None and spot is not None: + intrinsic = option_intrinsic( + option_side=option_side, strike=float(strike), spot=float(spot) + ) + + of_px = 0.0 + of_fee = 0.0 + of_slip = 0.0 + of_notional = 0.0 + + if pending_perp_only: + # 期权已在上次成交并入账；只读上次平期权 fill + prev = self.db.fetchone( + """SELECT fill_px, fee, notional, slip FROM fills + WHERE group_id=? AND leg='option' AND action='close' + ORDER BY id DESC LIMIT 1""", + (group_id,), + ) + if prev is None: + return CloseResult( + ok=False, + detail="option_closed_perp_pending 缺期权平仓记录，请人工核对", + ) + of_px = float(prev["fill_px"]) + of_fee = float(prev["fee"] or 0) + of_notional = float(prev["notional"] or (of_px * opt_qty)) + of_slip = float(prev["slip"] or 0) + else: + # 含到期：优先交易所真实平期权；失败且无内在价值时可本地结算 + try: + opt_live = client.place_market( + inst_id=option_inst_id, + side="sell", + sz=str(max(1, int(round(opt_contracts)))), + td_mode="cash", + reduce_only=True, + ) + of_px = float(opt_live.avg_px) + of_fee = float(opt_live.fee) + filled_c = float(opt_live.sz) if opt_live.sz and opt_live.sz > 0 else opt_contracts + opt_contracts = filled_c + opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id)) + of_notional = of_px * opt_qty + except Exception as e: + if is_expiry and intrinsic is not None: + # 到期后交易所可能已不能交易：用本地结算，仍进入 pending 再平永续 + of = option_expiry_settle( + intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate + ) + of_px, of_fee, of_slip, of_notional = ( + of.fill_px, + of.fee, + of.slip, + of.notional, + ) + logger.warning( + "expiry option exchange close failed, local settle: %s", e + ) + elif not bypass_liquidity: + return CloseResult( + ok=False, + detail=f"实盘平期权失败: {e}", + liquidity_wait=True, + ) + else: + return CloseResult(ok=False, detail=f"实盘平期权失败: {e}") + + # 期权已平（或到期本地结算）：立刻落 pending，避免永续失败后重试再卖期权 + self._mark_option_closed_perp_pending( + group_id=group_id, + option_inst_id=option_inst_id, + opt_qty=opt_qty, + opt_contracts=opt_contracts, + of_px=of_px, + of_fee=of_fee, + of_notional=of_notional, + of_slip=of_slip, + reason=reason, + ) + pending_perp_only = True + + try: + ct_val = client.get_ct_val(perp_inst, inst_type="SWAP") + perp_sz = max(1, int(round(perp_qty / ct_val))) + if perp_side == "long": + side, pos_side = "sell", "long" + else: + side, pos_side = "buy", "short" + perp_live = client.place_market( + inst_id=perp_inst, + side=side, + sz=str(perp_sz), + td_mode="cross", + pos_side=pos_side, + reduce_only=True, + ) + pf_px = float(perp_live.avg_px) + pf_fee = float(perp_live.fee) + except Exception as e: + return CloseResult( + ok=False, + detail=f"期权已平，永续待平(option_closed_perp_pending): {e}", + ) + + return self._finalize_dual_close( + pos=pos, + group_id=group_id, + option_inst_id=option_inst_id, + opt_qty=opt_qty, + opt_contracts=opt_contracts, + of_px=of_px, + of_fee=of_fee, + of_slip=of_slip, + of_notional=of_notional, + pf_px=pf_px, + pf_fee=pf_fee, + reason=reason, + option_fill_already_written=( + st == "option_closed_perp_pending" + or (pending_perp_only and not is_expiry) + ), + skip_option_cash=( + st == "option_closed_perp_pending" + or (pending_perp_only and not is_expiry) + ), + ) + + def _mark_option_closed_perp_pending( + self, + *, + group_id: str, + option_inst_id: str, + opt_qty: float, + opt_contracts: float, + of_px: float, + of_fee: float, + of_notional: float, + of_slip: float, + reason: str, + ) -> None: + self.ledger.apply_cash( + of_notional - of_fee, + kind="close_option", + group_id=group_id, + note=f"LIVE close option pending perp {reason}", + allow_negative=True, + ) + now = int(time.time() * 1000) + with self.db._lock: + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "option", + "close", + "flat", + option_inst_id, + opt_qty, + opt_contracts, + of_px, + of_px, + of_fee, + of_slip, + of_notional, + now, + "LIVE", + ), + ) + self.db._conn.execute( + "UPDATE positions SET status='option_closed_perp_pending' WHERE id=1" + ) + self.db._conn.execute( + "UPDATE groups SET fees=COALESCE(fees,0)+?, note=? WHERE group_id=?", + (of_fee, f"option_closed_perp_pending:{reason}", group_id), + ) + self.db._conn.commit() + + def _finalize_dual_close( + self, + *, + pos: dict, + group_id: str, + option_inst_id: str, + opt_qty: float, + opt_contracts: float, + of_px: float, + of_fee: float, + of_slip: float, + of_notional: float, + pf_px: float, + pf_fee: float, + reason: str, + option_fill_already_written: bool, + skip_option_cash: bool, + ) -> CloseResult: + s = live_settings() + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) + perp_side = str(pos["perp_side"]) + perp_qty = float(pos["perp_qty_eth"]) + opt_entry = float(pos["option_entry_px"]) + perp_entry = float(pos["perp_entry_px"] or pf_px) + opt_pnl = (of_px - opt_entry) * opt_qty + if perp_side == "long": + perp_pnl = (pf_px - perp_entry) * perp_qty + else: + perp_pnl = (perp_entry - pf_px) * perp_qty + + if not skip_option_cash: + self.ledger.apply_cash( + of_notional - of_fee, + kind="close_option", + group_id=group_id, + note=f"LIVE close option {reason}", + allow_negative=True, + ) + self.ledger.apply_cash( + perp_pnl - pf_fee, + kind="close_perp", + group_id=group_id, + note=f"LIVE close perp {reason}", + allow_negative=True, + ) + + now = int(time.time() * 1000) + g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) + base_fees = float((g["fees"] if g else 0) or 0) + fees = base_fees + (0.0 if skip_option_cash else of_fee) + pf_fee + slip = float((g["slip_cost"] if g else 0) or 0) + ( + 0.0 if option_fill_already_written else of_slip + ) + from ..sim.pnl import summarize_fills_pnl + + with self.db._lock: + if not option_fill_already_written: + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "option", + "close", + "flat", + option_inst_id, + opt_qty, + opt_contracts, + of_px, + of_px, + of_fee, + of_slip, + of_notional, + now, + "LIVE", + ), + ) + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "perp", + "close", + "flat", + perp_inst, + perp_qty, + None, + pf_px, + pf_px, + pf_fee, + 0.0, + pf_px * perp_qty, + now + 1, + "LIVE", + ), + ) + fills = self.db._conn.execute( + "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,) + ).fetchall() + summary = summarize_fills_pnl(list(fills)) + net = summary.get("net_pnl") + if net is None: + net = opt_pnl + perp_pnl - of_fee - pf_fee + self.db._conn.execute( + """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, + fees=?, slip_cost=? WHERE group_id=?""", + ("closed", now, reason, float(net), fees, slip, group_id), + ) + self.db._conn.execute( + """UPDATE positions SET + group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, + option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, + option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' + WHERE id=1""" + ) + self.db._conn.commit() + + from .live_pnl import reconcile_closed_group_pnl + + g2 = self.db.fetchone( + "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?", + (group_id,), + ) + net = reconcile_closed_group_pnl( + db=self.db, + client=self._client(), + exchange="okx", + group_id=group_id, + perp_inst_id=str((g2["perp_inst_id"] if g2 else None) or perp_inst), + open_at_ms=int(g2["open_at_ms"]) if g2 and g2["open_at_ms"] else None, + local_net=float(net) if net is not None else None, + ) + + return CloseResult( + ok=True, + detail="closed_live", + data={ + "group_id": group_id, + "reason": reason, + "net_pnl": net, + "exec_mode": "LIVE", + "pnl_source": "live_exchange", + }, + ) + + def close_perp_abandon_option( + self, *, reason: str = "target_perp_only", require_deep_otm: bool = True + ) -> CloseResult: + err = self._guard_live() + if err: + return CloseResult(ok=False, detail=err) + + pos = self.current_position() + st = str(pos.get("status") or "") + if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"): + return CloseResult(ok=False, detail="无持仓可平") + # 若期权已平只剩永续，走 close_group 续平即可 + if st == "option_closed_perp_pending": + return self.close_group(reason=reason, bypass_liquidity=True) + + # 优先尝试双腿全平（含交易所卖期权） + dual = self.close_group(reason=reason, bypass_liquidity=True) + if dual.ok: + return dual + + if require_deep_otm and not self.option_is_deep_otm(): + return CloseResult( + ok=False, + detail=f"期权非远虚且双腿全平失败，应人工处理: {dual.detail}", + ) + + s = live_settings() + group_id = str(pos["group_id"]) + perp_inst = resolve_perp_inst_id(self.db, group_id=group_id) + perp_side = str(pos["perp_side"]) + perp_qty = float(pos["perp_qty_eth"]) + perp_entry = float(pos["perp_entry_px"]) + client = self._client() + try: + ct_val = client.get_ct_val(perp_inst, inst_type="SWAP") + perp_sz = max(1, int(round(perp_qty / ct_val))) + if perp_side == "long": + side, pos_side = "sell", "long" + else: + side, pos_side = "buy", "short" + perp_live = client.place_market( + inst_id=perp_inst, + side=side, + sz=str(perp_sz), + td_mode="cross", + pos_side=pos_side, + reduce_only=True, + ) + except Exception as e: + return CloseResult(ok=False, detail=f"实盘平永续失败: {e}") + + pf_px = float(perp_live.avg_px) + pf_fee = float(perp_live.fee) + if perp_side == "long": + perp_pnl = (pf_px - perp_entry) * perp_qty + else: + perp_pnl = (perp_entry - pf_px) * perp_qty + + self.ledger.apply_cash( + perp_pnl - pf_fee, + kind="close_perp", + group_id=group_id, + note=f"LIVE close perp abandon option {reason}", + ) + + # 复用父类归档写入：临时改 fill 路径太重，直接调用父类会再平一次本地假价。 + # 因此把实盘价写入后走父类结构——这里内联父类 abandon 的 DB 段。 + option_inst_id = str(pos["option_inst_id"]) + option_side = str(pos["option_side"]) + strike = self._group_strike(group_id, option_inst_id) + g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,)) + expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None + expiry_ms = None + if expiry_ymd: + try: + from ..exchange.expiry import expiry_ms_from_ymd + + expiry_ms = int(expiry_ms_from_ymd(expiry_ymd)) + except Exception: + expiry_ms = None + + now = int(time.time() * 1000) + open_fees = float((g["fees"] if g else 0) or 0) + fees = open_fees + pf_fee + slip = float((g["slip_cost"] if g else 0) or 0) + interim_net = perp_pnl - open_fees - pf_fee + spot = self._close_spot_px(get_session().snapshot()) + + with self.db._lock: + self.db._conn.execute( + """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts, + base_px, fill_px, fee, slip, notional, ts_ms, exec_mode) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + "perp", + "close", + "flat", + perp_inst, + perp_qty, + None, + pf_px, + pf_px, + pf_fee, + 0.0, + pf_px * perp_qty, + now, + "LIVE", + ), + ) + self.db._conn.execute( + """INSERT INTO residual_options( + group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts, + option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px, + initial_premium, status, created_at_ms, note + ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + group_id, + option_inst_id, + option_side, + float(pos["option_qty_eth"]), + float(pos["option_qty_contracts"] or 0), + float(pos["option_entry_px"]), + float(strike) if strike is not None else None, + expiry_ymd, + expiry_ms, + float(pos["entry_index_px"] or 0), + float(pos["initial_premium"] or 0), + "pending", + now, + f"LIVE abandoned after {reason}; spot={spot}", + ), + ) + self.db._conn.execute( + """UPDATE groups SET status=?, close_reason=?, realized_pnl=?, + fees=?, slip_cost=?, note=?, exec_mode=? WHERE group_id=?""", + ( + "option_residual", + reason, + interim_net, + fees, + slip, + "LIVE perp_closed; option residual until expiry", + "LIVE", + group_id, + ), + ) + self.db._conn.execute( + """UPDATE positions SET + group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL, + option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0, + option_entry_px=NULL, entry_index_px=NULL, initial_premium=0, status='flat' + WHERE id=1""" + ) + self.db._conn.commit() + + return CloseResult( + ok=True, + detail="perp_closed_option_residual_live", + data={"group_id": group_id, "reason": reason, "mode": "target_perp_only", "exec_mode": "LIVE"}, + ) + + +def get_executor(db=None) -> Matcher: + """按 MODE + 交易所返回执行器。""" + from ..models.db import get_db + + database = db or get_db() + s = get_settings() + if s.is_sim: + return Matcher(database) + ex = load_runtime_settings().exchange + if ex == "binance": + from .binance_executor import BinanceLiveExecutor + + return BinanceLiveExecutor(database) + return OkxLiveExecutor(database) diff --git a/backend/app/live/live_pnl.py b/backend/app/live/live_pnl.py index 945505d..34fe95f 100644 --- a/backend/app/live/live_pnl.py +++ b/backend/app/live/live_pnl.py @@ -72,15 +72,32 @@ def enrich_live_unrealized( logger.warning("live unrealized exchange overlay failed: %s", e) option_upl = float(base.get("option_upl") or 0.0) # 期权净盈亏（本地） + # 保守预估平仓费：按现有名义×费率×2 腿，避免「刚达标、扣费后不够」 + try: + from ..config import get_settings + + fr = float(get_settings().fee_rate or 0.0005) + except Exception: + fr = 0.0005 + notional_est = abs(float(base.get("perp_notional") or 0.0)) + abs( + float(base.get("option_notional") or 0.0) + ) + if notional_est <= 0: + # 兜底：用标记价粗算 + notional_est = abs(float(base.get("spot") or 0.0)) * ( + abs(float(base.get("perp_qty_eth") or 0.0)) + + abs(float(base.get("option_qty_eth") or 0.0)) + ) + est_close = max(0.0, notional_est * fr * 2.0) # 资金费 signed：付出为负，直接加总 - net_pnl = perp_upl + option_upl - fees_paid + funding + net_pnl = perp_upl + option_upl - fees_paid + funding - est_close out = dict(base) out["perp_upl"] = perp_upl out["option_upl"] = option_upl out["fees_paid"] = fees_paid out["funding_usdt"] = funding - out["est_close_fees"] = 0.0 # LIVE 不估平仓费 + out["est_close_fees"] = est_close out["net_pnl"] = net_pnl out["pnl_source"] = "live_exchange" return out diff --git a/backend/app/live/okx_trade.py b/backend/app/live/okx_trade.py index 8f75e83..79ca047 100644 --- a/backend/app/live/okx_trade.py +++ b/backend/app/live/okx_trade.py @@ -126,9 +126,7 @@ class OkxTradeClient: if v and v > 0: self._ct_val_cache[inst_id] = float(v) return float(v) - default = 0.01 - self._ct_val_cache[inst_id] = default - return default + raise RuntimeError(f"OKX 无法取得合约面值 ctVal: {inst_id} instType={inst_type}") def place_market( self, @@ -159,7 +157,7 @@ class OkxTradeClient: fill = self._wait_fill(inst_id, ord_id) return fill - def _wait_fill(self, inst_id: str, ord_id: str, *, tries: int = 8) -> LiveFill: + def _wait_fill(self, inst_id: str, ord_id: str, *, tries: int = 20) -> LiveFill: path = f"/api/v5/trade/order?instId={inst_id}&ordId={ord_id}" last: dict[str, Any] = {} for _ in range(tries): @@ -168,11 +166,11 @@ class OkxTradeClient: last = rows[0] state = str(last.get("state") or "") avg = safe_float(last.get("avgPx")) - if state in ("filled", "partially_filled") and avg and avg > 0: + # 仅完全成交；部分成交继续等，避免账本张数与交易所不一致 + if state == "filled" and avg and avg > 0: sz = safe_float(last.get("accFillSz")) or safe_float(last.get("sz")) or 0.0 fee = abs(safe_float(last.get("fee")) or 0.0) fee_ccy = str(last.get("feeCcy") or "USDT") - # 订单上 fee 常为空，再查成交明细 if fee <= 0 and ord_id: fee, fee_ccy = self.sum_fill_fees(inst_id, ord_id) from .money import abs_fee_usdt @@ -188,8 +186,8 @@ class OkxTradeClient: ) if state in ("canceled", "failed"): raise RuntimeError(f"OKX 订单失败 state={state} {last}") - time.sleep(0.25) - raise RuntimeError(f"OKX 订单未成交 ordId={ord_id} last={last}") + time.sleep(0.3) + raise RuntimeError(f"OKX 订单未完全成交 ordId={ord_id} last={last}") def sum_fill_fees(self, inst_id: str, ord_id: str) -> tuple[float, str]: """成交明细手续费合计（原币种金额, 币种）。""" diff --git a/backend/app/live/symbols.py b/backend/app/live/symbols.py new file mode 100644 index 0000000..5ff3c58 --- /dev/null +++ b/backend/app/live/symbols.py @@ -0,0 +1,26 @@ +"""LIVE 下单用的运行时合约解析（禁止只用 env 默认 perp_inst_id）。""" + +from __future__ import annotations + +from ..config import Settings +from ..exchange.runtime import load_runtime_settings + + +def live_settings() -> Settings: + return load_runtime_settings() + + +def resolve_perp_inst_id(db, *, group_id: str | None = None) -> str: + """优先组内落库合约，否则 DB/交易所默认（load_runtime_settings）。""" + s = live_settings() + if group_id: + try: + row = db.fetchone( + "SELECT perp_inst_id FROM groups WHERE group_id=?", + (group_id,), + ) + if row and row["perp_inst_id"]: + return str(row["perp_inst_id"]) + except Exception: + pass + return str(s.perp_inst_id) diff --git a/backend/app/main.py b/backend/app/main.py index 4c14fa3..ec26bec 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -39,6 +39,17 @@ async def lifespan(app: FastAPI): settings = load_runtime_settings() engine = StrategyEngine() set_engine(engine) + # LIVE：进程启动后不自动真下单，须人工点「启动」 + if not get_settings().is_sim: + try: + db._conn.execute( + "UPDATE strategy_state SET running=0, phase=? WHERE id=1", + ("paused",), + ) + db._conn.commit() + logger.info("LIVE startup: forced strategy pause (manual start required)") + except Exception: + logger.exception("LIVE startup force-pause failed") engine.ensure_loop() session = bootstrap_session(settings) @@ -74,14 +85,17 @@ async def lifespan(app: FastAPI): app = FastAPI( title="比特骆驼自动化对冲系统", - version="0.3.0", + version="0.3.1", description="比特骆驼自动化对冲系统（eth_hedge_sim）", lifespan=lifespan, + docs_url=None if get_settings().disable_api_docs else "/docs", + redoc_url=None if get_settings().disable_api_docs else "/redoc", + openapi_url=None if get_settings().disable_api_docs else "/openapi.json", ) app.add_middleware( CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, + allow_origins=[], + allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) diff --git a/backend/tests/test_runtime_mode.py b/backend/tests/test_runtime_mode.py index ffa87f7..93522e7 100644 --- a/backend/tests/test_runtime_mode.py +++ b/backend/tests/test_runtime_mode.py @@ -6,11 +6,8 @@ from app.env_store import live_ready, mask_secret, okx_keys_configured def test_mask_secret() -> None: assert mask_secret(None) is None assert mask_secret("") is None - assert mask_secret("abcd") == "****" - m = mask_secret("abcdefghij") - assert m is not None - assert m.endswith("ghij") - assert m.startswith("*") + assert mask_secret("abcd") == "********" + assert mask_secret("abcdefghij") == "********" def test_live_ready_sim(monkeypatch) -> None: diff --git a/backend/tests/test_symbols.py b/backend/tests/test_symbols.py new file mode 100644 index 0000000..57840af --- /dev/null +++ b/backend/tests/test_symbols.py @@ -0,0 +1,33 @@ +"""LIVE 合约解析。""" + +from app.live.symbols import resolve_perp_inst_id + + +class _FakeDb: + def __init__(self, row=None): + self._row = row + + def fetchone(self, sql, args=()): + return self._row + + +def test_resolve_perp_prefers_group(monkeypatch) -> None: + import app.live.symbols as sym + + class S: + perp_inst_id = "ETH-USDT-SWAP" + + monkeypatch.setattr(sym, "live_settings", lambda: S()) + db = _FakeDb({"perp_inst_id": "ETHUSDT"}) + assert resolve_perp_inst_id(db, group_id="g1") == "ETHUSDT" + + +def test_resolve_perp_fallback_runtime(monkeypatch) -> None: + import app.live.symbols as sym + + class S: + perp_inst_id = "ETHUSDT" + + monkeypatch.setattr(sym, "live_settings", lambda: S()) + db = _FakeDb(None) + assert resolve_perp_inst_id(db) == "ETHUSDT" diff --git a/docs/更新说明.md b/docs/更新说明.md new file mode 100644 index 0000000..bb560a5 --- /dev/null +++ b/docs/更新说明.md @@ -0,0 +1,51 @@ +# 比特骆驼自动化对冲系统 — 更新说明 + +> 每次发版 / 合并重要改动后，在本文档顶部追加一条（日期新→旧）。 +> 关联审计结论写在同条「审计」小节。 + +--- + +## 2026-07-26 — 审计修复包（安全 / 下单 / 平仓） + +### 变更 + +1. **LIVE 永续合约号**：开平仓统一用 `load_runtime_settings()` / 组内 `perp_inst_id`，修复 UI 切币安后仍用 env `ETH-USDT-SWAP` 的问题。 +2. **到期平仓**：期权先走交易所真实平仓（失败且仓位已空才本地结算），并在永续失败前写入 `option_closed_perp_pending`。 +3. **成交完整性**：OKX/币安仅接受完全成交；账本数量按交易所 `accFillSz`/`executedQty` 回写。 +4. **`ctVal`**：查不到合约面值时直接失败，禁止静默默认 `0.01`。 +5. **放弃期权路径**：优先交易所卖期权；仅深虚且卖不出时才残留归档。 +6. **LIVE 重启**：启动时若为 LIVE，自动暂停策略，须人工再点启动。 +7. **安全**：登录限流；切 LIVE 须服务端校验确认串 `LIVE`；生产关闭 `/docs`；改密后抬升 token 版本作废旧票；`.env` 写入后尽量 `chmod 600`；CORS 改为同源优先。 +8. **出场**：LIVE 净利加入保守平仓手续费缓冲，减少「刚达标但扣费后不够」的提前出场。 + +### 审计（本包改后复审） + +| 项 | 状态 | +|----|------| +| 永续符号 env/DB 不一致 | **已修**（`symbols.resolve_perp_inst_id` + 开平仓全路径） | +| 到期只本地结算期权 | **已修**（先交易所卖；失败才本地结算） | +| 到期永续失败无 pending | **已修**（统一 `_mark_option_closed_perp_pending`） | +| partially_filled 当完成 | **已修**（OKX 仅 `filled`；币安仅 `FILLED`） | +| 放弃路径不卖期权 | **已缓解**（先 `close_group` 双腿；失败且深虚才残留） | +| 开仓崩溃双开 | **部分**：LIVE 启动强制 `running=0`；仍缺交易所对账 | +| ctVal 静默默认 | **已修**（查不到直接报错） | +| LIVE 出场不计平仓费 | **已修**（保守 `est_close_fees`） | +| 登录无限流 / LIVE 可 API 绕过确认 | **已修**（限流 + `confirm_live_phrase=LIVE`） | +| `/docs` 公开 / CORS `*` / mask 泄露后缀 | **已修** | +| 改密后旧 token | **已修**（`AUTH_TOKEN_VERSION`） | +| 默认口令 / 公网 TLS | **未改默认口令**（部署须改 `.env`）；TLS 仍靠反代 | +| 杠杆 set_leverage | **未做** | +| 平仓张数读交易所持仓 | **未做** | + +### 残留风险（下次优先） + +1. 启动时交易所↔本地仓位对账与开仓幂等键 +2. 平仓前查询交易所实际持仓张数 +3. 开仓前 `set_leverage` +4. 可选：拒绝默认 `AUTH_*` 启动（硬开关） + +### 测试 + +- `pytest`：46 passed（含 `test_symbols`、脱敏用例更新） + +--- diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 9fe9e50..726c251 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -210,6 +210,7 @@ export default function SettingsPage() { const body: Record = { mode, confirm_live: goingLive, + confirm_live_phrase: goingLive ? "LIVE" : undefined, }; if (okxKey.trim()) body.okx_api_key = okxKey.trim(); if (okxSecret.trim()) body.okx_api_secret = okxSecret.trim();