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

Replace 期期 equal-split with long/short bias sizing and env ratio controls.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-19 15:25:10 +08:00
parent bc797cb1db
commit e3cd2a75de
13 changed files with 330 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hedge_plan/hedge_plan_calc_lib.py
+116 -40
View File
@@ -150,6 +150,54 @@ def _cap_sheets_by_ask_depth(sheets: int, ask_sz: Any) -> int:
return min(n, int(math.floor(float(depth) + 1e-12)))
def _normalize_oo_sheets_mode(mode: str) -> str:
m = (mode or "same_sheets").strip().lower()
if m in ("long_bias", "bias_long", "long", "做多"):
return "long_bias"
if m in ("short_bias", "bias_short", "short", "做空"):
return "short_bias"
# 旧「均分」兼容:按预算 50/50(页面已移除)
if m in ("split", "equal_budget", "split_budget", "均分"):
return "split_budget"
return "same_sheets"
def _normalize_oo_bias_split_by(raw: Any) -> str:
v = str(raw or "budget").strip().lower()
if v in ("sheets", "qty", "quantity", "张数"):
return "sheets"
return "budget"
def _clamp_oo_bias_ratio(raw: Any, default: float = 0.7) -> float:
try:
r = float(raw)
except (TypeError, ValueError):
r = float(default)
if r <= 0 or r >= 1:
r = float(default)
return r
def _oo_call_put_leg_index(opt_type_a: str, opt_type_b: str) -> tuple[Optional[str], Optional[str], str]:
"""返回 (call_side, put_side, err);side 为 'a'/'b'."""
a = (opt_type_a or "").strip().upper()
b = (opt_type_b or "").strip().upper()
if a.startswith("C"):
a = "C"
elif a.startswith("P"):
a = "P"
if b.startswith("C"):
b = "C"
elif b.startswith("P"):
b = "P"
if {a, b} != {"C", "P"}:
return None, None, "做多/做空需一腿 Call、一腿 Put"
call_side = "a" if a == "C" else "b"
put_side = "b" if call_side == "a" else "a"
return call_side, put_side, ""
def suggest_oo_sheets(
*,
mode: str,
@@ -157,63 +205,89 @@ def suggest_oo_sheets(
ask_a: float,
ct_mult_a: float = 0.01,
ask_sz_a: Any = None,
opt_type_a: str = "",
ask_b: float,
ct_mult_b: float = 0.01,
ask_sz_b: Any = None,
opt_type_b: str = "",
bias_split_by: str = "budget",
bias_ratio: float = 0.7,
) -> dict[str, Any]:
"""期期建议张数:same_sheets(同张数,默认) / split_budget(均分预算)."""
"""期期建议张数:same_sheets / long_bias / short_bias(及旧 split_budget)."""
import math
m = (mode or "same_sheets").strip().lower()
if m in ("split", "equal_budget", "split_budget", "均分"):
m = "split_budget"
else:
m = "same_sheets"
m = _normalize_oo_sheets_mode(mode)
split_by = _normalize_oo_bias_split_by(bias_split_by)
ratio = _clamp_oo_bias_ratio(bias_ratio)
budget = max(0.0, float(budget_usdc or 0.0))
cost_a = option_unit_cost_usdc(ask=ask_a, ct_mult=ct_mult_a)
cost_b = option_unit_cost_usdc(ask=ask_b, ct_mult=ct_mult_b)
def _fail(msg: str, n_a: int = 0, n_b: int = 0) -> dict[str, Any]:
return {
"mode": m,
"sheets_a": n_a,
"sheets_b": n_b,
"cost_a": round(cost_a, 8),
"cost_b": round(cost_b, 8),
"premium_est": round(cost_a * n_a + cost_b * n_b, 6),
"ok": False,
"msg": msg,
"bias_split_by": split_by,
"bias_ratio": ratio,
}
if budget <= 0:
return {
"mode": m,
"sheets_a": 0,
"sheets_b": 0,
"cost_a": round(cost_a, 8),
"cost_b": round(cost_b, 8),
"premium_est": 0.0,
"ok": False,
"msg": "可用预算为 0",
}
return _fail("可用预算为 0")
if cost_a <= 0 or cost_b <= 0:
return {
"mode": m,
"sheets_a": 0,
"sheets_b": 0,
"cost_a": round(cost_a, 8),
"cost_b": round(cost_b, 8),
"premium_est": 0.0,
"ok": False,
"msg": "缺少有效卖一价,无法建议张数",
}
if m == "split_budget":
return _fail("缺少有效卖一价,无法建议张数")
pair = cost_a + cost_b
n_same = int(math.floor(budget / pair + 1e-12)) if pair > 0 else 0
n_same = _cap_sheets_by_ask_depth(n_same, ask_sz_a)
n_same = _cap_sheets_by_ask_depth(n_same, ask_sz_b)
if m == "same_sheets":
n_a = n_same
n_b = n_same
elif m == "split_budget":
half = budget / 2.0
n_a = int(math.floor(half / cost_a + 1e-12))
n_b = int(math.floor(half / cost_b + 1e-12))
n_a = _cap_sheets_by_ask_depth(n_a, ask_sz_a)
n_b = _cap_sheets_by_ask_depth(n_b, ask_sz_b)
else:
pair = cost_a + cost_b
n = int(math.floor(budget / pair + 1e-12)) if pair > 0 else 0
n_a = n
n_b = n
n_a = _cap_sheets_by_ask_depth(n_a, ask_sz_a)
n_b = _cap_sheets_by_ask_depth(n_b, ask_sz_b)
if m == "same_sheets":
n = min(n_a, n_b)
n_a = n
n_b = n
call_side, put_side, err = _oo_call_put_leg_index(opt_type_a, opt_type_b)
if err:
return _fail(err)
major_is_call = m == "long_bias"
if split_by == "sheets":
if n_same < 2:
return _fail("同张数规模不足 2,无法按比例拆分")
major_n = int(round(n_same * ratio))
major_n = max(1, min(major_n, n_same - 1))
minor_n = n_same - major_n
n_call = major_n if major_is_call else minor_n
n_put = minor_n if major_is_call else major_n
else:
maj_budget = budget * ratio
min_budget = budget * (1.0 - ratio)
cost_call = cost_a if call_side == "a" else cost_b
cost_put = cost_b if call_side == "a" else cost_a
if major_is_call:
n_call = int(math.floor(maj_budget / cost_call + 1e-12)) if cost_call > 0 else 0
n_put = int(math.floor(min_budget / cost_put + 1e-12)) if cost_put > 0 else 0
else:
n_put = int(math.floor(maj_budget / cost_put + 1e-12)) if cost_put > 0 else 0
n_call = int(math.floor(min_budget / cost_call + 1e-12)) if cost_call > 0 else 0
n_a = n_call if call_side == "a" else n_put
n_b = n_put if call_side == "a" else n_call
n_a = _cap_sheets_by_ask_depth(n_a, ask_sz_a)
n_b = _cap_sheets_by_ask_depth(n_b, ask_sz_b)
prem = cost_a * n_a + cost_b * n_b
ok = n_a >= 1 and n_b >= 1
msg = ""
if not ok:
msg = "预算不够开 1+1(或卖一深度不足)"
msg = "" if ok else "预算不够开 1+1(或卖一深度不足)"
return {
"mode": m,
"sheets_a": n_a,
@@ -223,6 +297,8 @@ def suggest_oo_sheets(
"premium_est": round(prem, 6),
"ok": ok,
"msg": msg,
"bias_split_by": split_by,
"bias_ratio": ratio,
}