当前位置：首页 > news >正文

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

news 2026/6/21 16:04:27

第 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？是不是市场分布明显异常？

可运行基础校验

股票池的结果必须能解释。当前统一用这条命令复现第 01-08 篇的基础能力：

uv run python -m scripts.chapter_examples foundation-check

本章对应输出如下：

示例输入里包含重复代码和 ST 名称，最后只留下600519.SH与000001.SZ。摘要同时给出市场、行业和来源分布，这比只打印股票列表更适合后续排查股票池是否异常。

本篇实战任务

拉取第 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