Add mutual-exclusion gate between hedge plans and standalone options.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-19 08:51:46 +08:00
parent 899a2de931
commit e11c13747a
8 changed files with 236 additions and 0 deletions
@@ -0,0 +1,103 @@
+"""对冲计划与单独期权互斥门控."""
+from __future__ import annotations
+
+import os
+import sqlite3
+import unittest
+from unittest import mock
+
+from lib.hedge_plan.hedge_options_exclusive_lib import (
+    block_hedge_plan_start_msg,
+    block_standalone_option_open_msg,
+    has_standalone_option_position,
+    mutual_exclusive_enabled,
+)
+from lib.hedge_plan.hedge_plan_calc_lib import gate_status
+from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables
+
+
+class HedgeOptionsExclusiveTests(unittest.TestCase):
+    def test_mutual_default_true(self):
+        with mock.patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE", None)
+            self.assertTrue(mutual_exclusive_enabled())
+
+    def test_mutual_can_disable(self):
+        with mock.patch.dict(os.environ, {"HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE": "false"}):
+            self.assertFalse(mutual_exclusive_enabled())
+
+    def test_block_open_when_active_plan(self):
+        conn = sqlite3.connect(":memory:")
+        conn.row_factory = sqlite3.Row
+        init_hedge_plan_tables(conn)
+        conn.execute(
+            "INSERT INTO hedge_plans (plan_type, underlying, status) "
+            "VALUES ('options_options', 'ETH', 'active')"
+        )
+        conn.commit()
+        with mock.patch.dict(os.environ, {"HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE": "true"}):
+            msg = block_standalone_option_open_msg(conn)
+        self.assertIsNotNone(msg)
+        self.assertIn("对冲计划", msg or "")
+        with mock.patch.dict(os.environ, {"HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE": "false"}):
+            self.assertIsNone(block_standalone_option_open_msg(conn))
+        conn.close()
+
+    def test_standalone_position_detection(self):
+        conn = sqlite3.connect(":memory:")
+        conn.row_factory = sqlite3.Row
+        init_hedge_plan_tables(conn)
+        raw = [{"instId": "ETH-USD_UM-260719-1890-C", "pos": "1"}]
+        with mock.patch(
+            "lib.instance.instance_dashboard_lib._resolve_options_source",
+            return_value=("option", "纯期权", None),
+        ):
+            self.assertTrue(has_standalone_option_position(conn, raw))
+        with mock.patch(
+            "lib.instance.instance_dashboard_lib._resolve_options_source",
+            return_value=("options_options", "期期对冲", 2),
+        ):
+            self.assertFalse(has_standalone_option_position(conn, raw))
+        conn.close()
+
+    def test_block_hedge_start_msg(self):
+        conn = sqlite3.connect(":memory:")
+        conn.row_factory = sqlite3.Row
+        init_hedge_plan_tables(conn)
+        raw = [{"instId": "ETH-USD_UM-260719-1890-C", "pos": "2"}]
+        with mock.patch.dict(os.environ, {"HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE": "true"}):
+            with mock.patch(
+                "lib.instance.instance_dashboard_lib._resolve_options_source",
+                return_value=("option", "纯期权", None),
+            ):
+                msg = block_hedge_plan_start_msg(conn, raw_positions=raw)
+        self.assertIsNotNone(msg)
+        self.assertIn("单独期权", msg or "")
+        conn.close()
+
+    def test_gate_status_blocks_start(self):
+        g = gate_status(
+            hedge_enabled=True,
+            sizing_mode="full_margin",
+            plan_type="options_options",
+            options_enabled=True,
+            live_order=True,
+            mutual_exclusive=True,
+            has_standalone_option=True,
+        )
+        self.assertFalse(g["can_start"])
+        self.assertTrue(any("单独期权" in r for r in g["reasons"]))
+        g2 = gate_status(
+            hedge_enabled=True,
+            sizing_mode="full_margin",
+            plan_type="options_options",
+            options_enabled=True,
+            live_order=True,
+            mutual_exclusive=False,
+            has_standalone_option=True,
+        )
+        self.assertTrue(g2["can_start"])
+
+
+if __name__ == "__main__":
+    unittest.main()