dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add OKX hedge-plan P0: env group, preview page, and PnL scenario math.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-14 11:00:15 +08:00
parent d27ccbea8f
commit 1fa98425e7
18 changed files with 2232 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hedge_plan/hedge_plan_calc_lib.py
+320
View File
@@ -0,0 +1,320 @@
"""对冲计划:情景测算与全仓建议仓(纯函数,无 IO)."""
from __future__ import annotations
from typing import Any, Optional
def _f(v: Any) -> Optional[float]:
if v is None or v == "":
return None
try:
return float(v)
except (TypeError, ValueError):
return None
def perp_coin_amount(*, contracts: float, contract_size: float) -> float:
return float(contracts) * float(contract_size or 1.0)
def perp_pnl(
*,
direction: str,
entry: float,
exit_px: float,
contracts: float,
contract_size: float,
) -> float:
coins = perp_coin_amount(contracts=contracts, contract_size=contract_size)
d = (direction or "long").strip().lower()
if d == "short":
return (float(entry) - float(exit_px)) * coins
return (float(exit_px) - float(entry)) * coins
def option_premium_total(*, ask: float, sheets: float, ct_mult: float) -> float:
"""卖一报价为每 1 币;权利金 = ask × 张数 × ct_mult."""
return float(ask) * float(sheets) * float(ct_mult or 0.01)
def option_expiry_pnl(
*,
opt_type: str,
strike: float,
spot: float,
sheets: float,
ct_mult: float,
premium_paid: float,
) -> float:
o = (opt_type or "").strip().upper()
intrinsic_per_coin = 0.0
if o in ("C", "CALL"):
intrinsic_per_coin = max(0.0, float(spot) - float(strike))
elif o in ("P", "PUT"):
intrinsic_per_coin = max(0.0, float(strike) - float(spot))
else:
return -float(premium_paid)
value = intrinsic_per_coin * float(sheets) * float(ct_mult or 0.01)
return value - float(premium_paid)
def suggest_contracts_from_notional(
*,
notional: float,
entry: float,
contract_size: float,
) -> float:
if entry <= 0 or contract_size <= 0 or notional <= 0:
return 0.0
return float(notional) / (float(entry) * float(contract_size))
def build_perp_options_preview(
*,
direction: str,
entry: float,
tp: float,
sl: float,
contracts: float,
contract_size: float,
opt_type: str,
strike: float,
sheets: float,
ct_mult: float,
premium_paid: float,
index_px: Optional[float] = None,
) -> dict[str, Any]:
"""
永期情景.
止盈账:永续止盈盈利 - 权利金.
止损账:期权到期内在(按 SL 价) - 永续止损亏损额.
"""
d = (direction or "long").strip().lower()
pnl_tp_perp = perp_pnl(
direction=d, entry=entry, exit_px=tp, contracts=contracts, contract_size=contract_size
)
pnl_sl_perp = perp_pnl(
direction=d, entry=entry, exit_px=sl, contracts=contracts, contract_size=contract_size
)
# 止盈统计口径
tp_total = float(pnl_tp_perp) - float(premium_paid)
# 止损:期权按 SL 价结算内在 - |永续亏损|
opt_at_sl = option_expiry_pnl(
opt_type=opt_type,
strike=strike,
spot=sl,
sheets=sheets,
ct_mult=ct_mult,
premium_paid=premium_paid,
)
sl_total = float(opt_at_sl) - abs(float(pnl_sl_perp)) if pnl_sl_perp < 0 else float(opt_at_sl) + float(
pnl_sl_perp
)
# 有符号相加更稳:期权盈亏 + 永续盈亏
sl_total_signed = float(opt_at_sl) + float(pnl_sl_perp)
spot = float(index_px) if index_px is not None else float(entry)
opt_flat = option_expiry_pnl(
opt_type=opt_type,
strike=strike,
spot=spot,
sheets=sheets,
ct_mult=ct_mult,
premium_paid=premium_paid,
)
flat_total = 0.0 + float(opt_flat)
opt_at_tp = option_expiry_pnl(
opt_type=opt_type,
strike=strike,
spot=tp,
sheets=sheets,
ct_mult=ct_mult,
premium_paid=premium_paid,
)
return {
"plan_type": "perp_options",
"direction": d,
"contracts": contracts,
"coin_amount": perp_coin_amount(contracts=contracts, contract_size=contract_size),
"premium_paid": round(float(premium_paid), 6),
"scenarios": [
{
"id": "tp",
"label": "止盈(计划结束口径)",
"spot": tp,
"perp_pnl": round(pnl_tp_perp, 4),
"options_pnl": round(-float(premium_paid), 4),
"total": round(tp_total, 4),
"note": "止盈盈利 − 权利金;期权可不强平",
},
{
"id": "sl",
"label": "止损(计划结束口径)",
"spot": sl,
"perp_pnl": round(pnl_sl_perp, 4),
"options_pnl": round(opt_at_sl, 4),
"total": round(sl_total_signed, 4),
"note": "期权盈利 − 永续亏损(有符号相加);期权须强平",
},
{
"id": "flat",
"label": "到期·现价附近",
"spot": spot,
"perp_pnl": 0.0,
"options_pnl": round(opt_flat, 4),
"total": round(flat_total, 4),
"note": "示意:永续未动,期权按到期内在",
},
{
"id": "expiry_tp",
"label": "到期·止盈价",
"spot": tp,
"perp_pnl": round(pnl_tp_perp, 4),
"options_pnl": round(opt_at_tp, 4),
"total": round(pnl_tp_perp + opt_at_tp, 4),
"note": "若期权拿到 TP 价到期(参考)",
},
{
"id": "expiry_sl",
"label": "到期·止损价",
"spot": sl,
"perp_pnl": round(pnl_sl_perp, 4),
"options_pnl": round(opt_at_sl, 4),
"total": round(pnl_sl_perp + opt_at_sl, 4),
"note": "与止损口径相近(期权用内在)",
},
],
"summary": {
"tp_total": round(tp_total, 4),
"sl_total": round(sl_total_signed, 4),
"premium_paid": round(float(premium_paid), 4),
"hedge_ratio_at_sl": _hedge_ratio(opt_at_sl, pnl_sl_perp),
},
}
def _hedge_ratio(opt_pnl: float, perp_pnl: float) -> Optional[float]:
loss = abs(float(perp_pnl)) if float(perp_pnl) < 0 else 0.0
if loss <= 1e-12:
return None
if float(opt_pnl) <= 0:
return 0.0
return round(float(opt_pnl) / loss * 100.0, 2)
def build_options_options_preview(
*,
target_price: float,
index_px: float,
leg_a: dict[str, Any],
leg_b: dict[str, Any],
) -> dict[str, Any]:
"""期期情景:目标价 / 到期现价 / 到期两边."""
def _leg_pnl(leg: dict[str, Any], spot: float) -> float:
return option_expiry_pnl(
opt_type=str(leg.get("opt_type") or ""),
strike=float(leg["strike"]),
spot=spot,
sheets=float(leg.get("sheets") or 0),
ct_mult=float(leg.get("ct_mult") or 0.01),
premium_paid=float(leg.get("premium_paid") or 0),
)
prem = float(leg_a.get("premium_paid") or 0) + float(leg_b.get("premium_paid") or 0)
a_t = _leg_pnl(leg_a, target_price)
b_t = _leg_pnl(leg_b, target_price)
at_target = a_t + b_t
win_leg = "a" if a_t >= b_t else "b"
a_flat = _leg_pnl(leg_a, index_px)
b_flat = _leg_pnl(leg_b, index_px)
flat_total = a_flat + b_flat
expiry_loss = flat_total if flat_total <= 0 else flat_total
return {
"plan_type": "options_options",
"premium_paid": round(prem, 6),
"target_price": target_price,
"winner_at_target": win_leg,
"scenarios": [
{
"id": "target",
"label": "到达目标价",
"spot": target_price,
"leg_a_pnl": round(a_t, 4),
"leg_b_pnl": round(b_t, 4),
"total": round(at_target, 4),
"note": f"盈利方≈腿{win_leg.upper()}(可平);亏损方默认到期",
},
{
"id": "expiry_flat",
"label": "到期·现价(无突破)",
"spot": index_px,
"leg_a_pnl": round(a_flat, 4),
"leg_b_pnl": round(b_flat, 4),
"total": round(flat_total, 4),
"note": "无盈利则记总亏损结束" if flat_total <= 0 else "到期仍可能有净值",
},
{
"id": "max_premium_loss",
"label": "最大保费损耗",
"spot": None,
"leg_a_pnl": round(-float(leg_a.get("premium_paid") or 0), 4),
"leg_b_pnl": round(-float(leg_b.get("premium_paid") or 0), 4),
"total": round(-prem, 4),
"note": "双腿权利金全部损失",
},
],
"summary": {
"at_target_total": round(at_target, 4),
"expiry_flat_total": round(expiry_loss, 4),
"premium_paid": round(prem, 4),
"expiry_is_loss": flat_total <= 0,
},
}
def gate_status(
*,
hedge_enabled: bool,
sizing_mode: str,
plan_type: str,
options_enabled: bool,
) -> dict[str, Any]:
from lib.trade.position_sizing_lib import is_full_margin_mode
full = is_full_margin_mode(sizing_mode)
pt = (plan_type or "").strip().lower()
can_preview = True
can_start = False
reasons: list[str] = []
if not hedge_enabled:
can_start = False
reasons.append("对冲计划未启用(HEDGE_PLAN_ENABLED)")
if not options_enabled:
can_preview = False
can_start = False
reasons.append("期权模块未启用")
if pt == "perp_options":
if not full:
can_start = False
reasons.append("永期开仓仅全仓模式可用(当前可测算)")
elif hedge_enabled and options_enabled:
can_start = False
reasons.append("P0 仅测算,真实开仓将在后续版本开放")
elif pt == "options_options":
if hedge_enabled and options_enabled:
can_start = False
reasons.append("P0 仅测算,真实开仓将在后续版本开放")
return {
"hedge_enabled": hedge_enabled,
"options_enabled": options_enabled,
"sizing_mode": sizing_mode,
"is_full_margin": full,
"plan_type": pt,
"can_preview": can_preview,
"can_start": can_start,
"reasons": reasons,
}