diff --git a/.env.example b/.env.example index 2c82968..540d191 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,8 @@ MODE=SIM ENV_NAME=test TZ=Asia/Shanghai +# 交易所模块：okx（已接入）| binance（占位） +EXCHANGE=okx # HTTP（前后端同端口，默认 5155） API_HOST=0.0.0.0 @@ -30,10 +32,10 @@ INDEX_INST_ID=ETH-USD FEE_RATE=0.0005 INITIAL_EQUITY=100000 -MAX_ROUNDS=3 -OPEN_HHMM=16:00 -STOP_OPEN_HHMM=08:00 -EXIT_MOVE_POINTS=30 +LEVERAGE=3 +MIN_OPTION_HOURS=12 +MIN_OPTION_LEVERAGE=100 +EXIT_MOVE_PCT=2 REST_SECONDS=300 PERP_QTY_ETH=1 OPTION_QTY_ETH=2 diff --git a/backend/app/api/sim.py b/backend/app/api/sim.py index 42536b2..0c795c7 100644 --- a/backend/app/api/sim.py +++ b/backend/app/api/sim.py @@ -52,7 +52,7 @@ async def sim_open_group( bias = "manual_" + force option_ask = pick.call_ask if force == "call" else pick.put_ask from ..config import get_settings - from ..market.instruments import option_leverage + from ..strategy.selection import option_leverage from ..sim.ledger import Ledger as Led s = get_settings() diff --git a/backend/app/config.py b/backend/app/config.py index 625e743..1568e7b 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -15,6 +15,7 @@ class Settings(BaseSettings): mode: str = "SIM" tz: str = "Asia/Shanghai" env_name: str = "test" # test / prod + exchange: str = "okx" # okx | binance（币安占位） api_host: str = "0.0.0.0" api_port: int = 5155 diff --git a/backend/app/exchange/__init__.py b/backend/app/exchange/__init__.py new file mode 100644 index 0000000..131a694 --- /dev/null +++ b/backend/app/exchange/__init__.py @@ -0,0 +1,16 @@ +"""交易所模块：OKX 已接入，币安占位。策略不直接依赖具体交易所。""" + +from .factory import build_exchange, get_exchange, set_exchange +from .protocol import ExchangeMarket +from .types import BookLevel, MarketSnapshot, OptionPair, Quote + +__all__ = [ + "BookLevel", + "ExchangeMarket", + "MarketSnapshot", + "OptionPair", + "Quote", + "build_exchange", + "get_exchange", + "set_exchange", +] diff --git a/backend/app/exchange/binance/__init__.py b/backend/app/exchange/binance/__init__.py new file mode 100644 index 0000000..ad31f01 --- /dev/null +++ b/backend/app/exchange/binance/__init__.py @@ -0,0 +1,3 @@ +from .adapter import BinanceExchange + +__all__ = ["BinanceExchange"] diff --git a/backend/app/exchange/binance/adapter.py b/backend/app/exchange/binance/adapter.py new file mode 100644 index 0000000..8c8560e --- /dev/null +++ b/backend/app/exchange/binance/adapter.py @@ -0,0 +1,62 @@ +"""币安交易所适配器占位：后期接入，接口与 OKX 对齐。""" + +from __future__ import annotations + +from typing import Any, Sequence + +from ...config import Settings, get_settings +from ..types import BookLevel, MarketSnapshot, OptionPair, Quote + + +class BinanceExchange: + name = "binance" + + def __init__(self, settings: Settings | None = None) -> None: + self.settings = settings or get_settings() + + async def start(self) -> None: + raise NotImplementedError("币安交易所模块尚未接入，请配置 EXCHANGE=okx") + + async def stop(self) -> None: + return + + def list_option_contracts(self, family: str) -> list[dict[str, Any]]: + raise NotImplementedError("BinanceExchange.list_option_contracts") + + def fetch_index(self, index_id: str) -> float | None: + raise NotImplementedError("BinanceExchange.fetch_index") + + def fetch_mark(self, inst_id: str) -> float | None: + raise NotImplementedError("BinanceExchange.fetch_mark") + + def fetch_book( + self, inst_id: str, depth: int = 5 + ) -> tuple[list[BookLevel], list[BookLevel], int | None]: + raise NotImplementedError("BinanceExchange.fetch_book") + + def get_ct_mult(self, option_inst_id: str, family: str, default: float) -> float: + return float(default) + + def set_pair(self, pair: OptionPair | None) -> None: + raise NotImplementedError("BinanceExchange.set_pair") + + def warm_and_subscribe(self, inst_ids: Sequence[str]) -> None: + raise NotImplementedError("BinanceExchange.warm_and_subscribe") + + async def resubscribe(self, inst_ids: Sequence[str]) -> None: + raise NotImplementedError("BinanceExchange.resubscribe") + + def quote(self, inst_id: str) -> Quote | None: + return None + + def snapshot(self, perp_inst_id: str) -> MarketSnapshot: + raise NotImplementedError("BinanceExchange.snapshot") + + def snapshot_dict(self, perp_inst_id: str) -> dict[str, Any]: + raise NotImplementedError("BinanceExchange.snapshot_dict") + + def set_index_px(self, px: float | None) -> None: + return + + def set_mark_px(self, inst_id: str, mark_px: float | None) -> None: + return diff --git a/backend/app/exchange/book_cache.py b/backend/app/exchange/book_cache.py new file mode 100644 index 0000000..e52e139 --- /dev/null +++ b/backend/app/exchange/book_cache.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import threading +import time +from typing import Iterable + +from .types import BookLevel, MarketSnapshot, OptionPair, Quote + + +class BookCache: + """内存盘口缓存：永续 + Call/Put。线程安全。""" + + def __init__(self) -> None: + self._lock = threading.RLock() + self._quotes: dict[str, Quote] = {} + self._index_px: float | None = None + self._pair: OptionPair | None = None + self._connected = False + self._updated_at_ms: int | None = None + + def set_connected(self, ok: bool) -> None: + with self._lock: + self._connected = bool(ok) + + def set_pair(self, pair: OptionPair | None) -> None: + with self._lock: + self._pair = pair + + def set_index_px(self, px: float | None) -> None: + with self._lock: + if px is not None and px > 0: + self._index_px = float(px) + self._touch() + + def upsert_book( + self, + inst_id: str, + *, + bids: list[BookLevel], + asks: list[BookLevel], + ts_ms: int | None = None, + ) -> None: + with self._lock: + q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) + q.bids = bids + q.asks = asks + q.bid = bids[0].px if bids else None + q.ask = asks[0].px if asks else None + q.bid_sz = bids[0].sz if bids else None + q.ask_sz = asks[0].sz if asks else None + if ts_ms is not None: + q.ts_ms = ts_ms + self._quotes[inst_id] = q + self._touch(ts_ms) + + def upsert_top( + self, + inst_id: str, + *, + bid: float | None, + ask: float | None, + bid_sz: float | None = None, + ask_sz: float | None = None, + ts_ms: int | None = None, + ) -> None: + with self._lock: + q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) + if bid is not None: + q.bid = bid + if ask is not None: + q.ask = ask + if bid_sz is not None: + q.bid_sz = bid_sz + if ask_sz is not None: + q.ask_sz = ask_sz + if ts_ms is not None: + q.ts_ms = ts_ms + # 同步一层盘口，便于 snapshot 展示 + if bid is not None and bid_sz is not None: + q.bids = [BookLevel(px=bid, sz=bid_sz)] + q.bids[1:] + if ask is not None and ask_sz is not None: + q.asks = [BookLevel(px=ask, sz=ask_sz)] + q.asks[1:] + self._quotes[inst_id] = q + self._touch(ts_ms) + + def set_mark_px(self, inst_id: str, mark_px: float | None, ts_ms: int | None = None) -> None: + with self._lock: + if mark_px is None or mark_px <= 0: + return + q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) + q.mark_px = float(mark_px) + if ts_ms is not None: + q.ts_ms = ts_ms + self._quotes[inst_id] = q + self._touch(ts_ms) + + def get(self, inst_id: str) -> Quote | None: + with self._lock: + return self._quotes.get(inst_id) + + def drop_except(self, keep: Iterable[str]) -> None: + keep_set = set(keep) + with self._lock: + for k in list(self._quotes): + if k not in keep_set: + del self._quotes[k] + + def snapshot(self, perp_inst_id: str) -> MarketSnapshot: + with self._lock: + pair = self._pair + call = self._quotes.get(pair.call_inst_id) if pair else None + put = self._quotes.get(pair.put_inst_id) if pair else None + return MarketSnapshot( + perp=self._quotes.get(perp_inst_id), + call=call, + put=put, + index_px=self._index_px, + pair=pair, + connected=self._connected, + updated_at_ms=self._updated_at_ms, + ) + + def _touch(self, ts_ms: int | None = None) -> None: + self._updated_at_ms = int(ts_ms) if ts_ms is not None else int(time.time() * 1000) diff --git a/backend/app/exchange/factory.py b/backend/app/exchange/factory.py new file mode 100644 index 0000000..86599a6 --- /dev/null +++ b/backend/app/exchange/factory.py @@ -0,0 +1,34 @@ +"""按配置创建交易所实例。""" + +from __future__ import annotations + +from ..config import Settings, get_settings +from .protocol import ExchangeMarket + +_exchange: ExchangeMarket | None = None + + +def build_exchange(settings: Settings | None = None) -> ExchangeMarket: + s = settings or get_settings() + name = (s.exchange or "okx").strip().lower() + if name == "okx": + from .okx.adapter import OkxExchange + + return OkxExchange(s) + if name in ("binance", "bn"): + from .binance.adapter import BinanceExchange + + return BinanceExchange(s) + raise ValueError(f"未知交易所 EXCHANGE={s.exchange!r}，支持 okx / binance") + + +def get_exchange() -> ExchangeMarket: + global _exchange + if _exchange is None: + _exchange = build_exchange() + return _exchange + + +def set_exchange(ex: ExchangeMarket | None) -> None: + global _exchange + _exchange = ex diff --git a/backend/app/exchange/okx/__init__.py b/backend/app/exchange/okx/__init__.py new file mode 100644 index 0000000..7e5742f --- /dev/null +++ b/backend/app/exchange/okx/__init__.py @@ -0,0 +1,3 @@ +from .adapter import OkxExchange + +__all__ = ["OkxExchange"] diff --git a/backend/app/exchange/okx/adapter.py b/backend/app/exchange/okx/adapter.py new file mode 100644 index 0000000..4f2a0f6 --- /dev/null +++ b/backend/app/exchange/okx/adapter.py @@ -0,0 +1,108 @@ +"""OKX 交易所适配器：只负责行情与合约，不含策略选约。""" + +from __future__ import annotations + +import logging +from typing import Any, Sequence + +from ...config import Settings, get_settings +from ..book_cache import BookCache +from ..types import BookLevel, MarketSnapshot, OptionPair, Quote +from .parse import rows_to_option_contracts, safe_float +from .rest import OkxRestClient +from .ws import OkxPublicWs + +logger = logging.getLogger(__name__) + + +class OkxExchange: + name = "okx" + + def __init__(self, settings: Settings | None = None) -> None: + self.settings = settings or get_settings() + self.cache = BookCache() + proxy = self.settings.okx_http_proxy or None + self.rest = OkxRestClient(self.settings.okx_rest_base, proxy=proxy) + self.ws = OkxPublicWs(self.settings.okx_ws_public, self.cache, proxy=proxy) + self._started = False + self._ct_cache: dict[str, float] = {} + + async def start(self) -> None: + if self._started: + return + self._started = True + await self.ws.start() + logger.info("OKX exchange started") + + async def stop(self) -> None: + self._started = False + await self.ws.stop() + self.rest.close() + logger.info("OKX exchange stopped") + + def list_option_contracts(self, family: str) -> list[dict[str, Any]]: + rows = self.rest.fetch_option_instruments(family) + contracts = rows_to_option_contracts(rows) + for c in contracts: + if c.get("ct_mult"): + self._ct_cache[str(c["inst_id"])] = float(c["ct_mult"]) + return contracts + + def fetch_index(self, index_id: str) -> float | None: + return self.rest.fetch_index_ticker(index_id) + + def fetch_mark(self, inst_id: str) -> float | None: + return self.rest.fetch_mark_price(inst_id) + + def fetch_book( + self, inst_id: str, depth: int = 5 + ) -> tuple[list[BookLevel], list[BookLevel], int | None]: + return self.rest.fetch_books(inst_id, sz=depth) + + def get_ct_mult(self, option_inst_id: str, family: str, default: float) -> float: + if option_inst_id in self._ct_cache: + return self._ct_cache[option_inst_id] + try: + rows = self.rest.fetch_instruments(inst_type="OPTION", inst_family=family) + for r in rows: + if str(r.get("instId")) == option_inst_id: + m = safe_float(r.get("ctMult")) + if m and m > 0: + self._ct_cache[option_inst_id] = float(m) + return float(m) + except Exception: + pass + return float(default) + + def set_pair(self, pair: OptionPair | None) -> None: + self.cache.set_pair(pair) + + def warm_and_subscribe(self, inst_ids: Sequence[str]) -> None: + ids = [i for i in inst_ids if i] + for inst in ids: + bids, asks, ts = self.rest.fetch_books(inst, sz=5) + self.cache.upsert_book(inst, bids=bids, asks=asks, ts_ms=ts) + mp = self.rest.fetch_mark_price(inst) + if mp: + self.cache.set_mark_px(inst, mp) + keep = set(ids) + self.cache.drop_except(keep) + self.ws.set_instruments(ids) + + async def resubscribe(self, inst_ids: Sequence[str]) -> None: + await self.ws.resubscribe([i for i in inst_ids if i]) + + def quote(self, inst_id: str) -> Quote | None: + return self.cache.get(inst_id) + + def snapshot(self, perp_inst_id: str) -> MarketSnapshot: + return self.cache.snapshot(perp_inst_id) + + def snapshot_dict(self, perp_inst_id: str) -> dict[str, Any]: + return self.snapshot(perp_inst_id).to_dict() + + def set_index_px(self, px: float | None) -> None: + self.cache.set_index_px(px) + + def set_mark_px(self, inst_id: str, mark_px: float | None) -> None: + self.cache.set_mark_px(inst_id, mark_px) diff --git a/backend/app/exchange/okx/parse.py b/backend/app/exchange/okx/parse.py new file mode 100644 index 0000000..8a27a0a --- /dev/null +++ b/backend/app/exchange/okx/parse.py @@ -0,0 +1,76 @@ +"""OKX 合约 ID / 到期解析（交易所专属）。""" + +from __future__ import annotations + +import re +from datetime import datetime, timezone +from typing import Any + +_DATE_RE = re.compile(r"^\d{6}$") + + +def safe_float(v: Any) -> float | None: + if v is None or v == "": + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +def parse_option_inst_id(inst_id: str) -> tuple[str | None, float | None, str | None]: + """ETH-USD_UM-YYMMDD-STRIKE-C → (YYMMDD, strike, C|P).""" + parts = (inst_id or "").strip().split("-") + if len(parts) < 5: + return None, None, None + ymd = parts[-3] + strike = safe_float(parts[-2]) + opt = parts[-1].upper() + if not _DATE_RE.fullmatch(ymd) or strike is None or opt not in ("C", "P"): + return None, None, None + return ymd, strike, opt + + +def expiry_ms_from_ymd(ymd: str) -> int: + """OKX 期权到期：当日 08:00 UTC = 上海 16:00。""" + yy, mm, dd = int(ymd[0:2]), int(ymd[2:4]), int(ymd[4:6]) + dt = datetime(2000 + yy, mm, dd, 8, 0, 0, tzinfo=timezone.utc) + return int(dt.timestamp() * 1000) + + +def rows_to_option_contracts(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """ + 归一化为策略层可用的中性结构： + {inst_id, expiry_ymd, strike, side, ct_mult} + """ + out: list[dict[str, Any]] = [] + for row in rows: + if not isinstance(row, dict): + continue + state = str(row.get("state") or "live").lower() + if state and state != "live": + continue + inst_id = str(row.get("instId") or "") + y, stk, opt = parse_option_inst_id(inst_id) + if y is None or stk is None or opt is None: + exp = safe_float(row.get("expTime")) + if exp: + ms = int(exp) if exp > 10_000_000_000 else int(exp * 1000) + y = datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%y%m%d") + stk = safe_float(row.get("stk")) + opt_raw = str(row.get("optType") or "").upper() + opt = opt_raw if opt_raw in ("C", "P") else None + if not inst_id or not y or stk is None or opt not in ("C", "P"): + continue + ct = safe_float(row.get("ctMult")) + out.append( + { + "inst_id": inst_id, + "expiry_ymd": y, + "expiry_ms": expiry_ms_from_ymd(y), + "strike": float(stk), + "side": opt, + "ct_mult": float(ct) if ct and ct > 0 else None, + } + ) + return out diff --git a/backend/app/exchange/okx/rest.py b/backend/app/exchange/okx/rest.py new file mode 100644 index 0000000..2cecf80 --- /dev/null +++ b/backend/app/exchange/okx/rest.py @@ -0,0 +1,99 @@ +"""OKX REST 只读行情。不调用任何交易类接口。""" + +from __future__ import annotations + +from typing import Any + +import httpx + +from ..types import BookLevel +from .parse import safe_float + + +class OkxRestClient: + def __init__( + self, + base_url: str = "https://www.okx.com", + timeout: float = 15.0, + proxy: str | None = None, + ) -> None: + self.base_url = base_url.rstrip("/") + self.proxy = (proxy or "").strip() or None + self._client = httpx.Client( + base_url=self.base_url, + timeout=timeout, + proxy=self.proxy, + headers={"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.1"}, + ) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> OkxRestClient: + return self + + def __exit__(self, *args: object) -> None: + self.close() + + def _get(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: + r = self._client.get(path, params=params or {}) + r.raise_for_status() + body = r.json() + if str(body.get("code")) != "0": + raise RuntimeError(f"OKX REST error code={body.get('code')} msg={body.get('msg')}") + data = body.get("data") or [] + return [x for x in data if isinstance(x, dict)] + + def fetch_instruments(self, *, inst_type: str, inst_family: str | None = None) -> list[dict[str, Any]]: + params: dict[str, Any] = {"instType": inst_type} + if inst_family: + params["instFamily"] = inst_family + return self._get("/api/v5/public/instruments", params) + + def fetch_option_instruments(self, inst_family: str) -> list[dict[str, Any]]: + rows = self.fetch_instruments(inst_type="OPTION", inst_family=inst_family) + return [r for r in rows if str(r.get("state") or "").lower() == "live"] + + def fetch_index_ticker(self, inst_id: str) -> float | None: + rows = self._get("/api/v5/market/index-tickers", {"instId": inst_id}) + if not rows: + return None + return safe_float(rows[0].get("idxPx")) + + def fetch_mark_price(self, inst_id: str) -> float | None: + rows = self._get("/api/v5/public/mark-price", {"instId": inst_id}) + if not rows: + t = self._get("/api/v5/market/ticker", {"instId": inst_id}) + if not t: + return None + return safe_float(t[0].get("markPx")) or safe_float(t[0].get("last")) + return safe_float(rows[0].get("markPx")) + + def fetch_books(self, inst_id: str, sz: int = 5) -> tuple[list[BookLevel], list[BookLevel], int | None]: + rows = self._get( + "/api/v5/market/books", + {"instId": inst_id, "sz": str(max(1, min(int(sz), 400)))}, + ) + if not rows: + return [], [], None + row = rows[0] + ts = safe_float(row.get("ts")) + ts_ms = int(ts) if ts is not None else None + return ( + _levels(row.get("bids") or []), + _levels(row.get("asks") or []), + ts_ms, + ) + + +def _levels(raw: list[Any]) -> list[BookLevel]: + out: list[BookLevel] = [] + for item in raw: + if not isinstance(item, (list, tuple)) or len(item) < 2: + continue + px = safe_float(item[0]) + sz = safe_float(item[1]) + if px is None or sz is None or px <= 0 or sz <= 0: + continue + out.append(BookLevel(px=px, sz=sz)) + return out diff --git a/backend/app/exchange/okx/ws.py b/backend/app/exchange/okx/ws.py new file mode 100644 index 0000000..b31ef70 --- /dev/null +++ b/backend/app/exchange/okx/ws.py @@ -0,0 +1,207 @@ +"""OKX 公共 WebSocket：永续 + 期权 books5 / mark-price。只读。""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Any +from urllib.parse import urlparse + +import websockets +from websockets.asyncio.client import ClientConnection + +from ..book_cache import BookCache +from ..types import BookLevel +from .parse import safe_float + +logger = logging.getLogger(__name__) + + +class OkxPublicWs: + def __init__( + self, + url: str, + cache: BookCache, + *, + proxy: str | None = None, + ping_interval: float = 20.0, + ) -> None: + self.url = url + self.cache = cache + self.proxy = (proxy or "").strip() or None + self.ping_interval = ping_interval + self._inst_ids: list[str] = [] + self._task: asyncio.Task[None] | None = None + self._stop = asyncio.Event() + self._subscribed: set[str] = set() + + def set_instruments(self, inst_ids: list[str]) -> None: + self._inst_ids = [i for i in inst_ids if i] + + async def start(self) -> None: + if self._task and not self._task.done(): + return + self._stop.clear() + self._task = asyncio.create_task(self._run_forever(), name="okx-public-ws") + + async def stop(self) -> None: + self._stop.set() + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + self.cache.set_connected(False) + + async def resubscribe(self, inst_ids: list[str]) -> None: + self.set_instruments(inst_ids) + self._stop.set() + await asyncio.sleep(0) + self._stop.clear() + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = asyncio.create_task(self._run_forever(), name="okx-public-ws") + + async def _open_connection(self) -> ClientConnection: + if not self.proxy: + return await websockets.connect( + self.url, + ping_interval=None, + max_size=2**22, + open_timeout=20, + ) + + from python_socks.async_.asyncio import Proxy + + parsed = urlparse(self.url) + host = parsed.hostname or "ws.okx.com" + port = parsed.port or (443 if parsed.scheme == "wss" else 80) + sock = await Proxy.from_url(self.proxy).connect(dest_host=host, dest_port=port) + return await websockets.connect( + self.url, + sock=sock, + server_hostname=host, + ping_interval=None, + max_size=2**22, + open_timeout=20, + ) + + async def _run_forever(self) -> None: + backoff = 1.0 + while not self._stop.is_set(): + try: + async with await self._open_connection() as ws: + self.cache.set_connected(True) + backoff = 1.0 + await self._subscribe(ws) + waiter = asyncio.create_task(self._stop.wait()) + reader = asyncio.create_task(self._read_loop(ws)) + pinger = asyncio.create_task(self._ping_loop(ws)) + done, pending = await asyncio.wait( + {waiter, reader, pinger}, + return_when=asyncio.FIRST_COMPLETED, + ) + for t in pending: + t.cancel() + for t in done: + exc = t.exception() + if exc and not isinstance(exc, asyncio.CancelledError): + raise exc + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning("OKX WS disconnected: %s", e) + self.cache.set_connected(False) + try: + await asyncio.wait_for(self._stop.wait(), timeout=backoff) + break + except asyncio.TimeoutError: + backoff = min(backoff * 2, 30.0) + + self.cache.set_connected(False) + + async def _subscribe(self, ws: ClientConnection) -> None: + args: list[dict[str, str]] = [] + for inst in self._inst_ids: + args.append({"channel": "books5", "instId": inst}) + args.append({"channel": "mark-price", "instId": inst}) + if not args: + return + payload = {"op": "subscribe", "args": args} + await ws.send(json.dumps(payload)) + self._subscribed = {a["instId"] for a in args} + logger.info("OKX WS subscribed: %s", sorted(self._subscribed)) + + async def _ping_loop(self, ws: ClientConnection) -> None: + while True: + await asyncio.sleep(self.ping_interval) + await ws.send("ping") + + async def _read_loop(self, ws: ClientConnection) -> None: + try: + async for raw in ws: + if raw == "pong": + continue + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="ignore") + if raw == "ping": + await ws.send("pong") + continue + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + self._handle_message(msg) + except websockets.exceptions.ConnectionClosed: + return + + def _handle_message(self, msg: dict[str, Any]) -> None: + if msg.get("event") in ("subscribe", "error", "channel-conn-count"): + if msg.get("event") == "error": + logger.error("OKX WS error: %s", msg) + return + arg = msg.get("arg") or {} + channel = str(arg.get("channel") or "") + inst_id = str(arg.get("instId") or "") + data = msg.get("data") or [] + if not inst_id or not data: + return + row = data[0] if isinstance(data[0], dict) else None + if row is None: + return + + if channel == "books5": + ts = safe_float(row.get("ts")) + self.cache.upsert_book( + inst_id, + bids=_levels(row.get("bids") or []), + asks=_levels(row.get("asks") or []), + ts_ms=int(ts) if ts is not None else None, + ) + elif channel == "mark-price": + ts = safe_float(row.get("ts")) + self.cache.set_mark_px( + inst_id, + safe_float(row.get("markPx")), + ts_ms=int(ts) if ts is not None else None, + ) + + +def _levels(raw: list[Any]) -> list[BookLevel]: + out: list[BookLevel] = [] + for item in raw: + if not isinstance(item, (list, tuple)) or len(item) < 2: + continue + px = safe_float(item[0]) + sz = safe_float(item[1]) + if px is None or sz is None or px <= 0 or sz <= 0: + continue + out.append(BookLevel(px=px, sz=sz)) + return out diff --git a/backend/app/exchange/protocol.py b/backend/app/exchange/protocol.py new file mode 100644 index 0000000..08333b3 --- /dev/null +++ b/backend/app/exchange/protocol.py @@ -0,0 +1,48 @@ +"""交易所行情适配器协议：策略/撮合只依赖此接口，不直接碰 OKX/币安。""" + +from __future__ import annotations + +from typing import Any, Protocol, Sequence, runtime_checkable + +from .types import BookLevel, MarketSnapshot, OptionPair, Quote + + +@runtime_checkable +class ExchangeMarket(Protocol): + name: str + + async def start(self) -> None: ... + + async def stop(self) -> None: ... + + def list_option_contracts(self, family: str) -> list[dict[str, Any]]: + """中性期权合约列表：inst_id/expiry_ymd/strike/side/ct_mult。""" + ... + + def fetch_index(self, index_id: str) -> float | None: ... + + def fetch_mark(self, inst_id: str) -> float | None: ... + + def fetch_book( + self, inst_id: str, depth: int = 5 + ) -> tuple[list[BookLevel], list[BookLevel], int | None]: ... + + def get_ct_mult(self, option_inst_id: str, family: str, default: float) -> float: ... + + def set_pair(self, pair: OptionPair | None) -> None: ... + + def warm_and_subscribe(self, inst_ids: Sequence[str]) -> None: + """REST 预热盘口 + 设置 WS 订阅列表。""" + ... + + async def resubscribe(self, inst_ids: Sequence[str]) -> None: ... + + def quote(self, inst_id: str) -> Quote | None: ... + + def snapshot(self, perp_inst_id: str) -> MarketSnapshot: ... + + def snapshot_dict(self, perp_inst_id: str) -> dict[str, Any]: ... + + def set_index_px(self, px: float | None) -> None: ... + + def set_mark_px(self, inst_id: str, mark_px: float | None) -> None: ... diff --git a/backend/app/exchange/types.py b/backend/app/exchange/types.py new file mode 100644 index 0000000..ca4b5a7 --- /dev/null +++ b/backend/app/exchange/types.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(slots=True) +class BookLevel: + px: float + sz: float # OKX 张数 / 合约张数口径 + + +@dataclass(slots=True) +class Quote: + inst_id: str + bid: float | None = None + ask: float | None = None + bid_sz: float | None = None + ask_sz: float | None = None + mark_px: float | None = None + ts_ms: int | None = None + bids: list[BookLevel] = field(default_factory=list) + asks: list[BookLevel] = field(default_factory=list) + + def to_dict(self, *, depth: int = 5) -> dict[str, Any]: + return { + "inst_id": self.inst_id, + "bid": self.bid, + "ask": self.ask, + "bid_sz": self.bid_sz, + "ask_sz": self.ask_sz, + "mark_px": self.mark_px, + "ts_ms": self.ts_ms, + "bids": [{"px": x.px, "sz": x.sz} for x in self.bids[:depth]], + "asks": [{"px": x.px, "sz": x.sz} for x in self.asks[:depth]], + } + + +@dataclass(slots=True) +class OptionPair: + expiry_ymd: str # YYMMDD + expiry_ms: int + strike: float + call_inst_id: str + put_inst_id: str + + def to_dict(self) -> dict[str, Any]: + return { + "expiry_ymd": self.expiry_ymd, + "expiry_ms": self.expiry_ms, + "strike": self.strike, + "call_inst_id": self.call_inst_id, + "put_inst_id": self.put_inst_id, + } + + +@dataclass(slots=True) +class MarketSnapshot: + perp: Quote | None + call: Quote | None + put: Quote | None + index_px: float | None + pair: OptionPair | None + connected: bool + updated_at_ms: int | None + + def to_dict(self) -> dict[str, Any]: + return { + "connected": self.connected, + "updated_at_ms": self.updated_at_ms, + "index_px": self.index_px, + "pair": self.pair.to_dict() if self.pair else None, + "perp": self.perp.to_dict() if self.perp else None, + "call": self.call.to_dict() if self.call else None, + "put": self.put.to_dict() if self.put else None, + "ask_compare": { + "call_ask": self.call.ask if self.call else None, + "put_ask": self.put.ask if self.put else None, + "bias": _ask_bias(self.call, self.put), + }, + } + + +def _ask_bias(call: Quote | None, put: Quote | None) -> str: + """卖一比价仅用于选向展示；相等则 wait。""" + ca = call.ask if call else None + pa = put.ask if put else None + if ca is None or pa is None: + return "unknown" + if ca > pa: + return "call_ask_gt_put" # 永续多 + 期权空（腿待拍板） + if ca < pa: + return "put_ask_gt_call" # 永续空 + 期权多（腿待拍板） + return "equal" diff --git a/backend/app/main.py b/backend/app/main.py index 64c06b6..37f0781 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -11,9 +11,9 @@ from fastapi.staticfiles import StaticFiles from .api import router as api_router from .config import get_settings -from .market import MarketGateway, set_gateway from .models.db import Database, set_db from .strategy import StrategyEngine, set_engine +from .strategy.session import bootstrap_session logging.basicConfig( level=logging.INFO, @@ -36,13 +36,15 @@ async def lifespan(app: FastAPI): engine = StrategyEngine() set_engine(engine) - gw = MarketGateway(settings) - set_gateway(gw) + session = bootstrap_session(settings) try: - await gw.start() - logger.info("market gateway started (SIM)") + await session.start() + logger.info( + "exchange=%s strategy session started (SIM)", + settings.exchange, + ) except Exception: - logger.exception("market gateway failed to start") + logger.exception("strategy session failed to start") yield @@ -53,8 +55,12 @@ async def lifespan(app: FastAPI): await engine._task except Exception: pass - await gw.stop() - set_gateway(None) + await session.stop() + from .exchange import set_exchange + from .strategy.session import set_session + + set_session(None) + set_exchange(None) set_engine(None) db.close() set_db(None) @@ -78,12 +84,12 @@ app.include_router(api_router) @app.get("/health") async def health() -> dict: - from .market import get_gateway from .strategy import get_engine + from .strategy.session import get_session settings = get_settings() - gw = get_gateway() - snap = gw.snapshot() + sess = get_session() + snap = sess.snapshot() try: st = get_engine().state() except Exception: @@ -92,6 +98,7 @@ async def health() -> dict: "ok": True, "mode": settings.mode, "env_name": settings.env_name, + "exchange": settings.exchange, "sim": settings.is_sim, "market_connected": snap.connected, "pair": snap.pair.to_dict() if snap.pair else None, diff --git a/backend/app/market/__init__.py b/backend/app/market/__init__.py index 6c6283a..b4cecba 100644 --- a/backend/app/market/__init__.py +++ b/backend/app/market/__init__.py @@ -1,22 +1,31 @@ -"""OKX 实盘只读行情网关。""" +"""兼容层：行情入口转发到 exchange + strategy.session。""" -from .book_cache import BookCache -from .gateway import MarketGateway, get_gateway, set_gateway -from .instruments import next_session_expiry_ymd, select_option_pair -from .okx_rest import OkxRestClient -from .okx_ws import OkxPublicWs -from .types import MarketSnapshot, OptionPair, Quote +from ..exchange.types import BookLevel, MarketSnapshot, OptionPair, Quote +from ..strategy.session import ( + MarketGateway, + OpenPick, + StrategySession, + get_gateway, + get_session, + set_gateway, + set_session, + bootstrap_session, +) +from ..strategy.selection import next_session_expiry_ymd, select_option_pair __all__ = [ - "BookCache", + "BookLevel", "MarketGateway", "MarketSnapshot", - "OkxPublicWs", - "OkxRestClient", + "OpenPick", "OptionPair", "Quote", + "StrategySession", + "bootstrap_session", "get_gateway", + "get_session", "next_session_expiry_ymd", "select_option_pair", "set_gateway", + "set_session", ] diff --git a/backend/app/market/book_cache.py b/backend/app/market/book_cache.py index e52e139..8513ebb 100644 --- a/backend/app/market/book_cache.py +++ b/backend/app/market/book_cache.py @@ -1,124 +1,5 @@ -from __future__ import annotations +"""兼容层：BookCache 在 exchange.book_cache。""" -import threading -import time -from typing import Iterable +from ..exchange.book_cache import BookCache -from .types import BookLevel, MarketSnapshot, OptionPair, Quote - - -class BookCache: - """内存盘口缓存：永续 + Call/Put。线程安全。""" - - def __init__(self) -> None: - self._lock = threading.RLock() - self._quotes: dict[str, Quote] = {} - self._index_px: float | None = None - self._pair: OptionPair | None = None - self._connected = False - self._updated_at_ms: int | None = None - - def set_connected(self, ok: bool) -> None: - with self._lock: - self._connected = bool(ok) - - def set_pair(self, pair: OptionPair | None) -> None: - with self._lock: - self._pair = pair - - def set_index_px(self, px: float | None) -> None: - with self._lock: - if px is not None and px > 0: - self._index_px = float(px) - self._touch() - - def upsert_book( - self, - inst_id: str, - *, - bids: list[BookLevel], - asks: list[BookLevel], - ts_ms: int | None = None, - ) -> None: - with self._lock: - q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) - q.bids = bids - q.asks = asks - q.bid = bids[0].px if bids else None - q.ask = asks[0].px if asks else None - q.bid_sz = bids[0].sz if bids else None - q.ask_sz = asks[0].sz if asks else None - if ts_ms is not None: - q.ts_ms = ts_ms - self._quotes[inst_id] = q - self._touch(ts_ms) - - def upsert_top( - self, - inst_id: str, - *, - bid: float | None, - ask: float | None, - bid_sz: float | None = None, - ask_sz: float | None = None, - ts_ms: int | None = None, - ) -> None: - with self._lock: - q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) - if bid is not None: - q.bid = bid - if ask is not None: - q.ask = ask - if bid_sz is not None: - q.bid_sz = bid_sz - if ask_sz is not None: - q.ask_sz = ask_sz - if ts_ms is not None: - q.ts_ms = ts_ms - # 同步一层盘口，便于 snapshot 展示 - if bid is not None and bid_sz is not None: - q.bids = [BookLevel(px=bid, sz=bid_sz)] + q.bids[1:] - if ask is not None and ask_sz is not None: - q.asks = [BookLevel(px=ask, sz=ask_sz)] + q.asks[1:] - self._quotes[inst_id] = q - self._touch(ts_ms) - - def set_mark_px(self, inst_id: str, mark_px: float | None, ts_ms: int | None = None) -> None: - with self._lock: - if mark_px is None or mark_px <= 0: - return - q = self._quotes.get(inst_id) or Quote(inst_id=inst_id) - q.mark_px = float(mark_px) - if ts_ms is not None: - q.ts_ms = ts_ms - self._quotes[inst_id] = q - self._touch(ts_ms) - - def get(self, inst_id: str) -> Quote | None: - with self._lock: - return self._quotes.get(inst_id) - - def drop_except(self, keep: Iterable[str]) -> None: - keep_set = set(keep) - with self._lock: - for k in list(self._quotes): - if k not in keep_set: - del self._quotes[k] - - def snapshot(self, perp_inst_id: str) -> MarketSnapshot: - with self._lock: - pair = self._pair - call = self._quotes.get(pair.call_inst_id) if pair else None - put = self._quotes.get(pair.put_inst_id) if pair else None - return MarketSnapshot( - perp=self._quotes.get(perp_inst_id), - call=call, - put=put, - index_px=self._index_px, - pair=pair, - connected=self._connected, - updated_at_ms=self._updated_at_ms, - ) - - def _touch(self, ts_ms: int | None = None) -> None: - self._updated_at_ms = int(ts_ms) if ts_ms is not None else int(time.time() * 1000) +__all__ = ["BookCache"] diff --git a/backend/app/market/gateway.py b/backend/app/market/gateway.py index 7e29b68..a5baa5f 100644 --- a/backend/app/market/gateway.py +++ b/backend/app/market/gateway.py @@ -1,330 +1,23 @@ -"""行情网关：REST 对齐合约 + WS 推送盘口。""" +"""兼容层：转发到 strategy.session。""" -from __future__ import annotations - -import asyncio -import logging -from dataclasses import dataclass -from typing import Any - -from ..config import Settings, get_settings -from .book_cache import BookCache -from .instruments import ( - hours_until_expiry, - list_eligible_expiry_ymds, - option_leverage, - select_option_pair, +from ..strategy.session import ( + MarketGateway, + OpenPick, + StrategySession, + bootstrap_session, + get_gateway, + get_session, + set_gateway, + set_session, ) -from .okx_rest import OkxRestClient -from .okx_ws import OkxPublicWs -from .types import MarketSnapshot, OptionPair -logger = logging.getLogger(__name__) - -# 展示用：现价偏离当前行权超过该点数则重选 ATM（空仓） -_ATM_DRIFT_POINTS = 5.0 - - -def _has_open_position() -> bool: - try: - from ..models.db import get_db - - row = get_db().fetchone("SELECT status FROM positions WHERE id=1") - return bool(row and row["status"] == "open") - except Exception: - return False - - -def _strategy_floats() -> tuple[float, float]: - """(min_option_hours, min_option_leverage)""" - 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 - ) - return hours, lev - except Exception: - return s.min_option_hours, s.min_option_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 - - -class MarketGateway: - def __init__(self, settings: Settings | None = None) -> None: - self.settings = settings or get_settings() - self.cache = BookCache() - proxy = self.settings.okx_http_proxy or None - self.rest = OkxRestClient(self.settings.okx_rest_base, proxy=proxy) - self.ws = OkxPublicWs(self.settings.okx_ws_public, self.cache, proxy=proxy) - 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 - - async def start(self) -> None: - if self._started: - return - self._started = True - await asyncio.to_thread(self.align_instruments) - await self.ws.start() - self._refresh_task = asyncio.create_task(self._refresh_loop(), name="market-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.ws.stop() - self.rest.close() - - def _apply_pair(self, pair: OptionPair, *, mark: float, idx: float | None) -> OptionPair: - s = self.settings - self._pair = pair - self.cache.set_pair(pair) - if idx is not None: - self.cache.set_index_px(idx) - - for inst in (s.perp_inst_id, pair.call_inst_id, pair.put_inst_id): - bids, asks, ts = self.rest.fetch_books(inst, sz=5) - self.cache.upsert_book(inst, bids=bids, asks=asks, ts_ms=ts) - mp = self.rest.fetch_mark_price(inst) - if mp: - self.cache.set_mark_px(inst, mp) - - keep = {s.perp_inst_id, pair.call_inst_id, pair.put_inst_id} - self.cache.drop_except(keep) - self.ws.set_instruments([s.perp_inst_id, pair.call_inst_id, pair.put_inst_id]) - logger.info( - "aligned pair expiry=%s strike=%s call=%s put=%s mark=%.2f hours=%.1f", - pair.expiry_ymd, - pair.strike, - pair.call_inst_id, - pair.put_inst_id, - mark, - hours_until_expiry(pair.expiry_ymd), - ) - return pair - - def align_instruments(self) -> OptionPair | None: - """空仓展示：选剩余时长合格的最近到期 ATM（不校验期权杠杆）。""" - s = self.settings - idx = self.rest.fetch_index_ticker(s.index_inst_id) - mark = self.rest.fetch_mark_price(s.perp_inst_id) or idx - if mark is None or mark <= 0: - raise RuntimeError("无法获取 ETH 标记/指数价格，无法选 ATM") - - min_hours, _ = _strategy_floats() - instruments = self.rest.fetch_option_instruments(s.option_inst_family) - pair = select_option_pair( - instruments, 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: - """ - 开仓选约： - 1) 剩余时长 ≥ min_hours 的到期日（由近到远） - 2) 该到期 ATM 平值 - 3) 卖一比价定方向后校验 现价/卖一 ≥ min_option_leverage - """ - s = self.settings - min_hours, min_lev = _strategy_floats() - idx = self.rest.fetch_index_ticker(s.index_inst_id) - mark = self.rest.fetch_mark_price(s.perp_inst_id) or idx - if mark is None or mark <= 0: - return None - underlying = float(mark) - instruments = self.rest.fetch_option_instruments(s.option_inst_family) - eligible = list_eligible_expiry_ymds(instruments, min_hours=min_hours) - if not eligible: - logger.info("no expiry with hours>=%.1f", min_hours) - return None - - from ..strategy.signal import decide - - for ymd in eligible: - pair = select_option_pair( - instruments, mark_px=underlying, expiry_ymd=ymd - ) - if pair is None: - continue - call_bids, call_asks, _ = self.rest.fetch_books(pair.call_inst_id, sz=5) - put_bids, put_asks, _ = self.rest.fetch_books(pair.put_inst_id, sz=5) - call_ask = call_asks[0].px if call_asks else None - put_ask = put_asks[0].px if put_asks else None - sig = decide(call_ask, put_ask) - 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) - 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) - # 写入刚拉的盘口，避免 WS 尚未推送 - self.cache.upsert_book(pair.call_inst_id, bids=call_bids, asks=call_asks) - self.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.ws.resubscribe( - [ - self.settings.perp_inst_id, - pair.call_inst_id, - pair.put_inst_id, - ] - ) - 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.ws.resubscribe( - [ - self.settings.perp_inst_id, - pick.pair.call_inst_id, - pick.pair.put_inst_id, - ] - ) - 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) + 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: - """空仓时按剩余时长+ATM 对齐。有持仓不切换。""" - if _has_open_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.cache.snapshot(self.settings.perp_inst_id) - - def snapshot_dict(self) -> dict[str, Any]: - return self.snapshot().to_dict() - - async def _refresh_loop(self) -> None: - while True: - await asyncio.sleep(30) - try: - idx = await asyncio.to_thread( - self.rest.fetch_index_ticker, self.settings.index_inst_id - ) - self.cache.set_index_px(idx) - mark = await asyncio.to_thread( - self.rest.fetch_mark_price, self.settings.perp_inst_id - ) - if mark: - self.cache.set_mark_px(self.settings.perp_inst_id, mark) - await self.ensure_atm_async(force=False) - except asyncio.CancelledError: - raise - except Exception as e: - logger.warning("market refresh failed: %s", e) - - -_gateway: MarketGateway | None = None - - -def get_gateway() -> MarketGateway: - global _gateway - if _gateway is None: - _gateway = MarketGateway() - return _gateway - - -def set_gateway(gw: MarketGateway | None) -> None: - global _gateway - _gateway = gw +__all__ = [ + "MarketGateway", + "OpenPick", + "StrategySession", + "bootstrap_session", + "get_gateway", + "get_session", + "set_gateway", + "set_session", +] diff --git a/backend/app/market/instruments.py b/backend/app/market/instruments.py index 91719ae..3a9bdb1 100644 --- a/backend/app/market/instruments.py +++ b/backend/app/market/instruments.py @@ -1,118 +1,36 @@ -"""合约选择：剩余时长过滤 + ATM 平值期权。""" +"""兼容层：选约逻辑已迁至 strategy.selection；OKX 解析在 exchange.okx.parse。""" from __future__ import annotations -import re -from datetime import datetime, timedelta, timezone from typing import Any -from zoneinfo import ZoneInfo -from .types import OptionPair +from ..exchange.okx.parse import ( + expiry_ms_from_ymd, + parse_option_inst_id, + safe_float, +) +from ..exchange.types import OptionPair +from ..strategy.selection import ( + hours_until_expiry, + list_eligible_expiry_ymds as _list_eligible, + next_session_expiry_ymd, + option_leverage, + pick_atm_strike, + select_option_pair as _select_pair, + normalize_contracts, +) -_SH = ZoneInfo("Asia/Shanghai") -_DATE_RE = re.compile(r"^\d{6}$") - - -def safe_float(v: Any) -> float | None: - if v is None or v == "": - return None - try: - return float(v) - except (TypeError, ValueError): - return None - - -def parse_option_inst_id(inst_id: str) -> tuple[str | None, float | None, str | None]: - """ETH-USD_UM-YYMMDD-STRIKE-C → (YYMMDD, strike, C|P).""" - parts = (inst_id or "").strip().split("-") - if len(parts) < 5: - return None, None, None - ymd = parts[-3] - strike = safe_float(parts[-2]) - opt = parts[-1].upper() - if not _DATE_RE.fullmatch(ymd) or strike is None or opt not in ("C", "P"): - return None, None, None - return ymd, strike, opt - - -def expiry_ms_from_ymd(ymd: str) -> int: - """OKX 期权到期：当日 08:00 UTC = 上海 16:00。""" - yy, mm, dd = int(ymd[0:2]), int(ymd[2:4]), int(ymd[4:6]) - dt = datetime(2000 + yy, mm, dd, 8, 0, 0, tzinfo=timezone.utc) - return int(dt.timestamp() * 1000) - - -def hours_until_expiry(ymd: str, now: datetime | None = None) -> float: - """距到期剩余小时（可为负）。""" - n = (now or datetime.now(tz=_SH)).astimezone(_SH) - left_ms = expiry_ms_from_ymd(ymd) - int(n.timestamp() * 1000) - return left_ms / 3_600_000.0 - - -def next_session_expiry_ymd(now: datetime | None = None) -> str: - """兼容旧逻辑：次日/当日 16:00 到期键（展示/测试用）。""" - now_sh = (now or datetime.now(tz=_SH)).astimezone(_SH) - open_today = now_sh.replace(hour=16, minute=0, second=0, microsecond=0) - if now_sh >= open_today: - target = now_sh.date() + timedelta(days=1) - else: - target = now_sh.date() - return target.strftime("%y%m%d") - - -def pick_atm_strike(strikes: list[float], mark_px: float) -> float | None: - if not strikes or mark_px <= 0: - return None - return min(strikes, key=lambda s: (abs(s - mark_px), s)) - - -def _complete_by_expiry( - instruments: list[dict[str, Any]], -) -> dict[str, dict[float, dict[str, str]]]: - """expiry_ymd -> strike -> {C|P: instId}，仅完整 Call+Put。""" - by_exp: dict[str, dict[float, dict[str, str]]] = {} - for row in instruments: - if not isinstance(row, dict): - continue - state = str(row.get("state") or "live").lower() - if state and state != "live": - continue - inst_id = str(row.get("instId") or "") - y, stk, opt = parse_option_inst_id(inst_id) - if y is None or stk is None or opt is None: - exp = safe_float(row.get("expTime")) - if exp: - ms = int(exp) if exp > 10_000_000_000 else int(exp * 1000) - y = datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%y%m%d") - stk = safe_float(row.get("stk")) - opt_raw = str(row.get("optType") or "").upper() - opt = opt_raw if opt_raw in ("C", "P") else None - if not inst_id or not y or stk is None or opt not in ("C", "P"): - continue - by_exp.setdefault(y, {}).setdefault(float(stk), {})[opt] = inst_id - - out: dict[str, dict[float, dict[str, str]]] = {} - for ymd, strikes in by_exp.items(): - complete = {s: v for s, v in strikes.items() if "C" in v and "P" in v} - if complete: - out[ymd] = complete - return out - - -def list_eligible_expiry_ymds( - instruments: list[dict[str, Any]], - *, - min_hours: float, - now: datetime | None = None, -) -> list[str]: - """剩余时间 >= min_hours 的到期日，由近到远。""" - complete = _complete_by_expiry(instruments) - eligible = [ - ymd - for ymd in complete - if hours_until_expiry(ymd, now) + 1e-9 >= float(min_hours) - ] - return sorted(eligible, key=lambda y: expiry_ms_from_ymd(y)) +__all__ = [ + "expiry_ms_from_ymd", + "hours_until_expiry", + "list_eligible_expiry_ymds", + "next_session_expiry_ymd", + "option_leverage", + "parse_option_inst_id", + "pick_atm_strike", + "safe_float", + "select_option_pair", +] def select_option_pair( @@ -121,54 +39,22 @@ def select_option_pair( mark_px: float, expiry_ymd: str | None = None, min_hours: float | None = None, - now: datetime | None = None, + now=None, ) -> OptionPair | None: - """ - 选 ATM Call/Put。 - - 若给 expiry_ymd：在该到期日选平值。 - - 若给 min_hours：选「剩余时长合格」中最近到期日的平值。 - - 否则回退 next_session_expiry_ymd。 - """ - complete = _complete_by_expiry(instruments) - if not complete: - return None - - if expiry_ymd: - ymd = expiry_ymd - if ymd not in complete: - return None - elif min_hours is not None: - eligible = list_eligible_expiry_ymds( - instruments, min_hours=min_hours, now=now - ) - if not eligible: - return None - ymd = eligible[0] - else: - ymd = next_session_expiry_ymd(now) - if ymd not in complete: - # 回退到最近合格到期 - eligible = list_eligible_expiry_ymds(instruments, min_hours=0, now=now) - if not eligible: - return None - ymd = eligible[0] - - strikes_map = complete[ymd] - atm = pick_atm_strike(list(strikes_map.keys()), mark_px) - if atm is None: - return None - legs = strikes_map[atm] - return OptionPair( - expiry_ymd=ymd, - expiry_ms=expiry_ms_from_ymd(ymd), - strike=atm, - call_inst_id=legs["C"], - put_inst_id=legs["P"], + contracts = normalize_contracts(instruments) + return _select_pair( + contracts, + mark_px=mark_px, + expiry_ymd=expiry_ymd, + min_hours=min_hours, + now=now, ) -def option_leverage(underlying_px: float, premium_ask: float) -> float | None: - """现价 / 卖一权利金。""" - if underlying_px <= 0 or premium_ask is None or premium_ask <= 0: - return None - return float(underlying_px) / float(premium_ask) +def list_eligible_expiry_ymds( + instruments: list[dict[str, Any]], + *, + min_hours: float, + now=None, +) -> list[str]: + return _list_eligible(normalize_contracts(instruments), min_hours=min_hours, now=now) diff --git a/backend/app/market/okx_rest.py b/backend/app/market/okx_rest.py index 6210ff8..00ef231 100644 --- a/backend/app/market/okx_rest.py +++ b/backend/app/market/okx_rest.py @@ -1,99 +1,5 @@ -"""OKX REST 只读行情。不调用任何交易类接口。""" +"""兼容层：OKX REST 在 exchange.okx.rest。""" -from __future__ import annotations +from ..exchange.okx.rest import OkxRestClient -from typing import Any - -import httpx - -from .instruments import safe_float -from .types import BookLevel - - -class OkxRestClient: - def __init__( - self, - base_url: str = "https://www.okx.com", - timeout: float = 15.0, - proxy: str | None = None, - ) -> None: - self.base_url = base_url.rstrip("/") - self.proxy = (proxy or "").strip() or None - self._client = httpx.Client( - base_url=self.base_url, - timeout=timeout, - proxy=self.proxy, - headers={"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.1"}, - ) - - def close(self) -> None: - self._client.close() - - def __enter__(self) -> OkxRestClient: - return self - - def __exit__(self, *args: object) -> None: - self.close() - - def _get(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: - r = self._client.get(path, params=params or {}) - r.raise_for_status() - body = r.json() - if str(body.get("code")) != "0": - raise RuntimeError(f"OKX REST error code={body.get('code')} msg={body.get('msg')}") - data = body.get("data") or [] - return [x for x in data if isinstance(x, dict)] - - def fetch_instruments(self, *, inst_type: str, inst_family: str | None = None) -> list[dict[str, Any]]: - params: dict[str, Any] = {"instType": inst_type} - if inst_family: - params["instFamily"] = inst_family - return self._get("/api/v5/public/instruments", params) - - def fetch_option_instruments(self, inst_family: str) -> list[dict[str, Any]]: - rows = self.fetch_instruments(inst_type="OPTION", inst_family=inst_family) - return [r for r in rows if str(r.get("state") or "").lower() == "live"] - - def fetch_index_ticker(self, inst_id: str) -> float | None: - rows = self._get("/api/v5/market/index-tickers", {"instId": inst_id}) - if not rows: - return None - return safe_float(rows[0].get("idxPx")) - - def fetch_mark_price(self, inst_id: str) -> float | None: - rows = self._get("/api/v5/public/mark-price", {"instId": inst_id}) - if not rows: - t = self._get("/api/v5/market/ticker", {"instId": inst_id}) - if not t: - return None - return safe_float(t[0].get("markPx")) or safe_float(t[0].get("last")) - return safe_float(rows[0].get("markPx")) - - def fetch_books(self, inst_id: str, sz: int = 5) -> tuple[list[BookLevel], list[BookLevel], int | None]: - rows = self._get( - "/api/v5/market/books", - {"instId": inst_id, "sz": str(max(1, min(int(sz), 400)))}, - ) - if not rows: - return [], [], None - row = rows[0] - ts = safe_float(row.get("ts")) - ts_ms = int(ts) if ts is not None else None - return ( - _levels(row.get("bids") or []), - _levels(row.get("asks") or []), - ts_ms, - ) - - -def _levels(raw: list[Any]) -> list[BookLevel]: - out: list[BookLevel] = [] - for item in raw: - if not isinstance(item, (list, tuple)) or len(item) < 2: - continue - px = safe_float(item[0]) - sz = safe_float(item[1]) - if px is None or sz is None or px <= 0 or sz <= 0: - continue - out.append(BookLevel(px=px, sz=sz)) - return out +__all__ = ["OkxRestClient"] diff --git a/backend/app/market/okx_ws.py b/backend/app/market/okx_ws.py index 229cde4..e76d0ac 100644 --- a/backend/app/market/okx_ws.py +++ b/backend/app/market/okx_ws.py @@ -1,207 +1,5 @@ -"""OKX 公共 WebSocket：永续 + 期权 books5 / mark-price。只读。""" +"""兼容层：OKX WS 在 exchange.okx.ws。""" -from __future__ import annotations +from ..exchange.okx.ws import OkxPublicWs -import asyncio -import json -import logging -from typing import Any -from urllib.parse import urlparse - -import websockets -from websockets.asyncio.client import ClientConnection - -from .book_cache import BookCache -from .instruments import safe_float -from .types import BookLevel - -logger = logging.getLogger(__name__) - - -class OkxPublicWs: - def __init__( - self, - url: str, - cache: BookCache, - *, - proxy: str | None = None, - ping_interval: float = 20.0, - ) -> None: - self.url = url - self.cache = cache - self.proxy = (proxy or "").strip() or None - self.ping_interval = ping_interval - self._inst_ids: list[str] = [] - self._task: asyncio.Task[None] | None = None - self._stop = asyncio.Event() - self._subscribed: set[str] = set() - - def set_instruments(self, inst_ids: list[str]) -> None: - self._inst_ids = [i for i in inst_ids if i] - - async def start(self) -> None: - if self._task and not self._task.done(): - return - self._stop.clear() - self._task = asyncio.create_task(self._run_forever(), name="okx-public-ws") - - async def stop(self) -> None: - self._stop.set() - if self._task: - self._task.cancel() - try: - await self._task - except asyncio.CancelledError: - pass - self._task = None - self.cache.set_connected(False) - - async def resubscribe(self, inst_ids: list[str]) -> None: - self.set_instruments(inst_ids) - self._stop.set() - await asyncio.sleep(0) - self._stop.clear() - if self._task and not self._task.done(): - self._task.cancel() - try: - await self._task - except asyncio.CancelledError: - pass - self._task = asyncio.create_task(self._run_forever(), name="okx-public-ws") - - async def _open_connection(self) -> ClientConnection: - if not self.proxy: - return await websockets.connect( - self.url, - ping_interval=None, - max_size=2**22, - open_timeout=20, - ) - - from python_socks.async_.asyncio import Proxy - - parsed = urlparse(self.url) - host = parsed.hostname or "ws.okx.com" - port = parsed.port or (443 if parsed.scheme == "wss" else 80) - sock = await Proxy.from_url(self.proxy).connect(dest_host=host, dest_port=port) - return await websockets.connect( - self.url, - sock=sock, - server_hostname=host, - ping_interval=None, - max_size=2**22, - open_timeout=20, - ) - - async def _run_forever(self) -> None: - backoff = 1.0 - while not self._stop.is_set(): - try: - async with await self._open_connection() as ws: - self.cache.set_connected(True) - backoff = 1.0 - await self._subscribe(ws) - waiter = asyncio.create_task(self._stop.wait()) - reader = asyncio.create_task(self._read_loop(ws)) - pinger = asyncio.create_task(self._ping_loop(ws)) - done, pending = await asyncio.wait( - {waiter, reader, pinger}, - return_when=asyncio.FIRST_COMPLETED, - ) - for t in pending: - t.cancel() - for t in done: - exc = t.exception() - if exc and not isinstance(exc, asyncio.CancelledError): - raise exc - except asyncio.CancelledError: - raise - except Exception as e: - logger.warning("OKX WS disconnected: %s", e) - self.cache.set_connected(False) - try: - await asyncio.wait_for(self._stop.wait(), timeout=backoff) - break - except asyncio.TimeoutError: - backoff = min(backoff * 2, 30.0) - - self.cache.set_connected(False) - - async def _subscribe(self, ws: ClientConnection) -> None: - args: list[dict[str, str]] = [] - for inst in self._inst_ids: - args.append({"channel": "books5", "instId": inst}) - args.append({"channel": "mark-price", "instId": inst}) - if not args: - return - payload = {"op": "subscribe", "args": args} - await ws.send(json.dumps(payload)) - self._subscribed = {a["instId"] for a in args} - logger.info("OKX WS subscribed: %s", sorted(self._subscribed)) - - async def _ping_loop(self, ws: ClientConnection) -> None: - while True: - await asyncio.sleep(self.ping_interval) - await ws.send("ping") - - async def _read_loop(self, ws: ClientConnection) -> None: - try: - async for raw in ws: - if raw == "pong": - continue - if isinstance(raw, bytes): - raw = raw.decode("utf-8", errors="ignore") - if raw == "ping": - await ws.send("pong") - continue - try: - msg = json.loads(raw) - except json.JSONDecodeError: - continue - self._handle_message(msg) - except websockets.exceptions.ConnectionClosed: - return - - def _handle_message(self, msg: dict[str, Any]) -> None: - if msg.get("event") in ("subscribe", "error", "channel-conn-count"): - if msg.get("event") == "error": - logger.error("OKX WS error: %s", msg) - return - arg = msg.get("arg") or {} - channel = str(arg.get("channel") or "") - inst_id = str(arg.get("instId") or "") - data = msg.get("data") or [] - if not inst_id or not data: - return - row = data[0] if isinstance(data[0], dict) else None - if row is None: - return - - if channel == "books5": - ts = safe_float(row.get("ts")) - self.cache.upsert_book( - inst_id, - bids=_levels(row.get("bids") or []), - asks=_levels(row.get("asks") or []), - ts_ms=int(ts) if ts is not None else None, - ) - elif channel == "mark-price": - ts = safe_float(row.get("ts")) - self.cache.set_mark_px( - inst_id, - safe_float(row.get("markPx")), - ts_ms=int(ts) if ts is not None else None, - ) - - -def _levels(raw: list[Any]) -> list[BookLevel]: - out: list[BookLevel] = [] - for item in raw: - if not isinstance(item, (list, tuple)) or len(item) < 2: - continue - px = safe_float(item[0]) - sz = safe_float(item[1]) - if px is None or sz is None or px <= 0 or sz <= 0: - continue - out.append(BookLevel(px=px, sz=sz)) - return out +__all__ = ["OkxPublicWs"] diff --git a/backend/app/market/types.py b/backend/app/market/types.py index ca4b5a7..1b2f350 100644 --- a/backend/app/market/types.py +++ b/backend/app/market/types.py @@ -1,94 +1,5 @@ -from __future__ import annotations +"""兼容层：类型定义在 exchange.types。""" -from dataclasses import dataclass, field -from typing import Any +from ..exchange.types import BookLevel, MarketSnapshot, OptionPair, Quote - -@dataclass(slots=True) -class BookLevel: - px: float - sz: float # OKX 张数 / 合约张数口径 - - -@dataclass(slots=True) -class Quote: - inst_id: str - bid: float | None = None - ask: float | None = None - bid_sz: float | None = None - ask_sz: float | None = None - mark_px: float | None = None - ts_ms: int | None = None - bids: list[BookLevel] = field(default_factory=list) - asks: list[BookLevel] = field(default_factory=list) - - def to_dict(self, *, depth: int = 5) -> dict[str, Any]: - return { - "inst_id": self.inst_id, - "bid": self.bid, - "ask": self.ask, - "bid_sz": self.bid_sz, - "ask_sz": self.ask_sz, - "mark_px": self.mark_px, - "ts_ms": self.ts_ms, - "bids": [{"px": x.px, "sz": x.sz} for x in self.bids[:depth]], - "asks": [{"px": x.px, "sz": x.sz} for x in self.asks[:depth]], - } - - -@dataclass(slots=True) -class OptionPair: - expiry_ymd: str # YYMMDD - expiry_ms: int - strike: float - call_inst_id: str - put_inst_id: str - - def to_dict(self) -> dict[str, Any]: - return { - "expiry_ymd": self.expiry_ymd, - "expiry_ms": self.expiry_ms, - "strike": self.strike, - "call_inst_id": self.call_inst_id, - "put_inst_id": self.put_inst_id, - } - - -@dataclass(slots=True) -class MarketSnapshot: - perp: Quote | None - call: Quote | None - put: Quote | None - index_px: float | None - pair: OptionPair | None - connected: bool - updated_at_ms: int | None - - def to_dict(self) -> dict[str, Any]: - return { - "connected": self.connected, - "updated_at_ms": self.updated_at_ms, - "index_px": self.index_px, - "pair": self.pair.to_dict() if self.pair else None, - "perp": self.perp.to_dict() if self.perp else None, - "call": self.call.to_dict() if self.call else None, - "put": self.put.to_dict() if self.put else None, - "ask_compare": { - "call_ask": self.call.ask if self.call else None, - "put_ask": self.put.ask if self.put else None, - "bias": _ask_bias(self.call, self.put), - }, - } - - -def _ask_bias(call: Quote | None, put: Quote | None) -> str: - """卖一比价仅用于选向展示；相等则 wait。""" - ca = call.ask if call else None - pa = put.ask if put else None - if ca is None or pa is None: - return "unknown" - if ca > pa: - return "call_ask_gt_put" # 永续多 + 期权空（腿待拍板） - if ca < pa: - return "put_ask_gt_call" # 永续空 + 期权多（腿待拍板） - return "equal" +__all__ = ["BookLevel", "MarketSnapshot", "OptionPair", "Quote"] diff --git a/backend/app/sim/matcher.py b/backend/app/sim/matcher.py index c5fea9a..2836a75 100644 --- a/backend/app/sim/matcher.py +++ b/backend/app/sim/matcher.py @@ -7,8 +7,9 @@ from dataclasses import dataclass from typing import Any from ..config import get_settings -from ..market import get_gateway +from ..exchange import get_exchange from ..models.db import Database, get_db +from ..strategy.session import get_session from .ledger import Ledger from .liquidity import bid_covers_eth, contracts_for_eth from .pricing import option_fill, perp_fill @@ -39,18 +40,11 @@ class Matcher: return self.ledger.get_setting_float("fee_rate", get_settings().fee_rate) def _ct_mult(self, option_inst_id: str) -> float: - # 尝试 REST meta；失败用默认 s = get_settings() try: - gw = get_gateway() - rows = gw.rest.fetch_instruments(inst_type="OPTION", inst_family=s.option_inst_family) - for r in rows: - if str(r.get("instId")) == option_inst_id: - from ..market.instruments import safe_float - - m = safe_float(r.get("ctMult")) - if m and m > 0: - return float(m) + return get_exchange().get_ct_mult( + option_inst_id, s.option_inst_family, s.option_ct_mult_default + ) except Exception: pass return float(s.option_ct_mult_default) @@ -77,11 +71,15 @@ class Matcher: if pos.get("status") == "open" and pos.get("group_id"): return OpenResult(ok=False, detail="已有持仓组，请先平仓") - gw = get_gateway() - snap = gw.snapshot() + sess = get_session() + snap = sess.snapshot() if not snap.perp or snap.perp.bid is None or snap.perp.ask is None: return OpenResult(ok=False, detail="永续盘口不可用") oq = snap.call if option_side == "call" else snap.put + # 若 ATM 对与持仓合约不一致，直接取持仓合约盘口 + held = get_exchange().quote(option_inst_id) + if held and held.ask is not None: + oq = held if not oq or oq.ask is None: return OpenResult(ok=False, detail="期权卖一不可用") @@ -228,14 +226,16 @@ class Matcher: return CloseResult(ok=False, detail="无持仓可平") group_id = str(pos["group_id"]) - gw = get_gateway() - snap = gw.snapshot() + sess = get_session() + snap = sess.snapshot() if not snap.perp or snap.perp.bid is None or snap.perp.ask is None: return CloseResult(ok=False, detail="永续盘口不可用") option_inst_id = str(pos["option_inst_id"]) option_side = str(pos["option_side"]) - oq = snap.call if option_side == "call" else snap.put + oq = get_exchange().quote(option_inst_id) or ( + snap.call if option_side == "call" else snap.put + ) if not oq or oq.bid is None: return CloseResult(ok=False, detail="期权买一不可用", liquidity_wait=True) @@ -388,8 +388,8 @@ class Matcher: "move_pct": 0.0, "premium_gap": None, } - gw = get_gateway() - snap = gw.snapshot() + sess = get_session() + snap = sess.snapshot() s = get_settings() index_px = snap.index_px if index_px is None and snap.perp: @@ -415,7 +415,7 @@ class Matcher: option_side = str(pos["option_side"]) # 优先用持仓合约盘口，避免 ATM 切换后盯错合约 opt_inst = str(pos.get("option_inst_id") or "") - oq = gw.cache.get(opt_inst) if opt_inst else None + oq = get_exchange().quote(opt_inst) if opt_inst else None if oq is None: oq = snap.call if option_side == "call" else snap.put opt_mark = None diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index f73945c..0910650 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -8,7 +8,7 @@ import time from typing import Any from ..config import get_settings -from ..market import get_gateway +from .session import get_session from ..models.db import get_db from ..sim.ledger import Ledger from ..sim.matcher import Matcher @@ -131,7 +131,7 @@ class StrategyEngine: continue async with self._lock: try: - await get_gateway().ensure_atm_async(force=False) + 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() @@ -191,8 +191,7 @@ class StrategyEngine: self._set_state(phase="idle") self._set_state(phase="wait_signal") - gw = get_gateway() - pick = await gw.pick_for_open_async() + pick = await get_session().pick_for_open_async() if pick is None: self._set_state( last_error="无合格期权：需剩余时长与杠杆倍数同时满足" diff --git a/backend/app/strategy/selection.py b/backend/app/strategy/selection.py new file mode 100644 index 0000000..c1806bf --- /dev/null +++ b/backend/app/strategy/selection.py @@ -0,0 +1,153 @@ +"""策略选约：剩余时长 + ATM 平值 + 期权杠杆（交易所无关）。""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any +from zoneinfo import ZoneInfo + +from ..exchange.types import OptionPair + +_SH = ZoneInfo("Asia/Shanghai") + + +def hours_until_ms(expiry_ms: int, now: datetime | None = None) -> float: + n = (now or datetime.now(tz=_SH)).astimezone(_SH) + return (int(expiry_ms) - int(n.timestamp() * 1000)) / 3_600_000.0 + + +def hours_until_expiry( + ymd: str, + now: datetime | None = None, + *, + expiry_ms: int | None = None, +) -> float: + if expiry_ms is not None: + return hours_until_ms(expiry_ms, now) + # 兼容测试：无 ms 时按 OKX 惯例（UTC 08:00）推算 + from ..exchange.okx.parse import expiry_ms_from_ymd + + return hours_until_ms(expiry_ms_from_ymd(ymd), now) + + +def next_session_expiry_ymd(now: datetime | None = None) -> str: + now_sh = (now or datetime.now(tz=_SH)).astimezone(_SH) + open_today = now_sh.replace(hour=16, minute=0, second=0, microsecond=0) + if now_sh >= open_today: + target = now_sh.date() + timedelta(days=1) + else: + target = now_sh.date() + return target.strftime("%y%m%d") + + +def pick_atm_strike(strikes: list[float], mark_px: float) -> float | None: + if not strikes or mark_px <= 0: + return None + return min(strikes, key=lambda s: (abs(s - mark_px), s)) + + +def option_leverage(underlying_px: float, premium_ask: float) -> float | None: + if underlying_px <= 0 or premium_ask is None or premium_ask <= 0: + return None + return float(underlying_px) / float(premium_ask) + + +def _complete_by_expiry( + contracts: list[dict[str, Any]], +) -> dict[str, tuple[int, dict[float, dict[str, str]]]]: + """ymd -> (expiry_ms, strike -> {C|P: instId})""" + by_exp: dict[str, dict[float, dict[str, str]]] = {} + ms_map: dict[str, int] = {} + for c in contracts: + y = str(c.get("expiry_ymd") or "") + stk = c.get("strike") + opt = str(c.get("side") or "").upper() + inst_id = str(c.get("inst_id") or "") + if not y or stk is None or opt not in ("C", "P") or not inst_id: + continue + by_exp.setdefault(y, {}).setdefault(float(stk), {})[opt] = inst_id + if c.get("expiry_ms") is not None: + ms_map[y] = int(c["expiry_ms"]) + out: dict[str, tuple[int, dict[float, dict[str, str]]]] = {} + for ymd, strikes in by_exp.items(): + complete = {s: v for s, v in strikes.items() if "C" in v and "P" in v} + if not complete: + continue + if ymd in ms_map: + ems = ms_map[ymd] + else: + from ..exchange.okx.parse import expiry_ms_from_ymd + + ems = expiry_ms_from_ymd(ymd) + out[ymd] = (ems, complete) + return out + + +def list_eligible_expiry_ymds( + contracts: list[dict[str, Any]], + *, + min_hours: float, + now: datetime | None = None, +) -> list[str]: + complete = _complete_by_expiry(contracts) + eligible = [ + ymd + for ymd, (ems, _) in complete.items() + if hours_until_ms(ems, now) + 1e-9 >= float(min_hours) + ] + return sorted(eligible, key=lambda y: complete[y][0]) + + +def select_option_pair( + contracts: list[dict[str, Any]], + *, + mark_px: float, + expiry_ymd: str | None = None, + min_hours: float | None = None, + now: datetime | None = None, +) -> OptionPair | None: + complete = _complete_by_expiry(contracts) + if not complete: + return None + + if expiry_ymd: + ymd = expiry_ymd + if ymd not in complete: + return None + elif min_hours is not None: + eligible = list_eligible_expiry_ymds(contracts, min_hours=min_hours, now=now) + if not eligible: + return None + ymd = eligible[0] + else: + ymd = next_session_expiry_ymd(now) + if ymd not in complete: + eligible = list_eligible_expiry_ymds(contracts, min_hours=0, now=now) + if not eligible: + return None + ymd = eligible[0] + + ems, strikes_map = complete[ymd] + atm = pick_atm_strike(list(strikes_map.keys()), mark_px) + if atm is None: + return None + legs = strikes_map[atm] + return OptionPair( + expiry_ymd=ymd, + expiry_ms=ems, + strike=atm, + call_inst_id=legs["C"], + put_inst_id=legs["P"], + ) + + +def normalize_contracts(contracts_or_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """若已是中性结构则原样返回；否则按 OKX 原始行解析（测试兼容）。""" + if not contracts_or_rows: + return [] + sample = contracts_or_rows[0] + if "inst_id" in sample and "expiry_ymd" in sample: + return contracts_or_rows + from ..exchange.okx.parse import rows_to_option_contracts + + return rows_to_option_contracts(contracts_or_rows) diff --git a/backend/app/strategy/session.py b/backend/app/strategy/session.py new file mode 100644 index 0000000..6113016 --- /dev/null +++ b/backend/app/strategy/session.py @@ -0,0 +1,336 @@ +"""策略行情会话：在交易所适配器之上做 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.protocol import ExchangeMarket +from ..exchange.types import MarketSnapshot, OptionPair +from .selection import ( + 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 + + row = get_db().fetchone("SELECT status FROM positions WHERE id=1") + return bool(row and row["status"] == "open") + except Exception: + return False + + +def _strategy_floats() -> tuple[float, float]: + 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 + ) + return hours, lev + except Exception: + return s.min_option_hours, s.min_option_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 + + +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 + + async def start(self) -> None: + if self._started: + return + self._started = True + await self.ex.start() + await asyncio.to_thread(self.align_instruments) + await self.ex.resubscribe( + [ + self.settings.perp_inst_id, + self._pair.call_inst_id if self._pair else "", + self._pair.put_inst_id if self._pair else "", + ] + ) + 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) + ids = [s.perp_inst_id, pair.call_inst_id, pair.put_inst_id] + self.ex.warm_and_subscribe(ids) + 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_instruments(self) -> OptionPair | None: + 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 = _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 + 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 + sig = decide(call_ask, put_ask) + 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.settings.perp_inst_id, + pair.call_inst_id, + pair.put_inst_id, + ] + ) + 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.settings.perp_inst_id, + pick.pair.call_inst_id, + pick.pair.put_inst_id, + ] + ) + 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(): + 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]: + return self.ex.snapshot_dict(self.settings.perp_inst_id) + + async def _refresh_loop(self) -> None: + while True: + await asyncio.sleep(30) + 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) + 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 启动：创建交易所 + 策略会话。""" + s = settings or get_settings() + ex = build_exchange(s) + set_exchange(ex) + sess = StrategySession(s, ex) + set_session(sess) + return sess