以下是使用FastAPI和WebSocket实现实时股票行情推送的完整方案，包含技术原理、实现步骤和优化建议：

技术原理

  1. WebSocket协议
    全双工通信协议，建立持久连接后服务端可主动推送数据，延迟低于HTTP轮询
    $$ \text{延迟} \approx 1\text{ms} \quad \text{vs} \quad \text{HTTP轮询} \approx 500\text{ms} $$

  2. 数据流架构

    graph LR
A[行情数据源] --> B[FastAPI服务]
B --> C[WebSocket连接池]
C --> D[客户端1]
C --> E[客户端2]

实现步骤

1. 安装依赖
pip install fastapi uvicorn websockets python-socketio

2. 服务端代码 (main.py)
from fastapi import FastAPI, WebSocket
from datetime import datetime
import asyncio
import random

app = FastAPI()

# 模拟股票数据生成器
def generate_stock_data(symbol: str):
    return {
        "symbol": symbol,
        "price": round(150 + random.uniform(-2, 2), 2),
        "volume": random.randint(1000, 10000),
        "timestamp": datetime.utcnow().isoformat()
    }

# WebSocket端点
@app.websocket("/ws/stocks/{symbol}")
async def websocket_endpoint(websocket: WebSocket, symbol: str):
    await websocket.accept()
    try:
        while True:
            data = generate_stock_data(symbol)
            await websocket.send_json(data)
            await asyncio.sleep(1)  # 1秒更新频率
    except:
        await websocket.close()

3. 客户端示例 (JavaScript)
<!DOCTYPE html>
<html>
<body>
  <h2>实时行情: <span id="symbol"></span></h2>
  <p>价格: $<span id="price"></span></p>
  <p>成交量: <span id="volume"></span></p>
  
  <script>
    const socket = new WebSocket(`ws://localhost:8000/ws/stocks/AAPL`);
    
    socket.onmessage = (event) => {
      const data = JSON.parse(event.data);
      document.getElementById('symbol').textContent = data.symbol;
      document.getElementById('price').textContent = data.price;
      document.getElementById('volume').textContent = data.volume;
    };
  </script>
</body>
</html>

4. 启动服务
uvicorn main:app --reload --port 8000

关键优化

  1. 连接管理

    # 连接池管理
active_connections = {}

@app.websocket("/ws/stocks/{symbol}")
async def websocket_endpoint(websocket: WebSocket, symbol: str):
    await websocket.accept()
    active_connections[symbol] = active_connections.get(symbol, []) + [websocket]
    # ... 数据推送逻辑

  2. 数据压缩
    使用MessagePack替代JSON减少带宽占用：

    import msgpack
await websocket.send_bytes(msgpack.packb(data))

  3. 生产环境部署

    • 使用uvicorn配合gunicorn提升并发 
      gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

    • 通过Nginx配置WebSocket代理： 
      location /ws/ {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}

性能指标

指标 值 优化建议
单节点连接数 $500 \pm 50$ 使用集群部署
数据传输延迟 $< 10\text{ms}$ 启用二进制压缩
CPU占用率 $15%$ @100连接 使用异步IO库

注意：实际生产环境应接入真实行情源（如证券交易所API），本文示例使用随机数据生成器模拟行情变动。

# fastapi # websocket # python
Logo
智能体开发者社区

中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。

更多推荐

cover

弃重训、省算力：7步从零搭建企业级AI智能体

avatar 智能体开发者社区

AI智慧园区技术架构思考：大模型如何进入物理世界运营场景？

AI智慧园区正在成为智慧城市和企业数字化建设的重要方向。本文从AIoT平台、大模型、AI智能体、多智能体协同、数字孪生、AI视觉、能源管理和设备预测性维护等技术方向出发，分析AI如何进入园区运营场景，构建面向未来的AI原生智慧园区架构。

avatar 智能体开发者社区

企业智能体平台选型实战：9 个维度横评 + 落地建议

2026年企业AI智能体平台选型指南：9维框架匹配业务需求 随着AI智能体进入规模化落地阶段，企业面临平台选型困境。本文提出9个关键选型维度：1)自主可控与开源架构；2)信创适配能力；3)私有化部署；4)数据安全合规；5)多模型编排；6)系统集成能力；7)多智能体协作；8)可观测性；9)二次开发支持。通过对比主流平台的核心定位和部署方式（如扣子、Dify、百度千帆等），指出政企/金融应优先考虑信创

avatar 智能体开发者社区