程序员量化交易实战 07：先做一个干净的 A 股股票池

第 6 篇把数据库表结构和迁移检查补上了。现在可以开始处理股票主数据。

这一篇只做一件事：把原始 A 股列表变成一个干净、稳定、可复用的股票池。别小看这一步。股票池如果随便拼，后面策略收益、回测覆盖率和模拟盘观察都会被污染。

股票池不是股票列表

股票列表是数据源给你的原始结果。股票池是系统决定“这一轮研究允许看哪些标的”的边界。

两者差别很大。

原始列表里可能有代码格式不统一、重复行、ST、退市整理、行业缺失、市场后缀缺失。股票池要做的是先把这些问题挡在策略外面。

代码格式先统一

A 股代码在不同系统里经常混着出现：

600519
600519.SH
300750.sz
000001.SZ

策略和数据库不能接受这种混乱。第 7 章新增 app/stock_universe.py，第一步就是把代码统一成 000000.SH/SZ：

def normalize_a_share_symbol(value: str) -> str | None:
    text = str(value or "").strip().upper()
    if not text:
        return None
    if "." in text:
        code, market = text.split(".", 1)
    else:
        code = text
        market = "SH" if code.startswith(("6", "9")) else "SZ"
    if not re.fullmatch(r"\d{6}", code):
        return None
    if market not in {"SH", "SZ"}:
        return None
    return f"{code}.{market}"

这不是完美的交易所识别逻辑，但对当前沪深股票池足够明确。后面如果加入北交所，可以在这里扩展，而不是让代码格式判断散落在策略里。

ST 和退市先过滤掉

实战早期不建议把 ST、退市整理标的混进公共股票池。

原因很简单：它们交易规则、风险暴露和流动性约束都更特殊。如果还没把普通 A 股链路跑稳，就先把高风险边界混进来，调试会变得很吵。

def is_tradeable_a_share_name(name: str) -> bool:
    normalized = str(name or "").strip().upper()
    if not normalized:
        return False
    return "退市" not in normalized and not normalized.startswith(("ST", "*ST"))

这不是投资建议，只是工程边界。等平台有了更细的风险分类，再把这些股票作为独立研究池处理。

从原始行变成候选对象

第 7 章定义了 StockCandidate：

@dataclass(frozen=True)
class StockCandidate:
    symbol: str
    name: str
    market: str
    sector: str
    lot_size: int = 100
    source: str = "manual"

然后用 stock_candidate_from_row() 接收不同字段名：

candidate = stock_candidate_from_row(
    {"code": "600036", "股票名称": "招商银行", "行业": "银行", "source": "eastmoney"}
)

返回结果会变成：

600036.SH / 招商银行 / SH / 银行 / source=eastmoney

这一步很适合放在纯函数里。真实供应商返回字段可能变，但只要入口函数能适配，后面的数据库写入和策略研究不需要跟着改。

构建公共股票池

公共股票池构建逻辑很短：

def build_public_universe(rows: Iterable[dict], limit: int = 500, default_source: str = "manual") -> list[StockCandidate]:
    out: list[StockCandidate] = []
    seen: set[str] = set()
    for row in rows:
        candidate = stock_candidate_from_row(row, default_source=default_source)
        if not candidate or candidate.symbol in seen:
            continue
        out.append(candidate)
        seen.add(candidate.symbol)
        if len(out) >= limit:
            break
    return out

这里有三个明确动作：过滤坏数据、按 symbol 去重、达到 limit 后停止。

早期我们用 500 只股票作为公共池规模，不是因为 500 有什么神奇意义，而是为了让本地回测、覆盖率检查和文章里的示例都能快速跑完。等链路稳定后，可以扩大到全市场。

还要能解释股票池

股票池不是构建完就结束。还要能解释它长什么样。

universe_summary() 会按市场、行业和来源做摘要：

{
    "count": 3,
    "by_market": {"SH": 1, "SZ": 2},
    "by_source": {"eastmoney": 2, "qveris": 1},
    "sample": ["600519.SH", "000001.SZ", "300750.SZ"],
}

这类摘要后面会进入数据质量报告。不是为了好看，而是为了能在回测前判断：这批股票是不是全是某个行业？是不是混入了 fallback？是不是市场分布明显异常？

本篇实战任务

拉取第 7 章代码：

git clone https://github.com/ax2/zi-quant-platform.git
cd zi-quant-platform
git checkout chapter-07
uv sync --extra dev
uv run pytest

只跑股票池测试：

uv run pytest tests/test_stock_universe.py

第 7 章本地全量测试通过：155 passed，仍只有既有 FastAPI deprecation warning。

本章更新与代码仓库

本章更新内容：

• 新增 app/stock_universe.py。
• 实现 A 股代码规范化、ST/退市过滤、去重、限额和股票池摘要。
• 新增 tests/test_stock_universe.py，覆盖股票池构建的关键边界。

代码仓库：

https://github.com/ax2/zi-quant-platform

本章代码：

git clone https://github.com/ax2/zi-quant-platform.git
cd zi-quant-platform
git checkout chapter-07
uv sync --extra dev
uv run pytest tests/test_stock_universe.py

本篇小结

股票池是量化系统的研究边界。

这一篇没有直接写策略，而是先把股票代码规范化、ST/退市过滤、去重、规模限制和来源摘要写成可测试代码。下一篇继续往下走：拿到股票池之后，怎样把原始 K 线清洗成统一的日线行情。