dekun
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""永期监控:假平仓防护与 TP/SL 分类."""
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from lib.hedge_plan.hedge_plan_monitor_lib import (
|
|
_classify_po_flat_reason,
|
|
_tick_po,
|
|
_within_open_grace,
|
|
)
|
|
|
|
|
|
class TestPoMonitorSafety(unittest.TestCase):
|
|
def test_classify_sl_before_tp(self):
|
|
self.assertEqual(
|
|
_classify_po_flat_reason(
|
|
direction="long", entry=1800, mark=1690, tp=1900, sl=1700
|
|
),
|
|
"perp_sl",
|
|
)
|
|
self.assertEqual(
|
|
_classify_po_flat_reason(
|
|
direction="long", entry=1800, mark=1910, tp=1900, sl=1700
|
|
),
|
|
"perp_tp",
|
|
)
|
|
|
|
def test_classify_ambiguous_prefers_sl(self):
|
|
# mark 在中间:偏 SL,避免误判 TP 跳过强平期权
|
|
self.assertEqual(
|
|
_classify_po_flat_reason(
|
|
direction="long", entry=1800, mark=1800, tp=1900, sl=1700
|
|
),
|
|
"perp_sl",
|
|
)
|
|
|
|
def test_classify_unknown_without_mark(self):
|
|
self.assertEqual(
|
|
_classify_po_flat_reason(
|
|
direction="long", entry=1800, mark=None, tp=1900, sl=1700
|
|
),
|
|
"perp_flat_unknown",
|
|
)
|
|
|
|
def test_tick_po_skips_when_live_none(self):
|
|
cfg = {"get_live_position_contracts": lambda *_a, **_k: None}
|
|
plan = {"id": 1, "direction": "long", "entry_mark": 1800, "tp": 1900, "sl": 1700}
|
|
legs = [{"leg_role": "perp", "status": "open", "symbol": "ETH/USDT:USDT", "size": 1}]
|
|
self.assertIsNone(_tick_po(cfg, MagicMock(), plan, legs))
|
|
|
|
def test_tick_po_skips_when_live_positive(self):
|
|
cfg = {"get_live_position_contracts": lambda *_a, **_k: 2.0}
|
|
plan = {"id": 1, "direction": "long", "entry_mark": 1800, "tp": 1900, "sl": 1700}
|
|
legs = [{"leg_role": "perp", "status": "open", "symbol": "ETH/USDT:USDT", "size": 1}]
|
|
self.assertIsNone(_tick_po(cfg, MagicMock(), plan, legs))
|
|
|
|
def test_within_grace_true_without_opened_at(self):
|
|
self.assertTrue(_within_open_grace({}))
|
|
|
|
def test_pending_sl_sticky(self):
|
|
"""perp_sl_pending_opt 粘滞,不因 mark 反弹改判 TP."""
|
|
cfg = {
|
|
"get_live_position_contracts": lambda *_a, **_k: 0.0,
|
|
"get_contract_size": lambda *_a, **_k: 0.01,
|
|
"exchange": None,
|
|
}
|
|
plan = {
|
|
"id": 9,
|
|
"direction": "long",
|
|
"entry_mark": 1800,
|
|
"tp": 1900,
|
|
"sl": 1700,
|
|
"premium_total": 5,
|
|
"opened_at": "2000-01-01 00:00:00",
|
|
"close_reason": "perp_sl_pending_opt",
|
|
}
|
|
legs = [
|
|
{"id": 1, "leg_role": "perp", "status": "open", "symbol": "ETH/USDT:USDT", "size": 1, "avg_open": 1800},
|
|
{
|
|
"id": 2,
|
|
"leg_role": "option_hedge",
|
|
"status": "open",
|
|
"inst_id": "ETH-USD-260731-1700-P",
|
|
"size": 1,
|
|
"avg_open": 10,
|
|
},
|
|
]
|
|
with patch("lib.hedge_plan.hedge_plan_monitor_lib._sell_option") as sell:
|
|
sell.return_value = {"ok": False, "msg": "thin book", "fully_closed": False}
|
|
with patch("lib.hedge_plan.hedge_plan_monitor_lib.notify_hedge"):
|
|
with patch("lib.hedge_plan.hedge_plan_monitor_lib.update_plan") as up:
|
|
conn = MagicMock()
|
|
out = _tick_po(cfg, conn, plan, legs)
|
|
self.assertTrue(out and out.get("retry"))
|
|
up.assert_called()
|
|
self.assertEqual(up.call_args.kwargs.get("close_reason"), "perp_sl_pending_opt")
|
|
self.assertNotEqual(up.call_args.kwargs.get("status"), "closed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|