dekun
95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
"""hold_timing unit tests."""
|
|
|
|
from backend.app.api.hold_timing import hold_timing
|
|
|
|
|
|
def test_hold_target_perp_only_uses_perp_close():
|
|
g = {
|
|
"open_at_ms": 1_000,
|
|
"close_at_ms": 9_000, # later residual settle would have overwritten
|
|
"status": "closed",
|
|
"close_reason": "target_perp_only",
|
|
}
|
|
fills = [
|
|
{"leg": "option", "action": "open", "ts_ms": 1_000},
|
|
{"leg": "perp", "action": "open", "ts_ms": 1_100},
|
|
{"leg": "perp", "action": "close", "ts_ms": 5_000},
|
|
{"leg": "option", "action": "close", "ts_ms": 9_000},
|
|
]
|
|
h = hold_timing(g, fills)
|
|
assert h["hold_open_at_ms"] == 1_000
|
|
assert h["hold_close_at_ms"] == 5_000
|
|
assert h["hold_ms"] == 4_000
|
|
assert h["hold_basis"] == "perp"
|
|
|
|
|
|
def test_hold_option_residual_uses_perp():
|
|
g = {
|
|
"open_at_ms": 100,
|
|
"close_at_ms": 500,
|
|
"status": "option_residual",
|
|
"close_reason": "target_perp_only",
|
|
}
|
|
fills = [
|
|
{"leg": "perp", "action": "close", "ts_ms": 500},
|
|
]
|
|
h = hold_timing(g, fills)
|
|
assert h["hold_close_at_ms"] == 500
|
|
assert h["hold_ms"] == 400
|
|
|
|
|
|
def test_hold_dual_leg_uses_group_close():
|
|
g = {
|
|
"open_at_ms": 100,
|
|
"close_at_ms": 800,
|
|
"status": "closed",
|
|
"close_reason": "fixed_usdt",
|
|
}
|
|
fills = [
|
|
{"leg": "option", "action": "close", "ts_ms": 790},
|
|
{"leg": "perp", "action": "close", "ts_ms": 800},
|
|
]
|
|
h = hold_timing(g, fills)
|
|
assert h["hold_close_at_ms"] == 800
|
|
assert h["hold_ms"] == 700
|
|
assert h["hold_basis"] == "group"
|
|
|
|
|
|
def test_hold_open_no_close():
|
|
g = {"open_at_ms": 100, "close_at_ms": None, "status": "open", "close_reason": None}
|
|
h = hold_timing(g, [])
|
|
assert h["hold_close_at_ms"] is None
|
|
assert h["hold_ms"] is None
|
|
assert h["hold_basis"] == "open"
|
|
|
|
|
|
def test_hold_sqlite_row_like_without_get():
|
|
"""sqlite3.Row 无 .get，需能转 dict。"""
|
|
|
|
class Row:
|
|
def __init__(self, d):
|
|
self._d = d
|
|
|
|
def keys(self):
|
|
return self._d.keys()
|
|
|
|
def __getitem__(self, k):
|
|
return self._d[k]
|
|
|
|
def __iter__(self):
|
|
return iter(self._d)
|
|
|
|
g = Row(
|
|
{
|
|
"open_at_ms": 100,
|
|
"close_at_ms": None,
|
|
"status": "option_residual",
|
|
"close_reason": "target_perp_only",
|
|
}
|
|
)
|
|
fills = [Row({"leg": "perp", "action": "close", "ts_ms": 400})]
|
|
h = hold_timing(g, fills)
|
|
assert h["hold_close_at_ms"] == 400
|
|
assert h["hold_ms"] == 300
|
|
|