crypto_monitor/tests/test_options_close_gate_lib.py

"""期权平仓门控:USDT 口径(权利金×倍数 / 净盈亏阈值)且持续持有."""
from __future__ import annotations

import os
import unittest

from lib.options.options_close_gate_lib import (
    clear_close_gate,
    is_close_gate_passed,
    mark_close_gate_passed,
    update_close_gate,
)


class OptionsCloseGateTests(unittest.TestCase):
    def setUp(self):
        clear_close_gate()
        self._env_backup = {
            k: os.environ.get(k)
            for k in (
                "OKX_OPTIONS_CLOSE_GATE_MODE",
                "OKX_OPTIONS_CLOSE_RECYCLE_MULT",
                "OKX_OPTIONS_CLOSE_RECYCLE_MULT_COIN",
                "OKX_OPTIONS_CLOSE_RECYCLE_MULT_USDC",
                "OKX_OPTIONS_CLOSE_NET_PNL_MIN_U",
            )
        }
        for k in self._env_backup:
            os.environ.pop(k, None)

    def tearDown(self):
        clear_close_gate()
        for k, v in self._env_backup.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v

    def test_usdc_below_2x_not_ready(self):
        g = update_close_gate(
            "ETH-X",
            recycle_usdc=15.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1000.0,
        )
        self.assertFalse(g["recycle_ok"])
        self.assertFalse(g["ready"])
        self.assertIn("U(估)", g["msg"])

    def test_usdc_meets_2x_needs_hold(self):
        g1 = update_close_gate(
            "ETH-X",
            recycle_usdc=20.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1000.0,
        )
        self.assertTrue(g1["recycle_ok"])
        self.assertFalse(g1["ready"])
        self.assertAlmostEqual(g1["remain_seconds"], 120.0)

        g2 = update_close_gate(
            "ETH-X",
            recycle_usdc=21.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1120.0,
        )
        self.assertTrue(g2["ready"])

    def test_coin_premium_gate_uses_usdt_and_default_105(self):
        # 0.0384 ETH * 2500 = 96U; ×1.05 = 100.8U
        g = update_close_gate(
            "ETH-P",
            recycle_usdc=0.036,
            premium_paid=0.0384,
            premium_ccy="ETH",
            index_px=2500.0,
            now=1000.0,
        )
        self.assertFalse(g["recycle_ok"])
        self.assertAlmostEqual(g["premium_usdt"], 96.0)
        self.assertAlmostEqual(g["need_recycle_usdt"], 100.8)
        self.assertIn("U(估)", g["msg"])

        g_ok = update_close_gate(
            "ETH-P",
            recycle_usdc=0.041,
            premium_paid=0.0384,
            premium_ccy="ETH",
            index_px=2500.0,
            now=1000.0,
        )
        self.assertTrue(g_ok["recycle_ok"])
        self.assertAlmostEqual(g_ok["recycle_usdt"], 102.5)

    def test_net_pnl_gate_mode(self):
        os.environ["OKX_OPTIONS_CLOSE_GATE_MODE"] = "net_pnl"
        os.environ["OKX_OPTIONS_CLOSE_NET_PNL_MIN_U"] = "1"
        g = update_close_gate(
            "ETH-N",
            recycle_usdc=0.039,
            premium_paid=0.0384,
            premium_ccy="ETH",
            index_px=2500.0,
            now=1000.0,
        )
        self.assertTrue(g["recycle_ok"])
        self.assertAlmostEqual(g["net_pnl_usdt"], 1.5)

        g2 = update_close_gate(
            "ETH-N2",
            recycle_usdc=0.0385,
            premium_paid=0.0384,
            premium_ccy="ETH",
            index_px=2500.0,
            now=1000.0,
        )
        self.assertFalse(g2["recycle_ok"])

    def test_break_resets_timer(self):
        update_close_gate(
            "ETH-X",
            recycle_usdc=20.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1000.0,
        )
        update_close_gate(
            "ETH-X",
            recycle_usdc=21.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1100.0,
        )
        g_break = update_close_gate(
            "ETH-X",
            recycle_usdc=12.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1110.0,
        )
        self.assertFalse(g_break["recycle_ok"])
        g_again = update_close_gate(
            "ETH-X",
            recycle_usdc=22.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1111.0,
        )
        self.assertTrue(g_again["recycle_ok"])
        self.assertFalse(g_again["ready"])
        self.assertAlmostEqual(g_again["held_seconds"], 0.0)

    def test_passed_latches_after_ready(self):
        update_close_gate(
            "ETH-Y",
            recycle_usdc=20.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1000.0,
        )
        g_ready = update_close_gate(
            "ETH-Y",
            recycle_usdc=21.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1120.0,
        )
        self.assertTrue(g_ready["ready"])
        self.assertTrue(g_ready["passed"])
        self.assertTrue(is_close_gate_passed("ETH-Y"))
        g_drop = update_close_gate(
            "ETH-Y",
            recycle_usdc=5.0,
            premium_paid=10.0,
            premium_ccy="USDC",
            now=1130.0,
        )
        self.assertFalse(g_drop["recycle_ok"])
        self.assertTrue(g_drop["passed"])
        self.assertFalse(g_drop["auto_close_blocked"])

    def test_mark_passed_manual(self):
        mark_close_gate_passed("ETH-Z")
        self.assertTrue(is_close_gate_passed("ETH-Z"))


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