Simon
9cc4bfccac
- index_weight 表（migration f5e0d1c2b3a4，MySQL 已应用；index_code+date+symbol 唯一） + IndexWeight 实体 + IndexConstituentRepository（members_at：取 <=as_of 最近一期快照， Survivorship-free / 无未来成分；latest_date） - UniverseSpec.index_code + universe.filter_stocks members 交集 + resolve_members； Research/Selection/Signal/Replay 服务注入 index repo（历史成分过滤，选股/回测共用） - tests/test_index_universe.py：快照历史成分（成分变更不入早期结果）、幂等、 空快照期空集、index_code 过滤下 as_of 一致性；全量 pytest 通过
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
"""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 resolve_members(index_repo, universe: UniverseSpec, as_of: date) -> set[str] | None:
|
||
"""若 universe 指定指数成分 → 取 as_of 当日历史成分；否则 None（不过滤）。"""
|
||
if index_repo is None or not universe.index_code:
|
||
return None
|
||
return index_repo.members_at(universe.index_code, as_of)
|
||
|
||
|
||
def filter_stocks(
|
||
stocks: Sequence[Stock],
|
||
universe: UniverseSpec,
|
||
as_of: date,
|
||
members: set[str] | None = None,
|
||
) -> list[Stock]:
|
||
"""按股票池口径过滤，返回 as_of 时点应纳入的股票列表。
|
||
|
||
members：指数历史成分集合（resolve_members 结果）；提供时取交集。
|
||
"""
|
||
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 members is not None and s.symbol not in members:
|
||
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
|