云上量化微服务：用 FastAPI + Redis 搭建多策略共享的“行情缓存网关”

┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  策略容器 A  │    │  策略容器 B  │    │  策略容器 C  │
└──────┬───────┘    └──────┬───────┘    └──────┬───────┘
       │                   │                   │
       └─────────────┬─────┴───────────────────┘
                     │ (内网 HTTP 请求)
                     ▼
             ┌──────────────┐       读取缓存 (0.1ms)   ┌──────────────┐
             │ 行情代理网关  ├─────────────────────────►│  腾讯云Redis │
             │  (FastAPI)   │◄─────────────────────────┤ (TencentDB)  │
             └──────┬───────┘       未命中则写入缓存   └──────────────┘
                    │
                    │ 缓存失效 / 未命中时回源 (公网 REST API)
                    ▼
             ┌──────────────┐
             │ QuantDash API│
             └──────────────┘

pip install fastapi uvicorn redis quantdash

import os
import json
import redis
from fastapi import FastAPI, HTTPException
from quantdash import QuantDash

app = FastAPI(title="Internal Quant Data Cache Gateway")

# 1. 连接本地或腾讯云 TencentDB for Redis
REDIS_HOST = os.getenv("REDIS_HOST", "127.0.0.1")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

# 2. 初始化外部行情源 QuantDash 客户端
QD_API_KEY = os.getenv("QUANTDASH_API_KEY", "your_api_key_here")
qd = QuantDash(api_key=QD_API_KEY)

# 定义盘中热行情缓存生存时间（TTL = 2秒）
CACHE_TTL_SECONDS = 2

@app.get("/v1/quote/{symbol}")
def get_quote_with_cache(symbol: str):
    """
    获取实时行情接口 (Cache-Aside 模式)
    """
    cache_key = f"realtime_quote:{symbol.upper()}"
    
    # 尝试一：读取 Redis 缓存（内网连接，时延 < 1ms）
    try:
        cached_data = r.get(cache_key)
        if cached_data:
            return {
                "status": "success",
                "source": "local_redis_cache",
                "data": json.loads(cached_data)
            }
    except Exception as redis_err:
        # 降级处理：若 Redis 宕机，记录日志并继续通过 API 回源，保证业务不中断
        print(f"Redis Connection Error: {redis_err}")

    # 尝试二：缓存未命中，回源公网 QuantDash API
    try:
        # 获取单只标的的盘中实时行情数据 (SDK 默认支持 A/港/美 统一代码)
        quote_res = qd.quotes.get(symbols=[symbol])
        
        if not quote_res:
            raise HTTPException(status_code=404, detail=f"Symbol {symbol} not found.")
        
        # 适配返回数据结构（QuantDash quotes 返回通常为字典或列表）
        single_quote = quote_res[0] if isinstance(quote_res, list) else quote_res
        
        # 异步/同步写入 Redis 并设置 2s 自动过期
        try:
            r.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(single_quote))
        except Exception as redis_err:
            print(f"Redis Write Error: {redis_err}")
            
        return {
            "status": "success",
            "source": "quantdash_origin_api",
            "data": single_quote
        }
        
    except Exception as api_err:
        raise HTTPException(
            status_code=500, 
            detail=f"Failed to fetch data from origin API: {str(api_err)}"
        )