Week 2-3：进阶 Python + FastAPI 实战

🎯 本周目标：掌握 Python 异步编程、Pydantic 数据模型、以及用 FastAPI 写 API。最终产出：在你的 electric-trading-ai 项目里跑通一个 FastAPI 后端。

1. 异步编程（你的强项！）

Python 的 async/await 几乎和 TypeScript 一模一样：

import asyncio

async def fetch_user(user_id: int) -> dict:
    """模拟异步请求"""
    await asyncio.sleep(1)    # 相当于 await new Promise(r => setTimeout(r, 1000))
    return {"id": user_id, "name": f"User{user_id}"}

async def main():
    # 并行执行多个请求
    tasks = [fetch_user(i) for i in range(3)]
    results = await asyncio.gather(*tasks)   # 相当于 Promise.all()
    print(results)

# 入口
asyncio.run(main())   # 相当于 Node.js 自带事件循环，需要显式启动

TS 对照

Python	TypeScript	说明
`async def foo():`	`async function foo() {}`	完全一样
`await foo()`	`await foo()`	完全一样
`asyncio.gather(a, b)`	`Promise.all([a, b])`	并行执行
`asyncio.run(main())`	直接运行 `main()`	Python 需要显式启动事件循环

💡 什么时候用异步？

FastAPI 接口处理网络请求 → 必须用 async
数据库查询（如 async sqlite / async postgres）→ 必须用 async
CPU 密集计算（如数学运算、AI 推理本体）→ 别用 async，用普通同步函数

异步 + 同步混用的坑

# ❌ 错误写法：async 函数里调同步阻塞函数
async def bad_handler():
    time.sleep(1)        # 阻塞整个事件循环！
    return {"ok": True}

# ✅ 正确写法 1：改成异步版本
async def good_handler():
    await asyncio.sleep(1)   # 不阻塞
    return {"ok": True}

# ✅ 正确写法 2：把同步函数放到线程池
async def good_handler_2():
    # run_in_executor 或用 anyio.to_thread.run_sync
    result = await asyncio.get_event_loop().run_in_executor(None, blocking_func)
    return result

🚨 这个坑在之前的真实项目中出现过：底层 DB 是异步实现但却调用了同步方法，导致运行时报错 run method is not supported with an async database. Please use arun method instead.。统一全用 arun() / astream() 全套异步链路，不要混用！

2. Pydantic v2（Python 的 zod）

Pydantic 是 FastAPI 的底层数据模型库，相当于 TypeScript 的 interface + zod 运行时校验。

基础用法

from pydantic import BaseModel, Field
from typing import Optional

class TradingRequest(BaseModel):
    symbol: str                              # 必填
    price: float                             # 必填
    quantity: int = 100                      # 可选 + 默认值
    note: Optional[str] = None               # 可选 + 默认 None
    tag: str = Field(default="normal", description="交易标签")

# 自动校验
req = TradingRequest(symbol="BTC", price=99999.99)
print(req.symbol)    # BTC
print(req.quantity)  # 100（默认值）

# 校验失败会抛异常
# req = TradingRequest(symbol="BTC")  # ❌ price 缺失 → ValidationError

TS 对照

// 用 zod 的话（运行时校验）
import { z } from 'zod';

const TradingRequestSchema = z.object({
  symbol: z.string(),
  price: z.number(),
  quantity: z.number().default(100),
  note: z.string().nullable().optional(),
  tag: z.string().default("normal"),
});

type TradingRequest = z.infer<typeof TradingRequestSchema>;

// 或者纯 TypeScript interface（只有类型检查，无运行时校验）
interface TradingRequest {
  symbol: string;
  price: number;
  quantity?: number;       // 可选
  note?: string | null;
  tag?: string;
}

Pydantic 高级特性

from pydantic import BaseModel, field_validator, model_validator

class User(BaseModel):
    email: str
    age: int

    # 字段校验器
    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if '@' not in v:
            raise ValueError('邮箱格式不正确')
        return v.lower()      # 自动转换

    # 多字段校验器
    @model_validator(mode='after')
    def check_age_email(self) -> 'User':
        if self.age < 18 and 'child' not in self.email:
            raise ValueError('未成年用户邮箱需包含 child 标识')
        return self

3. FastAPI 实战

📌 你的 electric-trading-ai 项目已经有 FastAPI 依赖了，直接用！

项目结构

