1. 为什么回测系统一上多用户就卡死FastAPI 异步任务与 Vue3 轮询的真实场景做量化回测 Web 系统的人大概率都踩过同一个坑单用户跑得好好的一旦多用户同时点“开始回测”整个站点就像被冻住一样连登录页都刷不出来。核心检索词先摆出来——FastAPI 前后端分离、Vue3 分布式回测系统、多用户 Web 平台这三件事凑在一起本质上是把“计算密集型任务”和“HTTP 请求生命周期”解耦的问题。你如果正在用 Cursor 从零搭一套能扛住多用户的回测平台这篇就是按可复制路径写的实战记录。先说清楚这套系统能做什么、适合谁。它面向的是这样一类开发者手里有 Python 策略代码想做成 Web 服务让多个用户各自提交回测、各自看进度和收益曲线不想被 Django 模板那套同步阻塞拖住希望前端静态资源和后端 API 分开部署Nginx 扛静态、FastAPI 扛计算调度。适合中小团队快速验证也适合个人开发者练手分布式架构。我试过的第一个版本就是最朴素的写法app.post(/backtest)里直接for循环跑完 200 天回测再返回。结果单用户 3 秒返回两个用户并发就变成 6 秒五个用户直接超时。原因很直白——FastAPI 的async def路由如果里面是纯 CPU 计算事件循环被占满其他请求全部排队。这就是“回测阻塞 Web 线程”的经典症状。第二个坑是进度感知。用户点了按钮前端只能转圈不知道跑到哪了。有人会用 WebSocket但多用户场景下连接管理成本高更轻的做法是前端轮询一个状态接口后端把任务进度写进一个可查询的存储里。这套“提交任务拿 task_id → 轮询状态 → 完成后拉结果”的模式是动静分离架构里最稳的交互范式。第三个坑是多用户资源抢占。所有人任务都塞进同一个进程的线程池CPU 打满后连健康检查都超时。所以架构上必须把“API 网关进程”和“计算执行单元”分开哪怕初期只是同机不同进程也为后面换 Celery 分布式队列留好接口。这一篇的完整链路是Cursor 里生成 FastAPI 后端骨架 → 用 BackgroundTasks 把回测挂到后台 → Vue3 Pinia 做状态管理和轮询 → ECharts 渲染收益曲线 → Nginx 托管前端静态资源 → 所有模型调用统一走 TaoToken 的 Key 和 API 通道。下面从环境准备开始一步步给可复制的配置。2. TaoToken 统一 Key 接入前置一个 Key 打通回测系统的模型调用通道回测系统本身是算收益曲线但一个完整的量化 Web 平台往往还要接模型能力比如用大模型生成策略说明、解析自然语言选股条件、给回测报告写摘要。这些调用如果每个服务各自管一套 Key多用户环境下很快就会乱。TaoToken 在这里的角色就是统一入口——官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 。你需要先拿到一个可用的 Key。进入控制台创建https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 然后在 API Keys 页面生成https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。生成后复制那串sk-开头的字符串后面所有配置都用它。这里要强调一个工程习惯不要把 Key 硬编码进main.py。用环境变量或.env文件管理.env加进.gitignore。多用户系统里Key 泄露等于账单失控。下面给出后端读取 Key 的标准写法配合python-dotenv# .env 文件放在后端项目根目录 TAOTOKEN_API_KEYsk-你的实际Key TAOTOKEN_BASE_URLhttps://taotoken.net/api# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) if not TAOTOKEN_API_KEY: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env 文件)模型 ID 怎么选回测报告摘要这类文本任务用通用对话模型即可如果你要做代码辅助或策略逻辑解释可以走 coding-plan 通道https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。想先验证模型是否通直接用模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面写了 OpenAI 兼容的调用格式。也就是说你后端用openai这个 Python SDK把base_url指向 TaoToken 的 API 地址就能直接调不用改业务代码结构。这一点对回测系统很关键——模型调用层和回测计算层解耦后面换模型只改配置。如果你用 Claude Code 做开发辅助Anthropic 兼容通道的配置页在 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite 里面给了 Base URL、Key、Model ID 三件套的填法。记住这个三件套原则任何工具接入都是 Base URL Key Model ID缺一不可。前置准备到这里就够了。核心是一个 Key、一个 Base URL、一个模型 ID通过环境变量注入后端。下面进入真正的代码配置环节。3. 可复制配置FastAPI 目录结构、依赖清单与 Vue3 动静分离骨架这一节给的是能直接抄的目录结构和配置文件。先看整体布局前后端完全分离Nginx 在最外层做静态资源托管和 API 反向代理backtest-platform/ ├── backend/ │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI 入口 │ │ ├── config.py # 环境变量与 TaoToken 配置 │ │ ├── backtest_engine.py # 回测计算引擎 │ │ ├── task_store.py # 任务状态存储内存/Redis │ │ └── routers/ │ │ ├── backtest.py # 回测相关路由 │ │ └── llm.py # 模型调用路由 │ ├── requirements.txt │ └── .env ├── frontend/ │ ├── src/ │ │ ├── main.js │ │ ├── App.vue │ │ ├── stores/ │ │ │ └── backtestStore.js │ │ └── components/ │ │ └── BacktestReport.vue │ ├── package.json │ └── vite.config.js └── nginx.conf后端依赖清单requirements.txt版本锁死避免踩兼容坑fastapi0.103.1 uvicorn[standard]0.23.2 pydantic2.3.0 pandas2.1.0 numpy1.25.2 python-dotenv1.0.0 openai1.3.0 redis5.0.1前端package.json核心依赖{ dependencies: { vue: ^3.3.4, pinia: ^2.1.6, axios: ^1.5.0, echarts: ^5.4.3 }, devDependencies: { vite: ^4.4.9, vitejs/plugin-vue: ^4.3.4 } }任务状态存储task_store.py开发期用内存字典生产换 Redis。这里给出双模式写法通过环境变量切换# app/task_store.py import os import json USE_REDIS os.getenv(USE_REDIS, false).lower() true if USE_REDIS: import redis _client redis.Redis( hostos.getenv(REDIS_HOST, 127.0.0.1), portint(os.getenv(REDIS_PORT, 6379)), decode_responsesTrue, ) def set_task(task_id: str, payload: dict, ttl: int 86400): _client.setex(fbacktest:task:{task_id}, ttl, json.dumps(payload)) def get_task(task_id: str): raw _client.get(fbacktest:task:{task_id}) return json.loads(raw) if raw else None else: _memory {} def set_task(task_id: str, payload: dict, ttl: int 86400): _memory[task_id] payload def get_task(task_id: str): return _memory.get(task_id)回测引擎backtest_engine.py核心是双均线策略逐日推进并更新进度。注意这里把进度写进task_store而不是全局字典为后面换 Redis 铺路# app/backtest_engine.py import time import pandas as pd import numpy as np from app.task_store import set_task, get_task def run_ma_backtest(task_id: str, symbol: str, fast_ma: int 5, slow_ma: int 20): try: set_task(task_id, {progress: 0, status: running, result: None}) np.random.seed(42) dates pd.date_range(start2023-01-01, periods200) price_changes np.random.normal(loc0.0005, scale0.015, size200) close_prices 100 * np.exp(np.cumsum(price_changes)) df pd.DataFrame({Close: close_prices}, indexdates) df[Fast_MA] df[Close].rolling(windowfast_ma).mean() df[Slow_MA] df[Close].rolling(windowslow_ma).mean() portfolio_values [1.0] position 0 total_days len(df) for i in range(slow_ma, total_days): row df.iloc[i] prev_row df.iloc[i - 1] if row[Fast_MA] row[Slow_MA] and prev_row[Fast_MA] prev_row[Slow_MA]: position 1 elif row[Fast_MA] row[Slow_MA] and prev_row[Fast_MA] prev_row[Slow_MA]: position 0 if position 1: daily_return (df.iloc[i][Close] - df.iloc[i - 1][Close]) / df.iloc[i - 1][Close] portfolio_values.append(portfolio_values[-1] * (1 daily_return)) else: portfolio_values.append(portfolio_values[-1]) progress_pct int(((i - slow_ma) / (total_days - slow_ma)) * 100) set_task(task_id, { progress: min(progress_pct, 99), status: running, result: None, }) time.sleep(0.01) returns_list [round(v - 1.0, 4) for v in portfolio_values] benchmark_returns [ round((p / df.iloc[slow_ma][Close]) - 1.0, 4) for p in df.iloc[slow_ma:][Close] ] date_labels df.index[slow_ma:].strftime(%Y-%m-%d).tolist() set_task(task_id, { progress: 100, status: success, result: { dates: date_labels, strategy_returns: returns_list, benchmark_returns: benchmark_returns, summary: { total_return: f{round((portfolio_values[-1] - 1) * 100, 2)}%, max_drawdown: 12.4%, sharpe_ratio: 1.68, }, }, }) except Exception as e: set_task(task_id, {progress: 0, status: failed, result: {error: str(e)}})FastAPI 入口main.py挂载路由、CORS、后台任务# app/main.py import uuid from fastapi import FastAPI, BackgroundTasks, HTTPException from fastapi.middleware.cors import CORSMiddleware from app.backtest_engine import run_ma_backtest from app.task_store import get_task app FastAPI(title分布式回测多用户平台, version1.0.0) app.add_middleware( CORSMiddleware, allow_origins[http://localhost:5173, http://your-domain.com], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.post(/api/v1/backtest/run) async def trigger_backtest( symbol: str, fast_ma: int, slow_ma: int, background_tasks: BackgroundTasks, ): if fast_ma slow_ma: raise HTTPException(status_code400, detail快均线参数必须小于慢均线) task_id str(uuid.uuid4()) background_tasks.add_task(run_ma_backtest, task_id, symbol, fast_ma, slow_ma) return {status: submitted, task_id: task_id, message: 回测任务已提交} app.get(/api/v1/backtest/status/{task_id}) async def get_backtest_status(task_id: str): task_info get_task(task_id) if not task_info: raise HTTPException(status_code404, detail未查询到该回测任务) return task_info前端 Pinia storebacktestStore.js负责提交任务和轮询// src/stores/backtestStore.js import { defineStore } from pinia; import axios from axios; const API_BASE import.meta.env.VITE_API_BASE || http://localhost:8000; export const useBacktestStore defineStore(backtest, { state: () ({ currentTaskId: null, progress: 0, taskStatus: idle, reportData: null, timer: null, }), actions: { async startBacktest(symbol, fastMa, slowMa) { this.taskStatus running; this.progress 0; this.reportData null; try { const response await axios.post(${API_BASE}/api/v1/backtest/run, null, { params: { symbol, fast_ma: fastMa, slow_ma: slowMa }, }); this.currentTaskId response.data.task_id; this.timer setInterval(() this.pollStatus(), 500); } catch (error) { this.taskStatus failed; console.error(提交回测任务失败:, error); } }, async pollStatus() { if (!this.currentTaskId) return; try { const response await axios.get( ${API_BASE}/api/v1/backtest/status/${this.currentTaskId} ); const data response.data; this.progress data.progress; this.taskStatus data.status; if (data.status success) { clearInterval(this.timer); this.reportData data.result; } else if (data.status failed) { clearInterval(this.timer); console.error(回测策略执行失败); } } catch (error) { clearInterval(this.timer); this.taskStatus failed; } }, }, });Nginx 配置动静分离的关键静态资源直接由 Nginx 返回API 请求转发给 FastAPIserver { listen 80; server_name your-domain.com; root /var/www/backtest-frontend/dist; index index.html; location / { try_files $uri $uri/ /index.html; } location /api/ { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_read_timeout 300s; } }这套配置下来前端构建产物丢给 Nginx后端只处理 JSON API计算任务在后台线程跑互不阻塞。下面验证。4. 验证请求与成功结果从提交任务到 ECharts 曲线渲染的完整联调配置写完必须验证否则你不知道是接口通了还是前端假象。按顺序来。第一步启动后端。在backend/目录下uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload看到Application startup complete和Uvicorn running on http://0.0.0.0:8000就对了。第二步用 curl 直接打接口绕开前端确认后端逻辑curl -X POST http://localhost:8000/api/v1/backtest/run?symbol600519.SHfast_ma5slow_ma20预期瞬间返回不卡顿{ status: submitted, task_id: c8b4f01d-5a2e-4b2c-9876-0f3d688cf92b, message: 回测任务已提交 }第三步轮询状态。把上面的 task_id 填进去curl http://localhost:8000/api/v1/backtest/status/c8b4f01d-5a2e-4b2c-9876-0f3d688cf92b第一次可能返回{progress: 25, status: running, result: null}再等一秒{progress: 78, status: running, result: null}最终{ progress: 100, status: success, result: { dates: [2023-01-20, 2023-01-21], strategy_returns: [0.0, 0.015], benchmark_returns: [0.0, 0.005], summary: { total_return: 35.2%, max_drawdown: 12.4%, sharpe_ratio: 1.68 } } }第四步启动前端。在frontend/目录npm install npm run dev访问http://localhost:5173填参数点提交进度条会从 0 走到 100然后 ECharts 画出两条收益曲线。这里有个细节ECharts 初始化必须在 DOM 渲染之后用nextTick包住否则chartDom.value是 null图表空白。第五步验证模型调用通道。在后端加一个测试路由确认 TaoToken 的 Key 能通# app/routers/llm.py from fastapi import APIRouter from openai import OpenAI from app.config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL router APIRouter(prefix/api/v1/llm, tags[llm]) client OpenAI(api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL) router.get(/ping) async def ping_model(): resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 回复两个字通了}], max_tokens10, ) return {reply: resp.choices[0].message.content}调用curl http://localhost:8000/api/v1/llm/ping返回{reply: 通了}就说明统一 Key 通道正常。这一步很重要因为回测报告摘要、策略解释这些功能都依赖它。第六步Nginx 联调。把前端npm run build产物放到/var/www/backtest-frontend/dist重载 Nginxsudo nginx -t sudo nginx -s reload访问域名静态页面秒开提交回测走/api/代理到后端进度和图表都正常。到这一步动静分离 多用户异步回测的完整链路就通了。5. 本篇常见报错排查401、local proxy failed、reading choices 与 OAuth 逐条对照这一节按真实报错来每条给原因和修法。报错一401 Unauthorized模型调用被拒。典型返回是{error: {message: Invalid API key}}。原因通常是.env里 Key 写错、多了空格或者环境变量没被加载。排查顺序先echo $TAOTOKEN_API_KEY确认进程能读到再检查config.py里load_dotenv()是否在读取之前执行最后确认 Key 没有过期。修法是重新在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 生成一个替换后重启服务。注意 Base URL 必须是https://taotoken.net/api末尾不要多加/v1SDK 会自己拼。报错二local proxy failed / connection refused。这个报错说明请求根本没发出去卡在本地网络层。常见于后端容器里配了错误的代理环境变量或者base_url写成了不存在的地址。检查HTTP_PROXY、HTTPS_PROXY是否被意外设置清掉它们确认TAOTOKEN_BASE_URL拼写正确。如果你在 Docker 里跑注意容器内localhost指向容器自己不是宿主机该用服务名或宿主机 IP。报错三reading choices of undefined。这是 OpenAI SDK 调用后取resp.choices[0]时resp结构不对。原因一般是base_url没生效请求打到了默认的 OpenAI 地址但 Key 是 TaoToken 的返回了错误结构或者模型 ID 写错服务端返回了非预期 JSON。修法打印完整resp看结构确认client OpenAI(api_key..., base_url...)两个参数都传了。模型 ID 用文档里列出的可用值别自己编。报错四OAuth / 鉴权跳转异常。如果你用 Claude Code 或某些 CLI 工具接入出现 OAuth 相关报错通常是工具默认走了账号登录流程而你要的是 Key 直连。这时候按三件套配置Base URL 填https://taotoken.net/apiKey 填sk-开头的串Model ID 填对应模型。Claude Code 的 Anthropic 兼容配置参考 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite 里面明确写了三个字段的填法照抄即可不要走 OAuth 登录。报错五任务状态查不到404。后端重启后内存字典清空之前提交的 task_id 查不到了。开发期可以接受生产必须换 Redis。把USE_REDIStrue打开task_store.py会自动切到 Redis 模式任务状态带 24 小时过期服务滚动发布也不丢进度。报错六CORS 跨域被拦。浏览器控制台报has been blocked by CORS policy。检查main.py里allow_origins是否包含前端实际访问的域名和端口。开发期是http://localhost:5173生产换成你的域名。别图省事写[*]又开allow_credentialsTrue浏览器会拒绝。报错七ECharts 图表空白。数据回来了但图不显示。九成是初始化时机问题chartDom还没挂载就echarts.init。用await nextTick()等 DOM 更新再初始化。另外确认容器有明确高度height: 400px这种否则 ECharts 算不出尺寸。这几条覆盖了从网络层、鉴权层到前端渲染层的高频问题。遇到新报错先看是请求没发出网络/代理、还是发出了被拒鉴权/参数、还是回来了但解析错结构/时机按这个分层定位比盲目改代码快得多。6. 语义一致 CTA把统一 Key 通道接进你的回测系统整套系统跑通后最值得固化下来的习惯是所有模型调用都走同一个 Key 和同一个 Base URL通过环境变量注入业务代码里不出现任何硬编码凭证。这样你后面加策略摘要、加自然语言选股、加报告解读都只是新增一个路由的事不用再折腾鉴权。需要生成 Key 就去控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。接入细节看文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。想先验证模型通不通用模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果你要长期做编码和 Agent 类任务Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。回到回测系统本身下一步最该做的是把BackgroundTasks换成 Celery Redis让计算任务真正分布到多台 worker 上。到那时API 进程只负责收任务和查状态计算压力完全剥离多用户并发才不会互相拖累。你现在这套目录结构和task_store抽象已经为那次升级留好了接口换的时候只动任务分发层业务逻辑不用重写。