dekun
104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""对冲计划与单独期权互斥门控."""
|
|
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()
|