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 本章知识体系
下面这张图总结了数据模型设计的核心流程:
从这张图可以看得很清楚:所有请求数据必须先过 Pydantic 这道关。过了,进业务逻辑;没过,直接返回错误。这个流程我用了好几年,从来没出过数据格式问题。
4.8 实战建议
最后分享几个我在实战中总结的经验:
- 模型和业务逻辑分离:Pydantic 模型只负责数据校验和序列化,不要在里面写业务代码。我曾经见过有人在校验器里查数据库,结果调试时根本分不清是校验报错还是业务报错。
- 统一错误格式:所有校验失败的错误,都返回同样的结构。比如
{"code": 400, "message": "参数错误", "errors": [{"field": "factor_name", "msg": "格式不正确"}]}。 - 版本管理:接口升级时,模型也要版本化。比如
FactorQueryRequestV1、FactorQueryRequestV2,不要在原模型上改来改去。
一个小工具:我写了个脚本,每次部署前自动扫描所有 Pydantic 模型,检查有没有字段类型不匹配、别名冲突之类的问题。这个脚本帮我抓出过不少低级错误。
数据模型设计,说白了就是给数据立规矩。规矩立好了,后面的开发就顺了。我见过太多项目因为数据格式问题反复返工,其实最开始花半小时把模型定义清楚,后面能省下几天的调试时间。
嗯,这一章就到这里。记住:好的模型设计,是高质量接口的基石。