crypto_monitor_user/tests/test_instance_dashboard_lib.py

"""instance_dashboard_lib 单元测试."""
from __future__ import annotations

import sqlite3
import unittest

from lib.instance.instance_dashboard_lib import build_instance_dashboard_payload


def _mem_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.executescript(
        """
        CREATE TABLE order_monitors (
            id INTEGER PRIMARY KEY,
            symbol TEXT,
            exchange_symbol TEXT,
            direction TEXT,
            status TEXT,
            monitor_type TEXT,
            key_signal_type TEXT,
            trigger_price REAL,
            stop_loss REAL,
            take_profit REAL
        );
        CREATE TABLE key_monitors (
            id INTEGER PRIMARY KEY,
            symbol TEXT,
            exchange_symbol TEXT,
            direction TEXT,
            signal_type TEXT,
            upper REAL,
            lower REAL,
            status TEXT
        );
        CREATE TABLE trend_pullback_plans (
            id INTEGER PRIMARY KEY,
            symbol TEXT,
            exchange_symbol TEXT,
            direction TEXT,
            status TEXT,
            entry_price REAL
        );
        CREATE TABLE roll_groups (
            id INTEGER PRIMARY KEY,
            order_monitor_id INTEGER,
            symbol TEXT,
            exchange_symbol TEXT,
            direction TEXT,
            status TEXT
        );
        """
    )
    return conn


class TestInstanceDashboardLib(unittest.TestCase):
    def test_empty_sections_and_conditional_hidden(self):
        conn = _mem_conn()
        payload = build_instance_dashboard_payload(conn, hedge_enabled=True)
        self.assertTrue(payload["ok"])
        self.assertEqual(payload["orders"]["count"], 0)
        self.assertEqual(payload["keys"]["count"], 0)
        self.assertEqual(payload["strategy"]["count"], 0)
        self.assertFalse(payload["options"]["visible"])
        self.assertFalse(payload["hedge_plan"]["visible"])
        conn.close()

    def test_orders_keys_strategy_and_options_visible(self):
        conn = _mem_conn()
        conn.execute(
            "INSERT INTO order_monitors (symbol, exchange_symbol, direction, status, monitor_type) "
            "VALUES ('BTC/USDT', 'BTC/USDT:USDT', 'long', 'active', 'manual')"
        )
        conn.execute(
            "INSERT INTO key_monitors (symbol, direction, signal_type, upper, lower, status) "
            "VALUES ('ETH/USDT', 'short', '箱体突破', 3000, 2800, 'active')"
        )
        conn.execute(
            "INSERT INTO trend_pullback_plans (symbol, direction, status, entry_price) "
            "VALUES ('SOL/USDT', 'long', 'active', 100)"
        )
        conn.execute(
            "INSERT INTO order_monitors (id, symbol, direction, status) VALUES (9, 'XRP/USDT', 'short', 'active')"
        )
        conn.execute(
            "INSERT INTO roll_groups (order_monitor_id, symbol, direction, status) "
            "VALUES (9, 'XRP/USDT', 'short', 'active')"
        )
        conn.commit()

        def fetch_opts():
            return [{"inst_id": "ETH-USD-260731-3000-C", "opt_type": "C", "pos": 1, "upl": 1.5}]

        payload = build_instance_dashboard_payload(
            conn,
            fetch_options_positions=fetch_opts,
            hedge_enabled=False,
        )
        self.assertEqual(payload["orders"]["count"], 2)
        self.assertEqual(payload["keys"]["count"], 1)
        self.assertEqual(payload["strategy"]["count"], 2)
        self.assertTrue(payload["options"]["visible"])
        self.assertEqual(payload["options"]["count"], 1)
        self.assertEqual(payload["options"]["items"][0]["source_label"], "纯期权")
        self.assertFalse(payload["hedge_plan"]["visible"])
        conn.close()

    def test_hedge_status_label_active(self):
        conn = _mem_conn()
        conn.executescript(
            """
            CREATE TABLE hedge_plans (
                id INTEGER PRIMARY KEY,
                underlying TEXT,
                plan_type TEXT,
                status TEXT
            );
            CREATE TABLE hedge_plan_legs (
                id INTEGER PRIMARY KEY,
                plan_id INTEGER,
                leg_role TEXT,
                symbol TEXT,
                inst_id TEXT,
                opt_type TEXT,
                status TEXT
            );
            """
        )
        conn.execute(
            "INSERT INTO hedge_plans (id, underlying, plan_type, status) "
            "VALUES (2, 'ETH', 'options_options', 'active')"
        )
        conn.execute(
            "INSERT INTO hedge_plan_legs (plan_id, leg_role, inst_id, opt_type, status) "
            "VALUES (2, 'option', 'ETH-USD-260719-1850-P', 'P', 'open')"
        )
        conn.commit()

        def fetch_opts():
            return [
                {
                    "inst_id": "ETH-USD-260719-1850-P",
                    "opt_type": "P",
                    "pos": 40,
                    "upl": 1.2,
                    "exp_time_ms": 1784505600000,
                    "hedge_plan_target": {
                        "plan_id": 2,
                        "opt_type": "P",
                        "target_index": 1800,
                    },
                }
            ]

        payload = build_instance_dashboard_payload(
            conn,
            fetch_options_positions=fetch_opts,
            hedge_enabled=True,
        )
        self.assertTrue(payload["hedge_plan"]["visible"])
        self.assertEqual(payload["hedge_plan"]["items"][0]["status_label"], "进行中")
        self.assertTrue(payload["hedge_plan"]["items"][0]["status_active"])
        opt = payload["options"]["items"][0]
        self.assertEqual(opt["source_label"], "期期对冲")
        self.assertIn("对冲#2", opt["target_monitor"])
        conn.close()


if __name__ == "__main__":
    unittest.main()