上一篇说数据比策略更重要。但数据不是“有一个下载脚本”就算落地真正的落点是数据库。从这一篇开始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(frozenTrue) 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_statusready说明核心表在 SQLAlchemy metadata 里完整migration_statusready说明 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 股列表变成可复用的公共股票池。