程序员量化交易实战 06：先把数据库表结构讲清楚

上一篇说数据比策略更重要。但数据不是“有一个下载脚本”就算落地，真正的落点是数据库。

从这一篇开始，ZiQuant 的主线项目进入第二组文章。第 6 篇先不急着拉更多行情，而是把 PostgreSQL 表结构、SQLAlchemy metadata 和 Alembic 迁移讲清楚。没有稳定表结构，后面的股票池、行情清洗、因子和回测都会变成临时代码。

为什么先看 schema

很多量化 demo 可以只用 CSV 和 DataFrame。这样写快，但很难追踪。

一个平台至少要回答这些问题：

• 某个策略用的是哪批股票？
• 回测那天用的是哪份行情？
• 行情来自哪个数据源？
• 因子值是不是后续重算过？
• 模拟盘订单有没有对应的策略和组合？

这些问题靠文件名和注释撑不住。要让系统能长期演进，必须把用户、股票、股票池、数据源、行情、财报、因子、策略、回测、模拟盘和任务记录都落到明确的表里。

ZiQuant 当前的核心表

当前项目已经有一批核心 ORM 模型，集中在app/models.py。

几类表最关键：

• zi_quant_stocks：股票主数据。
• zi_quant_stock_pools、zi_quant_stock_pool_members：股票池和成员。
• zi_quant_data_source_configs：数据源配置。
• zi_quant_market_bars：日线行情。
• zi_quant_financial_reports：财报数据。
• zi_quant_factor_definitions、zi_quant_factor_values：因子定义和因子值。
• zi_quant_strategies：策略配置。
• zi_quant_backtest_runs、zi_quant_backtest_trades：回测结果和交易明细。
• zi_quant_paper_portfolios、zi_quant_paper_positions、zi_quant_paper_orders：模拟盘组合、持仓和订单。

这些表不是为了“显得完整”。它们对应后续每一章要推进的能力。

把 schema 巡检写成代码

第 6 章新增了app/schema_checks.py，先做一件很小但很实用的事：从 SQLAlchemy metadata 里读出表、列、主键、外键和唯一约束，然后判断核心表是否完整。

关键数据结构是：

@dataclass(frozen=True) class TableSummary: name: str columns: tuple[str, ...] primary_key: tuple[str, ...] foreign_keys: tuple[str, ...] unique_constraints: tuple[str, ...]

这里没有连数据库。它检查的是“代码声明的 schema 是否满足平台最低要求”。这一步很适合放在单元测试里，跑得快，也不会依赖本地 PostgreSQL 是否启动。

核心函数是：

def schema_readiness(required_tables: Iterable[str] = CORE_TABLES, metadata: MetaData | None = None) -> dict[str, object]: summaries = summarize_metadata(metadata) present = {summary.name for summary in summaries} required = set(required_tables) missing = sorted(required - present) empty_primary_key = sorted(summary.name for summary in summaries if not summary.primary_key) return { "status": "ready" if not missing and not empty_primary_key else "degraded", "table_count": len(summaries), "required_table_count": len(required), "missing_tables": missing, "tables_without_primary_key": empty_primary_key, "tables": [summary.__dict__ for summary in summaries], }

这段代码的重点不是复杂，而是把“数据库结构是不是还能支撑平台”从口头判断变成自动检查。

Alembic 也要进入检查

只看 ORM 还不够。生产环境不能靠应用启动时自动建表，必须有明确迁移文件。

第 6 章也加了migration_readiness()：

def migration_readiness(project_root: Path | str = ".") -> dict[str, object]: root = Path(project_root) alembic_ini = root / "alembic.ini" versions_dir = root / "migrations" / "versions" versions = sorted(path.name for path in versions_dir.glob("*.py")) if versions_dir.exists() else [] missing: list[str] = [] if not alembic_ini.exists(): missing.append("alembic.ini") if not versions_dir.exists(): missing.append("migrations/versions") if not versions: missing.append("migration_versions") return { "status": "ready" if not missing else "degraded", "missing": missing, "revision_count": len(versions), "latest_revision_file": versions[-1] if versions else None, }

这仍然是轻量检查，但已经足够发现几类低级问题：忘了提交alembic.ini，迁移目录丢了，或者没有任何版本文件。

测试怎么写

对应测试在tests/test_schema_checks.py。

def test_schema_readiness_checks_required_tables_and_primary_keys(): ready = schema_readiness() assert ready["status"] == "ready" assert ready["missing_tables"] == [] assert ready["tables_without_primary_key"] == [] degraded = schema_readiness(required_tables={*CORE_TABLES, "zi_quant_missing_table"}) assert degraded["status"] == "degraded" assert degraded["missing_tables"] == ["zi_quant_missing_table"]

我故意让测试覆盖一个失败分支。只测 ready 状态没有太大意义，因为它不能证明检查真的会报警。

可运行基础校验

第 6 篇的核心是 schema readiness。当前统一用这条命令复现第 01-08 篇的基础能力：

uv run python -m scripts.chapter_examples foundation-check

本章对应输出如下：

schema_status=ready说明核心表在 SQLAlchemy metadata 里完整，migration_status=ready说明 Alembic 配置和版本文件也在。它不连接生产库，但能快速发现“模型或迁移文件没提交”的低级错误。

本篇实战任务

从 GitHub 拉取第 6 章代码：

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

也可以只跑本章测试：

uv run pytest tests/test_schema_checks.py

当前第 6 章 tag 对应的全量测试结果是150 passed。仍有两个 FastAPIon_eventdeprecation warning，这是既有启动生命周期写法带来的警告，不影响本章逻辑。

本章更新与代码仓库

本章更新内容：

• 新增app/schema_checks.py，从 SQLAlchemy metadata 汇总表、主键、外键和唯一约束。
• 新增 Alembic 迁移文件存在性检查。
• 新增tests/test_schema_checks.py，把 schema readiness 做成可回归测试。

代码仓库：

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

本章代码：

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

本篇小结

量化平台开始接真实数据之前，先要知道自己的数据库边界。

第 6 篇做的不是数据库性能优化，也不是复杂建模，而是把核心表、主键、外键、唯一约束和迁移文件变成可测试对象。下一篇我们在这个基础上处理股票池，把原始 A 股列表变成可复用的公共股票池。