"""策略行情会话:在交易所适配器之上做 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 _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() -> 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() 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]: """amplitude_pct, amplitude_hours, min_option_hours, min_leverage""" s = get_settings() try: from ..models.db import get_db db = get_db() 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), ) except Exception: return ( s.oo_amplitude_pct, s.oo_amplitude_hours, s.oo_min_option_hours, s.oo_min_leverage, ) @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._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 = _held_option_inst_id() if held: ids.append(held) # 去重保序 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: """有活跃仓时:监控对锁定为持仓合约的到期/行权价。""" held = _held_option_inst_id() if not held: return None pair = pair_from_option_inst(held) if pair is None: logger.warning("cannot rebuild pair from held option %s", held) return None mark = self._mark_for_atm() or float(pair.strike) idx = None try: idx = self.ex.fetch_index(self.settings.index_inst_id) except Exception: pass 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() 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 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 = _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 if float(amp.range_pct) + 1e-12 < float(amp_pct): logger.info( "oo: amplitude %.3f%% < need %.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, ) if picked is None: logger.info("oo: no OTM call/put pair for amplitude HL") 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 行权价构造假 pair(两腿不同 strike,call/put inst 正确) pair = OptionPair( expiry_ymd=ymd, expiry_ms=int(ems), strike=float(ck), call_inst_id=call_inst, put_inst_id=put_inst, ) 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 s = self.settings min_hours, min_lev, max_atm_off, atm_off_on = _strategy_floats() fixed_on, fixed_perp = _fixed_direction() opt_side_hint = _option_side_for_perp(fixed_perp) if fixed_on else 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: 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) return None skip_expiries = _skip_expiry_ymds_for_next() for ymd in eligible: if ymd in skip_expiries: logger.info( "skip expiry=%s: used today and/or residual pending", ymd, ) continue pair = select_option_pair( contracts, mark_px=underlying, expiry_ymd=ymd, option_side=opt_side_hint, ) if pair is None: continue if fixed_on: if not is_itm_or_atm( option_side=opt_side_hint or "", strike=pair.strike, mark_px=underlying, ): logger.info( "skip expiry=%s strike=%.0f not ITM/ATM for %s mark=%.2f", ymd, pair.strike, opt_side_hint, underlying, ) continue 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, ): logger.info( "skip expiry=%s strike=%.0f atm_offset=%.1f > max=%.1f", ymd, pair.strike, offset, max_atm_off, ) continue 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 # REST 被限流时回退 WS/缓存盘口 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: continue 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: 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, ) continue self._apply_pair(pair, mark=underlying, idx=idx) # warm_and_subscribe 已写盘口;再覆盖刚拉的 ask 侧 from ..exchange.book_cache import BookCache # 直接通过 exchange quote path:再 upsert 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) 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", ) 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 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() 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 async def ensure_atm_async(self, *, force: bool = False) -> OptionPair | None: if _has_open_position(): # 持仓期间:钉住持仓行权价(禁止漂到新 ATM) held = _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) ): return await asyncio.to_thread(self.align_to_held_position) return self._pair if force or self.atm_needs_realign(): logger.info( "ATM realign force=%s old_strike=%s old_exp=%s", 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 return d 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