crypto_monitor_user/tests/test_hedge_plan_notify_settle.py

"""对冲计划微信文案与到期结算."""
import sqlite3
import time
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock

from lib.exchange.okx_options_lib import expiry_ms_from_inst_id
from lib.hedge_plan.hedge_plan_db import get_plan, init_hedge_plan_tables, insert_leg, insert_plan
from lib.hedge_plan.hedge_plan_monitor_lib import _tick_oo_expiry, _settle_orphaned_after_tp
from lib.hedge_plan.hedge_plan_notify_lib import (
    build_hedge_end_message,
    build_hedge_start_message,
    notify_plan_end,
    notify_plan_start,
)
from lib.hedge_plan.hedge_plan_settle_lib import leg_is_expired, settle_option_leg_at_spot


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


class TestHedgeNotify(unittest.TestCase):
    def test_start_end_copy(self):
        plan = {
            "id": 7,
            "plan_type": "perp_options",
            "underlying": "ETH",
            "direction": "long",
            "entry_mark": 1800,
            "tp": 1900,
            "sl": 1700,
            "perp_size": 2,
            "leverage": 10,
            "premium_total": 1.5,
            "close_reason": "perp_tp",
            "realized_pnl_total": 12.3,
            "realized_pnl_perp": 15,
            "realized_pnl_options": -1.5,
            "opened_at": "2026-07-01 10:00:00",
            "closed_at": "2026-07-01 12:00:00",
        }
        s = build_hedge_start_message(plan)
        self.assertIn("启动 #7", s)
        self.assertIn("永期", s)
        e = build_hedge_end_message(plan)
        self.assertIn("结束 #7", e)
        self.assertIn("止盈", e)

    def test_idempotent_flags(self):
        conn = _mem_db()
        sent = []
        cfg = {"send_wechat": lambda c: sent.append(c)}
        pid = insert_plan(
            conn,
            {
                "plan_type": "options_options",
                "status": "active",
                "underlying": "ETH",
                "target_price": 2000,
                "premium_total": 2.0,
                "opened_at": "t0",
            },
        )
        plan = get_plan(conn, pid)
        self.assertTrue(notify_plan_start(cfg, conn, plan, []))
        plan = get_plan(conn, pid)
        self.assertEqual(int(plan["wechat_start_sent"]), 1)
        self.assertFalse(notify_plan_start(cfg, conn, plan, []))
        self.assertEqual(len(sent), 1)

        plan["status"] = "closed"
        plan["close_reason"] = "oo_expiry_loss"
        plan["realized_pnl_total"] = -2
        plan["closed_at"] = "t1"
        self.assertTrue(notify_plan_end(cfg, conn, plan))
        plan = get_plan(conn, pid)
        self.assertEqual(int(plan["wechat_end_sent"]), 1)
        self.assertFalse(notify_plan_end(cfg, conn, plan))
        self.assertEqual(len(sent), 2)


class TestHedgeSettle(unittest.TestCase):
    def test_put_expiry_otm(self):
        pnl = settle_option_leg_at_spot(
            {"opt_type": "P", "strike": 1700, "size": 2, "premium": 1.2, "ct_mult": 0.01},
            spot=1800,
        )
        self.assertAlmostEqual(pnl, -1.2)

    def test_call_expiry_itm(self):
        # intrinsic (1900-1800)*2*0.01 - 0.5 = 2 - 0.5
        pnl = settle_option_leg_at_spot(
            {"opt_type": "C", "strike": 1800, "size": 2, "premium": 0.5, "ct_mult": 0.01},
            spot=1900,
        )
        self.assertAlmostEqual(pnl, 1.5)

    def test_leg_expired_from_inst(self):
        # past date in inst_id
        past = datetime.now(timezone.utc) - timedelta(days=3)
        yy = past.year % 100
        tag = f"ETH-USD-{yy:02d}{past.month:02d}{past.day:02d}-1800-P"
        self.assertTrue(leg_is_expired({"inst_id": tag}))
        future = datetime.now(timezone.utc) + timedelta(days=10)
        tag2 = f"ETH-USD-{future.year % 100:02d}{future.month:02d}{future.day:02d}-1800-P"
        self.assertFalse(leg_is_expired({"inst_id": tag2}))
        self.assertIsNotNone(expiry_ms_from_inst_id(tag))


class TestHedgeMonitorExpiry(unittest.TestCase):
    def test_oo_expiry_loss_closes_plan(self):
        conn = _mem_db()
        past = datetime.now(timezone.utc) - timedelta(days=1)
        tag = f"ETH-USD-{past.year % 100:02d}{past.month:02d}{past.day:02d}-1800-P"
        pid = insert_plan(
            conn,
            {
                "plan_type": "options_options",
                "status": "active",
                "underlying": "ETH",
                "target_price": 2000,
                "premium_total": 2.0,
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_a",
                "inst_id": tag,
                "opt_type": "P",
                "strike": 1800,
                "size": 1,
                "premium": 1.0,
                "status": "open",
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_b",
                "inst_id": tag.replace("-P", "-C").replace("1800", "1900"),
                "opt_type": "C",
                "strike": 1900,
                "size": 1,
                "premium": 1.0,
                "status": "open",
            },
        )
        sent = []
        cfg = {
            "send_wechat": lambda c: sent.append(c),
            "fetch_index_price": lambda ex, u: 1850.0,
            "exchange_options": object(),
        }
        plan = get_plan(conn, pid)
        legs = [
            dict(r)
            for r in conn.execute("SELECT * FROM hedge_plan_legs WHERE plan_id=?", (pid,)).fetchall()
        ]
        r = _tick_oo_expiry(cfg, conn, plan, legs)
        self.assertIsNotNone(r)
        self.assertEqual(r["close_reason"], "oo_expiry_loss")
        plan2 = get_plan(conn, pid)
        self.assertEqual(plan2["status"], "closed")
        self.assertLessEqual(float(plan2["realized_pnl_total"]), 0)
        self.assertTrue(any("结束" in x for x in sent))

    def test_orphaned_option_does_not_rewrite_plan_total(self):
        conn = _mem_db()
        past = datetime.now(timezone.utc) - timedelta(days=1)
        tag = f"ETH-USD-{past.year % 100:02d}{past.month:02d}{past.day:02d}-1700-P"
        pid = insert_plan(
            conn,
            {
                "plan_type": "perp_options",
                "status": "closed",
                "underlying": "ETH",
                "direction": "long",
                "close_reason": "perp_tp",
                "realized_pnl_total": 10.0,
                "realized_pnl_options": -1.0,
                "stats_bucket": "tp",
            },
        )
        insert_leg(
            conn,
            {
                "plan_id": pid,
                "leg_role": "option_hedge",
                "inst_id": tag,
                "opt_type": "P",
                "strike": 1700,
                "size": 1,
                "premium": 1.0,
                "status": "hold_to_expiry",
                "close_reason": "orphaned_after_tp",
            },
        )
        cfg = {
            "fetch_index_price": lambda ex, u: 1800.0,
            "exchange_options": object(),
        }
        acted = _settle_orphaned_after_tp(cfg, conn)
        self.assertEqual(len(acted), 1)
        plan = get_plan(conn, pid)
        self.assertAlmostEqual(float(plan["realized_pnl_total"]), 10.0)
        leg = conn.execute("SELECT * FROM hedge_plan_legs WHERE plan_id=?", (pid,)).fetchone()
        self.assertEqual(leg["status"], "closed")
        self.assertEqual(leg["close_reason"], "expiry")


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