Comprobaciones que se ejecutan solas
Las pruebas solo te protegen si se ejecutan, siempre, por algo que no sea tu memoria. La integración continua (CI) es un servidor que, en cada push y pull request, descarga el código en un entorno limpio y ejecuta las comprobaciones definidas: lint, comprobación de tipos, pruebas, un pequeño backtest. Si algo falla, el cambio se bloquea. La entrega/despliegue continuo (CD) va más allá: cuando las comprobaciones pasan en la rama principal, la nueva versión se construye y se publica automáticamente. En trabajo cuantitativo lo que está en juego es concreto: una refactorización que desplaza en silencio un precio un punto básico debe ser detectada por una máquina antes de llegar a una cartera.
# 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')Qué comprueba una canalización cuantitativa
Una canalización típica tiene etapas que fallan rápido: el lint (estilo y errores evidentes) y la comprobación de tipos tardan segundos, luego las pruebas unitarias, luego comprobaciones más lentas como un pequeño backtest de regresión comparado con un resultado guardado. Haz las comprobaciones deterministas: semillas fijas, sin red, sin depender de la fecha de hoy y con datos de una pequeña instantánea versionada en lugar de un feed en vivo. Ejecútalas en una matriz de versiones de Python y sistemas que dices admitir. Para dar respuesta rápida, ejecuta solo las pruebas que un cambio puede afectar (selección de pruebas), con un respaldo seguro a toda la suite cuando no puedas saberlo.
# 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 anythingBarreras, secretos y el archivo de flujo de trabajo
El flujo se describe en un archivo guardado en el repositorio, por ejemplo .github/workflows/ci.yml para GitHub Actions: cuándo se ejecuta (on:), los trabajos, la matriz y los pasos, y cada paso uses: una acción ya hecha o run: un comando. Trata el archivo como código, revisado como cualquier otro. Añade una barrera de despliegue: solo la rama main despliega, y solo cuando todos los trabajos requeridos tuvieron éxito (un trabajo omitido o ausente no es un éxito). Protege la rama para que no se fusione sin comprobaciones superadas. Guarda las claves de API y contraseñas en los secretos del sistema de CI, nunca en el repositorio, y da a cada trabajo solo los permisos que necesita. Publica primero en un entorno de paper trading o de preproducción.
# 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- La CI ejecuta lint, comprobación de tipos y pruebas en un entorno limpio en cada push; la CD solo publica automáticamente si pasan en
main. - Haz las comprobaciones deterministas (semillas, instantáneas, sin red), ejecútalas en una matriz de versiones y sistemas y selecciona las pruebas según lo que cambió.
- Describe la canalización en un archivo de flujo revisado, condiciona los despliegues a
mainy a los trabajos requeridos, guarda los secretos en el sistema de CI y arregla las pruebas inestables.
A continuación: canalizaciones programadas, ejecutar tareas de datos y de trading en el momento y orden correctos.
Lab: qué se califica
- Expandir la matriz de pruebas con
matrix_jobs25 pts - Ejecutar solo lo que importa con
select_tests25 pts - Proteger el lanzamiento con
can_deploy25 pts - Generar el flujo de trabajo con
workflow_yaml25 pts