"""LIVE 开仓对账:占槽、交易所持仓核对、平永续数量解析、stuck opening 恢复。""" 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, CloseResult from .symbols import resolve_perp_inst_id logger = logging.getLogger(__name__) _PERP_EPS = 1e-8 _OPT_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 stamp_opening_intent( db, *, group_id: str, option_inst_id: str, option_side: str, perp_side: str, option_qty_eth: float, option_qty_contracts: float, entry_index_px: float | None = None, option_entry_px: float | None = None, ) -> None: """开仓意图落库:崩溃后仍可按 option_inst_id 恢复,禁止「opening 无元数据」。""" with db._lock: self_row = db._conn.execute( "SELECT status FROM positions WHERE id=1" ).fetchone() st = str(self_row["status"] or "") if self_row else "" if st != "opening": return db._conn.execute( """UPDATE positions SET group_id=?, option_inst_id=?, option_side=?, perp_side=?, option_qty_eth=?, option_qty_contracts=?, entry_index_px=COALESCE(?, entry_index_px), option_entry_px=COALESCE(?, option_entry_px), status='opening' WHERE id=1 AND status='opening'""", ( group_id, option_inst_id, option_side, perp_side, float(option_qty_eth), float(option_qty_contracts), entry_index_px, option_entry_px, ), ) db._conn.commit() def release_open_slot_if_opening(db) -> None: """开仓失败且未落 half_open/open 时,释放 opening 占槽。""" with db._lock: db._conn.execute( """UPDATE positions SET status='flat', group_id=NULL, option_inst_id=NULL, option_side=NULL, perp_side=NULL, option_qty_eth=0, option_qty_contracts=0, option_entry_px=NULL, perp_qty_eth=0, perp_entry_px=NULL, entry_index_px=NULL, initial_premium=0 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 in ("binance", "bn"): 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 exchange_option_abs_size( client: Any, option_inst_id: str ) -> float | None: try: return client.get_option_pos_sz(option_inst_id) except Exception as e: logger.warning("exchange_option_abs_size failed: %s", e) return None def exchange_any_option_abs(client: Any) -> float | None: try: return client.any_option_pos_abs() except Exception as e: logger.warning("exchange_any_option_abs failed: %s", 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) if st in ("flat", "", "opening"): sides = ("long", "short") else: sides = (str(pos.get("perp_side") or "long"),) total = 0.0 for side in sides: ex_sz = exchange_perp_abs_size(client, ex_name or "", perp_inst, side) if ex_sz is None: return False, "无法核对交易所持仓" total += float(ex_sz) if total > _PERP_EPS and st in ("flat", "", "opening"): return ( False, "交易所有永续仓但本地无持仓,禁止新开,请人工核对", ) # 期权:有具体合约则查该合约;flat 时查账户任意期权残留 opt_inst = str(pos.get("option_inst_id") or "") if opt_inst: opt_sz = exchange_option_abs_size(client, opt_inst) if opt_sz is None: return False, "无法核对交易所期权持仓" if opt_sz > _OPT_EPS and st in ("flat", "", "opening"): return ( False, f"交易所有期权仓({opt_inst})但本地未确认持仓,禁止新开,请人工核对", ) elif st in ("flat", "", "opening"): # opening 且尚未 stamp option_inst_id 时仍须扫任意期权残留 any_opt = exchange_any_option_abs(client) if any_opt is None: return False, "无法核对交易所期权持仓" if any_opt > _OPT_EPS: 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 OK") else: logger.warning("LIVE startup reconcile mismatch: %s", msg) # 启动时尝试恢复 stuck opening try: r = recover_stuck_opening(executor) if r is not None: logger.info("LIVE startup recover_opening: ok=%s detail=%s", r.ok, r.detail) except Exception: logger.exception("LIVE startup recover_opening failed") def perp_close_contracts_okx( client: Any, *, perp_inst: str, perp_side: str, perp_qty_eth: float, ct_val: float, allow_db_fallback: bool = True, ) -> int | None: """平永续张数:优先交易所持仓。 返回 >0 应下单;0=已确认空仓(仅 allow_db_fallback=False); None=查仓失败(调用方不得当空仓 finalize)。 """ ps = "long" if perp_side == "long" else "short" ex_sz = client.get_perp_pos_sz(perp_inst, pos_side=ps) if ex_sz is None: if not allow_db_fallback: return None return max(1, int(round(perp_qty_eth / ct_val))) if ex_sz > _PERP_EPS: return max(1, int(round(ex_sz))) if not allow_db_fallback: return 0 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, allow_db_fallback: bool = True, ) -> float | None: """平永续 ETH:>0 下单;0=已确认空;None=查仓失败。""" ps = "LONG" if perp_side == "long" else "SHORT" ex_sz = client.get_perp_pos_sz(perp_inst, position_side=ps) if ex_sz is None: if not allow_db_fallback: return None return float(perp_qty_eth) if ex_sz > _PERP_EPS: return float(ex_sz) if not allow_db_fallback: return 0.0 return float(perp_qty_eth) def perp_open_contracts_okx(*, perp_qty_eth: float, ct_val: float) -> int: """开仓张数:仅按设置名义/面值,不跟交易所残留。""" if ct_val <= 0: raise RuntimeError("ct_val invalid") return max(1, int(round(float(perp_qty_eth) / float(ct_val)))) def recover_stuck_opening(executor) -> CloseResult | None: """恢复本地 status=opening: - 交易所期权+永续皆空 → 清槽 - 仅期权 → half_open 并尝试 repair - 期权+永续 → 提升为 open(用本地已 stamp 的数量/均价) - 无元数据且交易所仍有仓 → 保持 opening,返回失败详情 """ if get_settings().is_sim: return None pos = executor.current_position() st = str(pos.get("status") or "") if st != "opening": return None client, ex_name = _executor_client_and_exchange(executor) if client is None: return CloseResult(ok=False, detail="recover_opening: 无交易客户端") option_inst = str(pos.get("option_inst_id") or "") perp_side = str(pos.get("perp_side") or "long") group_id = str(pos.get("group_id") or "") perp_inst = resolve_perp_inst_id(executor.db, group_id=group_id or None) perp_total = 0.0 for side in ("long", "short"): sz = exchange_perp_abs_size(client, ex_name or "", perp_inst, side) if sz is None: return CloseResult(ok=False, detail="recover_opening: 无法查永续") perp_total += float(sz) opt_sz = 0.0 if option_inst: raw = exchange_option_abs_size(client, option_inst) if raw is None: return CloseResult(ok=False, detail="recover_opening: 无法查期权") opt_sz = float(raw) else: any_opt = exchange_any_option_abs(client) if any_opt is None: return CloseResult(ok=False, detail="recover_opening: 无法查期权") if any_opt > _OPT_EPS: return CloseResult( ok=False, detail=( "recover_opening: opening 无 option_inst_id 但交易所有期权仓," "禁止自动清槽,请人工核对" ), ) opt_sz = 0.0 # 两边皆空 → 清槽 if opt_sz <= _OPT_EPS and perp_total <= _PERP_EPS: release_open_slot_if_opening(executor.db) return CloseResult(ok=True, detail="recover_opening: 交易所空仓,已释放 opening") # 无元数据但有仓 → 不自动处理 if not option_inst: return CloseResult( ok=False, detail="recover_opening: 交易所有仓但本地 opening 缺 option_inst_id", ) # 仅期权 → half_open + repair if opt_sz > _OPT_EPS and perp_total <= _PERP_EPS: persist = getattr(executor, "_persist_half_open", None) if not callable(persist): return CloseResult(ok=False, detail="recover_opening: 无 half_open 落库") of_px = float(pos.get("option_entry_px") or 0) or 0.0 opt_qty = float(pos.get("option_qty_eth") or 0) opt_contracts = float(pos.get("option_qty_contracts") or 0) if opt_contracts <= 0 and hasattr(executor, "_ct_mult"): from ..sim.liquidity import contracts_for_eth ct = executor._ct_mult(option_inst) opt_contracts = float(contracts_for_eth(opt_qty or opt_sz, ct)) if opt_qty else float(opt_sz) if opt_qty <= 0 and hasattr(executor, "_ct_mult"): from ..sim.liquidity import eth_from_contracts opt_qty = eth_from_contracts(opt_contracts or opt_sz, executor._ct_mult(option_inst)) persist( group_id=group_id or f"RCV-{option_inst[-12:]}", bias="recover", option_side=str(pos.get("option_side") or "call"), perp_side=perp_side, option_inst_id=option_inst, entry_index_px=float(pos.get("entry_index_px") or 0) or 0.0, strike=None, expiry_ymd=None, opt_qty=opt_qty, opt_contracts=opt_contracts or opt_sz, of_px=of_px, of_fee=0.0, detail="recover_opening option-only → half_open", ) repair = getattr(executor, "repair_half_open", None) if callable(repair): return repair() return CloseResult(ok=True, detail="recover_opening: 已落 half_open") # 期权+永续 → 提升为 open promote = getattr(executor, "_promote_opening_to_open", None) if callable(promote): return promote(pos=pos, perp_inst=perp_inst, opt_sz=opt_sz, perp_total=perp_total) return CloseResult( ok=False, detail="recover_opening: 双边有仓但执行器无 promote,请人工核对", ) 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