crypto_monitor/tests/test_hedge_plan_history_stats.py

"""对冲计划历史删除与分类型统计."""
import sqlite3
import unittest

from lib.hedge_plan.hedge_plan_db import (
    _metrics_from_pnls,
    active_options_targets_by_inst,
    delete_plan,
    init_hedge_plan_tables,
    insert_leg,
    insert_plan,
    legs_contract_summary,
    stats_summary,
)


def _mem():
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    init_hedge_plan_tables(conn)
    return conn


class TestHedgeHistoryStats(unittest.TestCase):
    def test_metrics_win_rate_pf_dd(self):
        rows = [
            {"realized_pnl_total": 10, "closed_at": "2026-01-01", "premium_total": 1},
            {"realized_pnl_total": -4, "closed_at": "2026-01-02", "premium_total": 1},
            {"realized_pnl_total": 6, "closed_at": "2026-01-03", "premium_total": 1},
            {"realized_pnl_total": -12, "closed_at": "2026-01-04", "premium_total": 1},
        ]
        m = _metrics_from_pnls(rows)
        self.assertEqual(m["count"], 4)
        self.assertEqual(m["wins"], 2)
        self.assertAlmostEqual(m["win_rate"], 0.5)
        # gross win 16 / gross loss 16 = 1
        self.assertAlmostEqual(m["profit_factor"], 1.0)
        self.assertAlmostEqual(m["max_profit"], 10)
        self.assertAlmostEqual(m["max_loss"], -12)
        # equity: 10 → 6 → 12 → 0; peak 12, dd to 0 = 12
        self.assertAlmostEqual(m["max_drawdown"], 12)

    def test_stats_by_type_and_delete(self):
        conn = _mem()
        po = insert_plan(
            conn,
            {
                "plan_type": "perp_options",
                "status": "closed",
                "underlying": "ETH",
                "realized_pnl_total": 5,
                "premium_total": 1,
                "close_reason": "perp_tp",
                "closed_at": "2026-07-01 10:00:00",
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": po,
                "leg_role": "perp",
                "symbol": "ETH/USDT:USDT",
                "status": "closed",
            },
        )
        oo = insert_plan(
            conn,
            {
                "plan_type": "options_options",
                "status": "closed",
                "underlying": "ETH",
                "realized_pnl_total": -2,
                "premium_total": 0.02,
                "close_reason": "oo_expiry_loss",
                "closed_at": "2026-07-02 10:00:00",
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": oo,
                "leg_role": "option_a",
                "inst_id": "ETH-USD_UM-260715-1900-C",
                "status": "closed",
            },
        )
        active = insert_plan(
            conn,
            {
                "plan_type": "perp_options",
                "status": "active",
                "underlying": "BTC",
                "realized_pnl_total": None,
            },
        )
        s = stats_summary(conn)
        self.assertEqual(s["closed_count"], 2)
        self.assertEqual(s["by_type"]["perp_options"]["count"], 1)
        self.assertEqual(s["by_type"]["options_options"]["count"], 1)
        self.assertAlmostEqual(s["by_type"]["perp_options"]["win_rate"], 1.0)
        self.assertAlmostEqual(s["by_type"]["options_options"]["max_loss"], -2)

        bad = delete_plan(conn, active)
        self.assertFalse(bad["ok"])
        ok = delete_plan(conn, oo)
        self.assertTrue(ok["ok"])
        s2 = stats_summary(conn)
        self.assertEqual(s2["closed_count"], 1)

    def test_contract_summary(self):
        s = legs_contract_summary(
            [
                {"leg_role": "perp", "symbol": "ETH/USDT:USDT"},
                {"leg_role": "option_hedge", "inst_id": "ETH-USD_UM-260715-1790-P"},
            ]
        )
        self.assertIn("永续 ETH/USDT:USDT", s)
        self.assertIn("ETH-USD_UM-260715-1790-P", s)

    def test_active_options_targets_are_read_only_plan_targets(self):
        conn = _mem()
        pid = insert_plan(
            conn,
            {
                "plan_type": "options_options",
                "status": "active",
                "underlying": "ETH",
                "target_price_up": 1950,
                "target_price_down": 1800,
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_a",
                "inst_id": "ETH-USD_UM-260719-1890-C",
                "opt_type": "C",
                "status": "open",
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_b",
                "inst_id": "ETH-USD_UM-260719-1850-P",
                "opt_type": "P",
                "status": "open",
            },
        )

        targets = active_options_targets_by_inst(conn)
        self.assertEqual(targets["ETH-USD_UM-260719-1890-C"]["target_index"], 1950)
        self.assertEqual(targets["ETH-USD_UM-260719-1850-P"]["target_index"], 1800)
        self.assertEqual(targets["ETH-USD_UM-260719-1890-C"]["managed_by"], "hedge_plan")

    def test_active_options_targets_profit_rr(self):
        conn = _mem()
        pid = insert_plan(
            conn,
            {
                "plan_type": "options_options",
                "status": "active",
                "underlying": "ETH",
                "profit_rr": 2,
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_a",
                "inst_id": "ETH-USD_UM-260719-1890-C",
                "opt_type": "C",
                "status": "open",
            },
        )
        targets = active_options_targets_by_inst(conn)
        self.assertEqual(targets["ETH-USD_UM-260719-1890-C"]["profit_rr"], 2)
        self.assertEqual(targets["ETH-USD_UM-260719-1890-C"]["exit_mode"], "profit_rr")
        self.assertIsNone(targets["ETH-USD_UM-260719-1890-C"]["target_index"])


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