dekun
f9c2a63cbc
Reject OO start when max active is under 2 free slots so limit=1 cannot open a half straddle. Co-authored-by: Cursor <cursoragent@cursor.com>
102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
"""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()
|