Simon
25a1d9531a
- quant/universe.py：filter_stocks 从 quant/service 迁出并集中（ST/上市天数/退市过滤）， as_of 当前/历史日语义由 delist/list_date 保证；exclude_suspended 依赖停牌表未建模， 由上层显式标注（选股结果 unimplemented） - research.UniverseSpec 增加 symbols 白名单（非空时仅白名单内参与，再叠加其余过滤） - quant/service re-export filter_stocks（外部引用不变）；SelectionService 已共用 - tests/test_universe.py：6 例覆盖当前/历史日、ST、上市天数、退市、白名单；全量 pytest 通过
71 lines
3.0 KiB
Python
71 lines
3.0 KiB
Python
"""M6.1 UniverseFilter 测试：as_of 当前/历史日语义、ST、上市天数、退市、symbols 白名单。
|
|
|
|
filter_stocks 从 quant.universe 引入（原 quant.service 语义，规则化集中）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
from app.domain.entities.market import Stock
|
|
from app.domain.entities.research import UniverseSpec
|
|
from app.quant.universe import filter_stocks
|
|
|
|
|
|
def _stocks() -> list[Stock]:
|
|
return [
|
|
Stock(symbol="600000.SH", name="正常股份", list_date=date(2000, 1, 1)),
|
|
Stock(symbol="600001.SH", name="ST 风险股份", list_date=date(2000, 1, 1)),
|
|
Stock(symbol="600002.SH", name="次新股", list_date=date(2024, 10, 1)),
|
|
Stock(symbol="600003.SH", name="已退市股", list_date=date(1995, 1, 1),
|
|
delist_date=date(2023, 6, 30)),
|
|
Stock(symbol="600004.SH", name="老股", list_date=date(1999, 1, 1)),
|
|
]
|
|
|
|
|
|
def _sym(rows: list[Stock]) -> set[str]:
|
|
return {s.symbol for s in rows}
|
|
|
|
|
|
class TestUniverseFilter:
|
|
def test_current_day(self) -> None:
|
|
rows = filter_stocks(_stocks(), UniverseSpec(), as_of=date(2025, 1, 1))
|
|
# ST、退市被剔除；次新股（上市<250 自然日）被剔除
|
|
assert _sym(rows) == {"600000.SH", "600004.SH"}
|
|
|
|
def test_historical_as_of_keeps_not_yet_delisted(self) -> None:
|
|
rows = filter_stocks(_stocks(), UniverseSpec(), as_of=date(2023, 1, 1))
|
|
# 2023-01 时 600003 尚未退市（2023-06 退市）→ 应纳入
|
|
assert "600003.SH" in _sym(rows)
|
|
assert "600002.SH" not in _sym(rows) # 2024-10 才上市，2023-01 尚不存在（上市天数不足被滤）
|
|
|
|
def test_delisted_before_as_of_excluded(self) -> None:
|
|
rows = filter_stocks(_stocks(), UniverseSpec(exclude_st=False), as_of=date(2024, 1, 1))
|
|
assert "600003.SH" not in _sym(rows) # 2023-06 已退市
|
|
|
|
def test_exclude_st_flag(self) -> None:
|
|
rows = filter_stocks(_stocks(), UniverseSpec(exclude_st=False), as_of=date(2025, 1, 1))
|
|
assert "600001.SH" in _sym(rows)
|
|
rows2 = filter_stocks(_stocks(), UniverseSpec(exclude_st=True), as_of=date(2025, 1, 1))
|
|
assert "600001.SH" not in _sym(rows2)
|
|
|
|
def test_min_listing_days_zero_disables(self) -> None:
|
|
rows = filter_stocks(
|
|
_stocks(), UniverseSpec(min_listing_days=0, exclude_st=True), as_of=date(2025, 1, 1)
|
|
)
|
|
assert "600002.SH" in _sym(rows)
|
|
|
|
def test_symbols_whitelist(self) -> None:
|
|
rows = filter_stocks(
|
|
_stocks(),
|
|
UniverseSpec(symbols=["600000.SH", "600003.SH"]),
|
|
as_of=date(2025, 1, 1),
|
|
)
|
|
# 白名单内的 ST/退市过滤仍然生效：600003 已退市被滤，仅剩 600000
|
|
assert _sym(rows) == {"600000.SH"}
|
|
|
|
def test_empty_whitelist_means_all(self) -> None:
|
|
assert UniverseSpec().symbols == []
|
|
rows = filter_stocks(_stocks(), UniverseSpec(symbols=[]), as_of=date(2025, 1, 1))
|
|
assert "600004.SH" in _sym(rows)
|