上一篇说数据比策略更重要。但数据不是“有一个下载脚本”就算落地,真正的落点是数据库。
从这一篇开始,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 股列表变成可复用的公共股票池。