yuxuanhui
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from zhixing_server.modules.market_data.domain.fingerprint import (
|
|
SnapshotChange,
|
|
compare_snapshots,
|
|
snapshot_fingerprint,
|
|
)
|
|
from zhixing_server.modules.market_data.domain.models import Bar, Stock, SyncWindow
|
|
from zhixing_server.modules.market_data.domain.rules import filter_current_hs_a_stocks
|
|
|
|
|
|
def make_bar(trade_date: date, close: str = "10") -> Bar:
|
|
return Bar(
|
|
ts_code="000001.SZ",
|
|
trade_date=trade_date,
|
|
open=Decimal("9"),
|
|
high=Decimal(close),
|
|
low=Decimal("8"),
|
|
close=Decimal(close),
|
|
pre_close=Decimal("9"),
|
|
change=Decimal("1"),
|
|
pct_chg=Decimal("11.11"),
|
|
vol=Decimal("100"),
|
|
amount=Decimal("1000"),
|
|
)
|
|
|
|
|
|
def test_window_uses_inclusive_calendar_boundary() -> None:
|
|
window = SyncWindow.from_target(date(2024, 2, 29))
|
|
|
|
assert window.start == date(2018, 2, 28)
|
|
assert window.end == date(2024, 2, 29)
|
|
assert window.contains(date(2018, 2, 28))
|
|
|
|
|
|
def test_universe_keeps_current_non_st_hs_a_stocks() -> None:
|
|
stocks = (
|
|
Stock("000001.SZ", "平安银行", exchange="SZSE", list_status="L"),
|
|
Stock("600000.SH", "浦发银行", exchange="SSE", list_status="L"),
|
|
Stock("300001.SZ", "特锐德", exchange="SZSE", list_status="L"),
|
|
Stock("600001.SH", "*ST风险", exchange="SSE", list_status="L"),
|
|
Stock("830001.BJ", "北交所", exchange="BSE", list_status="L"),
|
|
)
|
|
|
|
assert [stock.ts_code for stock in filter_current_hs_a_stocks(stocks)] == [
|
|
"000001.SZ",
|
|
"300001.SZ",
|
|
"600000.SH",
|
|
]
|
|
|
|
|
|
def test_fingerprint_is_order_independent_and_excludes_runtime_metadata() -> None:
|
|
rows = (make_bar(date(2024, 1, 2)), make_bar(date(2024, 1, 3)))
|
|
|
|
assert snapshot_fingerprint(rows) == snapshot_fingerprint(tuple(reversed(rows)))
|
|
assert compare_snapshots(rows, rows).change is SnapshotChange.UNCHANGED
|
|
assert (
|
|
compare_snapshots(rows, (*rows, make_bar(date(2024, 1, 4)))).change
|
|
is SnapshotChange.NEW_DATES
|
|
)
|
|
|
|
|
|
def test_fingerprint_detects_repairs_missing_rows_and_earlier_rows() -> None:
|
|
old = (make_bar(date(2024, 1, 2)), make_bar(date(2024, 1, 3)))
|
|
|
|
changed = (make_bar(date(2024, 1, 2), close="11"), make_bar(date(2024, 1, 3)))
|
|
missing = (make_bar(date(2024, 1, 2)),)
|
|
earlier = (make_bar(date(2024, 1, 1)), *old)
|
|
|
|
assert compare_snapshots(old, changed).change is SnapshotChange.CHANGED
|
|
assert compare_snapshots(old, missing).change is SnapshotChange.CHANGED
|
|
assert compare_snapshots(old, earlier).change is SnapshotChange.CHANGED
|