Initial CoinOX instance: coin-margined options + USDT hedge.

Forked from crypto_monitor OKX instance (copy-only). Keeps standalone options, perp-options and options-options hedge; drops standalone perp entrypoints; forces coin margin; adds sim wallets; default port 5010. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-14 00:18:40 +08:00
commit 88b7fb7a0a
381 changed files with 104628 additions and 0 deletions
@@ -0,0 +1,101 @@
+"""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_oo_needs_two_slots_when_max_one(self):
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_ids=["ETH-C", "ETH-P"],
+            max_active=1,
+            fetch_positions=lambda _ex: [],
+        )
+        self.assertIsNotNone(msg)
+        self.assertIn("期期", msg or "")
+
+    def test_oo_ok_when_max_two_empty(self):
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_ids=["ETH-C", "ETH-P"],
+            max_active=2,
+            fetch_positions=lambda _ex: [],
+        )
+        self.assertIsNone(msg)
+
+    def test_oo_block_when_one_slot_left(self):
+        rows = [{"instId": "OTHER", "pos": "1"}]
+        msg = option_position_limit_block_msg(
+            object(),
+            opening_inst_ids=["ETH-C", "ETH-P"],
+            max_active=2,
+            fetch_positions=lambda _ex: rows,
+        )
+        self.assertIsNotNone(msg)
+        self.assertIn("期期", msg or "")
+
+    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()