"""对冲计划下单路径校验(dry_run + 门禁).""" import unittest from unittest.mock import MagicMock from lib.hedge_plan.hedge_plan_calc_lib import gate_status from lib.hedge_plan.hedge_plan_orders_lib import ( build_oo_path_plan, build_po_path_plan, execute_options_options_start, execute_perp_options_start, validate_start_body, ) class TestHedgePlanOrderPath(unittest.TestCase): def test_po_path_options_first(self): body = { "opt_inst_id": "ETH-USD-260731-1800-P", "sheets": 2, "exchange_symbol": "ETH/USDT:USDT", "direction": "long", "contracts": 4.5, "tp": 1900, "sl": 1700, } path = build_po_path_plan(body) self.assertEqual(path[0]["step"], "options_buy_limit") self.assertEqual(path[0]["account"], "options") self.assertEqual(path[1]["step"], "perp_market_open") self.assertEqual(path[1]["account"], "swap") self.assertTrue(path[1]["attach_tpsl"]) def test_oo_path_two_option_buys(self): body = { "leg_a": {"inst_id": "ETH-USD-260731-1800-C", "sheets": 1}, "leg_b": {"inst_id": "ETH-USD-260731-1700-P", "sheets": 3}, } path = build_oo_path_plan(body) self.assertEqual(len(path), 2) self.assertEqual(path[0]["leg"], "a") self.assertEqual(path[1]["sheets"], 3) def test_validate_body(self): self.assertIsNotNone(validate_start_body("perp_options", {})) ok = validate_start_body( "perp_options", { "direction": "long", "entry": 1800, "tp": 1900, "sl": 1700, "contracts": 1, "opt_inst_id": "X", "sheets": 1, "exchange_symbol": "ETH/USDT:USDT", }, ) self.assertIsNone(ok) def test_gate_can_start_when_live(self): g = gate_status( hedge_enabled=True, sizing_mode="full_margin", plan_type="perp_options", options_enabled=True, live_order=True, live_trading=True, active_count=0, max_active=1, ) self.assertTrue(g["can_start"]) self.assertEqual(g["reasons"], []) def test_gate_oo_without_live_trading(self): g = gate_status( hedge_enabled=True, sizing_mode="risk", plan_type="options_options", options_enabled=True, live_order=True, live_trading=False, active_count=0, max_active=1, ) self.assertTrue(g["can_start"]) def test_dry_run_po_calls_quote_not_place(self): quote = MagicMock( return_value={ "ok": True, "ask": 12.5, "ask_sz": 10, "can_open": True, "ct_mult": 0.01, "tick_sz": "0.1", "strike": 1800, "exp_time": 1, "meta": {"optType": "P"}, } ) place_opt = MagicMock() place_perp = MagicMock() cfg = { "exchange_options": object(), "exchange": object(), "quote_option_contract": quote, "place_option_limit_order": place_opt, "place_exchange_order": place_perp, "td_mode_for_option_buy": lambda x: "isolated", "amount_to_precision": lambda s, a: a, "ensure_okx_live_ready": lambda: (True, ""), } body = { "direction": "long", "entry": 1800, "tp": 1900, "sl": 1700, "contracts": 4.5, "opt_inst_id": "ETH-USD-260731-1800-P", "sheets": 2, "exchange_symbol": "ETH/USDT:USDT", "leverage": 10, "underlying": "ETH", } out = execute_perp_options_start(cfg, body, dry_run=True) self.assertTrue(out["ok"]) self.assertTrue(out["dry_run"]) place_opt.assert_not_called() place_perp.assert_not_called() quote.assert_called() self.assertEqual(out["path"][0]["account"], "options") self.assertEqual(out["path"][1]["account"], "swap") def test_dry_run_oo(self): quote = MagicMock( side_effect=lambda _ex, inst_id: { "ok": True, "ask": 10, "ask_sz": 50, "can_open": True, "ct_mult": 0.01, "tick_sz": "0.1", "strike": 1800, "meta": {"optType": "C" if inst_id == "A" else "P"}, } ) cfg = { "exchange_options": object(), "quote_option_contract": quote, "place_option_limit_order": MagicMock(), "td_mode_for_option_buy": lambda x: "isolated", "trade_budget_usdc": 10, "budget_buffer": 0.95, } body = { "target_price": 1900, "target_price_up": 1950, "target_price_down": 1750, "oo_sheets_mode": "same_sheets", "leg_a": {"inst_id": "A", "sheets": 1, "opt_type": "C"}, "leg_b": {"inst_id": "B", "sheets": 1, "opt_type": "P"}, } from unittest import mock with mock.patch( "lib.exchange.okx_options_lib.fetch_options_trading_usdc", return_value=100.0, ): out = execute_options_options_start(cfg, body, dry_run=True) self.assertTrue(out["ok"], out) self.assertEqual(len(out["results"]), 2) self.assertTrue(out.get("refresh", {}).get("ok")) def test_buy_rejects_without_ask_depth(self): from lib.hedge_plan.hedge_plan_orders_lib import _buy_option quote = MagicMock( return_value={ "ok": True, "ask": None, "ask_sz": None, "mark": 11.2, "ref_ask": 11.2, "can_open": False, "open_block_msg": "暂无卖一深度,无法买入", "ct_mult": 0.01, } ) cfg = { "exchange_options": object(), "quote_option_contract": quote, "place_option_limit_order": MagicMock(), } out = _buy_option(cfg, inst_id="ETH-USD_UM-260717-1900-C", sheets=1, dry_run=True) self.assertFalse(out["ok"]) self.assertIn("卖一", out["msg"]) cfg["place_option_limit_order"].assert_not_called() def test_buy_caps_sheets_to_ask_depth(self): from lib.hedge_plan.hedge_plan_orders_lib import _buy_option quote = MagicMock( return_value={ "ok": True, "ask": 10, "ask_sz": 2, "can_open": True, "ct_mult": 0.01, "tick_sz": "0.1", "meta": {"optType": "C"}, } ) cfg = { "exchange_options": object(), "quote_option_contract": quote, "place_option_limit_order": MagicMock(), "td_mode_for_option_buy": lambda x: "isolated", } out = _buy_option(cfg, inst_id="X", sheets=9, dry_run=True) self.assertTrue(out["ok"]) self.assertEqual(out["sheets"], 2) if __name__ == "__main__": unittest.main()