crypto_monitor/tests/test_hedge_plan_orders.py

"""对冲计划下单路径校验(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()