"""策略行情会话：在交易所适配器之上做 ATM 对齐与开仓选约。""" from __future__ import annotations import asyncio import logging from dataclasses import dataclass from typing import Any from ..config import Settings, get_settings from ..exchange import get_exchange, set_exchange, build_exchange from ..exchange.option_ids import pair_from_option_inst from ..exchange.protocol import ExchangeMarket from ..exchange.types import MarketSnapshot, OptionPair from .selection import ( atm_allows_open, atm_open_offset, hours_until_expiry, is_itm_or_atm, list_eligible_expiry_ymds, option_leverage, select_option_pair, ) logger = logging.getLogger(__name__) _ATM_DRIFT_POINTS = 5.0 _session: StrategySession | None = None def _has_open_position() -> bool: try: from ..models.db import get_db from ..sim.matcher import BLOCKING_STATUSES row = get_db().fetchone("SELECT status, group_id, option_inst_id FROM positions WHERE id=1") if not row: return False st = str(row["status"] or "") if st not in BLOCKING_STATUSES: return False return bool(row["group_id"] or row["option_inst_id"]) except Exception: return False def _held_option_inst_id() -> str | None: """活跃持仓期权合约；无仓返回 None。""" try: from ..models.db import get_db row = get_db().fetchone( "SELECT status, option_inst_id FROM positions WHERE id=1" ) if not row or row["status"] not in ("open", "half_open", "option_closed_perp_pending"): return None # 期权已平待平永续：不再钉期权盘口 if row["status"] == "option_closed_perp_pending": return None inst = str(row["option_inst_id"] or "").strip() return inst or None except Exception: return None def _held_option_legs() -> tuple[str | None, str | None]: """期期持仓：返回 (call_inst, put_inst)；非期期或无仓则 put 为空。""" try: from ..models.db import get_db row = get_db().fetchone( """SELECT status, hedge_mode, option_inst_id, option2_inst_id FROM positions WHERE id=1""" ) if not row or row["status"] not in ("open", "half_open"): return None, None call_id = str(row["option_inst_id"] or "").strip() or None put_id = str(row["option2_inst_id"] or "").strip() or None if str(row["hedge_mode"] or "").strip().lower() != "option_option": return call_id, None return call_id, put_id except Exception: return None, None def _as_bool_setting(raw: str | None, default: bool) -> bool: if raw is None or raw == "": return default return str(raw).strip().lower() in ("1", "true", "yes", "on") def _skip_expiry_ymds_for_next(*, include_used: bool = True) -> set[str]: """ 空仓选约/监控应跳过的到期日： - 历史上已开过该到期（one_expiry_per_day，跨日）；半自动可关 - 仍有待结算残留期权的到期档（该档已「完成」开平，盯下一档） """ skip: set[str] = set() try: from ..models.db import get_db from .clock import pending_residual_expiry_ymds, used_expiry_ymds s = get_settings() db = get_db() if include_used: one_exp_day = _as_bool_setting( db.get_setting("one_expiry_per_day", str(s.one_expiry_per_day)), s.one_expiry_per_day, ) if one_exp_day: skip |= used_expiry_ymds(db) skip |= pending_residual_expiry_ymds(db) except Exception: logger.exception("skip-expiry lookup failed; continue without skip") return skip def _strategy_floats() -> tuple[float, float, float, bool]: """min_hours, min_leverage, max_atm_open_offset, atm_open_offset_enabled""" s = get_settings() try: from ..models.db import get_db db = get_db() hours = float( db.get_setting("min_option_hours", str(s.min_option_hours)) or s.min_option_hours ) lev = float( db.get_setting("min_option_leverage", str(s.min_option_leverage)) or s.min_option_leverage ) atm_off = float( db.get_setting("max_atm_open_offset", str(s.max_atm_open_offset)) or s.max_atm_open_offset ) atm_on = _as_bool_setting( db.get_setting("atm_open_offset_enabled", str(s.atm_open_offset_enabled)), s.atm_open_offset_enabled, ) return hours, lev, atm_off, atm_on except Exception: return ( s.min_option_hours, s.min_option_leverage, s.max_atm_open_offset, s.atm_open_offset_enabled, ) def _fixed_direction() -> tuple[bool, str]: """(enabled, perp_side long|short)。默认关。""" s = get_settings() try: from ..models.db import get_db db = get_db() enabled = _as_bool_setting( db.get_setting( "fixed_direction_enabled", str(s.fixed_direction_enabled) ), s.fixed_direction_enabled, ) side = str( db.get_setting("fixed_perp_side", s.fixed_perp_side) or s.fixed_perp_side ).strip().lower() if side not in ("long", "short"): side = "long" return enabled, side except Exception: side = str(s.fixed_perp_side or "long").strip().lower() if side not in ("long", "short"): side = "long" return bool(s.fixed_direction_enabled), side def _option_side_for_perp(perp_side: str) -> str: return "put" if (perp_side or "").strip().lower() == "long" else "call" def _hedge_mode() -> str: s = get_settings() try: from ..models.db import get_db raw = str( get_db().get_setting("hedge_mode", s.hedge_mode) or s.hedge_mode ).strip().lower() if raw in ("perp_option", "option_option"): return raw except Exception: pass return "perp_option" def _oo_settings() -> tuple[float, float, float, float, float, bool]: """amplitude_pct, amplitude_hours, min_option_hours, min_leverage, strike_max_dev_pct, amp_filter_on""" s = get_settings() try: from ..models.db import get_db db = get_db() filt_raw = db.get_setting( "oo_amplitude_filter_enabled", str(s.oo_amplitude_filter_enabled) ) filt_on = str(filt_raw or "").strip().lower() in ( "1", "true", "yes", "on", ) if filt_raw is None or filt_raw == "": filt_on = bool(s.oo_amplitude_filter_enabled) return ( float(db.get_setting("oo_amplitude_pct", str(s.oo_amplitude_pct)) or s.oo_amplitude_pct), float( db.get_setting("oo_amplitude_hours", str(s.oo_amplitude_hours)) or s.oo_amplitude_hours ), float( db.get_setting("oo_min_option_hours", str(s.oo_min_option_hours)) or s.oo_min_option_hours ), float(db.get_setting("oo_min_leverage", str(s.oo_min_leverage)) or s.oo_min_leverage), float( db.get_setting( "oo_strike_max_dev_pct", str(s.oo_strike_max_dev_pct) ) or s.oo_strike_max_dev_pct ), filt_on, ) except Exception: return ( s.oo_amplitude_pct, s.oo_amplitude_hours, s.oo_min_option_hours, s.oo_min_leverage, s.oo_strike_max_dev_pct, bool(s.oo_amplitude_filter_enabled), ) @dataclass(slots=True) class OpenPick: pair: OptionPair option_side: str perp_side: str bias: str call_ask: float put_ask: float option_ask: float option_leverage: float hours_left: float underlying_px: float hedge_mode: str = "perp_option" call_inst_id: str | None = None put_inst_id: str | None = None call_strike: float | None = None put_strike: float | None = None call_leverage: float | None = None put_leverage: float | None = None amplitude_high: float | None = None amplitude_low: float | None = None amplitude_range_pct: float | None = None oo_detail: str | None = None class StrategySession: """策略侧会话；交易所实现由 exchange 模块注入。""" def __init__( self, settings: Settings | None = None, exchange: ExchangeMarket | None = None, ) -> None: self.settings = settings or get_settings() self.ex = exchange or get_exchange() self._pair: OptionPair | None = None self._oo_amp: dict[str, Any] | None = None self._last_pick_fail: str | None = None self._refresh_task: asyncio.Task[None] | None = None self._started = False @property def pair(self) -> OptionPair | None: return self._pair def _watch_ids(self, pair: OptionPair | None = None) -> list[str]: """永续 + 监控对 + 持仓腿（有仓时绝不能 drop 持仓盘口）。""" s = self.settings p = pair if pair is not None else self._pair ids: list[str] = [s.perp_inst_id] if p is not None: ids.extend([p.call_inst_id, p.put_inst_id]) held, held2 = _held_option_legs() if held: ids.append(held) if held2: ids.append(held2) # 去重保序 out: list[str] = [] seen: set[str] = set() for i in ids: if i and i not in seen: seen.add(i) out.append(i) return out async def start(self) -> None: if self._started: return self._started = True await self.ex.start() try: # 有持仓时必须钉在持仓行权价，禁止重启后漂到新 ATM if _has_open_position(): await asyncio.to_thread(self.align_to_held_position) else: await asyncio.to_thread(self.align_instruments) except Exception as e: # eapi 418/429 时允许先起会话，后续 refresh 再对齐 logger.warning("initial ATM align failed (will retry): %s", e) await self.ex.resubscribe(self._watch_ids()) self._refresh_task = asyncio.create_task(self._refresh_loop(), name="strategy-align") async def stop(self) -> None: self._started = False if self._refresh_task: self._refresh_task.cancel() try: await self._refresh_task except asyncio.CancelledError: pass self._refresh_task = None await self.ex.stop() def _apply_pair(self, pair: OptionPair, *, mark: float, idx: float | None) -> OptionPair: s = self.settings self._pair = pair self.ex.set_pair(pair) if idx is not None: self.ex.set_index_px(idx) self.ex.warm_and_subscribe(self._watch_ids(pair)) logger.info( "aligned pair exchange=%s expiry=%s strike=%s mark=%.2f hours=%.1f", getattr(self.ex, "name", "?"), pair.expiry_ymd, pair.strike, mark, hours_until_expiry(pair.expiry_ymd, expiry_ms=pair.expiry_ms), ) return pair def align_to_held_position(self) -> OptionPair | None: """有活跃仓时：监控对锁定为持仓合约的到期/行权价。""" self.refresh_oo_amplitude() call_id, put_id = _held_option_legs() held = call_id or _held_option_inst_id() if not held: return None mark = self._mark_for_atm() idx = None try: idx = self.ex.fetch_index(self.settings.index_inst_id) except Exception: pass if put_id and call_id: # 期期：双腿分别钉住 try: from ..models.db import get_db row = get_db().fetchone( "SELECT strike2 FROM positions WHERE id=1" ) except Exception: row = None cpair = pair_from_option_inst(call_id) ppair = pair_from_option_inst(put_id) if cpair is None: logger.warning("cannot rebuild call pair from held %s", call_id) return None put_strike = None if row and row["strike2"] is not None: put_strike = float(row["strike2"]) elif ppair is not None: put_strike = float(ppair.strike) pair = OptionPair( expiry_ymd=cpair.expiry_ymd, expiry_ms=cpair.expiry_ms, strike=float(cpair.strike), call_inst_id=call_id, put_inst_id=put_id, put_strike=put_strike, ) logger.info( "pin OO watch call=%s put=%s C@%.0f P@%.0f", call_id, put_id, pair.strike, float(put_strike or pair.strike), ) return self._apply_pair( pair, mark=float(mark or pair.strike), idx=idx ) pair = pair_from_option_inst(held) if pair is None: logger.warning("cannot rebuild pair from held option %s", held) return None mark = mark or float(pair.strike) logger.info( "pin watch to held option %s strike=%.0f expiry=%s", held, pair.strike, pair.expiry_ymd, ) return self._apply_pair(pair, mark=float(mark), idx=idx) def align_instruments(self) -> OptionPair | None: # 重启/刷新时若仍有仓，绝不切到新 ATM if _has_open_position(): return self.align_to_held_position() if _hedge_mode() == "option_option": return self.align_oo_instruments() # 永期：刷新振幅供 Plan（过滤关也展示；开仓门禁在 pick） self.refresh_oo_amplitude() s = self.settings idx = self.ex.fetch_index(s.index_inst_id) mark = self.ex.fetch_mark(s.perp_inst_id) or idx if mark is None or mark <= 0: raise RuntimeError("无法获取标的标记/指数价格，无法选 ATM") min_hours, _, _, _ = _strategy_floats() fixed_on, fixed_perp = _fixed_direction() opt_side = _option_side_for_perp(fixed_perp) if fixed_on else None contracts = self.ex.list_option_contracts(s.option_inst_family) eligible = list_eligible_expiry_ymds(contracts, min_hours=min_hours) skip = _skip_expiry_ymds_for_next() pair = None for ymd in eligible: if ymd in skip: logger.info( "align skip expiry=%s (used today and/or residual pending)", ymd, ) continue pair = select_option_pair( contracts, mark_px=float(mark), expiry_ymd=ymd, option_side=opt_side, ) if pair is not None: break if pair is None and eligible: # 候选都被跳过时仍回退最近合格档，避免盘口空白 pair = select_option_pair( contracts, mark_px=float(mark), expiry_ymd=eligible[0], option_side=opt_side, ) if pair is None: kind = f"实值/平值 {opt_side}" if opt_side else "ATM" raise RuntimeError( f"未找到剩余≥{min_hours}h 的 {kind} Call/Put (family={s.option_inst_family})" ) return self._apply_pair(pair, mark=float(mark), idx=idx) def refresh_oo_amplitude(self) -> dict[str, Any] | None: """刷新振幅高低（永期/期期；有仓/无仓都要，否则 UI 指数/振幅会空）。""" try: from ..exchange.candles import fetch_amplitude_hl_for_runtime from .amplitude_gate import evaluate_amplitude_gate amp_pct, amp_hours, _, _, _, amp_filt = _oo_settings() amp = fetch_amplitude_hl_for_runtime(amp_hours) gate = evaluate_amplitude_gate( filter_enabled=bool(amp_filt), amp=amp, max_pct=float(amp_pct), hours=float(amp_hours), ) self._oo_amp = dict(gate["snapshot"]) return self._oo_amp except Exception: logger.exception("refresh_oo_amplitude failed") return self._oo_amp def _apply_amplitude_first_gate(self) -> bool: """ 振幅过滤为开仓第一关。返回 True=可继续选约；False=拒开。 无论是否开启过滤都写入 _oo_amp 供 Plan 展示。 """ from ..exchange.candles import fetch_amplitude_hl_for_runtime from .amplitude_gate import evaluate_amplitude_gate amp_pct, amp_hours, _, _, _, amp_filt = _oo_settings() amp = fetch_amplitude_hl_for_runtime(amp_hours) gate = evaluate_amplitude_gate( filter_enabled=bool(amp_filt), amp=amp, max_pct=float(amp_pct), hours=float(amp_hours), ) self._oo_amp = dict(gate["snapshot"]) if gate["blocked"]: logger.info("amplitude gate blocked: %s", gate.get("reason")) return False return True def last_pick_fail_reason(self) -> str | None: return self._last_pick_fail def amplitude_gate_fail_reason(self) -> str | None: """若最近一次振幅快照显示过滤开启且未过关，返回文案。""" amp = self._oo_amp if not amp or not amp.get("filter_enabled"): return None if amp.get("ok") is True: return None hours = amp.get("hours") range_pct = amp.get("range_pct") max_pct = amp.get("max_pct") if range_pct is None: return f"振幅未过关：无法获取近 {hours:g}h K 线高低" if hours is not None else "振幅未过关" try: return ( f"振幅未过关：{float(hours):g}h 内 " f"{float(range_pct):.2f}% > {float(max_pct):g}%" ) except (TypeError, ValueError): return "振幅未过关" def align_oo_instruments(self) -> OptionPair | None: """期期监控：按振幅高低点选虚值 Call/Put（展示用；振幅超限仍对齐候选）。""" from .oo_selection import select_oo_pair # 有仓也刷新振幅（钉仓不再走选约，否则 _oo_amp 一直空） self.refresh_oo_amplitude() if _has_open_position(): return self.align_to_held_position() s = self.settings amp_pct, amp_hours, min_hours, _min_lev, max_dev, _amp_filt = _oo_settings() idx = self.ex.fetch_index(s.index_inst_id) mark = self.ex.fetch_mark(s.perp_inst_id) or idx if mark is None or mark <= 0: raise RuntimeError("无法获取标的标记/指数价格，无法选期期虚值") underlying = float(mark) if self._oo_amp is None: raise RuntimeError("无法获取振幅 K 线高低点") amp_high = float(self._oo_amp["high"]) amp_low = float(self._oo_amp["low"]) contracts = self.ex.list_option_contracts(s.option_inst_family) skip = _skip_expiry_ymds_for_next() picked = select_oo_pair( contracts, spot=underlying, high=amp_high, low=amp_low, min_hours=float(min_hours), skip_expiry_ymds=skip, max_dev_pct=float(max_dev), ) if picked is None: raise RuntimeError( f"未找到剩余≥{min_hours}h 且贴高低≤{max_dev:g}% 的虚值 Call/Put" ) ymd, ems, ck, pk, call_inst, put_inst = picked pair = OptionPair( expiry_ymd=ymd, expiry_ms=int(ems), strike=float(ck), call_inst_id=call_inst, put_inst_id=put_inst, put_strike=float(pk), ) return self._apply_pair(pair, mark=underlying, idx=idx) def pick_for_open(self) -> OpenPick | None: if _hedge_mode() == "option_option": return self._pick_for_open_oo() return self._pick_for_open_perp() def _pick_for_open_oo(self) -> OpenPick | None: from ..exchange.candles import fetch_amplitude_hl_for_runtime from .oo_selection import ( pick_otm_call_strike, pick_otm_put_strike, select_oo_pair, ) from .selection import _complete_by_expiry, option_leverage s = self.settings amp_pct, amp_hours, min_hours, min_lev, max_dev, amp_filt = _oo_settings() idx = self.ex.fetch_index(s.index_inst_id) mark = self.ex.fetch_mark(s.perp_inst_id) or idx if mark is None or mark <= 0: return None underlying = float(mark) amp = fetch_amplitude_hl_for_runtime(amp_hours) if amp is None: logger.info("oo: amplitude candles unavailable") return None # 无论是否超限都写入，供「指数/振幅」面板展示 self._oo_amp = { "high": float(amp.high), "low": float(amp.low), "mid": float(amp.mid), "range_pct": float(amp.range_pct), "hours": float(amp_hours), "max_pct": float(amp_pct), "filter_enabled": bool(amp_filt), "ok": ( True if not amp_filt else float(amp.range_pct) <= float(amp_pct) + 1e-12 ), } if amp_filt and float(amp.range_pct) > float(amp_pct) + 1e-12: logger.info( "oo: amplitude %.3f%% > max %.3f%% (H=%.2f L=%.2f)", amp.range_pct, amp_pct, amp.high, amp.low, ) return None contracts = self.ex.list_option_contracts(s.option_inst_family) skip = _skip_expiry_ymds_for_next() picked = select_oo_pair( contracts, spot=underlying, high=float(amp.high), low=float(amp.low), min_hours=float(min_hours), skip_expiry_ymds=skip, max_dev_pct=float(max_dev), ) if picked is None: logger.info( "oo: no OTM call/put within %.2f%% of amplitude HL", max_dev, ) return None ymd, ems, ck, pk, call_inst, put_inst = picked call_bids, call_asks, _ = self.ex.fetch_book(call_inst, depth=5) put_bids, put_asks, _ = self.ex.fetch_book(put_inst, depth=5) call_ask = call_asks[0].px if call_asks else None put_ask = put_asks[0].px if put_asks else None if call_ask is None: cq = self.ex.quote(call_inst) call_ask = cq.ask if cq else None if put_ask is None: pq = self.ex.quote(put_inst) put_ask = pq.ask if pq else None if call_ask is None or put_ask is None or call_ask <= 0 or put_ask <= 0: logger.info("oo: missing ask call=%s put=%s", call_ask, put_ask) return None c_lev = option_leverage(underlying, float(call_ask)) p_lev = option_leverage(underlying, float(put_ask)) if ( c_lev is None or p_lev is None or c_lev + 1e-9 < min_lev or p_lev + 1e-9 < min_lev ): logger.info( "oo: leverage too low call=%s put=%s need>=%.0f", f"{c_lev:.1f}" if c_lev else "n/a", f"{p_lev:.1f}" if p_lev else "n/a", min_lev, ) return None # 监控用：Call/Put 不同行权价 pair = OptionPair( expiry_ymd=ymd, expiry_ms=int(ems), strike=float(ck), call_inst_id=call_inst, put_inst_id=put_inst, put_strike=float(pk), ) self._oo_amp = { "high": float(amp.high), "low": float(amp.low), "mid": float(amp.mid), "range_pct": float(amp.range_pct), "hours": float(amp_hours), "max_pct": float(amp_pct), "ok": True, } self._apply_pair(pair, mark=underlying, idx=idx) if hasattr(self.ex, "cache"): from ..exchange.book_cache import BookCache cache: BookCache = self.ex.cache # type: ignore[attr-defined] cache.upsert_book(call_inst, bids=call_bids, asks=call_asks) cache.upsert_book(put_inst, bids=put_bids, asks=put_asks) hours_left = hours_until_expiry(ymd, expiry_ms=ems) return OpenPick( pair=pair, option_side="call", perp_side="", bias="option_option", call_ask=float(call_ask), put_ask=float(put_ask), option_ask=float(call_ask), option_leverage=float(min(c_lev, p_lev)), hours_left=hours_left, underlying_px=underlying, hedge_mode="option_option", call_inst_id=call_inst, put_inst_id=put_inst, call_strike=float(ck), put_strike=float(pk), call_leverage=float(c_lev), put_leverage=float(p_lev), amplitude_high=float(amp.high), amplitude_low=float(amp.low), amplitude_range_pct=float(amp.range_pct), oo_detail=( f"amp={amp.range_pct:.2f}% H={amp.high:.2f} L={amp.low:.2f} " f"C@{ck:g} P@{pk:g}" ), ) def _pick_for_open_perp(self) -> OpenPick | None: from .signal import decide, decide_fixed from .semi_auto import is_armed, is_semi_auto, read_semi_params self._last_pick_fail = None s = self.settings min_hours, min_lev, max_atm_off, atm_off_on = _strategy_floats() fixed_on, fixed_perp = _fixed_direction() # 半自动：强制看法方向 + 行权类型(itm/atm/otm) + 半自动选约门槛（须已授权） semi_on = is_semi_auto() semi_mny: str | None = None semi_otm_off: float | None = None if semi_on: if not is_armed(): self._last_pick_fail = "半自动未授权" return None sp = read_semi_params() fixed_on = True fixed_perp = str(sp["perp_side"]) min_hours = float(sp["min_option_hours"]) min_lev = float(sp["min_option_leverage"]) atm_off_on = False semi_mny = str(sp.get("moneyness") or "otm") semi_otm_off = float(sp.get("otm_max_offset") or 0) opt_side_hint = _option_side_for_perp(fixed_perp) if fixed_on else None # 第一关：振幅过滤（默认关；开启则回看窗振幅须 ≤ 最大%） if not self._apply_amplitude_first_gate(): self._last_pick_fail = "振幅门未过" return None idx = self.ex.fetch_index(s.index_inst_id) mark = self.ex.fetch_mark(s.perp_inst_id) or idx if mark is None or mark <= 0: self._last_pick_fail = "无标的价" return None underlying = float(mark) contracts = self.ex.list_option_contracts(s.option_inst_family) eligible = list_eligible_expiry_ymds(contracts, min_hours=min_hours) if not eligible: logger.info("no expiry with hours>=%.1f", min_hours) self._last_pick_fail = f"无剩余≥{min_hours:g}h 的到期" return None # 半自动：人工授权可重复开同到期，不受 one_expiry_per_day 挡住； # 仍跳过有残余腿的到期。全自动保持原「一日一到期」。 skip_expiries = _skip_expiry_ymds_for_next(include_used=not semi_on) first_skip = "" last_skip = "" def _note_skip(msg: str) -> None: nonlocal first_skip, last_skip last_skip = msg if not first_skip: first_skip = msg from .selection import _complete_by_expiry, list_otm_strikes complete = _complete_by_expiry(contracts) def _try_pair(pair: OptionPair) -> OpenPick | None: """验盘口/方向/杠杆；不合格记 skip 并返回 None（调用方换下一档）。""" if fixed_on: from .selection import is_otm side = opt_side_hint or "" if semi_on and semi_mny == "otm": if not is_otm( option_side=side, strike=pair.strike, mark_px=underlying, ): _note_skip(f"{ymd} K{pair.strike:g} 非虚值") return None if ( atm_open_offset(pair.strike, underlying) > float(semi_otm_off or 0) + 1e-9 ): _note_skip( f"{ymd} K{pair.strike:g} 偏离>" f"{float(semi_otm_off or 0):g}" ) return None elif semi_on and semi_mny == "atm": pass elif not is_itm_or_atm( option_side=side, strike=pair.strike, mark_px=underlying, ): _note_skip(f"{ymd} K{pair.strike:g} 非实值/平值") logger.info( "skip expiry=%s strike=%.0f not ITM/ATM for %s mark=%.2f", ymd, pair.strike, opt_side_hint, underlying, ) return None else: offset = atm_open_offset(pair.strike, underlying) if not atm_allows_open( pair.strike, underlying, max_offset=max_atm_off, enabled=atm_off_on, ): _note_skip(f"{ymd} ATM偏离{offset:.1f}>{max_atm_off:g}") logger.info( "skip expiry=%s strike=%.0f atm_offset=%.1f > max=%.1f", ymd, pair.strike, offset, max_atm_off, ) return None call_bids, call_asks, _ = self.ex.fetch_book(pair.call_inst_id, depth=5) put_bids, put_asks, _ = self.ex.fetch_book(pair.put_inst_id, depth=5) call_ask = call_asks[0].px if call_asks else None put_ask = put_asks[0].px if put_asks else None if call_ask is None: cq = self.ex.quote(pair.call_inst_id) call_ask = cq.ask if cq else None if put_ask is None: pq = self.ex.quote(pair.put_inst_id) put_ask = pq.ask if pq else None if fixed_on: sig = decide_fixed(call_ask, put_ask, perp_side=fixed_perp) else: sig = decide( call_ask, put_ask, strike=pair.strike, mark_px=underlying, ) if sig is None: need = "Call" if (opt_side_hint == "call") else ( "Put" if opt_side_hint == "put" else "Call/Put" ) _note_skip(f"{ymd} K{pair.strike:g} 缺{need}卖一") return None opt_ask = sig.call_ask if sig.option_side == "call" else sig.put_ask lev = option_leverage(underlying, opt_ask) hours_left = hours_until_expiry(ymd, expiry_ms=pair.expiry_ms) if lev is None or lev + 1e-9 < min_lev: _note_skip( f"{ymd} {sig.option_side.upper()}@{pair.strike:g} " f"杠杆{(f'{lev:.0f}x' if lev else 'n/a')}<{min_lev:g}x" ) logger.info( "skip expiry=%s strike=%.0f side=%s lev=%s need>=%.0f hours=%.1f", ymd, pair.strike, sig.option_side, f"{lev:.1f}" if lev else "n/a", min_lev, hours_left, ) return None self._apply_pair(pair, mark=underlying, idx=idx) from ..exchange.book_cache import BookCache if hasattr(self.ex, "cache"): cache: BookCache = self.ex.cache # type: ignore[attr-defined] cache.upsert_book(pair.call_inst_id, bids=call_bids, asks=call_asks) cache.upsert_book(pair.put_inst_id, bids=put_bids, asks=put_asks) self._last_pick_fail = None return OpenPick( pair=pair, option_side=sig.option_side, perp_side=sig.perp_side, bias=sig.bias, call_ask=float(sig.call_ask), put_ask=float(sig.put_ask), option_ask=float(opt_ask), option_leverage=float(lev), hours_left=hours_left, underlying_px=underlying, hedge_mode="perp_option", ) for ymd in eligible: if ymd in skip_expiries: _note_skip(f"{ymd} 有残余期权（跳过该到期）") logger.info( "skip expiry=%s: residual pending (semi=%s)", ymd, semi_on, ) continue # 半自动虚值：同到期内由近到远试偏离内各档，近档杠杆不够再试 1930/1940 if semi_on and semi_mny == "otm" and opt_side_hint in ("call", "put"): if ymd not in complete: _note_skip(f"{ymd} 无完整对") continue ems, strikes_map = complete[ymd] otm_ks = list_otm_strikes( list(strikes_map.keys()), underlying, option_side=str(opt_side_hint), max_offset=float(semi_otm_off or 0), ) if not otm_ks: _note_skip( f"{ymd} 无{opt_side_hint}虚值" f"（偏离≤{float(semi_otm_off or 0):g}）" ) logger.info( "skip expiry=%s no OTM within offset=%.1f for %s mark=%.2f", ymd, float(semi_otm_off or 0), opt_side_hint, underlying, ) continue for k in otm_ks: legs = strikes_map.get(float(k)) or strikes_map.get(k) if not legs or "C" not in legs or "P" not in legs: continue pair = OptionPair( expiry_ymd=ymd, expiry_ms=ems, strike=float(k), call_inst_id=legs["C"], put_inst_id=legs["P"], ) picked = _try_pair(pair) if picked is not None: return picked continue pair = select_option_pair( contracts, mark_px=underlying, expiry_ymd=ymd, option_side=opt_side_hint, moneyness=semi_mny if semi_on else None, otm_max_offset=semi_otm_off if semi_on else None, ) if pair is None: _note_skip(f"{ymd} 无合格行权价") continue picked = _try_pair(pair) if picked is not None: return picked # 报最近到期（列表最前）的原因，避免只显示远月 261225 造成误会 why = first_skip or last_skip if why: hint = "" if semi_on: hint = ( f"（半自动{opt_side_hint or '?'}·" f"{semi_mny or '?'}·≥{min_lev:g}x·≥{min_hours:g}h）" ) self._last_pick_fail = f"{why}{hint}" else: self._last_pick_fail = "合格到期均被跳过（一日一到期/残余等）" return None async def realign_async(self) -> OptionPair | None: old = self._pair pair = await asyncio.to_thread(self.align_instruments) if old is None or ( pair and ( pair.call_inst_id != old.call_inst_id or pair.put_inst_id != old.put_inst_id ) ): await self.ex.resubscribe(self._watch_ids(pair)) return pair async def pick_for_open_async(self) -> OpenPick | None: old = self._pair pick = await asyncio.to_thread(self.pick_for_open) if pick and ( old is None or pick.pair.call_inst_id != old.call_inst_id or pick.pair.put_inst_id != old.put_inst_id ): await self.ex.resubscribe(self._watch_ids(pick.pair)) return pick def _mark_for_atm(self) -> float | None: snap = self.snapshot() if snap.perp and snap.perp.mark_px: return float(snap.perp.mark_px) if snap.index_px: return float(snap.index_px) if snap.perp and snap.perp.bid and snap.perp.ask: return (float(snap.perp.bid) + float(snap.perp.ask)) / 2 return None def atm_needs_realign(self, mark_px: float | None = None) -> bool: if _hedge_mode() == "option_option": return self.oo_needs_realign() if self._pair is None: return True min_hours, _, _, _ = _strategy_floats() if ( hours_until_expiry(self._pair.expiry_ymd, expiry_ms=self._pair.expiry_ms) + 1e-9 < min_hours ): return True skip = _skip_expiry_ymds_for_next() if str(self._pair.expiry_ymd or "") in skip: return True mark = mark_px if mark_px is not None else self._mark_for_atm() if mark is None or mark <= 0: return False fixed_on, fixed_perp = _fixed_direction() # 半自动虚值/平值：监控对齐勿按「必须实值」强行重钉，否则 OTM 会一直 realign try: from .semi_auto import is_semi_auto, read_semi_params from .selection import is_otm if is_semi_auto(): sp = read_semi_params() mny = str(sp.get("moneyness") or "otm") opt = _option_side_for_perp(str(sp["perp_side"])) if mny == "otm": off = float(sp.get("otm_max_offset") or 0) if not is_otm( option_side=opt, strike=float(self._pair.strike), mark_px=float(mark), ): return True if atm_open_offset(self._pair.strike, mark) > off + 1e-9: return True return False if mny == "atm": return ( abs(float(self._pair.strike) - float(mark)) >= _ATM_DRIFT_POINTS ) except Exception: logger.debug("semi atm_needs_realign check failed", exc_info=True) if fixed_on: opt = _option_side_for_perp(fixed_perp) if not is_itm_or_atm( option_side=opt, strike=float(self._pair.strike), mark_px=float(mark) ): return True return abs(float(self._pair.strike) - float(mark)) >= _ATM_DRIFT_POINTS def oo_needs_realign(self) -> bool: if self._pair is None: return True _amp_pct, amp_hours, min_hours, _, max_dev, amp_filt = _oo_settings() if ( hours_until_expiry(self._pair.expiry_ymd, expiry_ms=self._pair.expiry_ms) + 1e-9 < min_hours ): return True skip = _skip_expiry_ymds_for_next() if str(self._pair.expiry_ymd or "") in skip: return True try: from ..exchange.candles import fetch_amplitude_hl_for_runtime from .oo_selection import select_oo_pair mark = self._mark_for_atm() if mark is None or mark <= 0: return False amp = fetch_amplitude_hl_for_runtime(amp_hours) if amp is None: return False self._oo_amp = { "high": float(amp.high), "low": float(amp.low), "mid": float(amp.mid), "range_pct": float(amp.range_pct), "hours": float(amp_hours), "max_pct": float(_amp_pct), "filter_enabled": bool(amp_filt), "ok": ( True if not amp_filt else float(amp.range_pct) <= float(_amp_pct) + 1e-12 ), } contracts = self.ex.list_option_contracts(self.settings.option_inst_family) picked = select_oo_pair( contracts, spot=float(mark), high=float(amp.high), low=float(amp.low), min_hours=float(min_hours), skip_expiry_ymds=skip, max_dev_pct=float(max_dev), ) if picked is None: return False _ymd, _ems, _ck, _pk, call_inst, put_inst = picked return ( call_inst != self._pair.call_inst_id or put_inst != self._pair.put_inst_id ) except Exception: logger.exception("oo_needs_realign failed") return False async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None: if _has_open_position(): # 持仓期间：钉住持仓行权价（禁止漂到新 ATM/虚值）；仍刷新振幅供 UI await asyncio.to_thread(self.refresh_oo_amplitude) call_id, put_id = _held_option_legs() held = call_id or _held_option_inst_id() if held and ( self._pair is None or held not in (self._pair.call_inst_id, self._pair.put_inst_id) or ( put_id and put_id not in (self._pair.call_inst_id, self._pair.put_inst_id) ) ): return await asyncio.to_thread(self.align_to_held_position) return self._pair if force or self.atm_needs_realign(): logger.info( "%s realign force=%s old_strike=%s old_exp=%s", "OO" if _hedge_mode() == "option_option" else "ATM", force, self._pair.strike if self._pair else None, self._pair.expiry_ymd if self._pair else None, ) return await self.realign_async() return self._pair def snapshot(self) -> MarketSnapshot: return self.ex.snapshot(self.settings.perp_inst_id) def snapshot_dict(self) -> dict[str, Any]: d = self.ex.snapshot_dict(self.settings.perp_inst_id) d["exchange"] = getattr(self.ex, "name", self.settings.exchange) d["perp_inst_id"] = self.settings.perp_inst_id hm = _hedge_mode() d["hedge_mode"] = hm if self._pair is not None: pd = self._pair.to_dict() d["pair"] = pd if self._oo_amp is not None: d["oo_amplitude"] = dict(self._oo_amp) if hm == "option_option": ac = dict(d.get("ask_compare") or {}) ac["bias"] = "option_option" d["ask_compare"] = ac return d def option_ladder( self, *, wings: int = 5, side: str = "call", min_hours: float = 30.0, ) -> dict[str, Any]: """ 半自动单边报价：ATM 上下若干档。 到期：剩余时长 ≥ min_hours 的最近一档（与选约门槛一致）。 """ s = self.settings wings = max(1, min(12, int(wings))) min_h = max(1.0, float(min_hours)) opt_side = "put" if str(side).strip().lower() == "put" else "call" idx = self.ex.fetch_index(s.index_inst_id) mark = self.ex.fetch_mark(s.perp_inst_id) or idx if mark is None or float(mark) <= 0: return { "ok": False, "detail": "无标的价", "rows": [], "index_px": None, "side": opt_side, "min_hours": min_h, } underlying = float(mark) contracts = self.ex.list_option_contracts(s.option_inst_family) from .selection import _complete_by_expiry, pick_atm_strike # 与半自动选约一致：只考虑剩余 ≥ min_hours 的到期，取最近一档 eligible = list_eligible_expiry_ymds(contracts, min_hours=min_h) ymd = eligible[0] if eligible else None if not ymd: return { "ok": False, "detail": f"无剩余≥{min_h:g}h 的到期", "rows": [], "index_px": underlying, "side": opt_side, "min_hours": min_h, } complete = _complete_by_expiry(contracts) if ymd not in complete: return { "ok": False, "detail": f"到期 {ymd} 无完整对", "rows": [], "index_px": underlying, "expiry_ymd": ymd, "side": opt_side, "min_hours": min_h, } _ems, strikes_map = complete[ymd] hours_left = hours_until_expiry(ymd, expiry_ms=_ems) strikes = sorted(float(k) for k in strikes_map.keys()) atm = pick_atm_strike(strikes, underlying) if atm is None: return { "ok": False, "detail": "无 ATM", "rows": [], "index_px": underlying, "expiry_ymd": ymd, "side": opt_side, } atm_i = min(range(len(strikes)), key=lambda i: abs(strikes[i] - float(atm))) lo = max(0, atm_i - wings) hi = min(len(strikes), atm_i + wings + 1) def _quote_side(inst_id: str | None) -> tuple[float | None, float | None]: if not inst_id: return None, None q = self.ex.quote(str(inst_id)) ask = float(q.ask) if q and q.ask is not None else None ask_sz = float(q.ask_sz) if q and q.ask_sz is not None else None if ask is not None: return ask, ask_sz try: _bids, asks, _ = self.ex.fetch_book(str(inst_id), depth=1) if asks: return float(asks[0].px), ( float(asks[0].sz) if asks[0].sz is not None else None ) except Exception: logger.debug("ladder fetch_book failed inst=%s", inst_id, exc_info=True) return None, None rows: list[dict[str, Any]] = [] for k in strikes[lo:hi]: legs = strikes_map[k] inst = legs.get("C" if opt_side == "call" else "P") ask, ask_sz = _quote_side(inst) off = float(k) - underlying if abs(float(k) - float(atm)) < 1e-9: tag = "atm" elif opt_side == "call": tag = "itm" if float(k) < underlying - 1e-9 else "otm" else: tag = "itm" if float(k) > underlying + 1e-9 else "otm" rows.append( { "strike": float(k), "offset": round(off, 2), "tag": tag, "ask": ask, "ask_sz": ask_sz, "lev": option_leverage(underlying, ask) if ask else None, "inst_id": inst, } ) # Call：高行权价在上（虚值在上）；Put：低行权价在上（虚值在上） if opt_side == "call": rows.sort(key=lambda r: -float(r["strike"])) else: rows.sort(key=lambda r: float(r["strike"])) return { "ok": True, "detail": "", "side": opt_side, "index_px": underlying, "expiry_ymd": ymd, "hours_left": round(float(hours_left), 1) if hours_left is not None else None, "min_hours": min_h, "atm_strike": float(atm), "rows": rows, } async def _refresh_loop(self) -> None: while True: await asyncio.sleep(30 if self._pair is not None else 10) try: idx = await asyncio.to_thread( self.ex.fetch_index, self.settings.index_inst_id ) self.ex.set_index_px(idx) mark = await asyncio.to_thread( self.ex.fetch_mark, self.settings.perp_inst_id ) if mark: self.ex.set_mark_px(self.settings.perp_inst_id, mark) if self._pair is None: await self.ensure_atm_async(force=True) else: await self.ensure_atm_async(force=False) except asyncio.CancelledError: raise except Exception as e: logger.warning("strategy align refresh failed: %s", e) def get_session() -> StrategySession: global _session if _session is None: _session = StrategySession() return _session def set_session(s: StrategySession | None) -> None: global _session _session = s # 兼容旧名 MarketGateway = StrategySession get_gateway = get_session set_gateway = set_session def bootstrap_session(settings: Settings | None = None) -> StrategySession: """main 启动：创建交易所 + 策略会话。始终以 DB 覆盖后的 runtime 为准。""" from ..exchange.runtime import load_runtime_settings # 忽略裸 get_settings()：重启后必须跟 DB 里选的交易所一致 try: s = load_runtime_settings() except Exception: s = settings or get_settings() ex = build_exchange(s) set_exchange(ex) sess = StrategySession(s, ex) set_session(sess) return sess