dekun
5d161b4220
Lock one mode per instance/subaccount, cross-asset BTC option/ETH perp sizing, combo-U exit, and ship docs/tests/audit. Co-authored-by: Cursor <cursoragent@cursor.com>
297 lines
9.3 KiB
Python
297 lines
9.3 KiB
Python
"""永期对冲实例模式(env: HEDGE_PO_MODE).
|
||
|
||
一实例一种模式,对应一个 OKX 子账户;模拟盘与实盘同一执行路径.
|
||
|
||
eth → 期权 ETH + 永续 ETH
|
||
btc → 期权 BTC + 永续 BTC
|
||
btc_eth → 期权 BTC + 永续 ETH;永续名义 = 期权名义 × HEDGE_CROSS_NOTIONAL_K(默认 0.5)
|
||
|
||
方向(以期权为主):Call→永续空 / Put→永续多.
|
||
出场:组合盈利金额(U),扣开平双边手续费.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
from typing import Any, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
MODE_ETH = "eth"
|
||
MODE_BTC = "btc"
|
||
MODE_BTC_ETH = "btc_eth"
|
||
VALID_PO_MODES = frozenset({MODE_ETH, MODE_BTC, MODE_BTC_ETH})
|
||
|
||
DEFAULT_CROSS_NOTIONAL_K = 0.5
|
||
|
||
_ALIASES = {
|
||
"eth_eth": MODE_ETH,
|
||
"e": MODE_ETH,
|
||
"eth-only": MODE_ETH,
|
||
"btc_btc": MODE_BTC,
|
||
"b": MODE_BTC,
|
||
"btc-only": MODE_BTC,
|
||
"cross": MODE_BTC_ETH,
|
||
"btc-eth": MODE_BTC_ETH,
|
||
"btc/eth": MODE_BTC_ETH,
|
||
"跨币": MODE_BTC_ETH,
|
||
}
|
||
|
||
|
||
def _sf(v: Any) -> Optional[float]:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def normalize_hedge_po_mode(raw: Optional[str]) -> str:
|
||
s = str(raw or "").strip().lower().replace("-", "_").replace("/", "_")
|
||
if s in VALID_PO_MODES:
|
||
return s
|
||
if s in _ALIASES:
|
||
return _ALIASES[s]
|
||
return ""
|
||
|
||
|
||
def infer_hedge_po_mode_from_underly() -> str:
|
||
u = (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "ETH").strip().upper()
|
||
if u == "BTC":
|
||
return MODE_BTC
|
||
return MODE_ETH
|
||
|
||
|
||
def get_hedge_po_mode() -> str:
|
||
m = normalize_hedge_po_mode(os.getenv("HEDGE_PO_MODE"))
|
||
if m:
|
||
return m
|
||
return infer_hedge_po_mode_from_underly()
|
||
|
||
|
||
def cross_notional_k() -> float:
|
||
"""跨币:永续名义 / 期权名义.默认 0.5."""
|
||
v = _sf(os.getenv("HEDGE_CROSS_NOTIONAL_K"))
|
||
if v is None or v <= 0:
|
||
return DEFAULT_CROSS_NOTIONAL_K
|
||
# 合理钳制,避免误配把永续打到极端
|
||
return max(0.05, min(float(v), 5.0))
|
||
|
||
|
||
def default_option_perp_ratio(mode: Optional[str] = None) -> float:
|
||
"""UI/定仓用比例:期权名义 ÷ 比例 = 永续名义 → ratio = 1/k."""
|
||
m = mode or get_hedge_po_mode()
|
||
if m == MODE_BTC_ETH:
|
||
k = cross_notional_k()
|
||
return round(1.0 / k, 6) if k > 0 else 2.0
|
||
return 2.0
|
||
|
||
|
||
def option_underlying(mode: Optional[str] = None) -> str:
|
||
m = mode or get_hedge_po_mode()
|
||
if m in (MODE_BTC, MODE_BTC_ETH):
|
||
return "BTC"
|
||
return "ETH"
|
||
|
||
|
||
def perp_underlying(mode: Optional[str] = None) -> str:
|
||
m = mode or get_hedge_po_mode()
|
||
if m == MODE_BTC:
|
||
return "BTC"
|
||
return "ETH"
|
||
|
||
|
||
def is_cross_asset(mode: Optional[str] = None) -> bool:
|
||
return (mode or get_hedge_po_mode()) == MODE_BTC_ETH
|
||
|
||
|
||
def mode_label(mode: Optional[str] = None) -> str:
|
||
m = mode or get_hedge_po_mode()
|
||
return {
|
||
MODE_ETH: "ETH永期",
|
||
MODE_BTC: "BTC永期",
|
||
MODE_BTC_ETH: "跨币BTC期权/ETH永续",
|
||
}.get(m, m or "—")
|
||
|
||
|
||
def mode_short_label(mode: Optional[str] = None) -> str:
|
||
m = mode or get_hedge_po_mode()
|
||
return {
|
||
MODE_ETH: "ETH",
|
||
MODE_BTC: "BTC",
|
||
MODE_BTC_ETH: "BTC→ETH",
|
||
}.get(m, m or "—")
|
||
|
||
|
||
def clamp_hedge_po_mode_on_startup() -> dict[str, Any]:
|
||
"""启动钳制:规范化 HEDGE_PO_MODE,并同步默认期权标的.
|
||
|
||
不写回 .env 文件(避免热更/多实例踩踏);仅 os.environ 进程内生效.
|
||
返回审计摘要供日志/测试.
|
||
"""
|
||
raw = (os.getenv("HEDGE_PO_MODE") or "").strip()
|
||
normalized = normalize_hedge_po_mode(raw) if raw else ""
|
||
warnings: list[str] = []
|
||
if raw and not normalized:
|
||
warnings.append(f"非法 HEDGE_PO_MODE={raw!r},回退推断")
|
||
mode = infer_hedge_po_mode_from_underly()
|
||
elif normalized:
|
||
mode = normalized
|
||
else:
|
||
mode = infer_hedge_po_mode_from_underly()
|
||
warnings.append(f"未配置 HEDGE_PO_MODE,按 OKX_OPTIONS_DEFAULT_UNDERLY 推断为 {mode}")
|
||
|
||
os.environ["HEDGE_PO_MODE"] = mode
|
||
opt_u = option_underlying(mode)
|
||
prev_u = (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "").strip().upper()
|
||
if prev_u != opt_u:
|
||
os.environ["OKX_OPTIONS_DEFAULT_UNDERLY"] = opt_u
|
||
if prev_u:
|
||
warnings.append(f"同步 OKX_OPTIONS_DEFAULT_UNDERLY {prev_u}→{opt_u}")
|
||
|
||
k = cross_notional_k()
|
||
os.environ["HEDGE_CROSS_NOTIONAL_K"] = str(k)
|
||
summary = {
|
||
"ok": True,
|
||
"mode": mode,
|
||
"label": mode_label(mode),
|
||
"option_underlying": opt_u,
|
||
"perp_underlying": perp_underlying(mode),
|
||
"cross_notional_k": k,
|
||
"option_perp_ratio": default_option_perp_ratio(mode),
|
||
"warnings": warnings,
|
||
}
|
||
for w in warnings:
|
||
logger.warning("[HEDGE_PO_MODE] %s", w)
|
||
logger.info(
|
||
"[HEDGE_PO_MODE] mode=%s option=%s perp=%s k=%s (一实例一模式一子账户)",
|
||
mode,
|
||
opt_u,
|
||
summary["perp_underlying"],
|
||
k,
|
||
)
|
||
return summary
|
||
|
||
|
||
def apply_mode_to_start_body(body: dict[str, Any], *, mode: Optional[str] = None) -> dict[str, Any]:
|
||
"""启动前按实例模式钳制 body(模拟/实盘同路径)."""
|
||
m = mode or get_hedge_po_mode()
|
||
out = dict(body or {})
|
||
opt_u = option_underlying(m)
|
||
perp_u = perp_underlying(m)
|
||
out["po_mode"] = m
|
||
out["underlying"] = opt_u
|
||
out["option_underlying"] = opt_u
|
||
out["perp_underlying"] = perp_u
|
||
# 以期权为主 + 组合盈利金额出场
|
||
out["option_primary"] = True
|
||
out["exit_mode"] = "profit"
|
||
ratio = default_option_perp_ratio(m)
|
||
# 允许用户略调比例,但跨币默认锁到 1/k;若未填则写入默认
|
||
if out.get("option_perp_ratio") in (None, ""):
|
||
out["option_perp_ratio"] = ratio
|
||
elif m == MODE_BTC_ETH:
|
||
# 跨币强制名义系数,避免页内误改导致与子账户策略不一致
|
||
out["option_perp_ratio"] = ratio
|
||
# 永续 symbol:若未传或与模式不符则按 perp 标的重建
|
||
sym = str(out.get("exchange_symbol") or "").upper()
|
||
want_prefix = f"{perp_u}/"
|
||
if not sym or not sym.startswith(want_prefix):
|
||
out["exchange_symbol"] = f"{perp_u}/USDT:USDT"
|
||
return out
|
||
|
||
|
||
def validate_body_matches_mode(body: dict[str, Any], *, mode: Optional[str] = None) -> Optional[str]:
|
||
m = mode or get_hedge_po_mode()
|
||
opt_u = option_underlying(m)
|
||
perp_u = perp_underlying(m)
|
||
uly = str(body.get("underlying") or body.get("option_underlying") or "").upper()
|
||
if uly and uly != opt_u:
|
||
return f"本实例模式 {mode_label(m)} 期权标的须为 {opt_u},收到 {uly}"
|
||
inst = str(body.get("opt_inst_id") or "").upper()
|
||
if inst and not inst.startswith(f"{opt_u}-"):
|
||
return f"期权合约须为 {opt_u} 标的(本实例 {mode_label(m)})"
|
||
sym = str(body.get("exchange_symbol") or "").upper()
|
||
if sym and not sym.startswith(f"{perp_u}/"):
|
||
return f"永续合约须为 {perp_u}(本实例 {mode_label(m)})"
|
||
return None
|
||
|
||
|
||
def combo_net_usdt(
|
||
net: dict[str, Any] | None,
|
||
*,
|
||
option_index: float,
|
||
) -> Optional[float]:
|
||
"""币本位期权净利×期权指数 + 永续 U 净利 → 组合 U.
|
||
|
||
监控出场与持仓卡必须同口径;禁止把币与 U 直接相加.
|
||
"""
|
||
if not net:
|
||
return None
|
||
idx = float(option_index or 0)
|
||
if idx <= 0:
|
||
return None
|
||
opt_coin = float(net.get("opt_net") or 0)
|
||
perp_u = float(net.get("perp_net") or 0)
|
||
return round(opt_coin * idx + perp_u, 4)
|
||
|
||
|
||
def size_cross_notional(
|
||
*,
|
||
option_coin_qty: float,
|
||
option_index: float,
|
||
perp_index: float,
|
||
contract_size: float,
|
||
notional_k: Optional[float] = None,
|
||
) -> dict[str, Any]:
|
||
"""跨币/同币统一:期权币名义 → U 名义 → 永续张数.
|
||
|
||
同币时 option_index≈perp_index,等价于 coin_qty * k / cs.
|
||
"""
|
||
k = float(notional_k if notional_k is not None else cross_notional_k())
|
||
oc = float(option_coin_qty or 0)
|
||
oi = float(option_index or 0)
|
||
pi = float(perp_index or 0)
|
||
cs = float(contract_size or 0.01)
|
||
if oc <= 0 or oi <= 0 or pi <= 0 or cs <= 0 or k <= 0:
|
||
return {
|
||
"ok": False,
|
||
"msg": "跨币定仓参数无效",
|
||
"option_notional_usdt": 0.0,
|
||
"perp_notional_usdt": 0.0,
|
||
"perp_coin": 0.0,
|
||
"contracts": 0.0,
|
||
"notional_k": k,
|
||
}
|
||
opt_n = oc * oi
|
||
perp_n = opt_n * k
|
||
perp_coin = perp_n / pi
|
||
contracts = perp_coin / cs
|
||
return {
|
||
"ok": True,
|
||
"msg": "",
|
||
"option_notional_usdt": round(opt_n, 4),
|
||
"perp_notional_usdt": round(perp_n, 4),
|
||
"perp_coin": round(perp_coin, 8),
|
||
"contracts": contracts,
|
||
"notional_k": k,
|
||
}
|
||
|
||
|
||
def public_mode_payload(mode: Optional[str] = None) -> dict[str, Any]:
|
||
m = mode or get_hedge_po_mode()
|
||
return {
|
||
"po_mode": m,
|
||
"po_mode_label": mode_label(m),
|
||
"po_mode_short": mode_short_label(m),
|
||
"option_underlying": option_underlying(m),
|
||
"perp_underlying": perp_underlying(m),
|
||
"cross_asset": is_cross_asset(m),
|
||
"cross_notional_k": cross_notional_k() if is_cross_asset(m) else None,
|
||
"option_perp_ratio": default_option_perp_ratio(m),
|
||
"exit_mode": "profit",
|
||
"option_primary": True,
|
||
"instance_rule": "一实例一种模式对应一个子账户;模拟盘与实盘同路径;账户互不干涉",
|
||
}
|