"""对冲计划 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()