自動で走るチェック
テストは、あなたの記憶ではない何かによって毎回実行されて初めて、あなたを守ります。継続的インテグレーション(CI)は、push やプルリクエストのたびにクリーンな環境でコードをチェックアウトし、定義したチェックを実行するサーバーです。lint、型チェック、テスト、小さなバックテストなどです。何かが失敗すれば、その変更はブロックされます。継続的デリバリー/デプロイ(CD)はさらに進み、メインブランチでチェックが通ると新しいバージョンが自動的にビルド・公開されます。クオンツの仕事では、リスクは具体的です。価格を静かに1ベーシスポイントずらすリファクタリングは、ポートフォリオに届く前に機械が捕まえるべきです。
# 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 のバージョンと OS のマトリクスで実行します。フィードバックを速くするには、変更が影響しうるテストだけを実行し(テスト選択)、判断できないときは安全のためスイート全体に戻ります。
# 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: するか、コマンドを run: します。このファイルもコードとして扱い、他と同様にレビューします。デプロイゲートを設けます。デプロイするのは main ブランチだけで、必須のジョブがすべて成功したときだけです(スキップされたり存在しないジョブは成功ではありません)。チェックを通らないとマージできないようにブランチを保護します。API キーやパスワードは CI システムのシークレットに置き、決してリポジトリには置かず、各ジョブには必要な権限だけを与えます。まずペーパートレードやステージング環境にリリースします。
# 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 は push のたびにクリーンな環境で lint・型チェック・テストを走らせ、CD は
mainで通ったときだけ自動でリリースする - チェックは決定的に(シード、スナップショット、ネットワークなし)。バージョンと OS のマトリクスで実行し、変更内容に応じてテストを選ぶ
- パイプラインはレビューされるワークフローファイルに記述し、デプロイは
mainと必須ジョブでゲートし、シークレットは CI システムに置き、不安定なテストは直す
次はスケジュール実行されるパイプラインです。データ処理やトレーディングのジョブを適切な時刻に適切な順序で実行します。
ラボ:採点される内容
matrix_jobsでテストマトリクスを展開する25点select_testsで必要なテストだけ実行する25点can_deployでリリースを守る25点workflow_yamlでワークフローを生成する25点