feat(selection): M6.1 Universe 选股范围执行器(规则化 + symbols 白名单 + 历史日语义)

- 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 通过
This commit is contained in:
Simon
2026-09-09 00:13:28 +08:00
parent f3586adb25
commit 25a1d9531a
4 changed files with 123 additions and 22 deletions
+1 -20
View File
@@ -15,41 +15,22 @@ from datetime import date, timedelta
import pandas as pd
from app.domain.entities.market import Stock
from app.domain.entities.research import (
BacktestResult,
FactorTestReport,
ResearchSpec,
UniverseSpec,
)
from app.domain.repositories.market import (
DailyBarRepository,
StockRepository,
)
from app.quant.engine import QuantEngine
from app.quant.universe import filter_stocks # noqa: F401 —— 选股/回测共用范围过滤
# 流式路径每攒多少行落一个 DataFrame 分片(控制 concat 峰值)
_FRAME_CHUNK_ROWS = 50_000
def filter_stocks(stocks: list[Stock], universe: UniverseSpec, as_of: date) -> list[Stock]:
"""按股票池口径过滤(名称含 ST 判定 —— 名称快照为当日口径,属历史可追溯数据)。"""
out: list[Stock] = []
for s in stocks:
if s.delist_date is not None and s.delist_date < as_of:
continue
if universe.exclude_st and s.name and "ST" in s.name.upper():
continue
if (
universe.min_listing_days
and s.list_date
and (as_of - s.list_date).days < universe.min_listing_days
):
continue
out.append(s)
return out
def bars_to_daily_df(bars) -> pd.DataFrame:
"""DailyBar 列表 → 引擎长表 DataFrame(symbol/trade_date/ohlc/volume/amount)。
+42
View File
@@ -0,0 +1,42 @@
"""Universe:选股/回测的股票范围执行器(ARCHITECTURE_v2 §14/§20 Universe 输入)。
把 ResearchService.filter_stocks 的语义规则化并集中于此:
- 当前日与历史日(as_of)都必须正确:退市股(delist < as_of)、上市时间(list_date)
- exclude_st 按**当前名称快照**含 ST 判定(历史可追溯数据;历史改名无法回溯,属近似,
见结果 unimplemented 说明)
- exclude_suspended 依赖停牌数据表(尚未建模),此处不做剔除,由上层显式标注
- symbols 白名单:非空时仅这些 symbol 参与(自选池 / 测试用)
"""
from __future__ import annotations
from collections.abc import Sequence
from datetime import date
from app.domain.entities.market import Stock
from app.domain.entities.research import UniverseSpec
def filter_stocks(
stocks: Sequence[Stock],
universe: UniverseSpec,
as_of: date,
) -> list[Stock]:
"""按股票池口径过滤,返回 as_of 时点应纳入的股票列表。"""
symbols = set(universe.symbols) if universe.symbols else None
out: list[Stock] = []
for s in stocks:
if symbols is not None and s.symbol not in symbols:
continue
if s.delist_date is not None and s.delist_date < as_of:
continue
if universe.exclude_st and s.name and "ST" in s.name.upper():
continue
if (
universe.min_listing_days
and s.list_date
and (as_of - s.list_date).days < universe.min_listing_days
):
continue
out.append(s)
return out