Files
crypto_monitor/tests/test_options_target_lib.py
T
dekun d6823a2903 Add options index target monitors that auto limit-close on hit.
Position and order forms can arm a target; right-side and hub panels show active monitors; expiry remains the stop with no separate SL.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-15 15:27:46 +08:00

55 lines
1.7 KiB
Python

"""期权目标位委托单元测试."""
from __future__ import annotations
import sqlite3
import unittest
from lib.options.options_target_lib import (
ensure_target_tables,
list_active_targets,
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_upsert_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",
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"}
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)
if __name__ == "__main__":
unittest.main()