electric-trading-ai/
├── app/
│   ├── __init__.py
│   ├── main.py          # 入口文件
│   ├── routes/
│   │   ├── __init__.py
│   │   └── trading.py   # 交易相关路由
│   └── models/
│       ├── __init__.py
│       └── schemas.py   # Pydantic 数据模型
├── .env
├── pyproject.toml
└── .venv/

Step 1：定义数据模型

# app/models/schemas.py
from pydantic import BaseModel, Field
from typing import Literal

OrderSide = Literal["buy", "sell"]

class PlaceOrderRequest(BaseModel):
    symbol: str = Field(..., description="交易标的代码")
    side: OrderSide
    price: float = Field(gt=0, description="价格必须大于 0")
    quantity: int = Field(gt=0, default=100)

class OrderResponse(BaseModel):
    order_id: str
    symbol: str
    status: Literal["pending", "filled", "failed"]

class HealthResponse(BaseModel):
    status: Literal["ok", "error"]
    version: str

Step 2：写路由

# app/routes/trading.py
from fastapi import APIRouter, HTTPException
from app.models.schemas import PlaceOrderRequest, OrderResponse

router = APIRouter(prefix="/api/trading", tags=["交易"])

# 模拟内存存储
orders: dict[str, OrderResponse] = {}

@router.post("/orders", response_model=OrderResponse)
async def place_order(req: PlaceOrderRequest):
    """下单"""
    import uuid
    order_id = str(uuid.uuid4())[:8]
    order = OrderResponse(order_id=order_id, symbol=req.symbol, status="filled")
    orders[order_id] = order
    return order

@router.get("/orders/{order_id}", response_model=OrderResponse)
async def get_order(order_id: str):
    """查询订单"""
    if order_id not in orders:
        raise HTTPException(status_code=404, detail="订单不存在")
    return orders[order_id]

Step 3：主入口

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routes.trading import router as trading_router
from app.models.schemas import HealthResponse

app = FastAPI(title="Electric Trading AI", version="0.1.0")

# CORS 中间件（允许前端跨域）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(trading_router)

@app.get("/health", response_model=HealthResponse)
async def health_check():
    return HealthResponse(status="ok", version="0.1.0")

Step 4：启动服务

# 方式一：直接指定模块路径
.venv\Scripts\uvicorn app.main:app --reload

# 方式二（如果项目根目录有 main.py）
# .venv\Scripts\uvicorn main:app --reload

--reload 相当于 nodemon，代码改动后自动重启。

TS 对照（FastAPI vs Express.js）

FastAPI	Express.js (TS)	说明
`from fastapi import FastAPI`	`import express from 'express'`	导入框架
`app = FastAPI()`	`const app = express()`	创建应用实例
`@app.get("/path")`	`app.get("/path", handler)`	路由定义（装饰器 vs 链式调用）
`async def handler(req)`	`async (req, res) => {}`	异步处理函数
`Pydantic Model`	`zod schema` / `interface`	请求体校验
`response_model=XxxModel`	`res.json(data)`	响应序列化 + 文档生成
`uvicorn app:app --reload`	`nodemon ts-node src/index.ts`	启动命令

4. 自动生成 API 文档

FastAPI 开箱即用的 Swagger UI：

启动后打开 → http://localhost:8000/docs

你会看到一个完整的交互式 API 文档（类似 Swagger UI），可以直接在浏览器里点按钮测试每个接口。

这是 FastAPI 最大的优势之一：零配置自动生成文档。Express.js 里你得手动配 swagger-jsdoc + swagger-ui-express。

5. 本周作业

#	任务	验收标准
1	把上面的 FastAPI 代码跑起来	`GET /health` 返回正确
2	给 `/api/trading/orders` 加一个 mock 下单逻辑（不接真实交易所）	POST 能返回 OrderResponse
3	加一个 `GET /api/trading/orders` 列表接口	返回所有订单
4	加一个 `DELETE /api/trading/orders/{id}` 接口	能正确删除
5	读 FastAPI 官方 Tutorial（前 5 章即可）	理解路由、Pydantic、依赖注入

6. 常用第三方库速查

库	用途	TS 对照
`fastapi`	Web 框架	express / nestjs
`uvicorn`	ASGI 服务器	node / ts-node
`pydantic`	数据校验 + 序列化	zod / joi
`httpx`	异步 HTTP 客户端	axios / fetch
`sqlalchemy`	ORM	prisma / typeorm
`aiosqlite`	异步 SQLite	better-sqlite3-async
`python-dotenv`	读取 .env	dotenv
`loguru`	日志	winston / pino