程序员量化交易实战 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 状态没有太大意义，因为它不能证明检查真的会报警。

本篇实战任务

从 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。仍有两个 FastAPI on_event deprecation 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 股列表变成可复用的公共股票池。