4、数据模型与Schema设计:Pydantic模型定义、请求体与响应体规范、数据校验与序列化

做量化接口开发,最怕什么?

我最怕数据格式乱掉。你想想看,前端传过来一个因子值,到底是字符串还是浮点数?时间戳是秒还是毫秒?这些细节一旦搞错,整个策略就跑偏了。

Pydantic 就是来解决这个问题的。它帮我们把数据模型定死,让请求和响应都规规矩矩的。我个人习惯,写任何接口之前,先把数据模型定义好。模型定了,接口的骨架就有了。

4.1 为什么需要数据模型?

说白了,就是给数据上个「紧箍咒」。没有模型的时候,你收到一个 JSON,得自己手动判断字段对不对、类型对不对。有了 Pydantic,这些脏活累活它全包了。

我在项目中遇到过最典型的一个坑:有个同事传因子名称时用了中文全角冒号,结果数据库查询直接报错。从那以后,我要求所有接口必须用 Pydantic 做校验,格式不对直接打回。

核心价值:

  • 类型安全:传入的数据类型不对,立刻报错
  • 自动校验:字段范围、长度、格式自动检查
  • 序列化/反序列化:Python 对象 ↔ JSON 无缝转换
  • 文档自动生成:FastAPI 直接读取模型生成 OpenAPI 文档

4.2 基础模型定义

先看一个最简单的例子。我们定义一个因子查询的请求模型:

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional, List

class FactorQueryRequest(BaseModel):
    factor_name: str = Field(..., description="因子名称,如 'momentum_5d'")
    start_date: datetime = Field(..., description="开始日期")
    end_date: datetime = Field(..., description="结束日期")
    limit: int = Field(default=100, ge=1, le=10000, description="返回条数")
    offset: int = Field(default=0, ge=0, description="偏移量")

嗯,这里要注意几个点:

  • Field(...) 表示必填字段,三个点就是「必须提供」的意思
  • ge=1, le=10000 限制了 limit 的范围,防止有人一次拉几百万条数据
  • datetime 类型会自动解析 ISO 8601 格式的时间字符串

我的习惯:所有必填字段都用 Field(...) 显式标注,这样看代码的人一眼就知道哪些是必须的。可选的用 Optional[str] = None

4.3 响应体规范

响应体我一般分成三层:状态、数据、元信息。这样前端解析起来特别清晰。

class FactorValue(BaseModel):
    date: datetime
    value: float
    stock_code: str = Field(..., pattern=r'^\d{6}$')

class FactorResponse(BaseModel):
    code: int = Field(default=0, description="状态码,0表示成功")
    message: str = Field(default="success", description="提示信息")
    data: List[FactorValue]
    total: int = Field(..., description="数据总量")
    page: int = Field(default=1, description="当前页码")

为什么这么设计?

我曾经见过一个接口,成功时返回数组,失败时返回字符串。前端同学每次都要写 try-catch 来判断返回类型,苦不堪言。统一结构之后,前端只用判断 code 字段就行,省心多了。

避坑指南:千万不要在响应体里混用类型。比如 data 字段有时是对象、有时是数组、有时是 null。这种设计会让调用方崩溃。我见过一个老系统就是这么干的,后来重构时花了整整两周才理清楚。

4.4 数据校验进阶

基础校验不够用?Pydantic 提供了自定义校验器。比如我们要求因子名称必须是小写字母加下划线:

from pydantic import validator
import re

class FactorCreateRequest(BaseModel):
    factor_name: str
    formula: str
    category: str

    @validator('factor_name')
    def name_must_be_valid(cls, v):
        if not re.match(r'^[a-z_]+$', v):
            raise ValueError('因子名称只能包含小写字母和下划线')
        return v

    @validator('category')
    def category_must_exist(cls, v):
        valid_categories = ['momentum', 'volatility', 'value', 'quality']
        if v not in valid_categories:
            raise ValueError(f'分类必须是 {valid_categories} 之一')
        return v

这里有个小技巧:校验器可以链式调用。比如先校验格式,再校验业务逻辑。我习惯把格式校验和业务校验分开写,这样错误信息更明确。

4.5 序列化与别名

有时候 Python 里的字段名和 JSON 里的字段名不一样。比如数据库里存的是 factor_name,但前端习惯用 factorName。这时候用别名:

class FactorModel(BaseModel):
    factor_name: str = Field(alias='factorName')
    create_time: datetime = Field(alias='createTime')

    class Config:
        allow_population_by_field_name = True
        json_encoders = {
            datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S')
        }

allow_population_by_field_name = True 表示既可以用别名传参,也可以用原始字段名传参。我个人建议统一用别名,前后端各守各的约定。

时间格式的序列化也是个常见问题。默认 Pydantic 会输出 ISO 格式,但很多量化平台习惯用 YYYY-MM-DD HH:mm:ss。上面代码里的 json_encoders 就是干这个的。

4.6 嵌套模型与复杂结构

因子工厂里经常有复杂的嵌套结构。比如一个因子配置包含多个参数:

class Parameter(BaseModel):
    name: str
    value: float
    min_value: Optional[float] = None
    max_value: Optional[float] = None

class FactorConfig(BaseModel):
    factor_name: str
    parameters: List[Parameter]
    dependencies: Optional[List[str]] = None
    tags: List[str] = []

class BatchFactorRequest(BaseModel):
    factors: List[FactorConfig]
    overwrite: bool = False

这种嵌套模型,Pydantic 会自动递归校验。你传进来的 JSON 里,每个 Parameter 都会单独校验一遍。我在项目中用这个结构处理过上百个因子的批量导入,校验速度非常快。

性能提示:如果数据量很大(比如一次传几千个因子),建议用 model_validate 方法手动控制校验时机,而不是让 FastAPI 自动校验。我曾经优化过一个接口,把校验时间从 3 秒降到了 0.2 秒。

4.7 本章知识体系

下面这张图总结了数据模型设计的核心流程:

数据模型设计核心流程 客户端请求 JSON Pydantic 模型校验 校验通过 → 业务处理 校验失败 → 返回错误 统一响应体返回 错误信息返回 图:数据从请求到响应的完整校验流程

从这张图可以看得很清楚:所有请求数据必须先过 Pydantic 这道关。过了,进业务逻辑;没过,直接返回错误。这个流程我用了好几年,从来没出过数据格式问题。

4.8 实战建议

最后分享几个我在实战中总结的经验:

  • 模型和业务逻辑分离:Pydantic 模型只负责数据校验和序列化,不要在里面写业务代码。我曾经见过有人在校验器里查数据库,结果调试时根本分不清是校验报错还是业务报错。
  • 统一错误格式:所有校验失败的错误,都返回同样的结构。比如 {"code": 400, "message": "参数错误", "errors": [{"field": "factor_name", "msg": "格式不正确"}]}
  • 版本管理:接口升级时,模型也要版本化。比如 FactorQueryRequestV1FactorQueryRequestV2,不要在原模型上改来改去。

一个小工具:我写了个脚本,每次部署前自动扫描所有 Pydantic 模型,检查有没有字段类型不匹配、别名冲突之类的问题。这个脚本帮我抓出过不少低级错误。

数据模型设计,说白了就是给数据立规矩。规矩立好了,后面的开发就顺了。我见过太多项目因为数据格式问题反复返工,其实最开始花半小时把模型定义清楚,后面能省下几天的调试时间。

嗯,这一章就到这里。记住:好的模型设计,是高质量接口的基石。


蓝海数据掘金营,专注资料整理