dekun
338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""对冲计划:情景测算与全仓建议仓(纯函数,无 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 floor_contracts_to_precision(contracts: float, decimals: int) -> float:
|
||
"""按交易所张数精度向下取整,避免建议张数超过可用保证金."""
|
||
import math
|
||
|
||
raw = float(contracts or 0.0)
|
||
if raw <= 0:
|
||
return 0.0
|
||
try:
|
||
d = int(decimals)
|
||
except (TypeError, ValueError):
|
||
d = 0
|
||
if d <= 0:
|
||
return float(math.floor(raw + 1e-12))
|
||
scale = 10**d
|
||
return math.floor(raw * scale + 1e-12) / scale
|
||
|
||
|
||
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,
|
||
}
|