3. FastAPI 基础与 RESTful 设计

在实时交易风控系统中，API 是策略执行、数据查询与风控指令交互的核心通道。本章将深入 FastAPI 的核心机制，从路由设计、请求方法到数据模型验证，并建立一套适用于金融场景的 API 版本控制策略。

3.1 FastAPI 快速入门

FastAPI 基于 Python 3.7+ 的类型提示，自动生成 OpenAPI 文档，并具备异步原生支持。对于交易风控场景，其高性能与自动校验能力可显著降低接口开发与维护成本。

以下是一个最小化 FastAPI 应用示例，启动后即可访问 /docs 查看自动生成的 API 文档：

from fastapi import FastAPI

app = FastAPI(title="交易风控API", version="1.0.0")

@app.get("/")
def root():
    return {"message": "风控服务运行中"}

3.2 路由与请求方法

FastAPI 使用装饰器声明路由与 HTTP 方法。在风控系统中，不同操作对应不同语义：

路由参数支持路径参数与查询参数，FastAPI 自动进行类型转换与校验：

from fastapi import FastAPI, Query

app = FastAPI()

# 路径参数：获取指定风控规则
@app.get("/rules/{rule_id}")
def get_rule(rule_id: int):
    return {"rule_id": rule_id, "status": "active"}

# 查询参数：按阈值过滤规则
@app.get("/rules")
def list_rules(min_threshold: float = Query(0.0, ge=0.0)):
    return {"threshold_filter": min_threshold}

3.3 Pydantic 模型定义

Pydantic 是 FastAPI 的数据校验核心。在交易风控中，每个请求与响应都应定义明确的模型，确保数据完整性与类型安全。

以下是一个订单风控检查请求模型：

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

class OrderCheckRequest(BaseModel):
    symbol: str = Field(..., description="交易对，如 BTCUSDT")
    side: str = Field(..., pattern="^(BUY|SELL)$")
    quantity: float = Field(..., gt=0, description="数量")
    price: float = Field(..., gt=0, description="价格")
    client_order_id: Optional[str] = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

class OrderCheckResponse(BaseModel):
    passed: bool
    risk_score: float = Field(..., ge=0.0, le=1.0)
    reason: Optional[str] = None

在路由中使用模型后，FastAPI 自动完成请求体解析与校验：

@app.post("/orders/check", response_model=OrderCheckResponse)
def check_order(order: OrderCheckRequest):
    # 模拟风控检查逻辑
    risk_score = 0.1 if order.quantity < 100 else 0.8
    return OrderCheckResponse(
        passed=risk_score < 0.5,
        risk_score=risk_score,
        reason=None if risk_score < 0.5 else "数量超过风控阈值"
    )

3.4 API 版本控制策略

交易风控系统需要长期迭代，API 版本控制是保障向后兼容的关键。以下是三种主流策略的对比：

在 FastAPI 中，推荐使用 URL 路径 策略，通过 APIRouter 实现模块化版本管理：

from fastapi import APIRouter

# v1 版本路由
router_v1 = APIRouter(prefix="/v1")

@router_v1.get("/rules")
def list_rules_v1():
    return {"version": "v1", "rules": []}

# v2 版本路由（新增字段）
router_v2 = APIRouter(prefix="/v2")

@router_v2.get("/rules")
def list_rules_v2():
    return {"version": "v2", "rules": [], "metadata": {}}

在主应用中注册不同版本的路由：

app.include_router(router_v1)
app.include_router(router_v2)

3.5 架构总览：FastAPI 风控 API 请求处理流程

下图展示了从客户端请求到风控响应的完整链路，涵盖路由匹配、模型校验、业务处理与版本控制：

3.6 本章小结

• FastAPI 入门：利用类型提示与自动文档，快速搭建风控服务基础框架。
• 路由与请求方法：根据 CRUD 语义选择 GET/POST/PUT/DELETE，合理使用路径与查询参数。
• Pydantic 模型：定义严格的请求/响应模型，利用 Field 约束与正则校验保障数据质量。
• 版本控制：优先采用 URL 路径策略，通过 APIRouter 实现多版本共存，确保 API 平滑演进。