"""期权标的白名单:OKX_OPTIONS_UNDERLYINGS + DEFAULT_UNDERLY.""" from __future__ import annotations import os from typing import Any _ALLOWED = ("ETH", "BTC") def parse_options_underlyings() -> list[str]: """env 白名单;空或缺省为 ETH,BTC.仅认 ETH/BTC.""" raw = (os.getenv("OKX_OPTIONS_UNDERLYINGS") or "").strip() if not raw: return ["ETH", "BTC"] out: list[str] = [] for part in raw.replace(";", ",").split(","): u = part.strip().upper() if u in _ALLOWED and u not in out: out.append(u) return out or ["ETH", "BTC"] def resolve_default_underly(allowed: list[str] | None = None) -> str: """DEFAULT 落在白名单内;否则取白名单第一项.""" allow = list(allowed) if allowed is not None else parse_options_underlyings() if not allow: allow = ["ETH", "BTC"] d = (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "ETH").strip().upper() or "ETH" if d in allow: return d return allow[0] def underly_from_inst_id(inst_id: str) -> str: return (str(inst_id or "").split("-")[0] or "").strip().upper() def underly_not_allowed_msg(underlying: str, allowed: list[str] | None = None) -> str | None: """白名单外返回拒因;允许则 None.""" allow = list(allowed) if allowed is not None else parse_options_underlyings() u = (underlying or "").strip().upper() if u in allow: return None return f"标的 {u or '—'} 不在本机白名单({','.join(allow) or '—'})" def inst_underly_not_allowed_msg(inst_id: str, allowed: list[str] | None = None) -> str | None: return underly_not_allowed_msg(underly_from_inst_id(inst_id), allowed) def underly_cfg_snapshot() -> dict[str, Any]: allow = parse_options_underlyings() return { "allowed_underlys": allow, "default_underly": resolve_default_underly(allow), }