Files
crypto_monitor/tests/test_dashboard_position_source.py
T
dekun c4c8ad172a Label dashboard positions from monitor sources with priority.
Match hedge/roll/trend/order/key monitors for source badges, show option target monitors in green, and refresh the dashboard with the board 5s cycle.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-17 14:25:36 +08:00

64 lines
2.0 KiB
Python

"""数据看板仓位来源:监控匹配优先级."""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "manual_trading_hub"))
sys.path.insert(0, str(ROOT))
from hub_ai.context import ( # noqa: E402
_options_source_label,
resolve_position_monitor_source,
)
class TestDashboardPositionSource(unittest.TestCase):
def test_priority_hedge_over_roll(self):
hub = {
"ok": True,
"hedges": [
{
"plan_type": "perp_options",
"direction": "long",
"legs": [{"leg_role": "perp", "symbol": "ETH/USDT:USDT", "status": "open"}],
}
],
"rolls": [{"symbol": "ETH/USDT:USDT", "direction": "long"}],
}
self.assertEqual(
resolve_position_monitor_source({"symbol": "ETH/USDT:USDT", "side": "long"}, hub),
"永期对冲",
)
def test_unmatched_is_dash(self):
hub = {"ok": True, "orders": [], "trends": [], "rolls": [], "keys": [], "hedges": []}
self.assertEqual(
resolve_position_monitor_source({"symbol": "BTC/USDT:USDT", "side": "short"}, hub),
"",
)
def test_roll_beats_order(self):
hub = {
"ok": True,
"rolls": [{"symbol": "BTC/USDT:USDT", "direction": "short"}],
"orders": [{"symbol": "BTC/USDT:USDT", "direction": "short", "monitor_type": "下单监控"}],
}
self.assertEqual(
resolve_position_monitor_source({"symbol": "BTC/USDT:USDT", "side": "short"}, hub),
"顺势加仓",
)
def test_options_plain_is_dash(self):
self.assertEqual(_options_source_label({"source": "option", "source_label": "纯期权"}), "")
self.assertEqual(
_options_source_label({"source": "options_options", "source_label": "期期对冲"}),
"期期对冲",
)
if __name__ == "__main__":
unittest.main()