"""策略状态机：选向开仓 / 盯盘平仓 / 休息 / 限轮。""" from __future__ import annotations import asyncio import logging import time from typing import Any from ..config import get_settings from ..market import get_gateway from ..models.db import get_db from ..sim.ledger import Ledger from ..sim.matcher import Matcher from .clock import can_open_new, window_key from .exits import check_exits from .group import next_group_id from .signal import decide logger = logging.getLogger(__name__) class StrategyEngine: def __init__(self) -> None: self.db = get_db() self.matcher = Matcher(self.db) self.ledger = Ledger(self.db) self._task: asyncio.Task[None] | None = None self._lock = asyncio.Lock() 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_pts = self.ledger.get_setting_float("exit_move_points", s.exit_move_points) rest_sec = self.ledger.get_setting_int("rest_seconds", s.rest_seconds) max_rounds = self.ledger.get_setting_int("max_rounds", s.max_rounds) rest_until = row["rest_until_ms"] rest_left = 0 if rest_until: rest_left = max(0, int((int(rest_until) - time.time() * 1000) / 1000)) return { "running": bool(row["running"]), "phase": row["phase"], "rounds_done": int(row["rounds_done"] or 0), "max_rounds": max_rounds, "window_key": row["window_key"], "rest_until_ms": rest_until, "rest_left_sec": rest_left, "rest_seconds": rest_sec, "exit_move_points": exit_pts, "can_open": can_open_new(open_hhmm=s.open_hhmm, stop_hhmm=s.stop_open_hhmm), "last_error": row["last_error"], "position": upl, "ledger": self.ledger.snapshot(), } 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 start(self) -> dict[str, Any]: self._set_state(running=1, last_error=None, phase="idle") if self._task is None or self._task.done(): self._task = asyncio.create_task(self._loop(), name="strategy-engine") return self.state() async def pause(self) -> dict[str, Any]: self._set_state(running=0, phase="paused") return self.state() async def emergency_close(self) -> dict[str, Any]: async with self._lock: r = self.matcher.close_group(reason="emergency") if r.ok: self._after_close() return { "close": { "ok": r.ok, "detail": r.detail, "liquidity_wait": r.liquidity_wait, "data": r.data, }, "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) max_rounds = self.ledger.get_setting_int("max_rounds", s.max_rounds) rest_until = int(time.time() * 1000) + rest_sec * 1000 if rounds >= max_rounds: self._set_state( rounds_done=rounds, phase="stopped", rest_until_ms=None, ) else: self._set_state( rounds_done=rounds, phase="resting", rest_until_ms=rest_until, ) def _count_groups_for_window(self, wkey: str) -> int: # group_id like G-20260724-01 ; window_key is YYYYMMDD rows = self.db.fetchall( "SELECT group_id FROM groups WHERE group_id LIKE ?", (f"G-{wkey}-%",), ) return len(rows) 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") if not row or not int(row["running"]): await asyncio.sleep(1) continue async with self._lock: await asyncio.to_thread(self._tick) except asyncio.CancelledError: raise except Exception as e: logger.exception("strategy tick failed") self._set_state(last_error=str(e)) await asyncio.sleep(1) def _tick(self) -> None: 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, rounds_done=0, phase="idle", rest_until_ms=None) st = self.db.fetchone("SELECT * FROM strategy_state WHERE id=1") assert st is not None max_rounds = self.ledger.get_setting_int("max_rounds", s.max_rounds) exit_pts = self.ledger.get_setting_float("exit_move_points", s.exit_move_points) pos = self.matcher.current_position() # 有仓：盯平仓 if pos.get("status") == "open": self._set_state(phase="open", last_error=None) upl = self.matcher.unrealized() decision = check_exits( perp_upl=float(upl["perp_upl"]), initial_premium=float(upl["initial_premium"] or 0), move_points=float(upl["move_points"] or 0), exit_move_points=exit_pts, ) if decision.should_close: self._set_state(phase="closing") r = self.matcher.close_group(reason=decision.reason) if r.ok: self._after_close() elif r.liquidity_wait: self._set_state(phase="liquidity_wait", last_error=r.detail) else: self._set_state(last_error=r.detail) 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 int(st["rounds_done"] or 0) >= max_rounds: self._set_state(phase="stopped") return if not can_open_new(open_hhmm=s.open_hhmm, stop_hhmm=s.stop_open_hhmm): self._set_state(phase="outside_window") return if st["phase"] in ("stopped", "paused"): return # 尝试开仓 self._set_state(phase="wait_signal") gw = get_gateway() snap = gw.snapshot() if not snap.pair or not snap.call or not snap.put: return sig = decide(snap.call.ask, snap.put.ask) if sig is None: return self._set_state(phase="opening") count = self._count_groups_for_window(wkey) gid = next_group_id(count) option_inst = ( snap.pair.call_inst_id if sig.option_side == "call" else snap.pair.put_inst_id ) entry_idx = snap.index_px or (snap.perp.mark_px if snap.perp else None) if entry_idx is None: self._set_state(last_error="no index/mark for entry") return r = self.matcher.open_group( group_id=gid, bias=sig.bias, option_side=sig.option_side, perp_side=sig.perp_side, option_inst_id=option_inst, entry_index_px=float(entry_idx), strike=snap.pair.strike, expiry_ymd=snap.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(e: StrategyEngine | None) -> None: global _engine _engine = e