dekun
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""空仓选约：跳过当日已用到期与残留待结算到期。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.strategy.clock import (
|
|
expiry_blocked_by_one_per_day,
|
|
pending_residual_expiry_ymds,
|
|
used_expiry_ymds_for_day,
|
|
)
|
|
|
|
|
|
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:
|
|
like = params[0] if params else ""
|
|
prefix = like.replace("%", "")
|
|
return [
|
|
r
|
|
for r in self._groups
|
|
if str(r.get("group_id", "")).startswith(prefix)
|
|
]
|
|
if "from residual_options" in s:
|
|
return [r for r in self._residuals if r.get("status") == "pending"]
|
|
return []
|
|
|
|
|
|
def test_used_expiry_ymds_for_day() -> None:
|
|
db = _FakeDB(
|
|
groups=[
|
|
{"group_id": "G-20260802-01", "expiry_ymd": "260803"},
|
|
{"group_id": "G-20260801-01", "expiry_ymd": "260802"},
|
|
]
|
|
)
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
now = datetime(2026, 8, 2, 12, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
|
|
assert used_expiry_ymds_for_day(db, now) == {"260803"}
|
|
assert expiry_blocked_by_one_per_day("260803", {"260803"}, enabled=True)
|
|
assert not expiry_blocked_by_one_per_day("260804", {"260803"}, 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"}
|