4. FastAPI 框架入门:精算模型 API 的快速构建
FastAPI 是当前构建高性能精算模型 API 的首选框架。它原生支持异步处理、自动生成 OpenAPI 文档,并通过 Pydantic 模型实现严格的参数校验,完美契合精算模型对数据准确性和响应速度的双重要求。本节将带你从零搭建 FastAPI 环境,并完成第一个精算参数校验接口。
4.1 环境安装与项目初始化
FastAPI 依赖 Python 3.8+,推荐使用虚拟环境隔离项目依赖。安装命令如下:
# 创建虚拟环境(可选但推荐)
python -m venv actuarial_env
source actuarial_env/bin/activate # Linux/Mac
actuarial_env\Scripts\activate # Windows
# 安装 FastAPI 与生产级 ASGI 服务器
pip install fastapi uvicorn[standard]
# 验证安装
python -c "import fastapi; print(fastapi.__version__)"
精算环境提示: 生产环境建议锁定版本:
fastapi==0.110.0 与 uvicorn==0.27.0,避免因框架升级导致的接口行为变化。
4.2 路由定义:精算模型端点设计
路由是 API 的入口。精算模型通常按险种或计算类型划分端点,例如寿险准备金、非寿险赔付率。以下示例定义了一个基础的保费计算端点:
from fastapi import FastAPI
app = FastAPI(title="精算模型 API", version="1.0.0")
@app.get("/health")
async def health_check():
"""健康检查端点,用于负载均衡探活"""
return {"status": "healthy", "model_version": "1.0.0"}
@app.post("/api/v1/premium/calculate")
async def calculate_premium():
"""保费计算端点(暂未接入校验)"""
return {"message": "Premium calculation endpoint ready"}
| HTTP 方法 | 路由路径 | 用途 | 精算场景示例 |
|---|---|---|---|
GET |
/health |
服务探活 | K8s 就绪检查 |
POST |
/api/v1/premium/calculate |
保费计算 | 车险纯保费预测 |
POST |
/api/v1/reserve/run |
准备金评估 | 链梯法准备金计算 |
GET |
/api/v1/model/info |
模型元信息 | 返回模型版本、输入特征列表 |
4.3 依赖注入:精算上下文管理
FastAPI 的依赖注入系统允许将数据库连接、模型实例、配置对象等精算上下文优雅地注入到路由函数中,避免全局变量和重复初始化。
from fastapi import Depends, FastAPI
app = FastAPI()
# 模拟精算模型加载(实际场景为 pickle 或 ONNX 加载)
def get_actuarial_model():
# 实际代码:model = load_model("path/to/model.pkl")
class PremiumModel:
def predict(self, features: dict) -> float:
return features.get("age", 30) * 0.05 + 100
return PremiumModel()
@app.post("/api/v1/premium/calculate")
async def calculate_premium(
model: PremiumModel = Depends(get_actuarial_model)
):
# 模型已通过依赖注入自动获取
result = model.predict({"age": 35})
return {"premium": result}
性能要点: 将模型加载放在依赖函数中,并配合
lru_cache 或单例模式,确保模型只加载一次,后续请求复用内存中的模型实例,避免重复 I/O。
4.4 Pydantic 模型:精算参数校验
精算模型对输入参数有严格的业务规则:年龄不能为负、保额必须为正整数、性别只能为 'M' 或 'F'。Pydantic 模型通过类型注解和验证器,在请求到达业务逻辑前完成校验,拦截非法数据。
from pydantic import BaseModel, Field, field_validator
from typing import Literal
class PremiumInput(BaseModel):
"""保费计算输入参数"""
age: int = Field(ge=18, le=100, description="被保险人年龄(18-100)")
gender: Literal['M', 'F'] = Field(description="性别:M 男性,F 女性")
sum_assured: float = Field(gt=0, description="保额,必须大于 0")
smoking_status: Literal['smoker', 'non_smoker'] = Field(
description="吸烟状态"
)
@field_validator('sum_assured')
@classmethod
def check_sum_assured_precision(cls, v: float) -> float:
"""保额最多保留两位小数"""
if round(v, 2) != v:
raise ValueError('sum_assured 最多支持两位小数')
return v
@field_validator('age')
@classmethod
def check_age_range(cls, v: int) -> int:
"""精算业务规则:年龄不能为 0"""
if v == 0:
raise ValueError('年龄不能为 0')
return v
将 Pydantic 模型作为路由参数后,FastAPI 自动完成 JSON 解析、类型转换和校验:
from fastapi import FastAPI
app = FastAPI()
@app.post("/api/v1/premium/calculate")
async def calculate_premium(input_data: PremiumInput):
# 此时 input_data 已经过完整校验
base_premium = input_data.sum_assured * 0.02
age_factor = 1.0 if input_data.age < 50 else 1.5
smoking_factor = 1.8 if input_data.smoking_status == 'smoker' else 1.0
final_premium = base_premium * age_factor * smoking_factor
return {
"premium": round(final_premium, 2),
"input_summary": input_data.model_dump()
}
常见错误: 忘记在 Pydantic 模型中添加
@field_validator 装饰器,导致业务规则校验遗漏。精算场景下,务必对年龄、保额、费率因子等关键字段添加显式校验逻辑。
4.5 精算参数校验流程总览
下图展示了从 HTTP 请求到达 FastAPI 应用,到 Pydantic 模型完成校验,最终进入精算模型计算的完整流程:
4.6 启动服务与测试
使用 Uvicorn 启动服务,并验证参数校验效果:
# 启动服务(开发模式,热重载)
uvicorn main:app --reload --host 0.0.0.0 --port 8000
# 测试合法请求
curl -X POST "http://localhost:8000/api/v1/premium/calculate" \
-H "Content-Type: application/json" \
-d '{"age": 35, "gender": "M", "sum_assured": 100000.00, "smoking_status": "non_smoker"}'
# 测试非法请求(年龄超出范围)
curl -X POST "http://localhost:8000/api/v1/premium/calculate" \
-H "Content-Type: application/json" \
-d '{"age": 150, "gender": "M", "sum_assured": 100000.00, "smoking_status": "non_smoker"}'
低延迟技巧: 生产环境使用
uvicorn main:app --workers 4 启动多进程,配合 --limit-concurrency 100 控制并发数,避免精算模型计算时资源耗尽。
4.7 本章小结
| 知识点 | 精算应用价值 | 关键命令/代码 |
|---|---|---|
| FastAPI 安装 | 快速搭建精算 API 服务 | pip install fastapi uvicorn |
| 路由定义 | 按险种/计算类型划分端点 | @app.post("/api/v1/premium/calculate") |
| 依赖注入 | 复用模型实例,避免重复加载 | Depends(get_actuarial_model) |
| Pydantic 校验 | 拦截非法精算参数,保障模型安全 | Field(ge=18, le=100) + @field_validator |