eth_hedge_sim/backend/app/strategy/session.py

"""策略行情会话：在交易所适配器之上做 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,
    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 _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,
        )


@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


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()
        contracts = self.ex.list_option_contracts(s.option_inst_family)
        pair = select_option_pair(contracts, mark_px=float(mark), min_hours=min_hours)
        if pair is None:
            raise RuntimeError(
                f"未找到剩余≥{min_hours}h 的 ATM Call/Put (family={s.option_inst_family})"
            )
        return self._apply_pair(pair, mark=float(mark), idx=idx)

    def pick_for_open(self) -> OpenPick | None:
        from .signal import decide

        s = self.settings
        min_hours, min_lev, max_atm_off, atm_off_on = _strategy_floats()
        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

        for ymd in eligible:
            pair = select_option_pair(contracts, mark_px=underlying, expiry_ymd=ymd)
            if pair is None:
                continue
            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
            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,
            )
        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
        mark = mark_px if mark_px is not None else self._mark_for_atm()
        if mark is None or mark <= 0:
            return False
        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