dekun
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""空仓选约：跳过历史上已用到期与残留待结算到期。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.strategy.clock import (
|
|
expiry_blocked_by_one_per_day,
|
|
pending_residual_expiry_ymds,
|
|
used_expiry_ymds,
|
|
)
|
|
|
|
|
|
class _FakeDB:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
groups: list[dict] | None = None,
|
|
residuals: list[dict] | None = None,
|
|
) -> None:
|
|
self._groups = groups or []
|
|
self._residuals = residuals or []
|
|
|
|
def fetchall(self, sql: str, params: tuple = ()) -> list[dict]:
|
|
s = " ".join(sql.split()).lower()
|
|
if "from groups" in s:
|
|
return [
|
|
r
|
|
for r in self._groups
|
|
if str(r.get("expiry_ymd") or "").strip()
|
|
]
|
|
if "from residual_options" in s:
|
|
return [r for r in self._residuals if r.get("status") == "pending"]
|
|
return []
|
|
|
|
|
|
def test_used_expiry_ymds_across_calendar_days() -> None:
|
|
"""8.2 开过 260803 后，8.3 零点仍须拦截同到期。"""
|
|
db = _FakeDB(
|
|
groups=[
|
|
{"group_id": "G-20260802-01", "expiry_ymd": "260803"},
|
|
{"group_id": "G-20260801-01", "expiry_ymd": "260802"},
|
|
]
|
|
)
|
|
now_aug3 = datetime(2026, 8, 3, 0, 0, 2, tzinfo=ZoneInfo("Asia/Shanghai"))
|
|
used = used_expiry_ymds(db, now_aug3)
|
|
assert used == {"260803", "260802"}
|
|
assert expiry_blocked_by_one_per_day("260803", used, enabled=True)
|
|
assert not expiry_blocked_by_one_per_day("260804", used, enabled=True)
|
|
|
|
|
|
def test_pending_residual_expiry_ymds() -> None:
|
|
db = _FakeDB(
|
|
residuals=[
|
|
{"expiry_ymd": "260803", "status": "pending"},
|
|
{"expiry_ymd": "260802", "status": "settled"},
|
|
{"expiry_ymd": "", "status": "pending"},
|
|
]
|
|
)
|
|
assert pending_residual_expiry_ymds(db) == {"260803"}
|