代码能跑和系统能跑,是两件事
凌晨两点,老张被飞书消息炸醒。
他睡眼惺忪地摸出手机,看到的是合伙人的一连串感叹号:回测系统崩了,数据对不上,净值曲线全是尖刺——过去三周的历史数据全废。那是老李刚同步过去的,他用自己本地的 MySQL 导了一份,结果字段类型不兼容,timestamps 全乱了。
老张坐起来,叹了口气,打开电脑。
这不是技术问题。这是协作问题。
三个人的量化团队,技术栈可能不差——Python、Backtrader、FastAPI,甚至可能已经在用 TickDB 的历史 K 线接口做过几轮像样的回测。但从「个人的回测代码能跑」到「团队的生产系统稳定运行」,中间隔着的不是代码,是一整套基础设施协作规范。
数据怎么共享、怎么版本化、怎么保证三个人拉下来都是同一套?API Key 怎么传、怎么轮换、怎么防止某人的本地脚本里裸写了个明文 key 提交到 GitHub?权限怎么分——实习生能看到多少数据?风控模块该由谁review?
这些问题,单打独斗时永远不会出现。三个人一开始分头写代码,各自本地跑通,然后开始拼,才发现:数据对不上,代码合并一团糟,key 管理是裸的,权限是一刀切的。
本文拆解这三个问题的系统性解法:共享数据基础设施、Git 协作流程、API Key 管理,以及权限控制模型。覆盖从「本地能跑」到「团队能用」的全链路,给出一套可落地的生产级方案。
一、为什么 3 人团队的协作复杂度不是线性的
一个常见的误解是:3 人团队不过是 1 人团队的 3 倍难度。
不是的。复杂度是 O(n²) 级别的。
1 个人的时候,所有数据都在本地,API Key 写在 .env 里没人管,Git 只有一个分支,本地环境乱了重装 conda 就完事。
3 个人的时候,每个人本地环境可能都不一样(macOS / Ubuntu / Windows,Python 3.9 / 3.11 / 3.12),每个人同步数据的方式可能都不一样(老张用 Dropbox,老李用硬盘拷贝,新来的实习生不知道该同步哪个目录),API Key 存在不同地方(老张在 .env,老李在代码里,老李还把它提交到了某个公开仓库)。
这不是 3 倍的复杂度。这是 9 种潜在冲突。
量化的特殊性在于:数据是策略的起点,数据错了,后面的所有计算全废。而数据出错的原因,90% 不是存储引擎的 bug,而是「谁在什么时间用什么格式同步了什么数据」——一个纯粹的协作问题。
所以小型量化团队的基础设施建设,本质上是解决三个约束:
- 数据一致性:三个人看到的是同一套数据
- 代码可复现:任何人 clone 下来能跑通,不依赖「我记得装什么包」
- 协作安全性:API Key 不泄露,数据访问按角色分级
下面逐条拆解。
二、共享数据基础设施:让三个人拉下来都是同一套
2.1 共享数据的四种方案对比
小型量化团队在数据共享上有四条路:
| 方案 | 优点 | 致命缺陷 | 适用场景 |
|---|---|---|---|
| 局域网文件共享(SMB/NFS) | 简单,无需额外服务 | 无法跨地域,文件锁冲突,单点故障 | 纯本地、同城市团队 |
| 云对象存储(AWS S3 / 阿里云 OSS) | 跨地域,强一致性,版本控制 | 需要配置 IAM,有学习成本 | 有云基础设施的团队 |
| 专业数据 API(TickDB 等) | 统一接入,无需同步,实时+历史一体化 | 成本,数据版权 | 需要多市场数据的团队 |
| 自建 PostgreSQL + TimescaleDB | 完全可控,支持 SQL 查询 | 运维成本,冷数据清理需自行实现 | 有专职运维的团队 |
对于 3 人量化团队,推荐组合方案:TickDB 作为主力数据源(覆盖美股、港股、数字货币的历史 K 线 + 实时行情)+ 云对象存储作为本地缓存层。
这样设计的逻辑是:TickDB 负责「获取最新数据和长周期历史数据」,S3 负责「缓存每个人跑过的本地中间结果,避免重复拉取 API」。
2.2 统一数据层的设计
不要让每个人各自写数据拉取代码。应该在项目里建一个统一的 data/ 层,所有策略代码都从这里取数:
quant-system/
├── config/
│ ├── config.py # 全局配置,读取环境变量
│ └── symbols.yaml # 交易标的列表
├── data/
│ ├── fetcher.py # 统一数据获取层
│ ├── cache.py # S3 本地缓存管理
│ └── schemas.py # 数据schema定义
├── strategies/
├── backtest/
├── .env.example # 环境变量模板
└── requirements.txt
fetcher.py 的核心逻辑:优先从本地缓存(S3)读取,缓存未命中则调用 TickDB API 拉取,并写入缓存:
import os
import json
import boto3
import requests
from datetime import datetime, timedelta
from pathlib import Path
class DataFetcher:
"""
统一数据获取层。
读取顺序:本地缓存 → TickDB API → 报错。
缓存命中时直接返回,不调用 API,节省 rate limit。
"""
def __init__(self, api_key: str, cache_bucket: str = "quant-team-cache"):
self.api_key = api_key
self.cache_bucket = cache_bucket
self.s3 = boto3.client(
"s3",
aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
)
self.base_url = "https://api.tickdb.ai/v1/market"
def _cache_key(self, symbol: str, interval: str, start: str, end: str) -> str:
"""生成缓存 key:symbol/interval/start_end.json"""
return f"kline/{symbol}/{interval}/{start}_{end}.json"
def _read_cache(self, cache_key: str) -> dict | None:
"""从 S3 读取缓存,命中则返回数据,否则返回 None"""
local_cache = Path(f"/tmp/{cache_key.replace('/', '_')}")
if local_cache.exists():
# 本地缓存优先,避免每次都调 S3
with open(local_cache, "r") as f:
return json.load(f)
try:
self.s3.download_file(self.cache_bucket, cache_key, str(local_cache))
with open(local_cache, "r") as f:
return json.load(f)
except self.s3.exceptions.NoSuchKey:
return None
def _write_cache(self, cache_key: str, data: dict) -> None:
"""写入 S3 缓存,写入本地临时目录以加速下次读取"""
local_cache = Path(f"/tmp/{cache_key.replace('/', '_')}")
with open(local_cache, "w") as f:
json.dump(data, f)
self.s3.upload_file(str(local_cache), self.cache_bucket, cache_key)
def get_kline(
self,
symbol: str,
interval: str = "1h",
start: str | None = None,
end: str | None = None,
limit: int = 1000,
) -> dict:
"""
获取 K 线数据。
自动走缓存,API 未命中时才调用 TickDB。
"""
if end is None:
end = datetime.now().strftime("%Y-%m-%d")
if start is None:
start = (datetime.now() - timedelta(days=365)).strftime("%Y-%m-%d")
cache_key = self._cache_key(symbol, interval, start, end)
cached = self._read_cache(cache_key)
if cached is not None:
return cached
# ⚠️ 未命中缓存,调用 API
headers = {"X-API-Key": self.api_key}
params = {"symbol": symbol, "interval": interval, "start": start, "end": end, "limit": limit}
response = requests.get(
f"{self.base_url}/kline",
headers=headers,
params=params,
timeout=(3.05, 10),
)
if response.status_code == 200:
data = response.json()
self._write_cache(cache_key, data)
return data
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
raise RuntimeError(f"API 限频,等待 {retry_after} 秒后重试")
else:
raise RuntimeError(f"K线数据获取失败: {response.status_code} {response.text}")
关键设计点:
- 缓存 key 包含时间范围,不同时间范围的请求不会互相覆盖
- 本地临时目录 + S3 双写,本地缓存命中时完全不调用 API,保护 rate limit
- 超时设置
(3.05, 10)防止请求hang死 - 每个人 clone 项目后,只需要配置自己的
.env,数据获取逻辑完全一致
2.3 配置文件管理
所有环境相关的配置通过 .env 文件注入,.env.example 提交到 Git 仓库作为模板:
# .env.example(提交到 Git)
TICKDB_API_KEY=your_api_key_here
AWS_ACCESS_KEY_ID=your_aws_key
AWS_SECRET_ACCESS_KEY=your_aws_secret
S3_BUCKET=quant-team-cache
LOG_LEVEL=INFO
# .gitignore(不提交 .env)
.env
__pycache__/
*.pyc
data/raw/
data/cache/
*.log
这样任何人 clone 项目后,运行流程是:
cp .env.example .env
# 编辑 .env,填入自己的 key
pip install -r requirements.txt
python data/fetcher.py # 验证数据层是否正常
数据层一致性得到保证:三个人跑的是同一套代码、同一个缓存逻辑、同一个 API 调用路径。
三、Git 协作流程:让合并不再是噩梦
3.1 分支模型设计
3 人量化团队的 Git 分支模型不需要 GitFlow 那么重,但需要一个清晰的约定:
main # 生产代码,任何人不能直接 push
├── dev # 集成分支,所有人 merge 到这里
├── strategy/ # 每人一个 strategy/xxx 分支,策略开发
│ ├── strategy/zhang # 老张的因子研究
│ ├── strategy/li # 老李的信号逻辑
│ └── strategy/intern # 实习生的数据清洗
└── infra/ # 基础设施分支,修改 data/、config/ 需要 review
约定:
infra/分支的合并需要至少 1 人 review,防止单人不经意改坏数据层strategy/分支可以自由 push,合并前在本地跑通pytestdev分支每周合并一次到main,合并前运行完整的回测验证
3.2 Commit 规范:让代码审查有据可查
使用 Conventional Commits 规范,格式:type(scope): description
feat(data): 添加 S3 缓存层,减少 API 调用
fix(strategy): 修正移动止损的边界条件
refactor(fetcher): 将 API Key 从参数改为环境变量注入
docs(readme): 更新本地开发环境配置说明
perf(backtest): 向量化因子计算,耗时从 45s 降至 8s
对于量化项目,还有一个有价值的 commit 类型:
data: 补充 2020-2022 数字货币 1h K 线,修复周末缺失数据
这条 commit 记录的价值:当某天回测结果突然变了,首先查 data 相关的 commit——很可能是某人补充了历史数据,导致样本量变化。不用花两小时 debug 因子代码。
3.3 本地 pre-commit 规范
在 .pre-commit-config.yaml 中配置自动检查,防止低质量代码进入仓库:
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
args: ['--maxkb=5000'] # 数据文件超过 5MB 禁止提交
- id: detect-secrets # ⚠️ 防止 API Key 意外提交
args: ['--baseline', '.secrets.baseline']
- repo: https://github.com/psf/black
rev: 23.12.1
hooks:
- id: black
language_version: python3.11
- repo: https://github.com/pycqa/isort
rev: 5.13.2
hooks:
- id: isort
detect-secrets 是关键:如果有人把 TICKDB_API_KEY=sk-xxxx 写在代码里而不是 .env,pre-commit 会直接拒绝提交。这比事后发现 GitHub 已经通知你 key 泄露了要安全得多。
3.4 合并冲突的预防策略
量化项目中 merge 冲突的高发区是 config/symbols.yaml 和 requirements.txt。解决方案:按职责拆分配置文件,让「每个人改自己的东西」成为 Git 的天然约束:
config/
├── symbols.yaml # 老张负责,维护交易标的列表
├── risk.yaml # 老李负责,维护风控参数
└── schedule.yaml # 实习生可改,维护回测时间窗口
改到同一个文件的概率大幅降低。
四、API Key 管理:安全与效率的平衡
4.1 分层 Key 策略
3 人团队不建议共用一个 API Key。TickDB 支持按用途生成多个 Key,建议分配策略:
| Key 用途 | 持有者 | 权限范围 | 备注 |
|---|---|---|---|
| 生产数据 Key | 老张(技术负责人) | 全量数据读写 | 只在服务器端使用,不出现在代码仓库 |
| 回测 Key | 老李 | 只读 + 高频限制 | 专门用于回测系统的批量查询 |
| 研究 Key | 实习生/研究岗 | 只读 + 低频限制 | 可以开放给研究环境 |
不要让任何人在本地代码里写死 Key。正确做法:
# ✅ 正确:从环境变量读取
api_key = os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("环境变量 TICKDB_API_KEY 未设置,请检查 .env 文件")
# ❌ 错误:硬编码 Key
api_key = "sk-abc123xxxx" # 这行提交到 GitHub 就是安全事故
4.2 本地开发环境的安全配置
团队成员本地开发时,使用 1Password / Bitwarden 等密码管理器管理 .env 文件的 Key 值,而不是通过微信、Slack 传输 Key。
建议在项目的 docs/setup.md 中写清楚本地配置流程:
## 本地开发环境配置
1. 在 TickDB 控制台生成一个新的 API Key(不要用生产 Key)
2. 将 Key 存入本地密码管理器
3. 复制配置文件:`cp .env.example .env`
4. 在 `.env` 中填入你的 Key
5. 验证配置:`python -c "import os; print('OK' if os.environ.get('TICKDB_API_KEY') else 'MISSING')"`
⚠️ 永远不要通过聊天工具传输 API Key。
4.3 生产环境的 Key 轮换机制
每 90 天轮换一次生产 Key。轮换流程:
- 在 TickDB 控制台生成新 Key
- 更新服务器环境变量(不动代码)
- 观察 24 小时,确认所有服务正常
- 吊销旧 Key
- 团队内部通报(不含 Key 值,只说「已轮换,请确认本地 .env 是否更新」)
如果团队成员超过 3 人(或者有人离职),立即轮换 Key,而不是等到 90 天周期。
五、权限控制:不是信任问题,是风险管理
5.1 角色权限矩阵
| 操作 | 技术负责人(老张) | 策略开发(老李) | 研究实习生 |
|---|---|---|---|
修改 config/ 基础设施代码 |
✅ Review required | ❌ | ❌ |
修改 strategy/ 策略代码 |
✅ Review required | ✅ Self-review | ✅ Self-review |
修改 data/ 数据层代码 |
✅ Review required | ❌ | ❌ |
修改风控参数 risk.yaml |
✅ 独立审批 | ❌ | ❌ |
| 拉取 TickDB 历史数据 | ✅ | ✅ | ✅(限额) |
| 访问 S3 完整缓存 | ✅ | ✅(只读) | ❌ |
| 修改 S3 缓存 | ✅ | ❌ | ❌ |
合并到 main 分支 |
✅ | ❌ | ❌ |
关键原则:风控参数和基础设施代码的修改权必须集中,不能因为「大家都是合伙人」就开放所有权限。历史上无数量化灾难的源头,是「我只是想改一个小参数」。
5.2 S3 权限配置(IAM Policy)
使用 AWS IAM 控制各角色的 S3 访问范围:
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "技术负责人-完全控制",
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789:user/zhang"},
"Action": ["s3:*"],
"Resource": [
"arn:aws:s3:::quant-team-cache",
"arn:aws:s3:::quant-team-cache/*"
]
},
{
"Sid": "策略开发-只读缓存",
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789:user/li"},
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": [
"arn:aws:s3:::quant-team-cache",
"arn:aws:s3:::quant-team-cache/*"
]
},
{
"Sid": "研究实习生-只能读不能写",
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789:user/intern"},
"Action": ["s3:GetObject"],
"Resource": [
"arn:aws:s3:::quant-team-cache/data/approved/*"
]
}
]
}
实习生只能读到 data/approved/ 目录,无法写入任何缓存,也看不到其他中间结果。这是权限控制的意义:即使团队成员完全可信,系统设计也应该让「误操作」的后果最小化。
六、部署架构:让这套流程真正跑起来
6.1 开发环境 vs 生产环境分离
| 环境 | 运行位置 | 数据源 | 目的 |
|---|---|---|---|
| 开发环境 | 每人的本地机器 | TickDB + 本地缓存 | 因子研究、策略开发 |
| 回测环境 | 共享服务器(最低配云服务器) | TickDB 历史数据 | 全量回测、样本外验证 |
| 生产环境 | 独立服务器 | TickDB 实时数据 | 模拟交易、实盘监控 |
开发环境和生产环境的数据源完全隔离:本地跑回测用历史数据接口 /v1/market/kline,实盘监控用实时数据接口(WebSocket)。这个区分非常重要——有人在开发环境里用实时数据接口跑历史回测,数据是不完整的,结果自然失真。
6.2 一键环境重建
项目根目录的 setup.sh 让任何人在 5 分钟内从零搭建完整开发环境:
#!/bin/bash
# setup.sh - 一键重建开发环境
set -e
echo "=== TickDB 量化团队开发环境初始化 ==="
# 1. 检查环境变量
if [ -z "$TICKDB_API_KEY" ]; then
echo "❌ 错误:TICKDB_API_KEY 未设置"
echo "请复制 .env.example 为 .env 并填入 API Key"
exit 1
fi
# 2. 创建虚拟环境
python3.11 -m venv venv
source venv/bin/activate
# 3. 安装依赖
pip install --upgrade pip
pip install -r requirements.txt
# 4. 验证数据层
python -c "
from data.fetcher import DataFetcher
import os
fetcher = DataFetcher(api_key=os.environ['TICKDB_API_KEY'])
result = fetcher.get_kline('BTC.USDT', interval='1h', limit=10)
print(f'✅ 数据层验证成功:获取到 {len(result.get(\"data\", []))} 条 K 线')
"
# 5. 初始化 pre-commit
pip install pre-commit
pre-commit install
echo "=== 环境初始化完成 ==="
echo "运行 'source venv/bin/activate' 激活虚拟环境"
这条脚本的价值:当新成员加入,或者重装系统,不需要「找老张问这个怎么配」。任何人 clone 下来,两条命令搞定。
七、避坑清单:3 人团队最容易犯的错误
错误一:「我本地能跑就行」
这是所有协作灾难的起点。团队必须从第一天就约定:代码只有 push 到 dev 分支才算「能跑」,本地跑通不算数。这个约定能倒逼每个人在提交前验证代码在干净环境下的可复现性。
错误二:数据直接用 Dropbox / 微信传输
「我用微信给你发了一份数据」——这句话是量化团队数据事故的标准句式。数据文件没有版本,文件名没有规范,传过来的是哪天的数据谁也不知道。统一走 TickDB API + S3 缓存,数据来源永远可追溯。
错误三:所有人共用一个分支
3 个人在同一个分支上开发,不做分支隔离,最终结果一定是「我的代码被你的覆盖了」或者「不敢合并,因为合并太痛苦」。分支模型再简单也要有,否则协作效率会低到让团队放弃 Git。
错误四:把 API Key 写在代码注释里
这个不用多说了。pre-commit 配 detect-secrets,环境变量强制检查,从源头杜绝。
错误五:风控参数由策略开发者自己改
这是最危险的。策略开发者同时掌握风控参数,理论上可以在回测里把止损设得很宽来刷收益率。风控参数的修改必须走独立审批流程,代码层面也需要 review 机制。
八、结语
回到开头老张被飞书炸醒的故事。
那个凌晨,他花了两个小时定位数据不一致的原因,发现是老李上周五同步的本地 MySQL 导出文件里,timestamp 字段用了本地时区而不是 UTC。第二天老张的服务器跑了 UTC,两个文件 JOIN 在一起,时间戳差了 8 小时。
两个小时。三个人的团队,每个人损失两小时,就是六个人小时的沉默成本。这还没算上「这数据我不敢信了,我要重新跑一遍」的心理损耗。
这不是技术问题。这是一次基础设施事故。解决方案不是「下次小心点」,而是建立统一的数据层、Git 协作流程、Key 管理和权限控制。
三个人的团队,技术栈不一定差,但基础设施往往是一塌糊涂的。不是因为不会,是因为没花时间建。而建这套系统的投入,实际上比大多数人想象的要小:一个统一的数据层脚本 + 一套 Git 分支约定 + 一份 Key 管理规范 + 一个 setup.sh 脚本,四样东西,加起来两三天能搭完。
搭完之后,你会发现:团队里多一个人,少一个人;有人离职,有人加入;项目跑三个月,跑了三年——系统还在,数据还在,代码还在可复现的状态里。
这才是生产级的基础设施。
下一步行动
如果你的团队是 1-3 人,正在用「本地文件夹 + 微信传文件」的方式协作:
- 在 TickDB 控制台生成一个只读 API Key
- Clone 你们的代码仓库,按本文的结构改造
data/层 - 配置
setup.sh和 pre-commit,让新成员 5 分钟能跑起来
如果你的团队已经有基础协作流程,但 API Key 管理混乱:
- 审计现有代码仓库,确认没有任何 Key 硬编码(使用
git grep搜索sk-或api_key) - 为每个成员生成独立的 Key,按用途分配权限
- 将 Key 管理规范写入
docs/setup.md
如果你习惯用 AI 辅助开发,在 AI 助手中搜索安装 tickdb-market-data SKILL,可以让 AI 直接帮你写数据获取代码,降低从零搭建数据层的门槛。
本文不构成任何投资建议。市场有风险,投资需谨慎。