dekun
401 lines
15 KiB
Python
401 lines
15 KiB
Python
"""中控策略对比:同风险额下 合约 / 单期权 / 期期7:3 情景测算(纯函数)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
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 default_contract_size(base: str) -> float:
|
|
"""OKX 线性永续常用面值(币/张);与计算器缺省一致."""
|
|
b = (base or "ETH").strip().upper()
|
|
return 0.01
|
|
|
|
|
|
def default_ct_mult(base: str) -> float:
|
|
return 0.01
|
|
|
|
|
|
def floor_sheets(n: float, step: float = 1.0) -> float:
|
|
if n is None or not math.isfinite(n) or n <= 0:
|
|
return 0.0
|
|
s = float(step) if step and step > 0 else 1.0
|
|
return math.floor(n / s + 1e-12) * s
|
|
|
|
|
|
def option_unit_cost(*, ask: float, ct_mult: float) -> float:
|
|
return float(ask) * float(ct_mult or 0.01)
|
|
|
|
|
|
def option_intrinsic_value(
|
|
*,
|
|
opt_type: str,
|
|
strike: float,
|
|
spot: float,
|
|
sheets: float,
|
|
ct_mult: float,
|
|
) -> float:
|
|
o = (opt_type or "").strip().upper()
|
|
k = float(strike)
|
|
s = float(spot)
|
|
if o == "C":
|
|
intrinsic = max(0.0, s - k)
|
|
elif o == "P":
|
|
intrinsic = max(0.0, k - s)
|
|
else:
|
|
intrinsic = 0.0
|
|
return intrinsic * float(sheets) * float(ct_mult or 0.01)
|
|
|
|
|
|
def option_pnl_at_spot(
|
|
*,
|
|
opt_type: str,
|
|
strike: float,
|
|
spot: float,
|
|
sheets: float,
|
|
ct_mult: float,
|
|
premium_paid: float,
|
|
) -> float:
|
|
return option_intrinsic_value(
|
|
opt_type=opt_type,
|
|
strike=strike,
|
|
spot=spot,
|
|
sheets=sheets,
|
|
ct_mult=ct_mult,
|
|
) - float(premium_paid)
|
|
|
|
|
|
def perp_pnl(
|
|
*,
|
|
direction: str,
|
|
entry: float,
|
|
exit_px: float,
|
|
contracts: float,
|
|
contract_size: float,
|
|
) -> float:
|
|
coins = float(contracts) * float(contract_size or 0.01)
|
|
d = (direction or "long").strip().lower()
|
|
if d == "short":
|
|
return (float(entry) - float(exit_px)) * coins
|
|
return (float(exit_px) - float(entry)) * coins
|
|
|
|
|
|
def _validate_common(inp: dict[str, Any]) -> Optional[str]:
|
|
base = str(inp.get("base") or "ETH").strip().upper()
|
|
if base not in ("ETH", "BTC"):
|
|
return "标的仅支持 ETH / BTC"
|
|
direction = str(inp.get("direction") or "long").strip().lower()
|
|
if direction not in ("long", "short"):
|
|
return "方向须为 long / short"
|
|
s0 = _f(inp.get("entry"))
|
|
sl = _f(inp.get("sl"))
|
|
tp = _f(inp.get("tp"))
|
|
risk = _f(inp.get("risk_u"))
|
|
if s0 is None or s0 <= 0:
|
|
return "请填写有效入场价"
|
|
if sl is None or sl <= 0:
|
|
return "请填写有效止损价"
|
|
if tp is None or tp <= 0:
|
|
return "请填写有效止盈价"
|
|
if risk is None or risk <= 0:
|
|
return "请填写有效风险额 R"
|
|
if direction == "long" and not (sl < s0 < tp):
|
|
return "做多须满足 止损 < 入场 < 止盈"
|
|
if direction == "short" and not (tp < s0 < sl):
|
|
return "做空须满足 止盈 < 入场 < 止损"
|
|
return None
|
|
|
|
|
|
def _calc_perp(inp: dict[str, Any], *, contract_size: float) -> dict[str, Any]:
|
|
direction = str(inp.get("direction") or "long").strip().lower()
|
|
s0 = float(inp["entry"])
|
|
sl = float(inp["sl"])
|
|
tp = float(inp["tp"])
|
|
risk = float(inp["risk_u"])
|
|
per_sheet_sl = abs(s0 - sl) * contract_size
|
|
sheets = floor_sheets(risk / per_sheet_sl) if per_sheet_sl > 0 else 0.0
|
|
actual_sl_loss = abs(perp_pnl(
|
|
direction=direction, entry=s0, exit_px=sl, contracts=sheets, contract_size=contract_size
|
|
))
|
|
tp_pnl = perp_pnl(
|
|
direction=direction, entry=s0, exit_px=tp, contracts=sheets, contract_size=contract_size
|
|
)
|
|
# 路径 C:本单已止损 −actual;踏空未拿到 = 原止盈盈利
|
|
path_a = round(tp_pnl, 4)
|
|
path_b = round(-actual_sl_loss if sheets > 0 else -risk, 4)
|
|
path_c_realized = path_b
|
|
path_c_missed = path_a
|
|
return {
|
|
"kind": "perp",
|
|
"sheets": sheets,
|
|
"contract_size": contract_size,
|
|
"per_sheet_sl_u": round(per_sheet_sl, 6),
|
|
"risk_used_u": round(actual_sl_loss, 4),
|
|
"path_a_tp": path_a,
|
|
"path_b_sl": path_b,
|
|
"path_c_realized": path_c_realized,
|
|
"path_c_missed": path_c_missed,
|
|
"path_c_note": "本单已止损;踏空未拿到原止盈空间",
|
|
"worst_u": path_b,
|
|
}
|
|
|
|
|
|
def _calc_single_option(inp: dict[str, Any], *, ct_mult: float) -> dict[str, Any]:
|
|
direction = str(inp.get("direction") or "long").strip().lower()
|
|
risk = float(inp["risk_u"])
|
|
tp = float(inp.get("tp_opt") if inp.get("tp_opt") not in (None, "") else inp["tp"])
|
|
sl = float(inp["sl"])
|
|
opt = inp.get("option") if isinstance(inp.get("option"), dict) else {}
|
|
default_type = "C" if direction == "long" else "P"
|
|
opt_type = str(opt.get("opt_type") or default_type).strip().upper()
|
|
if opt_type not in ("C", "P"):
|
|
opt_type = default_type
|
|
strike = _f(opt.get("strike"))
|
|
ask = _f(opt.get("ask"))
|
|
if strike is None or strike <= 0:
|
|
return {"ok": False, "msg": "请填写单期权行权价"}
|
|
if ask is None or ask <= 0:
|
|
return {"ok": False, "msg": "请填写单期权卖一价"}
|
|
unit = option_unit_cost(ask=ask, ct_mult=ct_mult)
|
|
sheets = floor_sheets(risk / unit) if unit > 0 else 0.0
|
|
premium = option_unit_cost(ask=ask, ct_mult=ct_mult) * sheets if sheets else 0.0
|
|
# 若张数为 0
|
|
path_a = option_pnl_at_spot(
|
|
opt_type=opt_type, strike=strike, spot=tp, sheets=sheets, ct_mult=ct_mult, premium_paid=premium
|
|
)
|
|
path_b_at_sl = option_pnl_at_spot(
|
|
opt_type=opt_type, strike=strike, spot=sl, sheets=sheets, ct_mult=ct_mult, premium_paid=premium
|
|
)
|
|
path_b_worst = -premium
|
|
# 踏空路径:合约被洗后标的仍到 TP,期权仍持有 → 同止盈
|
|
path_c = path_a
|
|
return {
|
|
"ok": True,
|
|
"kind": "option",
|
|
"opt_type": opt_type,
|
|
"strike": strike,
|
|
"ask": ask,
|
|
"ct_mult": ct_mult,
|
|
"sheets": sheets,
|
|
"unit_cost_u": round(unit, 6),
|
|
"premium_u": round(premium, 4),
|
|
"path_a_tp": round(path_a, 4),
|
|
"path_b_sl": round(path_b_at_sl, 4),
|
|
"path_b_worst": round(path_b_worst, 4),
|
|
"path_c_hold_to_tp": round(path_c, 4),
|
|
"path_c_note": "合约踏空路径下期权仍持有至目标价(内在近似)",
|
|
"worst_u": round(path_b_worst, 4),
|
|
}
|
|
|
|
|
|
def _calc_hedge(inp: dict[str, Any], *, ct_mult: float) -> dict[str, Any]:
|
|
direction = str(inp.get("direction") or "long").strip().lower()
|
|
risk = float(inp["risk_u"])
|
|
tp = float(inp.get("tp_hedge") if inp.get("tp_hedge") not in (None, "") else inp["tp"])
|
|
sl = float(inp["sl"])
|
|
hedge = inp.get("hedge") if isinstance(inp.get("hedge"), dict) else {}
|
|
main_default = "C" if direction == "long" else "P"
|
|
side_default = "P" if direction == "long" else "C"
|
|
main = hedge.get("main") if isinstance(hedge.get("main"), dict) else {}
|
|
side = hedge.get("side") if isinstance(hedge.get("side"), dict) else {}
|
|
main_type = str(main.get("opt_type") or main_default).strip().upper()
|
|
side_type = str(side.get("opt_type") or side_default).strip().upper()
|
|
if main_type not in ("C", "P"):
|
|
main_type = main_default
|
|
if side_type not in ("C", "P"):
|
|
side_type = side_default
|
|
main_k = _f(main.get("strike"))
|
|
main_ask = _f(main.get("ask"))
|
|
side_k = _f(side.get("strike"))
|
|
side_ask = _f(side.get("ask"))
|
|
if None in (main_k, main_ask, side_k, side_ask) or min(
|
|
main_k or 0, main_ask or 0, side_k or 0, side_ask or 0
|
|
) <= 0:
|
|
return {"ok": False, "msg": "请填写期期对冲两腿的行权价与卖一"}
|
|
main_budget = 0.7 * risk
|
|
side_budget = 0.3 * risk
|
|
main_unit = option_unit_cost(ask=float(main_ask), ct_mult=ct_mult)
|
|
side_unit = option_unit_cost(ask=float(side_ask), ct_mult=ct_mult)
|
|
main_sheets = floor_sheets(main_budget / main_unit) if main_unit > 0 else 0.0
|
|
side_sheets = floor_sheets(side_budget / side_unit) if side_unit > 0 else 0.0
|
|
main_prem = main_unit * main_sheets
|
|
side_prem = side_unit * side_sheets
|
|
premium = main_prem + side_prem
|
|
|
|
def combo_at(spot: float) -> float:
|
|
a = option_pnl_at_spot(
|
|
opt_type=main_type,
|
|
strike=float(main_k),
|
|
spot=spot,
|
|
sheets=main_sheets,
|
|
ct_mult=ct_mult,
|
|
premium_paid=main_prem,
|
|
)
|
|
b = option_pnl_at_spot(
|
|
opt_type=side_type,
|
|
strike=float(side_k),
|
|
spot=spot,
|
|
sheets=side_sheets,
|
|
ct_mult=ct_mult,
|
|
premium_paid=side_prem,
|
|
)
|
|
return a + b
|
|
|
|
path_a = combo_at(tp)
|
|
path_b_at_sl = combo_at(sl)
|
|
path_b_worst = -premium
|
|
path_c = path_a
|
|
return {
|
|
"ok": True,
|
|
"kind": "hedge",
|
|
"ratio": "7:3",
|
|
"ct_mult": ct_mult,
|
|
"main": {
|
|
"opt_type": main_type,
|
|
"strike": main_k,
|
|
"ask": main_ask,
|
|
"sheets": main_sheets,
|
|
"premium_u": round(main_prem, 4),
|
|
"budget_u": round(main_budget, 4),
|
|
},
|
|
"side": {
|
|
"opt_type": side_type,
|
|
"strike": side_k,
|
|
"ask": side_ask,
|
|
"sheets": side_sheets,
|
|
"premium_u": round(side_prem, 4),
|
|
"budget_u": round(side_budget, 4),
|
|
},
|
|
"premium_u": round(premium, 4),
|
|
"path_a_tp": round(path_a, 4),
|
|
"path_b_sl": round(path_b_at_sl, 4),
|
|
"path_b_worst": round(path_b_worst, 4),
|
|
"path_c_hold_to_tp": round(path_c, 4),
|
|
"path_c_note": "合约踏空路径下对冲组合仍持有至目标价(内在近似)",
|
|
"worst_u": round(path_b_worst, 4),
|
|
}
|
|
|
|
|
|
def recommend(perp: dict[str, Any], opt: dict[str, Any], hedge: dict[str, Any], risk: float) -> dict[str, Any]:
|
|
"""可解释规则推荐."""
|
|
candidates: list[tuple[str, float, dict[str, Any]]] = []
|
|
if perp and perp.get("sheets", 0) > 0:
|
|
candidates.append(("合约", float(perp.get("path_a_tp") or 0), perp))
|
|
if opt and opt.get("ok") and opt.get("sheets", 0) > 0:
|
|
candidates.append(("单期权", float(opt.get("path_a_tp") or 0), opt))
|
|
if hedge and hedge.get("ok") and (hedge.get("premium_u") or 0) > 0:
|
|
candidates.append(("期期对冲", float(hedge.get("path_a_tp") or 0), hedge))
|
|
if not candidates:
|
|
return {
|
|
"choice": "—",
|
|
"reason": "输入不足,无法推荐",
|
|
"bullets": ["请检查风险额与卖一/止损距是否过小导致张数为 0"],
|
|
}
|
|
|
|
best_name, best_a, _ = max(candidates, key=lambda x: x[1])
|
|
perp_a = float(perp.get("path_a_tp") or 0) if perp else 0.0
|
|
opt_a = float(opt.get("path_a_tp") or 0) if opt and opt.get("ok") else 0.0
|
|
hedge_a = float(hedge.get("path_a_tp") or 0) if hedge and hedge.get("ok") else 0.0
|
|
|
|
# 踏空:合约 C 实现为亏损,期权/对冲 C 仍接近 A
|
|
perp_miss = float(perp.get("path_c_missed") or 0) if perp else 0.0
|
|
opt_c = float(opt.get("path_c_hold_to_tp") or 0) if opt and opt.get("ok") else None
|
|
hedge_c = float(hedge.get("path_c_hold_to_tp") or 0) if hedge and hedge.get("ok") else None
|
|
anti_whipsaw = False
|
|
if perp_miss > 0 and (
|
|
(opt_c is not None and opt_c > 0) or (hedge_c is not None and hedge_c > 0)
|
|
):
|
|
anti_whipsaw = True
|
|
|
|
# 合约止盈明显更高(>= 另两者 1.15 倍)且用户能接受踏空 → 推合约
|
|
others_max = max(opt_a, hedge_a, 0.0)
|
|
choice = best_name
|
|
if perp_a > 0 and perp_a >= others_max * 1.15 and perp_a >= best_a * 0.99:
|
|
choice = "合约"
|
|
if anti_whipsaw:
|
|
reason = "合约止盈赔付更高,但震荡易洗时存在踏空;能接受洗盘再走可选合约"
|
|
else:
|
|
reason = "同风险下合约干净止盈赔付最高"
|
|
elif anti_whipsaw and (opt_a > 0 or hedge_a > 0):
|
|
# 抗踏空优先期权类;期期与单腿接近时推期期
|
|
if hedge_a > 0 and (opt_a <= 0 or hedge_a >= opt_a * 0.85):
|
|
choice = "期期对冲"
|
|
reason = "震荡易洗时期权类更抗踏空;期期 7:3 兼顾方向与保护"
|
|
else:
|
|
choice = "单期权"
|
|
reason = "震荡易洗时单期权仍可持有到目标,抗踏空优于合约"
|
|
else:
|
|
reason = f"同风险下「{best_name}」干净止盈赔付最高"
|
|
|
|
bullets = [
|
|
f"止盈对比:合约 {perp_a:.2f}U / 单期权 {opt_a:.2f}U / 期期 {hedge_a:.2f}U(风险 R={risk:.2f}U)",
|
|
(
|
|
"止损与踏空:合约打止损即结束并可能踏空;"
|
|
"期权/对冲最坏约亏满权利金,踏空路径下常仍持有至目标"
|
|
if anti_whipsaw
|
|
else "止损与踏空:三者最坏接近 −R;关注合约是否易被洗后错过止盈"
|
|
),
|
|
f"选用建议:{reason}",
|
|
]
|
|
return {"choice": choice, "reason": reason, "bullets": bullets}
|
|
|
|
|
|
def run_compare(inp: dict[str, Any]) -> dict[str, Any]:
|
|
err = _validate_common(inp)
|
|
if err:
|
|
return {"ok": False, "msg": err}
|
|
base = str(inp.get("base") or "ETH").strip().upper()
|
|
risk = float(inp["risk_u"])
|
|
cs = _f(inp.get("contract_size")) or default_contract_size(base)
|
|
ct = _f(inp.get("ct_mult")) or default_ct_mult(base)
|
|
perp = _calc_perp(inp, contract_size=float(cs))
|
|
opt = _calc_single_option(inp, ct_mult=float(ct))
|
|
hedge = _calc_hedge(inp, ct_mult=float(ct))
|
|
rec = recommend(
|
|
perp,
|
|
opt if opt.get("ok") else {"ok": False},
|
|
hedge if hedge.get("ok") else {"ok": False},
|
|
risk,
|
|
)
|
|
warnings: list[str] = []
|
|
if perp.get("sheets", 0) <= 0:
|
|
warnings.append("合约张数为 0:止损距过大或 R 过小")
|
|
if isinstance(opt, dict) and opt.get("ok") and opt.get("sheets", 0) <= 0:
|
|
warnings.append("单期权张数为 0:卖一过高或 R 过小")
|
|
if isinstance(hedge, dict) and hedge.get("ok") and hedge.get("premium_u", 0) <= 0:
|
|
warnings.append("期期对冲未开出张数:卖一过高或 R 过小")
|
|
if isinstance(opt, dict) and not opt.get("ok"):
|
|
warnings.append(str(opt.get("msg") or "单期权输入不完整"))
|
|
if isinstance(hedge, dict) and not hedge.get("ok"):
|
|
warnings.append(str(hedge.get("msg") or "期期对冲输入不完整"))
|
|
return {
|
|
"ok": True,
|
|
"base": base,
|
|
"direction": str(inp.get("direction") or "long").strip().lower(),
|
|
"entry": float(inp["entry"]),
|
|
"sl": float(inp["sl"]),
|
|
"tp": float(inp["tp"]),
|
|
"risk_u": risk,
|
|
"contract_size": float(cs),
|
|
"ct_mult": float(ct),
|
|
"perp": perp,
|
|
"option": opt,
|
|
"hedge": hedge,
|
|
"recommend": rec,
|
|
"warnings": warnings,
|
|
"notes": [
|
|
"期权止盈按标的到价的内在价值近似,非盘口卖出价",
|
|
"到期小盈/小亏未纳入主表与推荐",
|
|
"仅本地测算,不下单",
|
|
],
|
|
}
|