dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""期权挂单超时撤单单测."""
|
|
from unittest import TestCase
|
|
|
|
from lib.options.options_pending_lib import (
|
|
cancel_stale_close_pending_orders,
|
|
enrich_pending_orders,
|
|
is_close_pending_order,
|
|
order_age_seconds,
|
|
)
|
|
|
|
|
|
class OptionsPendingLibTests(TestCase):
|
|
def test_order_age_and_close_detect(self):
|
|
now = 1_700_000_600_000
|
|
age = order_age_seconds({"c_time": now - 90_000}, now_ms=now)
|
|
self.assertAlmostEqual(age, 90.0, places=3)
|
|
self.assertTrue(is_close_pending_order({"side": "sell"}))
|
|
self.assertTrue(is_close_pending_order({"side": "buy", "reduce_only": True}))
|
|
self.assertFalse(is_close_pending_order({"side": "buy"}))
|
|
|
|
def test_enrich_expire(self):
|
|
now = 1_700_000_600_000
|
|
rows = enrich_pending_orders(
|
|
[
|
|
{"ord_id": "1", "inst_id": "A", "side": "sell", "c_time": now - 700_000},
|
|
{"ord_id": "2", "inst_id": "B", "side": "buy", "c_time": now - 700_000},
|
|
{"ord_id": "3", "inst_id": "C", "side": "sell", "c_time": now - 30_000},
|
|
],
|
|
ttl_seconds=600,
|
|
now_ms=now,
|
|
)
|
|
by_id = {r["ord_id"]: r for r in rows}
|
|
self.assertTrue(by_id["1"]["stale"])
|
|
self.assertTrue(by_id["1"]["auto_cancel_enabled"])
|
|
self.assertFalse(by_id["2"]["auto_cancel_enabled"])
|
|
self.assertFalse(by_id["3"]["stale"])
|
|
self.assertAlmostEqual(by_id["3"]["expire_in_sec"], 570.0, places=0)
|
|
|
|
def test_cancel_stale_only_close(self):
|
|
now = 1_700_000_600_000
|
|
pending = [
|
|
{"ord_id": "s1", "inst_id": "A", "side": "sell", "c_time": now - 700_000},
|
|
{"ord_id": "b1", "inst_id": "B", "side": "buy", "c_time": now - 700_000},
|
|
{"ord_id": "s2", "inst_id": "C", "side": "sell", "c_time": now - 10_000},
|
|
]
|
|
cancelled = []
|
|
|
|
def fetch(_ex=None):
|
|
return pending
|
|
|
|
def cancel(_ex=None, inst_id=None, ord_id=None):
|
|
cancelled.append((inst_id, ord_id))
|
|
return {"ok": True}
|
|
|
|
out = cancel_stale_close_pending_orders(
|
|
fetch_pending=fetch,
|
|
cancel_order=cancel,
|
|
ttl_seconds=60,
|
|
now_ms=now,
|
|
ex=object(),
|
|
)
|
|
self.assertEqual(out["cancelled"], 1)
|
|
self.assertEqual(cancelled, [("A", "s1")])
|