eth_hedge_sim/backend/app/strategy/engine.py

"""策略状态机：选向开仓 / 盯盘平仓 / 休息（无开仓窗、无轮次上限）。"""

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

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

    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)
        leverage = self.ledger.get_setting_float("leverage", s.leverage)
        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
        )
        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)
        return {
            "running": bool(row["running"]),
            "phase": row["phase"],
            "rounds_done": int(row["rounds_done"] or 0),
            "window_key": row["window_key"],
            "rest_until_ms": rest_until,
            "rest_left_sec": rest_left,
            "rest_seconds": rest_sec,
            "skip_weekends": skip_weekends,
            "exit_mode": exit_mode,
            "net_profit_target": net_target,
            "premium_exit_multiple": prem_mult,
            "exit_target_usdt": exit_amt,
            "leverage": leverage,
            "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,
            "can_open": allow_open,
            "last_error": last_error,
            "position": upl,
            "residuals": self.matcher.list_residual_options(pending_only=True),
            "ledger": self.ledger.snapshot(),
            "mode": "SIM" if s.is_sim else "LIVE",
            "sim": s.is_sim,
            "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"),
        }

    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)
        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)
                return self.state()
        self._set_state(running=1, last_error=None, phase="idle")
        self.ensure_loop()
        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._after_close()
            elif st in ("open", "option_closed_perp_pending"):
                # A：双腿（或续平永续）
                r = self.matcher.close_group(reason="emergency", bypass_liquidity=True)
                if not r.ok and st == "open":
                    # 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._after_close()

            residuals = self.matcher.settle_all_residuals_now()
            return {
                "close": {
                    "ok": ok,
                    "detail": detail,
                    "data": close_data,
                    "residuals_settled": residuals,
                },
                "state": self.state(),
            }

    def _after_close(self) -> None:
        s = get_settings()
        row = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1")
        assert row is not None
        rounds = int(row["rounds_done"] or 0) + 1
        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 _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)
                self._after_close()
                self._set_state(
                    last_error=None,
                    phase="resting",
                )
            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()
        elif r.liquidity_wait and not bypass_liquidity:
            # 等待期间若已变成远虚，下一 tick 走归档
            if 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._after_close()
                    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_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._maybe_expiry_close()
                    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)
            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_async(self) -> None:
        # 残留期权到期结算（与活跃组隔离，不挡开仓）
        await self._settle_residuals()

        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
        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 == "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._after_close()
                    self._set_state(phase="resting", last_error=None)
                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 == "open":
            upl = self.matcher.unrealized()
            expired = check_expiry_close(expiry_ms=self._position_expiry_ms(upl))
            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),
            )
            pending_close = st["phase"] in ("liquidity_wait", "closing")
            if expired.should_close or decision.should_close or pending_close:
                if expired.should_close:
                    reason = "expiry"
                    bypass = True
                    abandon = False
                    rkind = "expiry"
                else:
                    reason = decision.reason or "liquidity_retry"
                    bypass = False
                    # 目标达标（或流动性等待重试）时：远虚走只平永续
                    abandon = bool(decision.should_close or pending_close)
                    rkind = "liquidity" if pending_close else "close"
                await self._close_open_position(
                    reason=reason,
                    bypass_liquidity=bypass,
                    pending_close=pending_close,
                    abandon_if_deep_otm=abandon,
                    retry_kind=rkind,
                )
            else:
                self._set_state(phase="open", last_error=None)
            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")

        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

        self._set_state(phase="wait_signal")
        pick = await get_session().pick_for_open_async()
        if pick is None:
            self._set_state(
                last_error="无合格期权：需剩余时长、杠杆（及已开启的ATM偏差）同时满足"
            )
            return

        self._set_state(phase="opening", last_error=None)
        count = self._count_groups_for_day(wkey)
        gid = next_group_id(count)
        option_inst = (
            pick.pair.call_inst_id if pick.option_side == "call" else pick.pair.put_inst_id
        )
        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)
                return
        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:
            self._set_state(phase="open", last_error=None)
        else:
            self._set_state(phase="idle", last_error=r.detail)


_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