dekun
342 lines
11 KiB
Python
342 lines
11 KiB
Python
"""中控永期对冲计算器:永续 1 币 + 按目标盈利反推期权仓位/波动点数(纯函数)."""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Optional, Tuple
|
||
|
||
from lib.trade.trade_fee_lib import estimate_roundtrip_fee_usdt, taker_fee_rate
|
||
|
||
DEFAULT_CT_MULT = 0.01
|
||
PERP_COINS = 1.0
|
||
|
||
|
||
def _f(v: Any) -> Optional[float]:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _parse_base_common(
|
||
*,
|
||
base: str,
|
||
spot: Any,
|
||
capital_usdt: Any,
|
||
target_profit_u: Any,
|
||
perp_leverage: Any,
|
||
option_leverage: Any,
|
||
ct_mult: Any,
|
||
) -> Tuple[Optional[dict[str, float]], Optional[str]]:
|
||
b = (base or "ETH").strip().upper()
|
||
if b not in ("ETH", "BTC"):
|
||
return None, "币种仅支持 BTC / ETH"
|
||
s = _f(spot)
|
||
capital = _f(capital_usdt)
|
||
target = _f(target_profit_u)
|
||
p_lev = _f(perp_leverage)
|
||
o_lev = _f(option_leverage)
|
||
ct = _f(ct_mult)
|
||
if s is None or capital is None or target is None or p_lev is None or o_lev is None:
|
||
return None, "参数格式错误"
|
||
if ct is None or ct <= 0:
|
||
ct = DEFAULT_CT_MULT
|
||
if s <= 0 or capital <= 0 or p_lev <= 0 or o_lev <= 0:
|
||
return None, "现价、资金、杠杆须大于 0"
|
||
if target < 0:
|
||
return None, "目标盈利不能为负"
|
||
prem_per_coin = s / o_lev
|
||
if prem_per_coin <= 0:
|
||
return None, "单币权利金无效"
|
||
margin = (s * PERP_COINS) / p_lev
|
||
return {
|
||
"base_ok": 1.0,
|
||
"spot": s,
|
||
"capital": capital,
|
||
"target": target,
|
||
"p_lev": p_lev,
|
||
"o_lev": o_lev,
|
||
"ct": ct,
|
||
"prem_per_coin": prem_per_coin,
|
||
"margin": margin,
|
||
"fee_rate": taker_fee_rate(),
|
||
}, None
|
||
|
||
|
||
def _move_for_perp_correct(*, spot: float, target: float, premium: float, fee_rate: float) -> float:
|
||
"""净利 = move − premium − fee(move) = target → 解 move.
|
||
|
||
fee = (2*spot + move) * fee_rate
|
||
move*(1-fee_rate) = target + premium + 2*spot*fee_rate
|
||
"""
|
||
denom = 1.0 - float(fee_rate)
|
||
if denom <= 0:
|
||
return 0.0
|
||
return (float(target) + float(premium) + 2.0 * float(spot) * float(fee_rate)) / denom
|
||
|
||
|
||
def calc_perp_options_hedge(
|
||
*,
|
||
base: str = "ETH",
|
||
spot: float,
|
||
capital_usdt: float,
|
||
target_profit_u: float,
|
||
move_mode: str = "points",
|
||
move_value: float,
|
||
perp_leverage: float,
|
||
option_leverage: float,
|
||
ct_mult: float = DEFAULT_CT_MULT,
|
||
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
||
"""由波动反推期权开仓币数/张数(calc_mode=size)."""
|
||
common, err = _parse_base_common(
|
||
base=base,
|
||
spot=spot,
|
||
capital_usdt=capital_usdt,
|
||
target_profit_u=target_profit_u,
|
||
perp_leverage=perp_leverage,
|
||
option_leverage=option_leverage,
|
||
ct_mult=ct_mult,
|
||
)
|
||
if err or not common:
|
||
return None, err
|
||
|
||
s = common["spot"]
|
||
capital = common["capital"]
|
||
target = common["target"]
|
||
p_lev = common["p_lev"]
|
||
o_lev = common["o_lev"]
|
||
ct = common["ct"]
|
||
prem_per_coin = common["prem_per_coin"]
|
||
margin = common["margin"]
|
||
fee_rate = common["fee_rate"]
|
||
b = (base or "ETH").strip().upper()
|
||
|
||
move = _f(move_value)
|
||
mode = (move_mode or "points").strip().lower()
|
||
if mode not in ("points", "pct", "percent", "rate"):
|
||
return None, "波动模式须为 points 或 pct"
|
||
if mode in ("percent", "rate"):
|
||
mode = "pct"
|
||
if move is None:
|
||
return None, "参数格式错误"
|
||
if move <= 0:
|
||
return None, "现价、资金、波动、杠杆须大于 0"
|
||
|
||
if mode == "pct":
|
||
move_points = s * (move / 100.0)
|
||
else:
|
||
move_points = move
|
||
if move_points <= 0:
|
||
return None, "波动对应价格变动须大于 0"
|
||
|
||
exit_px = s + move_points
|
||
perp_gross = move_points * PERP_COINS
|
||
fee = estimate_roundtrip_fee_usdt(s, exit_px, qty=PERP_COINS, contract_size=1.0)
|
||
|
||
premium_budget = perp_gross - target - fee
|
||
if premium_budget <= 0:
|
||
return None, "波动收益不足以覆盖目标盈利+手续费，无法开期权"
|
||
|
||
opt_coins = premium_budget / prem_per_coin
|
||
opt_sheets = opt_coins / ct
|
||
premium_total = opt_coins * prem_per_coin
|
||
|
||
case_a_net = perp_gross - premium_total - fee
|
||
opt_intrinsic = opt_coins * move_points
|
||
opt_net = opt_intrinsic - premium_total
|
||
perp_loss = -perp_gross
|
||
portfolio_net = opt_net + perp_loss
|
||
|
||
return {
|
||
"calc_mode": "size",
|
||
"base": b,
|
||
"spot": round(s, 8),
|
||
"capital_usdt": round(capital, 8),
|
||
"target_profit_u": round(target, 8),
|
||
"move_mode": mode,
|
||
"move_value": round(move, 8),
|
||
"move_points": round(move_points, 8),
|
||
"exit_price": round(exit_px, 8),
|
||
"perp_coins": PERP_COINS,
|
||
"perp_leverage": round(p_lev, 8),
|
||
"option_leverage": round(o_lev, 8),
|
||
"ct_mult": ct,
|
||
"prem_per_coin": round(prem_per_coin, 8),
|
||
"perp_gross_u": round(perp_gross, 8),
|
||
"perp_fee_u": round(fee, 8),
|
||
"fee_rate": fee_rate,
|
||
"premium_budget_u": round(premium_budget, 8),
|
||
"opt_coins": round(opt_coins, 8),
|
||
"opt_sheets": round(opt_sheets, 8),
|
||
"premium_total_u": round(premium_total, 8),
|
||
"perp_margin_u": round(margin, 8),
|
||
"capital_ok": bool(capital >= margin),
|
||
"case_a": {
|
||
"label": "永续方向对",
|
||
"perp_pnl_u": round(perp_gross, 8),
|
||
"premium_u": round(premium_total, 8),
|
||
"fee_u": round(fee, 8),
|
||
"net_u": round(case_a_net, 8),
|
||
},
|
||
"case_b": {
|
||
"label": "期权方向对",
|
||
"opt_intrinsic_u": round(opt_intrinsic, 8),
|
||
"premium_u": round(premium_total, 8),
|
||
"opt_net_u": round(opt_net, 8),
|
||
"perp_pnl_u": round(perp_loss, 8),
|
||
"portfolio_net_u": round(portfolio_net, 8),
|
||
},
|
||
}, None
|
||
|
||
|
||
def calc_perp_options_points(
|
||
*,
|
||
base: str = "ETH",
|
||
spot: float,
|
||
capital_usdt: float,
|
||
target_profit_u: float,
|
||
perp_leverage: float,
|
||
option_leverage: float,
|
||
ratio_perp: float = 1.0,
|
||
ratio_opt: float = 2.0,
|
||
ct_mult: float = DEFAULT_CT_MULT,
|
||
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
||
"""按永续:期权比例 + 目标盈利,反推两套情景所需波动点数.
|
||
|
||
永续币数固定为 ratio 归一后的 1 币侧(perp_coins = PERP_COINS).
|
||
期权币数 = PERP_COINS * (ratio_opt / ratio_perp),例 1:2 → 2 币.
|
||
|
||
A 永续方向对: move − premium − fee(move) = 目标盈利
|
||
B 期权方向对:
|
||
- 期权净利达目标: opt_coins*move − premium = 目标
|
||
- 组合净利达目标: move*(opt_coins − perp_coins) − premium = 目标
|
||
"""
|
||
common, err = _parse_base_common(
|
||
base=base,
|
||
spot=spot,
|
||
capital_usdt=capital_usdt,
|
||
target_profit_u=target_profit_u,
|
||
perp_leverage=perp_leverage,
|
||
option_leverage=option_leverage,
|
||
ct_mult=ct_mult,
|
||
)
|
||
if err or not common:
|
||
return None, err
|
||
|
||
rp = _f(ratio_perp)
|
||
ro = _f(ratio_opt)
|
||
if rp is None or ro is None or rp <= 0 or ro <= 0:
|
||
return None, "永续:期权比例须大于 0"
|
||
|
||
s = common["spot"]
|
||
capital = common["capital"]
|
||
target = common["target"]
|
||
p_lev = common["p_lev"]
|
||
o_lev = common["o_lev"]
|
||
ct = common["ct"]
|
||
prem_per_coin = common["prem_per_coin"]
|
||
margin = common["margin"]
|
||
fee_rate = common["fee_rate"]
|
||
b = (base or "ETH").strip().upper()
|
||
|
||
opt_coins = PERP_COINS * (ro / rp)
|
||
premium_total = opt_coins * prem_per_coin
|
||
opt_sheets = opt_coins / ct
|
||
|
||
move_a = _move_for_perp_correct(spot=s, target=target, premium=premium_total, fee_rate=fee_rate)
|
||
if move_a <= 0:
|
||
return None, "无法解出永续方向对所需点数"
|
||
|
||
fee_a = estimate_roundtrip_fee_usdt(s, s + move_a, qty=PERP_COINS, contract_size=1.0)
|
||
net_a = move_a * PERP_COINS - premium_total - fee_a
|
||
|
||
# 期权净利 = 目标
|
||
move_b_opt = (target + premium_total) / opt_coins
|
||
opt_net_at_b_opt = opt_coins * move_b_opt - premium_total
|
||
portfolio_at_b_opt = opt_net_at_b_opt - move_b_opt * PERP_COINS
|
||
|
||
# 组合净利 = 目标
|
||
edge = opt_coins - PERP_COINS
|
||
if edge <= 0:
|
||
move_b_port = None
|
||
port_err = "期权币数须大于永续币数,组合才能在方向对时赚到目标盈利"
|
||
else:
|
||
move_b_port = (target + premium_total) / edge
|
||
port_err = None
|
||
if move_b_port is not None:
|
||
opt_net_at_b_port = opt_coins * move_b_port - premium_total
|
||
portfolio_at_b_port = opt_net_at_b_port - move_b_port * PERP_COINS
|
||
else:
|
||
opt_net_at_b_port = None
|
||
portfolio_at_b_port = None
|
||
|
||
return {
|
||
"calc_mode": "points",
|
||
"base": b,
|
||
"spot": round(s, 8),
|
||
"capital_usdt": round(capital, 8),
|
||
"target_profit_u": round(target, 8),
|
||
"ratio_perp": round(rp, 8),
|
||
"ratio_opt": round(ro, 8),
|
||
"ratio_label": f"{_fmt_ratio(rp)}:{_fmt_ratio(ro)}",
|
||
"perp_coins": PERP_COINS,
|
||
"opt_coins": round(opt_coins, 8),
|
||
"opt_sheets": round(opt_sheets, 8),
|
||
"perp_leverage": round(p_lev, 8),
|
||
"option_leverage": round(o_lev, 8),
|
||
"ct_mult": ct,
|
||
"prem_per_coin": round(prem_per_coin, 8),
|
||
"premium_total_u": round(premium_total, 8),
|
||
"fee_rate": fee_rate,
|
||
"perp_margin_u": round(margin, 8),
|
||
"capital_ok": bool(capital >= margin),
|
||
"case_a": {
|
||
"label": "永续方向对",
|
||
"move_points": round(move_a, 8),
|
||
"move_pct": round(move_a / s * 100.0, 8),
|
||
"perp_pnl_u": round(move_a * PERP_COINS, 8),
|
||
"premium_u": round(premium_total, 8),
|
||
"fee_u": round(fee_a, 8),
|
||
"net_u": round(net_a, 8),
|
||
},
|
||
"case_b": {
|
||
"label": "期权方向对",
|
||
"move_points_opt_net": round(move_b_opt, 8),
|
||
"move_pct_opt_net": round(move_b_opt / s * 100.0, 8),
|
||
"opt_net_u": round(opt_net_at_b_opt, 8),
|
||
"portfolio_net_at_opt_target_u": round(portfolio_at_b_opt, 8),
|
||
"move_points_portfolio": None if move_b_port is None else round(move_b_port, 8),
|
||
"move_pct_portfolio": None
|
||
if move_b_port is None
|
||
else round(move_b_port / s * 100.0, 8),
|
||
"opt_net_at_portfolio_target_u": None
|
||
if opt_net_at_b_port is None
|
||
else round(opt_net_at_b_port, 8),
|
||
"portfolio_net_u": None if portfolio_at_b_port is None else round(portfolio_at_b_port, 8),
|
||
"portfolio_error": port_err,
|
||
"premium_u": round(premium_total, 8),
|
||
},
|
||
}, None
|
||
|
||
|
||
def _fmt_ratio(v: float) -> str:
|
||
if abs(v - round(v)) < 1e-9:
|
||
return str(int(round(v)))
|
||
s = f"{v:.4f}".rstrip("0").rstrip(".")
|
||
return s
|
||
|
||
|
||
def calc_perp_options(
|
||
*,
|
||
calc_mode: str = "size",
|
||
**kwargs: Any,
|
||
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
||
"""统一入口:size=由波动推仓位;points=由比例推点数."""
|
||
mode = (calc_mode or "size").strip().lower()
|
||
if mode in ("points", "ratio", "move"):
|
||
return calc_perp_options_points(**kwargs)
|
||
# size mode: ignore ratio kwargs if present
|
||
kwargs.pop("ratio_perp", None)
|
||
kwargs.pop("ratio_opt", None)
|
||
return calc_perp_options_hedge(**kwargs)
|