dekun
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""公开行情辅助：补历史到期结算指数展示（不发明成交现金）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_CACHE: dict[int, float] = {}
|
|
_CACHE_MAX = 256
|
|
|
|
|
|
def looks_binance_option(inst_id: str | None) -> bool:
|
|
return "USD_UM" in str(inst_id or "")
|
|
|
|
|
|
def eth_usdt_close_at_ms(ts_ms: int | None) -> float | None:
|
|
"""币安 ETHUSDT 1m K 线收盘价（近似期权结算指数）。失败返回 None。"""
|
|
if ts_ms is None:
|
|
return None
|
|
try:
|
|
ms = int(ts_ms)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if ms <= 0:
|
|
return None
|
|
minute = (ms // 60_000) * 60_000
|
|
cached = _CACHE.get(minute)
|
|
if cached is not None:
|
|
return cached
|
|
url = (
|
|
"https://api.binance.com/api/v3/klines"
|
|
f"?symbol=ETHUSDT&interval=1m&startTime={minute}&limit=1"
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=4) as resp:
|
|
raw = resp.read().decode("utf-8", "replace")
|
|
rows = json.loads(raw)
|
|
if not rows:
|
|
return None
|
|
close_px = float(rows[0][4])
|
|
if close_px <= 0:
|
|
return None
|
|
if len(_CACHE) >= _CACHE_MAX:
|
|
_CACHE.clear()
|
|
_CACHE[minute] = close_px
|
|
return close_px
|
|
except (urllib.error.URLError, TimeoutError, ValueError, TypeError, IndexError) as e:
|
|
logger.debug("eth_usdt_close_at_ms failed ms=%s: %s", minute, e)
|
|
return None
|
|
|
|
|
|
def maybe_public_settle_index(g: dict[str, Any]) -> float | None:
|
|
"""库内无结算价时，币安期权到期组用公开 ETHUSDT 收盘近似。"""
|
|
if str(g.get("close_reason") or "") != "expiry":
|
|
return None
|
|
inst = g.get("option_inst_id") or g.get("option2_inst_id")
|
|
if not looks_binance_option(str(inst) if inst else None):
|
|
return None
|
|
ts = g.get("close_at_ms") or g.get("hold_close_at_ms")
|
|
return eth_usdt_close_at_ms(ts if ts is not None else None)
|