"""永期实例模式 HEDGE_PO_MODE:钳制/定仓/校验/组合U口径.""" from __future__ import annotations import os import unittest from unittest.mock import patch from lib.hedge_plan.hedge_plan_option_primary_lib import ( contracts_from_option_sheets, estimate_combo_net_pnl, size_from_premium, ) from lib.hedge_plan.hedge_po_mode_lib import ( MODE_BTC, MODE_BTC_ETH, MODE_ETH, apply_mode_to_start_body, clamp_hedge_po_mode_on_startup, combo_net_usdt, default_option_perp_ratio, get_hedge_po_mode, normalize_hedge_po_mode, option_underlying, perp_underlying, public_mode_payload, size_cross_notional, validate_body_matches_mode, ) class TestHedgePoMode(unittest.TestCase): def test_normalize_aliases(self): self.assertEqual(normalize_hedge_po_mode("ETH"), MODE_ETH) self.assertEqual(normalize_hedge_po_mode("btc-eth"), MODE_BTC_ETH) self.assertEqual(normalize_hedge_po_mode("跨币"), MODE_BTC_ETH) self.assertEqual(normalize_hedge_po_mode("nope"), "") def test_underlyings(self): self.assertEqual(option_underlying(MODE_ETH), "ETH") self.assertEqual(perp_underlying(MODE_ETH), "ETH") self.assertEqual(option_underlying(MODE_BTC), "BTC") self.assertEqual(perp_underlying(MODE_BTC), "BTC") self.assertEqual(option_underlying(MODE_BTC_ETH), "BTC") self.assertEqual(perp_underlying(MODE_BTC_ETH), "ETH") def test_cross_ratio_from_k(self): with patch.dict(os.environ, {"HEDGE_CROSS_NOTIONAL_K": "0.5"}, clear=False): self.assertAlmostEqual(default_option_perp_ratio(MODE_BTC_ETH), 2.0) def test_clamp_startup_and_sync_underly(self): with patch.dict( os.environ, {"HEDGE_PO_MODE": "btc_eth", "OKX_OPTIONS_DEFAULT_UNDERLY": "ETH"}, clear=False, ): summary = clamp_hedge_po_mode_on_startup() self.assertEqual(summary["mode"], MODE_BTC_ETH) self.assertEqual(os.environ["HEDGE_PO_MODE"], MODE_BTC_ETH) self.assertEqual(os.environ["OKX_OPTIONS_DEFAULT_UNDERLY"], "BTC") self.assertEqual(get_hedge_po_mode(), MODE_BTC_ETH) def test_clamp_invalid_falls_back(self): with patch.dict( os.environ, {"HEDGE_PO_MODE": "weird", "OKX_OPTIONS_DEFAULT_UNDERLY": "BTC"}, clear=False, ): summary = clamp_hedge_po_mode_on_startup() self.assertEqual(summary["mode"], MODE_BTC) self.assertTrue(summary["warnings"]) def test_apply_and_validate_body(self): with patch.dict(os.environ, {"HEDGE_PO_MODE": "btc_eth", "HEDGE_CROSS_NOTIONAL_K": "0.5"}): clamp_hedge_po_mode_on_startup() body = apply_mode_to_start_body( { "direction": "long", "underlying": "ETH", "opt_inst_id": "BTC-USD-260831-90000-C", "exchange_symbol": "ETH/USDT:USDT", "profit_target_usdt": 20, } ) self.assertEqual(body["underlying"], "BTC") self.assertEqual(body["perp_underlying"], "ETH") self.assertEqual(body["exit_mode"], "profit") self.assertTrue(body["option_primary"]) self.assertAlmostEqual(float(body["option_perp_ratio"]), 2.0) self.assertIsNone(validate_body_matches_mode(body)) bad = dict(body, opt_inst_id="ETH-USD-260831-2000-C") self.assertIn("BTC", validate_body_matches_mode(bad) or "") def test_size_cross_notional_half(self): # 0.1 BTC @ 100k → 10k U 期权名义; k=0.5 → 5k ETH 名义 @ 2500 → 2 ETH → 200 张(cs=0.01) sz = size_cross_notional( option_coin_qty=0.1, option_index=100_000, perp_index=2500, contract_size=0.01, notional_k=0.5, ) self.assertTrue(sz["ok"]) self.assertAlmostEqual(sz["option_notional_usdt"], 10_000.0) self.assertAlmostEqual(sz["perp_notional_usdt"], 5_000.0) self.assertAlmostEqual(sz["perp_coin"], 2.0) self.assertAlmostEqual(sz["contracts"], 200.0) def test_size_from_premium_cross_matches_notional(self): # ask=0.01 BTC → 预算 1 BTC 权利金(币); usable 0.95 → coin=0.95 → sheets # 用整张回写后再跨币 sized = size_from_premium( premium_budget=1000, # 若 ask 用币报价则预算单位需一致;此处用大预算保证 ≥1 张 ask=0.01, ct_mult=0.01, ratio=2, contract_size=0.01, option_index=100_000, perp_index=2_500, ) self.assertTrue(sized["ok"], sized.get("msg")) self.assertTrue(sized.get("cross_asset") or sized.get("option_notional_usdt")) # coin_qty = sheets * 0.01; 名义路径 self.assertGreater(float(sized["contracts"]), 0) def test_contracts_from_sheets_same_vs_cross(self): same = contracts_from_option_sheets( sheets=10, ct_mult=0.01, ratio=2, contract_size=0.01 ) self.assertTrue(same["ok"]) # 0.1 币 / 2 / 0.01 = 5 张 self.assertAlmostEqual(same["contracts"], 5.0) cross = contracts_from_option_sheets( sheets=10, ct_mult=0.01, ratio=2, contract_size=0.01, option_index=100_000, perp_index=2_500, ) self.assertTrue(cross["ok"]) # 0.1 BTC * 100k * 0.5 / 2500 / 0.01 = 200 self.assertAlmostEqual(cross["contracts"], 200.0) def test_combo_net_usdt_not_mix_units(self): net = estimate_combo_net_pnl( view_side="long", strike=90000, index_px=91000, ask_open=0.01, bid=0.012, sheets=1, ct_mult=0.01, perp_direction="short", perp_entry=2500, perp_mark=2450, contracts=10, contract_size=0.01, fee=0.0, option_fee=0.0, ) # opt_net 为币;perp_net 为 U;禁止直接用 net['net'] 出门控 u = combo_net_usdt(net, option_index=91000) self.assertIsNotNone(u) self.assertAlmostEqual(u, float(net["opt_net"]) * 91000 + float(net["perp_net"]), places=4) def test_public_payload(self): with patch.dict(os.environ, {"HEDGE_PO_MODE": "eth"}): clamp_hedge_po_mode_on_startup() p = public_mode_payload() self.assertEqual(p["po_mode"], MODE_ETH) self.assertEqual(p["exit_mode"], "profit") self.assertIn("子账户", p["instance_rule"]) if __name__ == "__main__": unittest.main()