代码能跑和系统能跑,是两件事

凌晨两点,老张被飞书消息炸醒。

他睡眼惺忪地摸出手机,看到的是合伙人的一连串感叹号:回测系统崩了,数据对不上,净值曲线全是尖刺——过去三周的历史数据全废。那是老李刚同步过去的,他用自己本地的 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,合并前在本地跑通 pytest
  • dev 分支每周合并一次到 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.yamlrequirements.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。轮换流程:

  1. 在 TickDB 控制台生成新 Key
  2. 更新服务器环境变量(不动代码)
  3. 观察 24 小时,确认所有服务正常
  4. 吊销旧 Key
  5. 团队内部通报(不含 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 人,正在用「本地文件夹 + 微信传文件」的方式协作

  1. 在 TickDB 控制台生成一个只读 API Key
  2. Clone 你们的代码仓库,按本文的结构改造 data/
  3. 配置 setup.sh 和 pre-commit,让新成员 5 分钟能跑起来

如果你的团队已经有基础协作流程,但 API Key 管理混乱

  1. 审计现有代码仓库,确认没有任何 Key 硬编码(使用 git grep 搜索 sk-api_key
  2. 为每个成员生成独立的 Key,按用途分配权限
  3. 将 Key 管理规范写入 docs/setup.md

如果你习惯用 AI 辅助开发,在 AI 助手中搜索安装 tickdb-market-data SKILL,可以让 AI 直接帮你写数据获取代码,降低从零搭建数据层的门槛。


本文不构成任何投资建议。市场有风险,投资需谨慎。