dekun
237 lines
8.3 KiB
Python
237 lines
8.3 KiB
Python
"""对冲计划 P0 测算口径单测."""
|
|
import unittest
|
|
|
|
from lib.hedge_plan.hedge_plan_calc_lib import (
|
|
build_options_options_preview,
|
|
build_perp_options_preview,
|
|
floor_contracts_to_precision,
|
|
gate_status,
|
|
option_expiry_pnl,
|
|
option_premium_total,
|
|
perp_pnl,
|
|
resolve_oo_budget_usdc,
|
|
suggest_oo_sheets,
|
|
)
|
|
from lib.hedge_plan.hedge_plan_monitor_lib import resolve_oo_rest_close_mode
|
|
|
|
|
|
class TestHedgePlanCalc(unittest.TestCase):
|
|
def test_perp_tp_accounting_is_profit_minus_premium(self):
|
|
p = build_perp_options_preview(
|
|
direction="long",
|
|
entry=3200,
|
|
tp=3400,
|
|
sl=3000,
|
|
contracts=50,
|
|
contract_size=0.01,
|
|
opt_type="P",
|
|
strike=3100,
|
|
sheets=10,
|
|
ct_mult=0.01,
|
|
premium_paid=8,
|
|
index_px=3200,
|
|
)
|
|
self.assertEqual(p["summary"]["tp_total"], 92.0)
|
|
self.assertEqual(p["scenarios"][0]["options_pnl"], -8.0)
|
|
|
|
def test_perp_sl_accounting_is_option_plus_perp_signed(self):
|
|
p = build_perp_options_preview(
|
|
direction="long",
|
|
entry=3200,
|
|
tp=3400,
|
|
sl=3000,
|
|
contracts=50,
|
|
contract_size=0.01,
|
|
opt_type="P",
|
|
strike=3100,
|
|
sheets=10,
|
|
ct_mult=0.01,
|
|
premium_paid=8,
|
|
index_px=3200,
|
|
)
|
|
self.assertEqual(p["summary"]["sl_total"], -98.0)
|
|
self.assertEqual(p["summary"]["hedge_ratio_at_sl"], 2.0)
|
|
|
|
def test_option_premium_and_expiry(self):
|
|
self.assertEqual(option_premium_total(ask=80, sheets=1, ct_mult=0.01), 0.8)
|
|
self.assertEqual(
|
|
option_expiry_pnl(
|
|
opt_type="P", strike=3100, spot=3000, sheets=10, ct_mult=0.01, premium_paid=8
|
|
),
|
|
2.0,
|
|
)
|
|
|
|
def test_gate_perp_requires_full_margin_for_start_message(self):
|
|
g = gate_status(
|
|
hedge_enabled=True,
|
|
sizing_mode="risk",
|
|
plan_type="perp_options",
|
|
options_enabled=True,
|
|
)
|
|
self.assertTrue(g["can_preview"])
|
|
self.assertFalse(g["can_start"])
|
|
self.assertTrue(any("全仓" in r for r in g["reasons"]))
|
|
|
|
def test_gate_hidden_plan_type_blocks_preview(self):
|
|
g = gate_status(
|
|
hedge_enabled=True,
|
|
sizing_mode="full_margin",
|
|
plan_type="perp_options",
|
|
options_enabled=True,
|
|
live_order=True,
|
|
live_trading=True,
|
|
show_perp_options=False,
|
|
)
|
|
self.assertFalse(g["can_preview"])
|
|
self.assertFalse(g["can_start"])
|
|
self.assertTrue(any("隐藏" in r for r in g["reasons"]))
|
|
self.assertFalse(g["show_perp_options"])
|
|
self.assertTrue(g["show_options_options"])
|
|
|
|
g2 = gate_status(
|
|
hedge_enabled=True,
|
|
sizing_mode="risk",
|
|
plan_type="options_options",
|
|
options_enabled=True,
|
|
show_options_options=False,
|
|
)
|
|
self.assertFalse(g2["can_preview"])
|
|
self.assertTrue(any("期期" in r for r in g2["reasons"]))
|
|
|
|
def test_oo_expiry_loss_flag(self):
|
|
a = {"opt_type": "C", "strike": 3300, "sheets": 1, "ct_mult": 0.01, "premium_paid": 5}
|
|
b = {"opt_type": "P", "strike": 3100, "sheets": 1, "ct_mult": 0.01, "premium_paid": 5}
|
|
p = build_options_options_preview(
|
|
profit_rr=2,
|
|
index_px=3200,
|
|
leg_a=a,
|
|
leg_b=b,
|
|
)
|
|
self.assertEqual(p["summary"]["premium_paid"], 10)
|
|
self.assertTrue(p["summary"]["expiry_is_loss"])
|
|
self.assertEqual(p["summary"]["profit_rr"], 2)
|
|
self.assertEqual(p["summary"]["at_rr_a_full_total"], 15) # 盈利=2*10, 亏腿-5
|
|
self.assertEqual(len(p["scenarios"]), 5)
|
|
self.assertEqual(p["scenarios"][0]["id"], "rr_leg_a_full")
|
|
self.assertEqual(p["scenarios"][1]["id"], "rr_leg_b_full")
|
|
# 到期实值反推:Call 盈利20 → 价值25 → 每币2500 → spot=3300+2500
|
|
self.assertEqual(p["scenarios"][0]["spot"], 5800.0)
|
|
# Put 盈利20 → spot=3100-2500
|
|
self.assertEqual(p["scenarios"][1]["spot"], 600.0)
|
|
self.assertEqual(p["scenarios"][2]["spot"], 5800.0) # 残值情景同腿A反推
|
|
|
|
def test_oo_legacy_single_target_still_works(self):
|
|
a = {"opt_type": "C", "strike": 3300, "sheets": 1, "ct_mult": 0.01, "premium_paid": 5}
|
|
b = {"opt_type": "P", "strike": 3100, "sheets": 1, "ct_mult": 0.01, "premium_paid": 5}
|
|
p = build_options_options_preview(target_price=3500, index_px=3200, leg_a=a, leg_b=b)
|
|
self.assertEqual(p["target_price_up"], 3500)
|
|
self.assertEqual(p["target_price_down"], 3500)
|
|
|
|
def test_perp_short_pnl(self):
|
|
self.assertEqual(
|
|
perp_pnl(direction="short", entry=100, exit_px=90, contracts=1, contract_size=1),
|
|
10,
|
|
)
|
|
|
|
def test_floor_contracts_to_precision(self):
|
|
self.assertEqual(floor_contracts_to_precision(4.569713, 4), 4.5697)
|
|
self.assertEqual(floor_contracts_to_precision(4.569713, 0), 4.0)
|
|
self.assertEqual(floor_contracts_to_precision(0, 4), 0.0)
|
|
|
|
def test_oo_budget_min_trading_and_cap(self):
|
|
b = resolve_oo_budget_usdc(trading_usdc=11.07, trade_budget_usdc=10, buffer_ratio=0.95)
|
|
self.assertTrue(b["ok"])
|
|
self.assertAlmostEqual(b["trading_cap"], 11.07 * 0.95, places=4)
|
|
self.assertEqual(b["budget_usdc"], 10.0)
|
|
|
|
def test_suggest_oo_same_sheets_default(self):
|
|
# cost_a=1, cost_b=1 → pair=2; budget=10 → n=5
|
|
s = suggest_oo_sheets(
|
|
mode="same_sheets",
|
|
budget_usdc=10,
|
|
ask_a=100,
|
|
ct_mult_a=0.01,
|
|
ask_b=100,
|
|
ct_mult_b=0.01,
|
|
)
|
|
self.assertEqual(s["mode"], "same_sheets")
|
|
self.assertEqual(s["sheets_a"], 5)
|
|
self.assertEqual(s["sheets_b"], 5)
|
|
self.assertTrue(s["ok"])
|
|
|
|
def test_suggest_oo_long_bias_budget(self):
|
|
# cost_call=1, cost_put=1; budget 10 → call 7U / put 3U → 7 / 3
|
|
s = suggest_oo_sheets(
|
|
mode="long_bias",
|
|
budget_usdc=10,
|
|
ask_a=100,
|
|
ct_mult_a=0.01,
|
|
opt_type_a="C",
|
|
ask_b=100,
|
|
ct_mult_b=0.01,
|
|
opt_type_b="P",
|
|
bias_split_by="budget",
|
|
bias_ratio=0.7,
|
|
)
|
|
self.assertEqual(s["mode"], "long_bias")
|
|
self.assertEqual(s["sheets_a"], 7)
|
|
self.assertEqual(s["sheets_b"], 3)
|
|
self.assertTrue(s["ok"])
|
|
|
|
def test_suggest_oo_short_bias_sheets(self):
|
|
# 同张数 n=5 → 总张数 10; short → put 7 / call 3
|
|
s = suggest_oo_sheets(
|
|
mode="short_bias",
|
|
budget_usdc=10,
|
|
ask_a=100,
|
|
ct_mult_a=0.01,
|
|
opt_type_a="C",
|
|
ask_b=100,
|
|
ct_mult_b=0.01,
|
|
opt_type_b="P",
|
|
bias_split_by="sheets",
|
|
bias_ratio=0.7,
|
|
)
|
|
self.assertEqual(s["mode"], "short_bias")
|
|
self.assertEqual(s["sheets_a"], 3)
|
|
self.assertEqual(s["sheets_b"], 7)
|
|
|
|
def test_suggest_oo_bias_requires_call_put(self):
|
|
s = suggest_oo_sheets(
|
|
mode="long_bias",
|
|
budget_usdc=10,
|
|
ask_a=100,
|
|
ct_mult_a=0.01,
|
|
opt_type_a="C",
|
|
ask_b=100,
|
|
ct_mult_b=0.01,
|
|
opt_type_b="C",
|
|
bias_split_by="budget",
|
|
)
|
|
self.assertFalse(s["ok"])
|
|
self.assertIn("Call", s["msg"])
|
|
|
|
def test_suggest_oo_depth_cap(self):
|
|
s = suggest_oo_sheets(
|
|
mode="same_sheets",
|
|
budget_usdc=100,
|
|
ask_a=100,
|
|
ct_mult_a=0.01,
|
|
ask_sz_a=2,
|
|
ask_b=100,
|
|
ct_mult_b=0.01,
|
|
ask_sz_b=50,
|
|
)
|
|
self.assertEqual(s["sheets_a"], 2)
|
|
self.assertEqual(s["sheets_b"], 2)
|
|
|
|
def test_oo_rest_close_mode_default_close_all(self):
|
|
self.assertEqual(resolve_oo_rest_close_mode({"oo_close_mode": "close_all"}), "close_all")
|
|
self.assertEqual(resolve_oo_rest_close_mode({"oo_close_mode": "hold_expiry"}), "hold_expiry")
|
|
# 旧计划无字段:保持到期平,避免部署后误清残腿
|
|
self.assertEqual(resolve_oo_rest_close_mode({}), "hold_expiry")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|