dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

目标平仓门控改为 USDT 口径：env 可配权利金倍数与净盈亏阈值，币本位默认×1.05

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-25 09:01:06 +08:00
parent 5f9901db0f
commit fe346571b1
8 changed files with 362 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_options_close_gate_lib.py
+142 -15
View File
@@ -1,6 +1,7 @@
"""期权平仓门控:可回收≥2×权利金且持续持有."""
"""期权平仓门控:USDT 口径(权利金×倍数 / 净盈亏阈值)且持续持有."""
from __future__ import annotations
import os
import unittest
from lib.options.options_close_gate_lib import (
@@ -14,43 +15,169 @@ from lib.options.options_close_gate_lib import (
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_below_2x_not_ready(self):
g = update_close_gate("ETH-X", recycle_usdc=15.0, premium_paid=10.0, now=1000.0)
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_meets_2x_needs_hold(self):
g1 = update_close_gate("ETH-X", recycle_usdc=20.0, premium_paid=10.0, now=1000.0)
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, now=1120.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"])
self.assertGreaterEqual(g2["held_seconds"], 120.0)
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, now=1000.0)
update_close_gate("ETH-X", recycle_usdc=21.0, premium_paid=10.0, now=1100.0)
g_break = update_close_gate("ETH-X", recycle_usdc=12.0, premium_paid=10.0, now=1110.0)
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, now=1111.0)
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, now=1000.0)
g_ready = update_close_gate("ETH-Y", recycle_usdc=21.0, premium_paid=10.0, now=1120.0)
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"))
# 后续回收跌破 2×:计时重置,但 passed 仍保留供续批只验流动性
g_drop = update_close_gate("ETH-Y", recycle_usdc=5.0, premium_paid=10.0, now=1130.0)
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"])