企业官网建设流程全解析

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基址是 https://taotoken.net/api 。

你需要先拿到一个可用的 Key。进入控制台创建：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，然后在 API Keys 页面生成：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。生成后复制那串sk-开头的字符串，后面所有配置都用它。

这里要强调一个工程习惯：不要把 Key 硬编码进main.py。用环境变量或.env文件管理，.env加进.gitignore。多用户系统里，Key 泄露等于账单失控。下面给出后端读取 Key 的标准写法，配合python-dotenv：

# .env 文件，放在后端项目根目录 TAOTOKEN_API_KEY=sk-你的实际Key TAOTOKEN_BASE_URL=https://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_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。想先验证模型是否通，直接用模型对话页：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面写了 OpenAI 兼容的调用格式。也就是说，你后端用openai这个 Python SDK，把base_url指向 TaoToken 的 API 地址，就能直接调，不用改业务代码结构。这一点对回测系统很关键——模型调用层和回测计算层解耦，后面换模型只改配置。

如果你用 Claude Code 做开发辅助，Anthropic 兼容通道的配置页在 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode-anthropic&utm_campaign=rewrite ，里面给了 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，版本锁死避免踩兼容坑：

fastapi==0.103.1 uvicorn[standard]==0.23.2 pydantic==2.3.0 pandas==2.1.0 numpy==1.25.2 python-dotenv==1.0.0 openai==1.3.0 redis==5.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( host=os.getenv("REDIS_HOST", "127.0.0.1"), port=int(os.getenv("REDIS_PORT", 6379)), decode_responses=True, ) def set_task(task_id: str, payload: dict, ttl: int = 86400): _client.setex(f"backtest:task:{task_id}", ttl, json.dumps(payload)) def get_task(task_id: str): raw = _client.get(f"backtest: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(start="2023-01-01", periods=200) price_changes = np.random.normal(loc=0.0005, scale=0.015, size=200) close_prices = 100 * np.exp(np.cumsum(price_changes)) df = pd.DataFrame({"Close": close_prices}, index=dates) df["Fast_MA"] = df["Close"].rolling(window=fast_ma).mean() df["Slow_MA"] = df["Close"].rolling(window=slow_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="分布式回测多用户平台", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://your-domain.com"], allow_credentials=True, 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_code=400, 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_code=404, 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 请求转发给 FastAPI：

server { 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?symbol=600519.SH&fast_ma=5&slow_ma=20"

预期瞬间返回，不卡顿：

{ "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_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL) @router.get("/ping") async def ping_model(): resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "回复两个字：通了"}], max_tokens=10, ) 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，重载 Nginx：

sudo 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_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 生成一个，替换后重启服务。注意 Base URL 必须是https://taotoken.net/api，末尾不要多加/v1，SDK 会自己拼。

报错二：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/api，Key 填sk-开头的串，Model ID 填对应模型。Claude Code 的 Anthropic 兼容配置参考 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode-anthropic&utm_campaign=rewrite ，里面明确写了三个字段的填法，照抄即可，不要走 OAuth 登录。

报错五：任务状态查不到，404。后端重启后内存字典清空，之前提交的 task_id 查不到了。开发期可以接受，生产必须换 Redis。把USE_REDIS=true打开，task_store.py会自动切到 Redis 模式，任务状态带 24 小时过期，服务滚动发布也不丢进度。

报错六：CORS 跨域被拦。浏览器控制台报has been blocked by CORS policy。检查main.py里allow_origins是否包含前端实际访问的域名和端口。开发期是http://localhost:5173，生产换成你的域名。别图省事写["*"]又开allow_credentials=True，浏览器会拒绝。

报错七：ECharts 图表空白。数据回来了但图不显示。九成是初始化时机问题，chartDom还没挂载就echarts.init。用await nextTick()等 DOM 更新，再初始化。另外确认容器有明确高度，height: 400px这种，否则 ECharts 算不出尺寸。

这几条覆盖了从网络层、鉴权层到前端渲染层的高频问题。遇到新报错，先看是请求没发出（网络/代理）、还是发出了被拒（鉴权/参数）、还是回来了但解析错（结构/时机），按这个分层定位，比盲目改代码快得多。

6. 语义一致 CTA：把统一 Key 通道接进你的回测系统

整套系统跑通后，最值得固化下来的习惯是：所有模型调用都走同一个 Key 和同一个 Base URL，通过环境变量注入，业务代码里不出现任何硬编码凭证。这样你后面加策略摘要、加自然语言选股、加报告解读，都只是新增一个路由的事，不用再折腾鉴权。

需要生成 Key 就去控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，Key 管理在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。接入细节看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。想先验证模型通不通，用模型对话页：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。如果你要长期做编码和 Agent 类任务，Coding Plan 在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。

回到回测系统本身，下一步最该做的是把BackgroundTasks换成 Celery + Redis，让计算任务真正分布到多台 worker 上。到那时，API 进程只负责收任务和查状态，计算压力完全剥离，多用户并发才不会互相拖累。你现在这套目录结构和task_store抽象，已经为那次升级留好了接口，换的时候只动任务分发层，业务逻辑不用重写。