自动运行的检查
测试只有在每次都由你的记忆之外的东西来运行时才能保护你。持续集成（CI）是一台服务器：每次 push 和 pull request 时，它在干净的环境中检出代码，并运行你定义的检查：lint、类型检查、测试、一个小型回测。任何一项失败，这次修改就会被拦下。持续交付/部署（CD）更进一步：当主分支上的检查通过后，新版本会被自动构建并发布。在量化工作中，风险是具体的：一次悄悄让定价结果偏移一个基点的重构，应当在到达投资组合之前被机器发现。
# A test matrix: every operating system with every Python version, minus the combinations you skip
from itertools import product
os_list = ["linux", "macos", "windows"]
py_versions = ["3.11", "3.12"]
exclude = {("windows", "3.11")}
jobs = [(os_name, version) for os_name, version in product(os_list, py_versions) if (os_name, version) not in exclude]
print(len(jobs)) # 5
print(jobs[0], jobs[-1]) # ('linux', '3.11') ('windows', '3.12')量化流水线检查什么
典型的流水线有会快速失败的阶段：lint（风格和明显错误）与类型检查只需几秒，然后是单元测试，再是较慢的检查，比如与已存结果对比的小型回归回测。让检查具有确定性：固定随机种子，不用网络，不依赖今天的日期，并使用提交到仓库的小快照数据而不是实时数据流。在你声称支持的 Python 版本和操作系统的矩阵上运行它们。为了让反馈更快，只运行某次修改可能影响到的测试（测试选择），在无法判断时安全地回退到整套测试。
# Test selection: run only the tests a change can affect; an unknown file means "run everything"
def select_tests(changed_files, mapping):
chosen = set()
for path in changed_files:
matches = [tests for prefix, tests in mapping.items() if path.startswith(prefix)]
if not matches:
return sorted({test for tests in mapping.values() for test in tests})
for tests in matches:
chosen.update(tests)
return sorted(chosen)
mapping = {"src/pricing/": ["tests/test_bs.py", "tests/test_greeks.py"], "src/risk/": ["tests/test_var.py"]}
print(select_tests(["src/risk/var.py"], mapping)) # ['tests/test_var.py']
print(select_tests(["src/risk/var.py", "setup.cfg"], mapping)) # all three tests: setup.cfg could affect anything关卡、密钥与工作流文件
流水线用保存在仓库里的文件来描述，例如 GitHub Actions 的 .github/workflows/ci.yml：何时运行（on:）、有哪些任务、矩阵和步骤，每个步骤要么 uses: 一个现成的 action，要么 run: 一条命令。把这个文件当作代码，像其他代码一样评审。增加部署关卡：只有 main 分支才部署，并且只有当每个必需任务都成功时才部署（被跳过或缺失的任务不算成功）。保护分支，使其在检查未通过时无法合并。API 密钥和密码放在 CI 系统的机密（secrets）里，绝不要放进仓库，并且只给每个任务它需要的权限。先发布到模拟交易或预发布环境。
# The deploy gate: main branch only, and every required job must have succeeded
def can_deploy(results, required, branch):
return branch == "main" and all(results.get(job) == "success" for job in required)
print(can_deploy({"lint": "success", "tests": "success"}, ["lint", "tests"], "main")) # True
print(can_deploy({"lint": "success", "tests": "skipped"}, ["lint", "tests"], "main")) # False skipped is not success
print(can_deploy({"lint": "success"}, ["lint", "tests"], "main")) # False a missing job is not success
print(can_deploy({"lint": "success", "tests": "success"}, ["lint", "tests"], "dev")) # False only main deploys# The workflow file is plain text, so it can be generated and checked like any other output
def workflow_yaml(name, python_versions, command):
versions = ", ".join(f'"{version}"' for version in python_versions)
lines = [f"name: {name}", "on: [push, pull_request]", "jobs:", " test:", " runs-on: ubuntu-latest",
" strategy:", " matrix:", f" python-version: [{versions}]", " steps:",
" - uses: actions/checkout@v4", " - uses: actions/setup-python@v5", " with:",
" python-version: ${{ matrix.python-version }}", " - run: pip install -e .[test]",
f" - run: {command}"]
return "\n".join(lines) + "\n"
text = workflow_yaml("ci", ["3.11", "3.12"], "pytest -q")
print(text.splitlines()[7]) # python-version: ["3.11", "3.12"]
print(text.splitlines()[-1]) # - run: pytest -q流水线通过的价值取决于测试本身CI 证明的是检查通过了，而不是策略好。让测试套件保持快速，大家才会运行它；不稳定的测试要么修好要么删掉（随机失败的测试会教会所有人无视红灯）；构建失败要作为团队的首要任务。固定 CI 中使用的工具版本，并保留一种在自己机器上运行完全相同检查的方法，这样“我这里能跑”和“CI 里失败”就永远不会成为谜团。
- CI 在每次 push 时于干净环境中运行 lint、类型检查和测试；只有在
main上通过后，CD 才自动发布。 - 让检查具有确定性（种子、快照、无网络），在版本与系统矩阵上运行，并按改动选择测试。
- 把流水线写进经过评审的工作流文件，用
main和必需任务给部署设关卡，把机密放在 CI 系统里，并修复不稳定的测试。
下一节：定时流水线，在正确的时间、以正确的顺序运行数据和交易任务。
实验：评分内容
- 用
matrix_jobs展开测试矩阵25 分 - 用
select_tests只运行相关测试25 分 - 用
can_deploy守住发布关口25 分 - 用
workflow_yaml生成工作流25 分