dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from lib.trade.force_close_lib import (
|
|
apply_force_close_display_result,
|
|
build_force_close_state,
|
|
coerce_force_close_result,
|
|
compute_next_force_close_at_ms,
|
|
force_close_label,
|
|
format_force_close_countdown,
|
|
infer_force_close_result,
|
|
is_close_at_force_close_window,
|
|
is_force_close_active_hour,
|
|
is_force_close_executing,
|
|
)
|
|
|
|
TZ = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def _ms(y, m, d, hh, mm=0):
|
|
return int(datetime(y, m, d, hh, mm, tzinfo=TZ).timestamp() * 1000)
|
|
|
|
|
|
def test_force_close_label():
|
|
assert force_close_label(0) == "强制清仓 00:00"
|
|
assert force_close_label(8) == "强制清仓 08:00"
|
|
|
|
|
|
def test_next_force_close_at_midnight():
|
|
# 2026-07-05 23:30 -> next 2026-07-06 00:00
|
|
now = _ms(2026, 7, 5, 23, 30)
|
|
assert compute_next_force_close_at_ms(bj_hour=0, now_ms=now, tz_name="Asia/Shanghai") == _ms(
|
|
2026, 7, 6, 0, 0
|
|
)
|
|
|
|
|
|
def test_next_force_close_same_day_before_hour():
|
|
now = _ms(2026, 7, 6, 15, 0)
|
|
assert compute_next_force_close_at_ms(bj_hour=0, now_ms=now, tz_name="Asia/Shanghai") == _ms(
|
|
2026, 7, 7, 0, 0
|
|
)
|
|
|
|
|
|
def test_next_force_close_after_trigger_same_day():
|
|
now = _ms(2026, 7, 7, 0, 18)
|
|
assert compute_next_force_close_at_ms(bj_hour=0, now_ms=now, tz_name="Asia/Shanghai") == _ms(
|
|
2026, 7, 8, 0, 0
|
|
)
|
|
|
|
|
|
def test_executing_window_and_countdown():
|
|
now = _ms(2026, 7, 6, 0, 14)
|
|
assert is_force_close_executing(0, now_ms=now, tz_name="Asia/Shanghai")
|
|
assert is_force_close_active_hour(0, now_ms=now, tz_name="Asia/Shanghai")
|
|
state = build_force_close_state(
|
|
True, 0, now_ms=now, tz_name="Asia/Shanghai", has_active_positions=True
|
|
)
|
|
assert state["enabled"] is True
|
|
assert state["active"] is True
|
|
assert state["countdown"] == "执行中"
|
|
assert state["next_at_ms"] == _ms(2026, 7, 7, 0, 0)
|
|
|
|
|
|
def test_not_executing_after_grace_without_positions():
|
|
now = _ms(2026, 7, 7, 0, 18)
|
|
assert not is_force_close_executing(0, now_ms=now, tz_name="Asia/Shanghai")
|
|
state = build_force_close_state(
|
|
True, 0, now_ms=now, tz_name="Asia/Shanghai", has_active_positions=False
|
|
)
|
|
assert state["active"] is False
|
|
assert state["countdown"] != "执行中"
|
|
assert state["next_at_ms"] == _ms(2026, 7, 8, 0, 0)
|
|
|
|
|
|
def test_disabled_state():
|
|
state = build_force_close_state(False, 0)
|
|
assert state["enabled"] is False
|
|
assert state["next_at_ms"] is None
|
|
|
|
|
|
def test_format_countdown():
|
|
assert format_force_close_countdown(3661) == "01:01:01"
|
|
assert format_force_close_countdown(0, active=True) == "执行中"
|
|
|
|
|
|
def test_infer_force_close_from_closed_at():
|
|
assert is_close_at_force_close_window("2026-07-07 00:00", 0)
|
|
assert infer_force_close_result("2026-07-07 00:00", enabled=True, bj_hour=0) == "强制清仓"
|
|
assert infer_force_close_result("2026-07-07 00:20", enabled=True, bj_hour=0) is None
|
|
|
|
|
|
def test_coerce_and_display_external_close_at_midnight():
|
|
res, note = coerce_force_close_result(
|
|
"外部平仓",
|
|
"2026-07-07 00:00",
|
|
enabled=True,
|
|
bj_hour=0,
|
|
)
|
|
assert res == "强制清仓"
|
|
assert "00:00" in note
|
|
assert (
|
|
apply_force_close_display_result(
|
|
"手动平仓",
|
|
"2026-07-07 00:00",
|
|
enabled=True,
|
|
bj_hour=0,
|
|
)
|
|
== "强制清仓"
|
|
)
|