Add OKX options max active positions env gate.

Enforce concurrent option contract count on standalone and hedge buys; editable in env UI with hot reload (0 = unlimited). Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-30 08:12:50 +08:00
parent 644dcdf092
commit 94c65cbd6e
8 changed files with 203 additions and 0 deletions
@@ -0,0 +1,71 @@
+"""OKX 期权持仓笔数上限."""
+
+import os
+import unittest
+from unittest.mock import patch
+
+from lib.options.options_position_limit_lib import (
+    count_live_option_positions,
+    option_position_limit_block_msg,
+    options_max_active_positions,
+)
+
+
+class OptionsPositionLimitTests(unittest.TestCase):
+    def test_default_unlimited(self):
+        with patch.dict(os.environ, {}, clear=False):
+            os.environ.pop("OKX_OPTIONS_MAX_ACTIVE_POSITIONS", None)
+            self.assertEqual(options_max_active_positions(), 0)
+
+    def test_parse_max(self):
+        with patch.dict(os.environ, {"OKX_OPTIONS_MAX_ACTIVE_POSITIONS": "2"}):
+            self.assertEqual(options_max_active_positions(), 2)
+
+    def test_count_live(self):
+        rows = [{"instId": "A", "pos": "1"}, {"instId": "B", "pos": "0"}, {"instId": "C", "pos": "-2"}]
+        self.assertEqual(count_live_option_positions(rows), 2)
+
+    def test_block_when_at_limit(self):
+        rows = [{"instId": "ETH-C", "pos": "1"}, {"instId": "ETH-P", "pos": "2"}]
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_id="ETH-NEW",
+            max_active=2,
+            fetch_positions=lambda _ex: rows,
+        )
+        self.assertIsNotNone(msg)
+        self.assertIn("上限", msg or "")
+
+    def test_allow_add_to_existing(self):
+        rows = [{"instId": "ETH-C", "pos": "1"}, {"instId": "ETH-P", "pos": "2"}]
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_id="ETH-C",
+            max_active=2,
+            fetch_positions=lambda _ex: rows,
+        )
+        self.assertIsNone(msg)
+
+    def test_allow_under_limit(self):
+        rows = [{"instId": "ETH-C", "pos": "1"}]
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_id="ETH-P",
+            max_active=2,
+            fetch_positions=lambda _ex: rows,
+        )
+        self.assertIsNone(msg)
+
+    def test_fail_closed_when_fetch_none(self):
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_id="ETH-C",
+            max_active=1,
+            fetch_positions=lambda _ex: None,
+        )
+        self.assertIsNotNone(msg)
+        self.assertIn("无法获取", msg or "")
+
+
+if __name__ == "__main__":
+    unittest.main()