"""策略状态机：选向开仓 / 盯盘平仓 / 休息（无开仓窗、无轮次上限）。""" from __future__ import annotations import asyncio import logging import time from typing import Any from ..config import get_settings from .session import get_session from ..models.db import get_db from ..sim.ledger import Ledger from ..live import get_executor from ..live.rate_limit import LiveRetryGate, is_rate_limit_error from ..env_store import live_ready from .clock import can_open_new, window_key from .exits import check_expiry_close, check_exits, resolve_exit_target from .group import next_group_id from .open_capacity import maybe_notify_funds_short logger = logging.getLogger(__name__) class StrategyEngine: def __init__(self) -> None: self.db = get_db() self.ledger = Ledger(self.db) self.matcher = get_executor(self.db) self._task: asyncio.Task[None] | None = None self._lock = asyncio.Lock() self._retry_gate = LiveRetryGate() self._extra_sleep_sec = 0.0 self._last_residual_premium_check_ms = 0 def refresh_executor(self) -> None: """MODE 变更后刷新执行器。""" self.matcher = get_executor(self.db) def _gate_key(self, kind: str) -> str: pos = self.matcher.current_position() gid = str(pos.get("group_id") or "none") return f"{kind}:{gid}" def _note_retry_result(self, kind: str, *, ok: bool, detail: str = "") -> None: key = self._gate_key(kind) if ok: self._retry_gate.success(key) return rl = is_rate_limit_error(detail) delay = self._retry_gate.fail(key, rate_limited=rl) if rl: self._extra_sleep_sec = max(self._extra_sleep_sec, min(delay, 60.0)) logger.warning( "live retry backoff kind=%s fails=%s delay=%.1fs rate_limited=%s detail=%s", kind, self._retry_gate.fails(key), delay, rl, (detail or "")[:160], ) def _retry_allowed(self, kind: str) -> tuple[bool, float]: return self._retry_gate.allow(self._gate_key(kind)) def state(self) -> dict[str, Any]: row = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert row is not None upl = self.matcher.unrealized() s = get_settings() exit_mode = self.ledger.get_setting_str("exit_mode", s.exit_mode) net_target = self.ledger.get_setting_float( "net_profit_target", s.net_profit_target ) prem_mult = self.ledger.get_setting_float( "premium_exit_multiple", s.premium_exit_multiple ) exit_amt, _ = resolve_exit_target( exit_mode=exit_mode, net_profit_target=net_target, premium_exit_multiple=prem_mult, initial_premium=float(upl.get("initial_premium") or 0), ) rest_sec = self.ledger.get_setting_int("rest_seconds", s.rest_seconds) skip_weekends = self.ledger.get_setting_bool("skip_weekends", s.skip_weekends) one_expiry_per_day = self.ledger.get_setting_bool( "one_expiry_per_day", s.one_expiry_per_day ) leverage = self.ledger.get_setting_float("leverage", s.leverage) perp_mm = str( self.ledger.get_setting_str("perp_margin_mode", s.perp_margin_mode) or s.perp_margin_mode or "cross" ).strip().lower() if perp_mm not in ("cross", "isolated"): perp_mm = "cross" min_hours = self.ledger.get_setting_float("min_option_hours", s.min_option_hours) min_opt_lev = self.ledger.get_setting_float( "min_option_leverage", s.min_option_leverage ) atm_off_on = self.ledger.get_setting_bool( "atm_open_offset_enabled", s.atm_open_offset_enabled ) max_atm_off = self.ledger.get_setting_float( "max_atm_open_offset", s.max_atm_open_offset ) fixed_dir_on = self.ledger.get_setting_bool( "fixed_direction_enabled", s.fixed_direction_enabled ) fixed_perp = str( self.ledger.get_setting_str("fixed_perp_side", s.fixed_perp_side) or s.fixed_perp_side or "long" ).strip().lower() if fixed_perp not in ("long", "short"): fixed_perp = "long" 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) oo_put_qty = self.ledger.get_setting_float("oo_put_qty_eth", opt_qty) sizing_mode = ( sm if ( sm := str( self.ledger.get_setting_str("sizing_mode", "manual") or "manual" ) .strip() .lower() ) in ("manual", "risk_based") else "manual" ) risk_perp_unit = self.ledger.get_setting_float("risk_perp_unit", 1.0) risk_option_unit = self.ledger.get_setting_float("risk_option_unit", 2.0) risk_exit_unit = self.ledger.get_setting_float("risk_exit_unit", 15.0) risk_loss_pct = self.ledger.get_setting_float("risk_loss_pct", 1.0) risk_last_k = self.ledger.get_setting_float("risk_last_k", 0.0) risk_preview: dict[str, Any] | None = None pos_status = str(upl.get("status") or "flat") trade_locked = pos_status in ( "open", "half_open", "option_closed_perp_pending", "opening", ) locked_exit = None try: from .exits import read_locked_exit_target locked_exit = read_locked_exit_target(upl) except Exception: locked_exit = None if trade_locked and locked_exit is not None: # 持仓中：出场目标锁定，不再用盘口重算覆盖 net_target = float(locked_exit) exit_amt = float(locked_exit) # 以损定仓：仅空仓时用实时估算覆盖展示；持仓中保持开仓锁定名义/k/目标 if sizing_mode == "risk_based" and not trade_locked: try: from .risk_sizing import preview_risk_sizing risk_preview = preview_risk_sizing(self.db) if risk_preview.get("ok"): if risk_preview.get("perp_qty_eth") is not None: perp_qty = float(risk_preview["perp_qty_eth"]) if risk_preview.get("call_qty_eth") is not None: opt_qty = float(risk_preview["call_qty_eth"]) elif risk_preview.get("option_qty_eth") is not None: opt_qty = float(risk_preview["option_qty_eth"]) if risk_preview.get("put_qty_eth") is not None: oo_put_qty = float(risk_preview["put_qty_eth"]) if risk_preview.get("net_profit_target") is not None: net_target = float(risk_preview["net_profit_target"]) exit_amt = net_target if risk_preview.get("k") is not None: risk_last_k = float(risk_preview["k"]) if risk_preview.get("perp_unit") is not None: risk_perp_unit = float(risk_preview["perp_unit"]) if risk_preview.get("option_unit") is not None: risk_option_unit = float(risk_preview["option_unit"]) if risk_preview.get("exit_unit") is not None: risk_exit_unit = float(risk_preview["exit_unit"]) except Exception: logger.exception("risk sizing preview for state() failed") risk_preview = {"ok": False, "detail": "以损定仓预览失败"} elif sizing_mode == "risk_based" and trade_locked: risk_preview = { "ok": True, "locked": True, "detail": "持仓中已锁定本组成交目标与名义，平仓后再自动计算", "net_profit_target": net_target, "k": risk_last_k if risk_last_k > 0 else None, "perp_qty_eth": perp_qty, "option_qty_eth": opt_qty, } try: from .risk_sizing import resolve_martingale martingale = resolve_martingale( self.db, ledger=self.ledger, base_pct=risk_loss_pct ) except Exception: logger.exception("resolve_martingale for state() failed") martingale = { "enabled": False, "eligible": False, "doubles": 0, "loss_days": 0, "start_after_loss_days": 2, "max_doubles": 3, "effective_pct": risk_loss_pct, } rest_until = row["rest_until_ms"] rest_left = 0 if rest_until: rest_left = max(0, int((int(rest_until) - time.time() * 1000) / 1000)) last_error = row["last_error"] if last_error and "PriceResult" in str(last_error) and "__dict__" in str(last_error): self._set_state(last_error=None) last_error = None allow_open = can_open_new(skip_weekends=skip_weekends) try: # 以损定仓（尤其选约杠杆）：用预览名义+定仓卖一，勿用监控盘口贵卖一×账本名义虚高「需USDC」 from .auto_usdc import preview_capacity_for_convert open_cap = preview_capacity_for_convert(self.db) except Exception: logger.exception("preview_capacity_for_convert failed") open_cap = { "perp_can_open": None, "option_can_open": None, "perp_label": f"永续{int(round(leverage))}x —", "option_label": "期权 —", "funds_ok": False, "leverage": leverage, } out: dict[str, Any] = { "running": bool(row["running"]), "phase": row["phase"], "rounds_done": self._closed_rounds(), "window_key": row["window_key"], "rest_until_ms": rest_until, "rest_left_sec": rest_left, "rest_seconds": rest_sec, "skip_weekends": skip_weekends, "one_expiry_per_day": one_expiry_per_day, "exit_mode": exit_mode, "net_profit_target": net_target, "premium_exit_multiple": prem_mult, "exit_target_usdt": exit_amt, "leverage": leverage, "perp_margin_mode": perp_mm, "perp_qty_eth": perp_qty, "option_qty_eth": opt_qty, "oo_put_qty_eth": oo_put_qty, "sizing_mode": sizing_mode, "risk_based": sizing_mode == "risk_based", "fee_rate": float( self.ledger.get_setting_float("fee_rate", s.fee_rate) or s.fee_rate ), "hedge_mode": ( hm if ( hm := str( self.ledger.get_setting_str( "hedge_mode", s.hedge_mode ) or s.hedge_mode or "perp_option" ) .strip() .lower() ) in ("perp_option", "option_option") else "perp_option" ), "oo_amplitude_pct": float( self.ledger.get_setting_float( "oo_amplitude_pct", s.oo_amplitude_pct ) or s.oo_amplitude_pct ), "oo_amplitude_hours": float( self.ledger.get_setting_float( "oo_amplitude_hours", s.oo_amplitude_hours ) or s.oo_amplitude_hours ), "oo_amplitude_filter_enabled": str( self.ledger.get_setting_str( "oo_amplitude_filter_enabled", str(s.oo_amplitude_filter_enabled), ) or s.oo_amplitude_filter_enabled ) .strip() .lower() in ("1", "true", "yes", "on"), "oo_min_option_hours": float( self.ledger.get_setting_float( "oo_min_option_hours", s.oo_min_option_hours ) or s.oo_min_option_hours ), "oo_min_leverage": float( self.ledger.get_setting_float( "oo_min_leverage", s.oo_min_leverage ) or s.oo_min_leverage ), "oo_reward_ratio": float( self.ledger.get_setting_float( "oo_reward_ratio", s.oo_reward_ratio ) or s.oo_reward_ratio ), "oo_strike_max_dev_pct": float( self.ledger.get_setting_float( "oo_strike_max_dev_pct", s.oo_strike_max_dev_pct ) or s.oo_strike_max_dev_pct ), "risk_perp_unit": risk_perp_unit, "risk_option_unit": risk_option_unit, "risk_exit_unit": risk_exit_unit, "risk_loss_pct": risk_loss_pct, "risk_last_k": risk_last_k if risk_last_k > 0 else None, "risk_sizing_preview": risk_preview, "risk_sizing_locked": bool(trade_locked and sizing_mode == "risk_based"), "martingale_enabled": bool(martingale.get("enabled")), "martingale_eligible": bool(martingale.get("eligible")), "martingale_doubles": int(martingale.get("doubles") or 0), "martingale_loss_days": int(martingale.get("loss_days") or 0), "martingale_start_after_loss_days": int( martingale.get("start_after_loss_days") or 2 ), "martingale_max_doubles": int(martingale.get("max_doubles") or 3), "risk_effective_loss_pct": float( martingale.get("effective_pct") or risk_loss_pct ), "min_option_hours": min_hours, "min_option_leverage": min_opt_lev, "atm_open_offset_enabled": atm_off_on, "max_atm_open_offset": max_atm_off, "fixed_direction_enabled": fixed_dir_on, "fixed_perp_side": fixed_perp, "can_open": allow_open, "open_capacity": open_cap, "last_error": last_error, "position": upl, "residuals": self.matcher.list_residual_options_enriched(), "ledger": self.ledger.snapshot(), "mode": "SIM" if s.is_sim else "LIVE", "sim": s.is_sim, "exchange": str(s.exchange or "okx").lower(), "live_ready": (live_ready()[0] if not s.is_sim else True), "live_ready_reason": (live_ready()[1] if not s.is_sim else "sim"), "show_manual_trade_buttons": self.ledger.get_setting_bool( "show_manual_trade_buttons", False ), } try: from .semi_auto import read_semi_params sp = read_semi_params(self.ledger) k_eff = float(risk_last_k) if risk_last_k and float(risk_last_k) > 0 else 1.0 out["semi_auto_enabled"] = bool(sp.get("enabled")) out["semi_armed"] = bool(sp.get("armed")) out["semi_view_side"] = sp.get("view_side") out["semi_option_side"] = sp.get("option_side") out["semi_perp_side"] = sp.get("perp_side") out["semi_option_move_points"] = sp.get("option_move_points") out["semi_perp_exit_unit"] = sp.get("perp_exit_unit") out["semi_min_option_hours"] = sp.get("min_option_hours") out["semi_min_option_leverage"] = sp.get("min_option_leverage") out["semi_moneyness"] = sp.get("moneyness") out["semi_otm_max_offset"] = sp.get("otm_max_offset") out["semi_perp_unit"] = sp.get("perp_unit") out["semi_option_unit"] = sp.get("option_unit") out["semi_net_exit_target"] = float(sp["perp_exit_unit"]) * k_eff except Exception: logger.exception("semi params for state() failed") return out def _set_state(self, **kwargs: Any) -> None: cols = [] vals: list[Any] = [] for k, v in kwargs.items(): cols.append(f"{k}=?") vals.append(v) cols.append("updated_at_ms=?") vals.append(int(time.time() * 1000)) sql = f"UPDATE strategy_state SET {', '.join(cols)} WHERE id=1" self.db.execute(sql, tuple(vals)) async def pause(self) -> dict[str, Any]: self._set_state(running=0, phase="paused", last_error=None) try: from ..notify import wecom wecom.notify_pause() except Exception: logger.exception("wecom notify_pause failed") return self.state() async def start(self) -> dict[str, Any]: self.refresh_executor() s = get_settings() if not s.is_sim: ok, reason = live_ready() if not ok: self._set_state(running=0, phase="paused", last_error=reason) try: from ..notify import wecom wecom.notify_fault(title="启动失败", detail=reason, dedupe_key="start_fail") except Exception: pass return self.state() self._set_state(running=1, last_error=None, phase="idle") self.ensure_loop() try: from ..notify import wecom wecom.notify_start() except Exception: logger.exception("wecom notify_start failed") return self.state() def ensure_loop(self) -> None: """保证后台循环在跑（暂停时仍盯目标/到期，不新开）。""" if self._task is None or self._task.done(): self._task = asyncio.create_task(self._loop(), name="strategy-engine") async def emergency_close(self) -> dict[str, Any]: async with self._lock: close_data: dict[str, Any] | None = None detail = "flat" ok = True pos = self.matcher.current_position() st = str(pos.get("status") or "flat") if st == "half_open": repair = getattr(self.matcher, "repair_half_open", None) if callable(repair): r = repair() else: r = self.matcher.close_group(reason="emergency", bypass_liquidity=True) ok = r.ok detail = r.detail close_data = r.data if r.ok: self.enter_rest_after_close() elif st == "opening": recover = getattr(self.matcher, "recover_opening", None) if callable(recover): r = recover() ok = r.ok detail = r.detail close_data = r.data st2 = str(self.matcher.current_position().get("status") or "") if r.ok and st2 in ("flat",): self.enter_rest_after_close() else: ok = False detail = ( "stuck opening：请核对交易所期权/永续后人工处理" "（已占槽防重复开）" ) close_data = {"status": "opening"} try: from ..notify import wecom wecom.notify_fault( title="紧急全平遇到 stuck opening", detail=detail, dedupe_key="emergency:opening", ) except Exception: pass elif st == "closing": close_oo = getattr( self.matcher, "close_winning_oo_leave_residual", None ) if callable(close_oo): r = close_oo(reason="emergency_closing") else: r = self.matcher.close_group( reason="emergency", bypass_liquidity=True ) ok = r.ok detail = r.detail close_data = r.data if r.ok: self.enter_rest_after_close() elif st in ("open", "option_closed_perp_pending"): is_oo = ( str(pos.get("hedge_mode") or "") == "option_option" or bool(pos.get("option2_inst_id")) ) if is_oo and st == "open": close_full = getattr(self.matcher, "close_oo_full", None) if callable(close_full): r = close_full(reason="emergency", bypass_liquidity=True) else: r = self.matcher.close_group( reason="emergency", bypass_liquidity=True ) else: # A：双腿（或续平永续） r = self.matcher.close_group( reason="emergency", bypass_liquidity=True ) if not r.ok and st == "open" and not is_oo: # B：砸不出期权时强制只平永续（不要求远虚） abandon = getattr( self.matcher, "close_perp_abandon_option", None ) if callable(abandon): try: r2 = abandon( reason="emergency_perp", require_deep_otm=False ) except TypeError: r2 = abandon(reason="emergency_perp") if r2.ok: r = r2 ok = r.ok detail = r.detail close_data = r.data if r.ok: self.enter_rest_after_close() residuals = self.matcher.settle_all_residuals_now() try: if ok: from ..notify import wecom wecom.notify_close( reason="emergency", detail=detail, data=close_data or {}, ) except Exception: logger.exception("wecom emergency notify failed") return { "close": { "ok": ok, "detail": detail, "data": close_data, "residuals_settled": residuals, }, "state": self.state(), } def enter_rest_after_close(self) -> None: """全平成功后进入组间休息（自动 / 手动 / 紧急共用）。""" self._after_close() def _closed_rounds(self) -> int: """已完成组数 = 已平仓组数量（与顶栏总交易口径一致，避免计数漂移）。""" row = self.db.fetchone( "SELECT COUNT(*) AS n FROM groups WHERE status='closed'" ) return int(row["n"] or 0) if row else 0 def _after_close(self) -> None: from .semi_auto import ( PHASE_WAIT_HUMAN, clear_trade_lock, is_semi_auto, set_armed, ) s = get_settings() rounds = self._closed_rounds() clear_trade_lock(self.db) if is_semi_auto(self.ledger): # 半自动：平完停，清授权，等人工再开下一单 set_armed(self.db, False) self._set_state( rounds_done=rounds, phase=PHASE_WAIT_HUMAN, rest_until_ms=None, last_error=None, ) return rest_sec = self.ledger.get_setting_int("rest_seconds", s.rest_seconds) rest_until = int(time.time() * 1000) + rest_sec * 1000 self._set_state( rounds_done=rounds, phase="resting", rest_until_ms=rest_until, ) def arm_semi(self, *, armed: bool = True) -> dict[str, Any]: """首页半自动：授权/取消本单盯开。""" from .semi_auto import PHASE_WAIT_HUMAN, is_semi_auto, set_armed if not is_semi_auto(self.ledger): self._set_state(last_error="未开启半自动模式（系统设置）") return self.state() hm = str( self.ledger.get_setting_str("hedge_mode", "perp_option") or "perp_option" ).strip().lower() if hm == "option_option": self._set_state(last_error="期期模式不支持半自动") return self.state() if self.matcher.has_open_position(): self._set_state(last_error="有持仓时不能改授权；请先平仓") return self.state() st_row = self.db.fetchone("SELECT phase FROM strategy_state WHERE id=1") cur_phase = str(st_row["phase"] or "") if st_row else "" if not armed and cur_phase == "opening": self._set_state(last_error="开仓落单中，无法取消授权") return self.state() if armed: s = get_settings() if not s.is_sim: ok, reason = live_ready() if not ok: self._set_state( running=0, phase=PHASE_WAIT_HUMAN, last_error=f"授权失败（LIVE 未就绪）：{reason}", ) return self.state() set_armed(self.db, bool(armed)) if armed: if not self.state().get("running"): # 授权时自动拉起循环（仅盯开；未授权不会开）；LIVE 已过 live_ready self._set_state(running=1, phase="idle", last_error=None) self.ensure_loop() else: self._set_state(phase="idle", last_error=None) else: self._set_state(phase=PHASE_WAIT_HUMAN, last_error=None) return self.state() def _count_groups_for_day(self, wkey: str) -> int: rows = self.db.fetchall( "SELECT group_id FROM groups WHERE group_id LIKE ?", (f"G-{wkey}-%",), ) return len(rows) def _position_expiry_ms(self, upl: dict[str, Any]) -> int | None: raw = upl.get("expiry_ms") if raw is not None: try: return int(raw) except (TypeError, ValueError): pass ymd = upl.get("expiry_ymd") if ymd: try: from ..exchange.expiry import expiry_ms_from_ymd return int(expiry_ms_from_ymd(str(ymd))) except Exception: return None return None async def _close_open_position( self, *, reason: str, bypass_liquidity: bool, pending_close: bool, abandon_if_deep_otm: bool = False, retry_kind: str | None = None, ) -> None: kind = retry_kind or ( "perp_pending" if reason == "perp_pending_retry" else ("liquidity" if pending_close or not bypass_liquidity else "close") ) allowed, left = self._retry_allowed(kind) if not allowed: self._set_state( phase="liquidity_wait" if kind == "liquidity" else "closing", last_error=f"限流/失败退避中，{left:.0f}s 后再试 ({kind})", ) return if not pending_close: self._set_state(phase="closing", last_error=None) # 目标平仓 B：远虚 → 只平永续，期权归档 if abandon_if_deep_otm and reason != "expiry" and self.matcher.option_is_deep_otm(): abandon = self.matcher.close_perp_abandon_option try: r = await asyncio.to_thread( abandon, reason="target_perp_only", require_deep_otm=True, ) except TypeError: r = await asyncio.to_thread(abandon, reason="target_perp_only") if r.ok: self._note_retry_result(kind, ok=True) # _after_close 已写入 resting / wait_human，勿再覆盖 phase self.enter_rest_after_close() try: from ..notify import wecom wecom.notify_close( reason="target_perp_only", detail=r.detail, data=r.data or {}, ) except Exception: logger.exception("wecom notify_close failed") else: self._note_retry_result(kind, ok=False, detail=r.detail) self._set_state(phase="closing", last_error=r.detail) return r = await asyncio.to_thread( self.matcher.close_group, reason=reason, bypass_liquidity=bypass_liquidity, ) if r.ok: self._note_retry_result(kind, ok=True) self._after_close() try: from ..notify import wecom wecom.notify_close( reason=reason, detail=r.detail, data=r.data or {}, ) except Exception: logger.exception("wecom notify_close failed") elif r.liquidity_wait and not bypass_liquidity: # 等待期间若已变成远虚，下一 tick 走归档（半自动要求双腿全平，不归档） if ( reason not in ( "semi_target_points", "semi_perp_exit", ) and self.matcher.option_is_deep_otm() ): r2 = await asyncio.to_thread( self.matcher.close_perp_abandon_option, reason="target_perp_only", ) if r2.ok: self._note_retry_result(kind, ok=True) self.enter_rest_after_close() try: from ..notify import wecom wecom.notify_close( reason="target_perp_only", detail=r2.detail, data=r2.data or {}, ) except Exception: logger.exception("wecom notify_close failed") return self._note_retry_result("liquidity", ok=False, detail=r.detail) self._set_state(phase="liquidity_wait", last_error=r.detail) else: self._note_retry_result(kind, ok=False, detail=r.detail) self._set_state(phase="closing", last_error=r.detail) async def _settle_residuals(self) -> None: await asyncio.to_thread(self.matcher.settle_due_residuals) async def _maybe_close_residuals_by_premium(self) -> None: """残留期权：权利金回升达标时周期性尝试中途平仓。""" s = get_settings() interval_sec = int( self.ledger.get_setting_int( "residual_close_check_sec", s.residual_close_check_sec ) ) interval_sec = max(30, interval_sec) now_ms = int(time.time() * 1000) if now_ms - self._last_residual_premium_check_ms < interval_sec * 1000: return self._last_residual_premium_check_ms = now_ms fn = getattr(self.matcher, "try_close_pending_residuals", None) if not callable(fn): return try: closed = await asyncio.to_thread(fn) except Exception: logger.exception("try_close_pending_residuals failed") return if not closed: return for item in closed: if isinstance(item, dict) and item.get("fully_done") is False: continue try: from ..notify import wecom wecom.notify_close( reason="residual_premium_close", detail="残留期权权利金回收中途平", data=item if isinstance(item, dict) else {}, ) except Exception: logger.exception("wecom notify residual premium close failed") async def _maybe_expiry_close(self) -> bool: """若持仓已到期则强制全平。返回是否触发到期平仓。""" await self._settle_residuals() pos = self.matcher.current_position() if pos.get("status") not in ("open", "option_closed_perp_pending"): return False upl = self.matcher.unrealized() expired = check_expiry_close(expiry_ms=self._position_expiry_ms(upl)) if not expired.should_close: return False st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None pending = st["phase"] in ("liquidity_wait", "closing") await self._close_open_position( reason="expiry", bypass_liquidity=True, pending_close=pending, abandon_if_deep_otm=False, retry_kind="expiry", ) return True async def _loop(self) -> None: logger.info("strategy engine loop started") while True: try: row = self.db.fetchone("SELECT running FROM strategy_state WHERE id=1") running = bool(row and int(row["running"])) if not running: # 暂停：不新开仓；仍盯目标平仓 + 到期平仓 async with self._lock: await self._tick_manage_positions() await asyncio.sleep(1) continue async with self._lock: try: await get_session().ensure_atm_async(force=False) except Exception as e: logger.warning("ATM ensure before tick failed: %s", e) await self._tick_async() except asyncio.CancelledError: raise except Exception as e: err = str(e) if is_rate_limit_error(err): logger.warning("strategy tick rate-limited: %s", err[:200]) self._set_state(last_error="交易/行情接口限流，稍后自动重试") await asyncio.sleep(20) continue logger.exception("strategy tick failed") self._set_state(last_error=err) try: from ..notify import wecom wecom.notify_fault( title="策略循环异常", detail=err, dedupe_key=f"tick:{err[:80]}", ) except Exception: pass sleep_for = 1.0 + max(0.0, self._extra_sleep_sec) self._extra_sleep_sec = 0.0 await asyncio.sleep(min(sleep_for, 60.0)) async def _tick_manage_positions(self) -> None: """有仓时的盯盘：残留结算 / 半仓修复 / 目标平 / 到期平。不新开仓。""" await self._settle_residuals() await self._maybe_close_residuals_by_premium() s = get_settings() st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None exit_mode = self.ledger.get_setting_str("exit_mode", s.exit_mode) net_target = self.ledger.get_setting_float( "net_profit_target", s.net_profit_target ) prem_mult = self.ledger.get_setting_float( "premium_exit_multiple", s.premium_exit_multiple ) pos = self.matcher.current_position() st_pos = str(pos.get("status") or "flat") if st_pos == "opening": allowed, left = self._retry_allowed("opening") if not allowed: self._set_state( phase="opening_stuck", last_error=f"opening 恢复退避中，{left:.0f}s 后再试", ) return recover = getattr(self.matcher, "recover_opening", None) if callable(recover): r = await asyncio.to_thread(recover) self._note_retry_result("opening", ok=r.ok, detail=r.detail) if r.ok: try: from ..notify import wecom wecom.notify_fault( title="opening 已自动恢复", detail=r.detail, dedupe_key=f"opening_ok:{r.detail[:60]}", ) except Exception: pass # 恢复后按新状态继续本 tick（half_open/open/flat） pos = self.matcher.current_position() st_pos = str(pos.get("status") or "flat") if st_pos == "opening": self._set_state(phase="opening_stuck", last_error=r.detail) return if st_pos == "flat": self._set_state(phase="idle", last_error=None) return # half_open / open / pending：落入下方分支 self._set_state( phase="open" if st_pos == "open" else "closing", last_error=None, ) else: self._set_state(phase="opening_stuck", last_error=r.detail) try: from ..notify import wecom wecom.notify_fault( title="opening 恢复失败", detail=r.detail, dedupe_key=f"opening_fail:{r.detail[:60]}", ) except Exception: pass return else: return if st_pos == "half_open": allowed, left = self._retry_allowed("half_open") if not allowed: self._set_state( phase="closing", last_error=f"half_open 修复退避中，{left:.0f}s 后再试", ) return repair = getattr(self.matcher, "repair_half_open", None) if callable(repair): r = await asyncio.to_thread(repair) if r.ok: self._note_retry_result("half_open", ok=True) self.enter_rest_after_close() try: from ..notify import wecom wecom.notify_close( reason="emergency", detail=r.detail or "half_open repaired", data=r.data or {}, ) except Exception: pass else: self._note_retry_result("half_open", ok=False, detail=r.detail) self._set_state(phase="closing", last_error=r.detail) return if st_pos == "option_closed_perp_pending": await self._close_open_position( reason="perp_pending_retry", bypass_liquidity=True, pending_close=True, abandon_if_deep_otm=False, retry_kind="perp_pending", ) return if st_pos == "closing": allowed, left = self._retry_allowed("closing") if not allowed: self._set_state( phase="closing", last_error=f"closing 收尾退避中，{left:.0f}s 后再试", ) return close_oo = getattr(self.matcher, "close_winning_oo_leave_residual", None) if callable(close_oo): r = await asyncio.to_thread(close_oo, reason="closing_retry") self._note_retry_result("closing", ok=r.ok, detail=r.detail) if r.ok: self.enter_rest_after_close() else: self._set_state(phase="closing", last_error=r.detail) return if st_pos == "open": upl = self.matcher.unrealized() from .exits import lock_trade_exit_target, read_locked_exit_target from .semi_auto import check_semi_exits, is_semi_auto, read_semi_params locked_exit = read_locked_exit_target(upl) if locked_exit is None and upl.get("group_id") and not is_semi_auto( self.ledger ): try: locked_exit = lock_trade_exit_target( self.db, group_id=str(upl["group_id"]), initial_premium=float(upl.get("initial_premium") or 0), ) except Exception: logger.exception("backfill exit lock failed") expired = check_expiry_close(expiry_ms=self._position_expiry_ms(upl)) if is_semi_auto(self.ledger): sp = read_semi_params( self.ledger, group_id=str(upl["group_id"]) if upl.get("group_id") else None, ) rk = float( self.ledger.get_setting_float("risk_last_k", 1.0) or 1.0 ) if rk <= 0: rk = 1.0 semi_d = check_semi_exits( net_pnl=float(upl.get("net_pnl") or 0), strike=( float(upl["strike"]) if upl.get("strike") is not None else None ), index_px=( float(upl["index_px"]) if upl.get("index_px") is not None else None ), view_side=str(sp["view_side"]), option_move_points=float(sp["option_move_points"]), perp_exit_unit=float(sp["perp_exit_unit"]), risk_k=rk, entry_index=( float(upl["entry_index_px"]) if upl.get("entry_index_px") is not None else None ), ) # 复用 ExitDecision 形态 from .exits import ExitDecision should_semi = bool(semi_d.should_close) if should_semi: # 出场前要求持仓期权有买一，避免无对手盘硬平 opt_id = str(upl.get("option_inst_id") or "") oq = None try: oq = self.matcher._quote_held_option(opt_id) except Exception: oq = None if oq is None or oq.bid is None or float(oq.bid) <= 0: should_semi = False self._set_state( phase="liquidity_wait", last_error="半自动已达标但期权无买一，等待流动性", ) return decision = ExitDecision( should_semi, str(semi_d.reason or ""), float(semi_d.net_target or 0), ) if semi_d.detail and not semi_d.should_close: # 到点但净利≤0 等提示，不刷屏：仅非空时写入 if "净利≤0" in semi_d.detail: self._set_state(last_error=semi_d.detail) else: decision = check_exits( net_pnl=float(upl.get("net_pnl") or 0), exit_mode=exit_mode, net_profit_target=net_target, premium_exit_multiple=prem_mult, initial_premium=float(upl.get("initial_premium") or 0), locked_exit_target=locked_exit, ) pending_close = st["phase"] in ("liquidity_wait", "closing") # 曾因达标进入流动性等待，但当前估价已跌破目标：取消挂起，避免硬平出亏损却仍记「达标」 if ( pending_close and not expired.should_close and not decision.should_close ): self._set_state( phase="open", last_error="达标后流动性等待期间浮盈回落，已取消平仓挂起", ) return if expired.should_close or decision.should_close or pending_close: is_oo = ( str(upl.get("hedge_mode") or "") == "option_option" or bool(upl.get("option2_inst_id")) ) if expired.should_close: reason = "expiry" bypass = True abandon = False rkind = "expiry" else: reason = decision.reason or "liquidity_retry" bypass = False # 半自动要求双腿全平（先期权后永续），禁止远虚只平永续 semi_full = reason in ( "semi_target_points", "semi_perp_exit", ) abandon = ( bool(decision.should_close or pending_close) and not semi_full ) rkind = "liquidity" if pending_close else "close" if is_oo and decision.should_close and not expired.should_close: # 期期达标：只平盈利腿，亏损腿残留 close_oo = getattr( self.matcher, "close_winning_oo_leave_residual", None ) if close_oo is not None: r = await asyncio.to_thread( close_oo, reason="target_oo_win" ) if r.ok: self.enter_rest_after_close() else: self._set_state( phase="liquidity_wait", last_error=r.detail or "期期盈利腿暂不可平", ) return if is_oo and ( expired.should_close or reason in ("expiry", "emergency", "manual") or bypass ): close_full = getattr(self.matcher, "close_oo_full", None) if close_full is not None and ( expired.should_close or bypass or reason == "emergency" ): oo_reason = ( reason if reason != "liquidity_retry" else "expiry" ) # LIVE：到期不卖（交易所自动结算）；紧急等才卖两腿 if oo_reason != "expiry": for sell_fn_name in ( "_live_sell_oo_both", "live_sell_oo_both", ): sell_both = getattr(self.matcher, sell_fn_name, None) if callable(sell_both): try: await asyncio.to_thread( sell_both, bypass_liquidity=bypass, reason=oo_reason, ) except TypeError: await asyncio.to_thread( sell_both, bypass_liquidity=bypass ) except Exception: logger.exception("live sell oo both failed") break r = await asyncio.to_thread( close_full, reason=oo_reason, bypass_liquidity=True, ) if r.ok: self.enter_rest_after_close() else: self._set_state( phase="liquidity_wait", last_error=r.detail or "期期全平失败", ) return await self._close_open_position( reason=reason, bypass_liquidity=bypass, pending_close=pending_close, abandon_if_deep_otm=abandon and not is_oo, retry_kind=rkind, ) else: self._set_state(phase="open", last_error=None) async def _tick_async(self) -> None: # 先盯仓（平仓路径） await self._tick_manage_positions() s = get_settings() st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None wkey = window_key() if st["window_key"] != wkey: self._set_state(window_key=wkey, phase="idle") st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None pos = self.matcher.current_position() st_pos = str(pos.get("status") or "flat") # 仍有活跃仓或半仓：本 tick 不再开新仓 if st_pos in ("open", "half_open", "option_closed_perp_pending", "opening"): return if st["phase"] == "resting" and st["rest_until_ms"]: if int(time.time() * 1000) < int(st["rest_until_ms"]): return self._set_state(phase="idle", rest_until_ms=None) st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None if st["phase"] in ("paused",): return # 旧「轮次停开」状态：自动恢复为空闲以便继续 if st["phase"] in ("stopped", "outside_window"): self._set_state(phase="idle") from .semi_auto import PHASE_WAIT_HUMAN, is_armed, is_semi_auto # 半自动未授权：停在 wait_human，不进入选约/开仓 if is_semi_auto(self.ledger) and not is_armed(self.ledger): if st["phase"] != PHASE_WAIT_HUMAN: self._set_state(phase=PHASE_WAIT_HUMAN, last_error=None) return skip_weekends = self.ledger.get_setting_bool("skip_weekends", s.skip_weekends) if not can_open_new(skip_weekends=skip_weekends): self._set_state( phase="weekend_skip", last_error="周六/周日跳过开仓（上海时区）；持仓仍可平仓", ) return if st["phase"] == "weekend_skip": self._set_state(phase="idle", last_error=None) # 双保险：账本仍显示有仓则不开 if self.matcher.has_open_position(): self._set_state(phase="open", last_error="有未平仓，禁止开下一组") return # OKX：等待选约时仅预览名义+兑 USDC（不落库改 qty/exit） try: from .open_pipeline import prepare_usdc_while_waiting prep = prepare_usdc_while_waiting(self.db) conv = prep.get("convert") or {} if conv.get("acted"): logger.info("auto_usdc while waiting: %s", conv.get("detail")) elif conv.get("ok") is False: logger.warning("auto_usdc while waiting: %s", conv.get("detail")) except Exception: logger.exception("prepare USDC while waiting failed") self._set_state(phase="wait_signal") pick = await get_session().pick_for_open_async() one_expiry_per_day = self.ledger.get_setting_bool( "one_expiry_per_day", s.one_expiry_per_day ) # 半自动人工授权：允许同到期再开；全自动仍受一日一到期约束 if ( pick is not None and one_expiry_per_day and not is_semi_auto(self.ledger) ): from .clock import ( expiry_blocked_by_one_per_day, used_expiry_ymds, ) used = used_expiry_ymds(self.db) if expiry_blocked_by_one_per_day( pick.pair.expiry_ymd, used, enabled=True ): self._set_state( phase="idle", last_error=( f"每个到期只开一次：已用过 {pick.pair.expiry_ymd}，" "请等下一到期日" ), ) return if pick is None: amp_fail = None try: amp_fail = get_session().amplitude_gate_fail_reason() except Exception: amp_fail = None if amp_fail: self._set_state(last_error=amp_fail) return try: from .auto_usdc import preview_capacity_for_convert from .open_capacity import funds_gate_blocks # 与自动兑 USDC 同一口径（选约杠杆用隐含卖一估需），避免无合格约时用盘口贵卖一误报缺 USDC cap = preview_capacity_for_convert(self.db) blocked, why = funds_gate_blocks(cap) if blocked and cap.get("option_can_open") is False: self._set_state( phase="wait_funds", last_error=( "无合格期权；且按拟定仓估算交易账户 USDC 不够开仓" f"（需≈{cap.get('option_need_usdc')}U / 有{cap.get('option_have_usdc')}U）。" "系统会在冷却后自动市价兑 USDC。" ), ) return if blocked and cap.get("option_can_open") is None: self._set_state( phase="wait_funds", last_error=why, ) return except Exception: pass pick_why = None try: pick_why = get_session().last_pick_fail_reason() except Exception: pick_why = None if pick_why: self._set_state(last_error=f"无合格期权：{pick_why}") else: self._set_state( last_error="无合格期权：需剩余时长、杠杆（及已开启的ATM偏差）同时满足" ) return # 半自动：选约异步窗口后再次确认授权，防止取消授权后仍开仓 from .semi_auto import ( PHASE_WAIT_HUMAN, is_armed, is_semi_auto, lock_trade_params, read_semi_params, ) if is_semi_auto(self.ledger) and not is_armed(self.ledger): self._set_state( phase=PHASE_WAIT_HUMAN, last_error="半自动授权已取消，已中止开仓", ) return # 选约后：定仓落库 → 兑 USDC → 资金门 fail-closed（与手动开仓同一管道） from .open_pipeline import size_and_gate oo = getattr(pick, "hedge_mode", "perp_option") == "option_option" prep = size_and_gate( index_px=float(pick.underlying_px), option_ask=float(pick.option_ask), db=self.db, call_ask=float(pick.call_ask) if oo else None, put_ask=float(pick.put_ask) if oo else None, hedge_mode="option_option" if oo else "perp_option", ) if not prep.ok: phase = "wait_funds" if prep.capacity is not None else "idle" self._set_state(phase=phase, last_error=prep.detail) if prep.capacity is not None: maybe_notify_funds_short(prep.capacity) if "以损定仓" in (prep.detail or ""): try: from ..notify import wecom wecom.notify_fault( title="以损定仓失败", detail=prep.detail, dedupe_key=f"risk_sizing:{prep.detail[:80]}", ) except Exception: pass return if st["phase"] == "wait_funds": self._set_state(phase="idle", last_error=None) if is_semi_auto(self.ledger) and not is_armed(self.ledger): self._set_state( phase=PHASE_WAIT_HUMAN, last_error="半自动授权已取消，已中止开仓", ) return self._set_state(phase="opening", last_error=None) wkey = window_key() count = self._count_groups_for_day(wkey) gid = next_group_id(count) 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) try: from ..notify import wecom wecom.notify_fault( title="开仓前对账拒绝", detail=safe_msg, dedupe_key=f"recon:{safe_msg[:80]}", ) except Exception: pass return if is_semi_auto(self.ledger) and not is_armed(self.ledger): self._set_state( phase=PHASE_WAIT_HUMAN, last_error="半自动授权已取消，已中止开仓", ) return if oo: open_fn = getattr(self.matcher, "open_oo_group", None) if open_fn is None: self._set_state(phase="idle", last_error="当前执行器不支持期期开仓") return r = await asyncio.to_thread( open_fn, group_id=gid, call_inst_id=str(pick.call_inst_id or pick.pair.call_inst_id), put_inst_id=str(pick.put_inst_id or pick.pair.put_inst_id), call_strike=float(pick.call_strike or pick.pair.strike), put_strike=float(pick.put_strike or pick.pair.strike), entry_index_px=float(entry_idx), expiry_ymd=pick.pair.expiry_ymd, ) option_inst = str(pick.call_inst_id or pick.pair.call_inst_id) else: option_inst = ( pick.pair.call_inst_id if pick.option_side == "call" else pick.pair.put_inst_id ) r = await asyncio.to_thread( self.matcher.open_group, group_id=gid, bias=pick.bias, option_side=pick.option_side, perp_side=pick.perp_side, option_inst_id=option_inst, entry_index_px=float(entry_idx), strike=pick.pair.strike, expiry_ymd=pick.pair.expiry_ymd, ) if r.ok: if is_semi_auto(self.ledger): sp_lock = read_semi_params(self.ledger) lock_trade_params( self.db, group_id=gid, view_side=str(sp_lock["view_side"]), option_move_points=float(sp_lock["option_move_points"]), perp_exit_unit=float(sp_lock["perp_exit_unit"]), moneyness=str(sp_lock.get("moneyness") or "otm"), otm_max_offset=float(sp_lock.get("otm_max_offset") or 25), perp_unit=float(sp_lock.get("perp_unit") or 1), option_unit=float(sp_lock.get("option_unit") or 4), ) self._set_state(phase="open", last_error=None) try: from ..notify import wecom wecom.notify_open( group_id=gid, detail=r.detail, extra={ "bias": pick.bias, "option_side": pick.option_side, "perp_side": pick.perp_side, "option_inst_id": option_inst, "strike": pick.pair.strike, "expiry_ymd": pick.pair.expiry_ymd, **(r.data or {}), }, ) except Exception: logger.exception("wecom notify_open failed") else: # 保留 opening 时勿伪装 idle pos_after = self.matcher.current_position() if str(pos_after.get("status") or "") == "opening": self._set_state(phase="opening_stuck", last_error=r.detail) else: self._set_state(phase="idle", last_error=r.detail) try: from ..notify import wecom wecom.notify_fault( title="开仓失败", detail=r.detail, dedupe_key=f"open_fail:{r.detail[:80]}", ) except Exception: pass _engine: StrategyEngine | None = None def get_engine() -> StrategyEngine: global _engine if _engine is None: _engine = StrategyEngine() return _engine def set_engine(engine: StrategyEngine | None) -> None: global _engine _engine = engine