crypto_monitor/tests/test_options_target_lib.py

"""期权目标委托单元测试(盈亏比 + 旧指数兼容)."""
from __future__ import annotations

import sqlite3
import unittest

from lib.options.options_target_lib import (
    ensure_target_tables,
    list_active_targets,
    list_closing_targets,
    profit_rr_hit,
    run_options_target_closes,
    target_hit,
    upsert_target_monitor,
)


class OptionsTargetLibTests(unittest.TestCase):
    def test_target_hit_call_put(self):
        self.assertTrue(target_hit(opt_type="C", index_px=2000, target_index=1950))
        self.assertFalse(target_hit(opt_type="C", index_px=1900, target_index=1950))
        self.assertTrue(target_hit(opt_type="P", index_px=1800, target_index=1850))
        self.assertFalse(target_hit(opt_type="P", index_px=1900, target_index=1850))

    def test_profit_rr_hit(self):
        # premium=10, rr=2 → need pnl≥20 → recycle≥30 → bid*sheets*ct ≥30
        self.assertTrue(
            profit_rr_hit(premium=10, bid=30, sheets=1, ct_mult=1, profit_rr=2)
        )
        self.assertFalse(
            profit_rr_hit(premium=10, bid=29.9, sheets=1, ct_mult=1, profit_rr=2)
        )
        self.assertFalse(
            profit_rr_hit(premium=10, bid=None, sheets=1, ct_mult=1, profit_rr=2)
        )

    def test_upsert_rr_and_trigger_close(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        ensure_target_tables(conn)
        out = upsert_target_monitor(
            conn,
            inst_id="ETH-USD_UM-260717-1900-C",
            profit_rr=2,
            opt_type="C",
            sheets=1,
        )
        self.assertTrue(out["ok"])
        self.assertEqual(out.get("profit_rr"), 2.0)
        self.assertEqual(len(list_active_targets(conn)), 1)

        closed = []

        def close_fn(inst_id: str):
            closed.append(inst_id)
            return {
                "ok": True,
                "submitted_sheets": 1,
                "premium_received": 30.0,
                "close_ord_id": "oid1",
                "fully_closed": True,
                "remaining_sheets": 0,
            }

        # bid=30, ct=1 → pnl=20 ≥ 2*10; premium 来自持仓字段
        n = run_options_target_closes(
            conn,
            [
                {
                    "inst_id": "ETH-USD_UM-260717-1900-C",
                    "idx_px": 1885,
                    "opt_type": "C",
                    "pos": 1,
                    "ct_mult": 1,
                    "premium_paid": 10,
                }
            ],
            close_fn=close_fn,
            bid_fn=lambda _i: 30.0,
        )
        self.assertEqual(n, 1)
        self.assertEqual(closed, ["ETH-USD_UM-260717-1900-C"])
        self.assertEqual(len(list_active_targets(conn)), 0)

    def test_upsert_and_trigger_close_legacy_index(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        ensure_target_tables(conn)
        out = upsert_target_monitor(
            conn,
            inst_id="ETH-USD_UM-260717-1900-C",
            target_index=1880,
            opt_type="C",
            sheets=1,
        )
        self.assertTrue(out["ok"])
        self.assertEqual(len(list_active_targets(conn)), 1)

        closed = []

        def close_fn(inst_id: str):
            closed.append(inst_id)
            return {
                "ok": True,
                "submitted_sheets": 1,
                "premium_received": 1.2,
                "close_ord_id": "oid1",
                "fully_closed": True,
                "remaining_sheets": 0,
            }

        n = run_options_target_closes(
            conn,
            [{"inst_id": "ETH-USD_UM-260717-1900-C", "idx_px": 1885, "opt_type": "C"}],
            close_fn=close_fn,
        )
        self.assertEqual(n, 1)
        self.assertEqual(closed, ["ETH-USD_UM-260717-1900-C"])
        self.assertEqual(len(list_active_targets(conn)), 0)

    def test_partial_fill_notifies_once_then_closing_retry_silent(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        ensure_target_tables(conn)
        upsert_target_monitor(
            conn,
            inst_id="ETH-USD_UM-260715-1870-P",
            target_index=1872,
            opt_type="P",
            sheets=1,
        )
        conn.commit()
        notices: list[str] = []
        calls = {"n": 0}

        def close_fn(inst_id: str):
            calls["n"] += 1
            if calls["n"] == 1:
                return {
                    "ok": True,
                    "submitted_sheets": 1,
                    "premium_received": 0.032,
                    "close_ord_id": "oid-a",
                    "fully_closed": False,
                    "remaining_sheets": 1,
                    "stopped_reason": "order_not_filled",
                }
            return {
                "ok": True,
                "submitted_sheets": 1,
                "premium_received": 0.032,
                "close_ord_id": "oid-b",
                "fully_closed": True,
                "remaining_sheets": 0,
                "already_flat": True,
            }

        pos = [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1867.5, "opt_type": "P"}]
        n1 = run_options_target_closes(
            conn,
            pos,
            close_fn=close_fn,
            send_wechat=notices.append,
            account_label="主账户·期权",
        )
        self.assertEqual(n1, 1)
        self.assertEqual(len(notices), 1)
        self.assertEqual(len(list_active_targets(conn)), 0)
        self.assertEqual(len(list_closing_targets(conn)), 1)

        n2 = run_options_target_closes(
            conn,
            pos,
            close_fn=close_fn,
            send_wechat=notices.append,
            account_label="主账户·期权",
        )
        self.assertEqual(n2, 0)
        self.assertEqual(len(notices), 1)
        self.assertEqual(len(list_closing_targets(conn)), 0)

    def test_commit_before_wechat_survives_later_rollback(self):
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        ensure_target_tables(conn)
        upsert_target_monitor(
            conn,
            inst_id="ETH-USD_UM-260715-1870-P",
            target_index=1872,
            opt_type="P",
        )
        conn.commit()
        notices: list[str] = []

        def close_fn(inst_id: str):
            return {
                "ok": True,
                "submitted_sheets": 1,
                "premium_received": 0.03,
                "close_ord_id": "oid1",
                "fully_closed": True,
                "remaining_sheets": 0,
            }

        run_options_target_closes(
            conn,
            [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1860, "opt_type": "P"}],
            close_fn=close_fn,
            send_wechat=notices.append,
        )
        conn.rollback()
        self.assertEqual(len(notices), 1)
        self.assertEqual(len(list_active_targets(conn)), 0)

        n2 = run_options_target_closes(
            conn,
            [{"inst_id": "ETH-USD_UM-260715-1870-P", "idx_px": 1860, "opt_type": "P"}],
            close_fn=close_fn,
            send_wechat=notices.append,
        )
        self.assertEqual(n2, 0)
        self.assertEqual(len(notices), 1)


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