diff --git a/backend/app/live/binance_executor.py b/backend/app/live/binance_executor.py index 3640a6a..d4358dd 100644 --- a/backend/app/live/binance_executor.py +++ b/backend/app/live/binance_executor.py @@ -12,6 +12,12 @@ 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 .reconcile import ( + assert_safe_to_open_live, + claim_open_slot, + perp_close_qty_eth_binance, + release_open_slot_if_opening, +) from .symbols import live_settings, resolve_perp_inst_id logger = logging.getLogger(__name__) @@ -83,14 +89,16 @@ class BinanceLiveExecutor(Matcher): 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})，请先修复或平仓", - ) + claimed, claim_msg = claim_open_slot(self.db) + if not claimed: + return OpenResult(ok=False, detail=claim_msg) + safe, safe_msg = assert_safe_to_open_live(self) + if not safe: + release_open_slot_if_opening(self.db) + return OpenResult(ok=False, detail=safe_msg) + + s = live_settings() 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) @@ -106,6 +114,7 @@ class BinanceLiveExecutor(Matcher): ) except Exception as e: logger.exception("binance live open option failed") + release_open_slot_if_opening(self.db) 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( @@ -120,6 +129,11 @@ class BinanceLiveExecutor(Matcher): side, pos_side = "BUY", "LONG" else: side, pos_side = "SELL", "SHORT" + leverage = self.ledger.get_setting_float("leverage", s.leverage) + try: + client.set_leverage(perp_inst, leverage) + except Exception as e_lev: + logger.warning("binance set_leverage failed: %s", e_lev) perp_fill_live = client.place_perp_market( symbol=perp_inst, side=side, @@ -157,6 +171,7 @@ class BinanceLiveExecutor(Matcher): group_id=group_id, detail=f"永续开仓失败(保证金)且期权回滚失败，已标记 half_open: {e} / {e2}", ) + release_open_slot_if_opening(self.db) return OpenResult( ok=False, detail=f"永续开仓失败(多为保证金不足)，已回滚期权: {e}", @@ -598,10 +613,16 @@ class BinanceLiveExecutor(Matcher): side, pos_side = "SELL", "LONG" else: side, pos_side = "BUY", "SHORT" + perp_qty_close = perp_close_qty_eth_binance( + client, + perp_inst=perp_inst, + perp_side=perp_side, + perp_qty_eth=perp_qty, + ) perp_live = client.place_perp_market( symbol=perp_inst, side=side, - qty_eth=perp_qty, + qty_eth=perp_qty_close, position_side=pos_side, reduce_only=True, ) @@ -901,10 +922,16 @@ class BinanceLiveExecutor(Matcher): side, pos_side = "SELL", "LONG" else: side, pos_side = "BUY", "SHORT" + perp_qty_close = perp_close_qty_eth_binance( + client, + perp_inst=perp_inst, + perp_side=perp_side, + perp_qty_eth=perp_qty, + ) perp_live = client.place_perp_market( symbol=perp_inst, side=side, - qty_eth=perp_qty, + qty_eth=perp_qty_close, position_side=pos_side, reduce_only=True, ) diff --git a/backend/app/live/binance_trade.py b/backend/app/live/binance_trade.py index 8530e92..c8d17a5 100644 --- a/backend/app/live/binance_trade.py +++ b/backend/app/live/binance_trade.py @@ -314,6 +314,39 @@ class BinanceTradeClient: return to_usdt(float(upl), "USDT") return 0.0 + def get_perp_pos_sz(self, symbol: str, *, position_side: str | None = None) -> float | None: + """当前永续绝对持仓（ETH）。""" + try: + rows = self._signed( + self._fapi, "GET", "/fapi/v2/positionRisk", {"symbol": symbol} + ) + except Exception as e: + logger.warning("binance get_perp_pos_sz failed: %s", e) + return None + if isinstance(rows, dict): + rows = [rows] + want = (position_side or "").strip().upper() + for row in rows: + amt = safe_float(row.get("positionAmt")) or 0.0 + if abs(amt) < 1e-12: + continue + ps = str(row.get("positionSide") or "").upper() + if want and ps and ps not in ("BOTH",) and ps != want: + continue + return abs(float(amt)) + return 0.0 + + def set_leverage(self, symbol: str, leverage: int | float) -> None: + lev = int(round(float(leverage))) + if lev < 1: + lev = 1 + self._signed( + self._fapi, + "POST", + "/fapi/v1/leverage", + {"symbol": symbol, "leverage": lev}, + ) + def get_funding_usdt( self, symbol: str, *, begin_ms: int, end_ms: int | None = None ) -> float: diff --git a/backend/app/live/executor.py b/backend/app/live/executor.py index b1371c2..c2a1f94 100644 --- a/backend/app/live/executor.py +++ b/backend/app/live/executor.py @@ -14,6 +14,12 @@ 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 .reconcile import ( + assert_safe_to_open_live, + claim_open_slot, + perp_close_contracts_okx, + release_open_slot_if_opening, +) from .symbols import live_settings, resolve_perp_inst_id logger = logging.getLogger(__name__) @@ -87,14 +93,16 @@ class OkxLiveExecutor(Matcher): 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})，请先修复或平仓", - ) + claimed, claim_msg = claim_open_slot(self.db) + if not claimed: + return OpenResult(ok=False, detail=claim_msg) + safe, safe_msg = assert_safe_to_open_live(self) + if not safe: + release_open_slot_if_opening(self.db) + return OpenResult(ok=False, detail=safe_msg) + + s = live_settings() 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) @@ -112,6 +120,7 @@ class OkxLiveExecutor(Matcher): ) except Exception as e: logger.exception("live open option failed") + release_open_slot_if_opening(self.db) return OpenResult(ok=False, detail=f"实盘开期权失败: {e}") # 以交易所实际成交张数回写名义 @@ -124,11 +133,24 @@ class OkxLiveExecutor(Matcher): # 永续市价：按产品假设，失败原因实质为保证金不足 → 必须回滚期权 try: ct_val = client.get_ct_val(perp_inst, inst_type="SWAP") - perp_sz = max(1, int(round(perp_qty / ct_val))) + perp_sz = perp_close_contracts_okx( + client, + perp_inst=perp_inst, + perp_side=perp_side, + perp_qty_eth=perp_qty, + ct_val=ct_val, + ) if perp_side == "long": side, pos_side = "buy", "long" else: side, pos_side = "sell", "short" + leverage = self.ledger.get_setting_float("leverage", s.leverage) + try: + client.set_leverage( + perp_inst, leverage, mgn_mode="cross", pos_side=pos_side + ) + except Exception as e_lev: + logger.warning("okx set_leverage failed: %s", e_lev) perp_fill_live = client.place_market( inst_id=perp_inst, side=side, @@ -168,6 +190,7 @@ class OkxLiveExecutor(Matcher): group_id=group_id, detail=f"永续开仓失败(保证金)且期权回滚失败，已标记 half_open: {e} / {e2}", ) + release_open_slot_if_opening(self.db) return OpenResult( ok=False, detail=f"永续开仓失败(多为保证金不足)，已回滚期权: {e}", @@ -604,7 +627,13 @@ class OkxLiveExecutor(Matcher): try: ct_val = client.get_ct_val(perp_inst, inst_type="SWAP") - perp_sz = max(1, int(round(perp_qty / ct_val))) + perp_sz = perp_close_contracts_okx( + client, + perp_inst=perp_inst, + perp_side=perp_side, + perp_qty_eth=perp_qty, + ct_val=ct_val, + ) if perp_side == "long": side, pos_side = "sell", "long" else: @@ -883,7 +912,13 @@ class OkxLiveExecutor(Matcher): 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))) + perp_sz = perp_close_contracts_okx( + client, + perp_inst=perp_inst, + perp_side=perp_side, + perp_qty_eth=perp_qty, + ct_val=ct_val, + ) if perp_side == "long": side, pos_side = "sell", "long" else: diff --git a/backend/app/live/okx_trade.py b/backend/app/live/okx_trade.py index 79ca047..313578e 100644 --- a/backend/app/live/okx_trade.py +++ b/backend/app/live/okx_trade.py @@ -232,6 +232,43 @@ class OkxTradeClient: return to_usdt(float(upl), ccy) return 0.0 + def get_perp_pos_sz(self, inst_id: str, *, pos_side: str | None = None) -> float | None: + """当前永续绝对持仓张数。""" + try: + rows = self._request( + "GET", f"/api/v5/account/positions?instId={inst_id}" + ) + except Exception as e: + logger.warning("okx get_perp_pos_sz failed: %s", e) + return None + want = (pos_side or "").strip().lower() + for row in rows: + ps = str(row.get("posSide") or "").lower() + pos = safe_float(row.get("pos")) or 0.0 + if abs(pos) < 1e-12: + continue + if want and want not in ("net", "") and ps and ps != want and ps != "net": + continue + return abs(float(pos)) + return 0.0 + + def set_leverage( + self, + inst_id: str, + leverage: float, + *, + mgn_mode: str = "cross", + pos_side: str | None = None, + ) -> None: + body: dict[str, Any] = { + "instId": inst_id, + "lever": str(leverage), + "mgnMode": mgn_mode, + } + if pos_side: + body["posSide"] = pos_side + self._request("POST", "/api/v5/account/set-leverage", body) + def get_funding_usdt( self, inst_id: str, *, begin_ms: int, end_ms: int | None = None ) -> float: diff --git a/backend/app/live/reconcile.py b/backend/app/live/reconcile.py new file mode 100644 index 0000000..b4ab094 --- /dev/null +++ b/backend/app/live/reconcile.py @@ -0,0 +1,148 @@ +"""LIVE 开仓对账：占槽、交易所持仓核对、平永续数量解析。""" + +from __future__ import annotations + +import logging +from typing import Any + +from ..config import get_settings +from ..exchange.runtime import load_runtime_settings +from ..sim.matcher import BLOCKING_STATUSES +from .symbols import resolve_perp_inst_id + +logger = logging.getLogger(__name__) + +_PERP_EPS = 1e-8 + + +def claim_open_slot(db) -> tuple[bool, str]: + """原子占用开仓槽：flat/空 → opening；已有 blocking 状态则拒绝。""" + with db._lock: + row = db._conn.execute("SELECT status FROM positions WHERE id=1").fetchone() + st = str(row["status"] or "") if row else "" + if st in BLOCKING_STATUSES: + return False, f"已有持仓/半仓状态({st})，请先修复或平仓" + cur = db._conn.execute( + """UPDATE positions SET status='opening' + WHERE id=1 AND (status IS NULL OR status='' OR status='flat')""" + ) + if cur.rowcount != 1: + st2 = st or "unknown" + return False, f"无法占用开仓槽(当前 status={st2})" + db._conn.commit() + return True, "ok" + + +def release_open_slot_if_opening(db) -> None: + """开仓失败且未落 half_open/open 时，释放 opening 占槽。""" + with db._lock: + db._conn.execute( + "UPDATE positions SET status='flat' WHERE id=1 AND status='opening'" + ) + db._conn.commit() + + +def exchange_perp_abs_size( + client: Any, + exchange: str, + perp_inst_id: str, + perp_side: str, +) -> float | None: + """查询交易所永续绝对持仓：OKX 张数，Binance ETH。""" + ex = (exchange or "").strip().lower() + try: + if ex == "binance": + ps = "LONG" if perp_side == "long" else "SHORT" + return client.get_perp_pos_sz(perp_inst_id, position_side=ps) + ps = "long" if perp_side == "long" else "short" + return client.get_perp_pos_sz(perp_inst_id, pos_side=ps) + except Exception as e: + logger.warning("exchange_perp_abs_size failed exchange=%s: %s", ex, e) + return None + + +def assert_safe_to_open_live(executor) -> tuple[bool, str]: + """LIVE 开仓前：本地无 blocking 仓，且交易所无残留永续（flat/opening 时）。""" + if get_settings().is_sim: + return True, "ok" + + pos = executor.current_position() + st = str(pos.get("status") or "") + if st in BLOCKING_STATUSES and st != "opening": + return False, f"已有持仓/半仓状态({st})，请先修复或平仓" + + client, ex_name = _executor_client_and_exchange(executor) + if client is None: + return False, "无法核对交易所持仓" + + perp_inst = resolve_perp_inst_id(executor.db) + perp_side = str(pos.get("perp_side") or "long") + ex_sz = exchange_perp_abs_size(client, ex_name or "", perp_inst, perp_side) + if ex_sz is None: + return False, "无法核对交易所持仓" + + if ex_sz > _PERP_EPS and st in ("flat", "", "opening"): + return ( + False, + "交易所有永续仓但本地无持仓，禁止新开，请人工核对", + ) + return True, "ok" + + +def log_exchange_db_mismatch(executor) -> None: + """LIVE 启动时记录交易所 vs 本地持仓不一致（仅日志，不阻断）。""" + if get_settings().is_sim: + return + ok, msg = assert_safe_to_open_live(executor) + if ok: + logger.info("LIVE startup reconcile: exchange/DB perp OK") + else: + logger.warning("LIVE startup reconcile mismatch: %s", msg) + + +def perp_close_contracts_okx( + client: Any, + *, + perp_inst: str, + perp_side: str, + perp_qty_eth: float, + ct_val: float, +) -> int: + """平永续张数：优先交易所持仓，否则 DB qty/ct_val。""" + ps = "long" if perp_side == "long" else "short" + ex_sz = client.get_perp_pos_sz(perp_inst, pos_side=ps) + if ex_sz is not None and ex_sz > _PERP_EPS: + return max(1, int(round(ex_sz))) + return max(1, int(round(perp_qty_eth / ct_val))) + + +def perp_close_qty_eth_binance( + client: Any, + *, + perp_inst: str, + perp_side: str, + perp_qty_eth: float, +) -> float: + """平永续 ETH 数量：优先交易所持仓，否则 DB qty。""" + ps = "LONG" if perp_side == "long" else "SHORT" + ex_sz = client.get_perp_pos_sz(perp_inst, position_side=ps) + if ex_sz is not None and ex_sz > _PERP_EPS: + return float(ex_sz) + return float(perp_qty_eth) + + +def _executor_client_and_exchange(executor) -> tuple[Any | None, str | None]: + if get_settings().is_sim: + return None, None + try: + ex_name = load_runtime_settings().exchange + except Exception: + ex_name = get_settings().exchange + client = None + if hasattr(executor, "_client"): + try: + client = executor._client() + except Exception as e: + logger.warning("live executor client unavailable: %s", e) + return None, ex_name + return client, ex_name diff --git a/backend/app/main.py b/backend/app/main.py index ec26bec..d17f517 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -50,6 +50,13 @@ async def lifespan(app: FastAPI): logger.info("LIVE startup: forced strategy pause (manual start required)") except Exception: logger.exception("LIVE startup force-pause failed") + try: + from .live import get_executor + from .live.reconcile import log_exchange_db_mismatch + + log_exchange_db_mismatch(get_executor(db)) + except Exception: + logger.exception("LIVE startup reconcile log failed") engine.ensure_loop() session = bootstrap_session(settings) diff --git a/backend/app/sim/matcher.py b/backend/app/sim/matcher.py index 38d091c..50192cd 100644 --- a/backend/app/sim/matcher.py +++ b/backend/app/sim/matcher.py @@ -23,7 +23,7 @@ from .pricing import ( # 禁止新开仓的本地仓位状态（实盘防卡） BLOCKING_STATUSES = frozenset( - {"open", "half_open", "option_closed_perp_pending"} + {"open", "half_open", "option_closed_perp_pending", "opening"} ) @@ -72,11 +72,13 @@ class Matcher: return dict(row) def has_open_position(self) -> bool: - """是否禁止新开：含 open / half_open / option_closed_perp_pending。""" + """是否禁止新开：含 open / half_open / option_closed_perp_pending / opening。""" pos = self.current_position() st = str(pos.get("status") or "") if st not in BLOCKING_STATUSES: return False + if st == "opening": + return True return bool(pos.get("group_id") or pos.get("option_inst_id")) def position_status(self) -> str: diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index 4a6f989..5d3cb1d 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -513,6 +513,13 @@ class StrategyEngine: pick.pair.call_inst_id if pick.option_side == "call" else pick.pair.put_inst_id ) entry_idx = pick.underlying_px + if not get_settings().is_sim: + from ..live.reconcile import assert_safe_to_open_live + + safe, safe_msg = assert_safe_to_open_live(self.matcher) + if not safe: + self._set_state(phase="idle", last_error=safe_msg) + return r = await asyncio.to_thread( self.matcher.open_group, group_id=gid, diff --git a/backend/tests/test_live_recovery.py b/backend/tests/test_live_recovery.py index 30a6840..d705474 100644 --- a/backend/tests/test_live_recovery.py +++ b/backend/tests/test_live_recovery.py @@ -10,6 +10,7 @@ def test_blocking_statuses_include_repair_states() -> None: assert "half_open" in BLOCKING_STATUSES assert "option_closed_perp_pending" in BLOCKING_STATUSES assert "open" in BLOCKING_STATUSES + assert "opening" in BLOCKING_STATUSES def test_has_open_position_blocks_half_open(tmp_path, monkeypatch) -> None: diff --git a/backend/tests/test_reconcile_claim.py b/backend/tests/test_reconcile_claim.py new file mode 100644 index 0000000..e78c5d9 --- /dev/null +++ b/backend/tests/test_reconcile_claim.py @@ -0,0 +1,72 @@ +"""claim_open_slot / release_open_slot 单元测试。""" + +from __future__ import annotations + +from app.live.reconcile import claim_open_slot, release_open_slot_if_opening +from app.sim.matcher import BLOCKING_STATUSES, Matcher + + +def test_blocking_statuses_include_opening() -> None: + assert "opening" in BLOCKING_STATUSES + + +def test_claim_open_slot_from_flat(tmp_path) -> None: + from app.models.db import Database + + db = Database(tmp_path / "claim.db") + m = Matcher(db) + assert m.has_open_position() is False + + ok, msg = claim_open_slot(db) + assert ok is True + assert msg == "ok" + assert m.position_status() == "opening" + assert m.has_open_position() is True + + ok2, _ = claim_open_slot(db) + assert ok2 is False + + release_open_slot_if_opening(db) + assert m.position_status() == "flat" + assert m.has_open_position() is False + + ok3, _ = claim_open_slot(db) + assert ok3 is True + release_open_slot_if_opening(db) + db.close() + + +def test_claim_rejects_blocking_states(tmp_path) -> None: + from app.models.db import Database + + db = Database(tmp_path / "block.db") + for st in ("open", "half_open", "option_closed_perp_pending"): + with db._lock: + db._conn.execute( + "UPDATE positions SET status=?, group_id=? WHERE id=1", + (st, "G-test"), + ) + db._conn.commit() + ok, msg = claim_open_slot(db) + assert ok is False + assert st in msg + with db._lock: + db._conn.execute( + "UPDATE positions SET status='flat', group_id=NULL WHERE id=1" + ) + db._conn.commit() + db.close() + + +def test_release_only_when_opening(tmp_path) -> None: + from app.models.db import Database + + db = Database(tmp_path / "rel.db") + with db._lock: + db._conn.execute("UPDATE positions SET status='open' WHERE id=1") + db._conn.commit() + release_open_slot_if_opening(db) + row = db.fetchone("SELECT status FROM positions WHERE id=1") + assert row is not None + assert row["status"] == "open" + db.close()