Los datos: seis mil millones de partidas y un recorte justificado

Qué vas a construir

El recorte de datos con el que se entrena todo el curso y, lo que importa más, la prueba de qué entró en él. El centro es una consulta SQL que se puede leer antes de descargar nada.

Lichess publica todo lo que se juega en su servidor. Los volcados mensuales, las copias en Hugging Face (Lichess/standard-chess-games, más de 6 700 millones de partidas), los seis millones de puzles y los casi 400 millones de posiciones evaluadas con Stockfish están bajo CC0CC0Dedicación al dominio público de Creative Commons: el autor renuncia a todos los derechos. Lichess publica sus partidas, puzles y evaluaciones bajo CC0, así que se pueden recortar, redistribuir y usar para entrenar sin pedir permiso (se cita por cortesía).: dominio público.

Un mes reciente ocupa unos 73 GB en parquetParquetFormato de fichero columnar y comprimido. Lichess publica sus partidas así en Hugging Face, y DuckDB puede filtrarlas en remoto leyendo solo las columnas y los bloques necesarios (predicate pushdown), lo que hace posible recortar 73 GB a unos pocos sin descargarlo todo.. No vamos a descargarlo.

La configuración: qué partidas entran en el recorte

# Two recent months of rated standard games, filtered remotely (see docs/spec/01-datos.md).
dataset: Lichess/standard-chess-games
months: ["2025-01", "2025-02"]
min_elo: 1800
min_base_seconds: 180
terminations: ["Normal", "Time forfeit"]
min_plies: 20
exclude_variants: true
out_dir: data/raw
limit: null

↗configs/data/lichess-2025-01-02.yamllíneas 1-10 · p0

El comentario de la primera línea apunta al documento de diseño de datos del repositorio de referencia. No lo tecleas y tu repositorio no lo tiene (en main ya se retiró); si quieres leerlo, está en el tag p0, o en tu terminal con git show p0:docs/spec/01-datos.md. Todo lo que decide el recorte está en la tabla de abajo.

Cada clave es una decisión que condiciona lo que el modelo va a aprender:

Ese último punto es el que evita la fugaFuga de datosCualquier camino por el que información del conjunto de validación llega al de entrenamiento. No falla: los números mejoran, y se descubre cuando el modelo sale al mundo. En Rukh se evita partiendo por mes (enero entrena; de febrero, desde M2 parte 4, validan las primeras 100 000 partidas y el resto del mes entrena), por PuzzleId con semilla en los puzles, entrenando el BPE solo con datos de enero y, en M3, repartiendo la tabla de posiciones por game_id con un CRC-32 de la semilla y el id, nunca por posición: dos posiciones consecutivas de la misma partida se diferencian en una jugada. El umbral del detector se elige además en una mitad tune y se mide en otra score, partidas también por partida. más tonta de este dominio. Si se partiera al azar, las partidas de los mismos jugadores con las mismas aperturas caerían a los dos lados, y la validación mediría “¿reconoces a este jugador?” en vez de “¿sabes jugar?”.

manifest.py: la procedencia, como un modelo

Los dos ficheros de esta lección, manifest.py y fetch.py, viven en un paquete nuevo, rukh.data. Lo primero es crearlo con su __init__.py:

"""Data pipeline: fetch, filter and describe Lichess games."""

↗src/rukh/data/__init__.pylínea 1 · p0

Solo lleva el docstring: quien use el paquete importa cada módulo por su nombre. Dentro, antes del código que descarga, va el que registra.

"""Manifest written next to fetched data: what was asked for, what came back."""

from __future__ import annotations

from datetime import UTC, datetime

from pydantic import BaseModel, ConfigDict, Field

from rukh import __version__


class FileHash(BaseModel):
    """One materialized file with its content hash and size."""

    model_config = ConfigDict(extra="forbid")

    path: str
    sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    bytes: int = Field(ge=0)


class Manifest(BaseModel):
    """Provenance record for a fetched dataset slice."""

    model_config = ConfigDict(extra="forbid")

    dataset: str
    months: list[str]
    filters: dict[str, object]
    counts: dict[str, int]
    files: list[FileHash]
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    rukh_version: str = __version__

↗src/rukh/data/manifest.pylíneas 1-33 · p0

Los tres campos del medio hacen el trabajo. filters son los parámetros con los que el paso corrió de verdad, no los que dice el README. counts es cuántas filas salieron por mes, y convierte «descartamos el bullet» en algo comprobable. files es un sha256 por salida, y hace de “el mismo dataset” una afirmación que un test puede tumbar.

El patrón del sha256 valida la forma del hash aunque lo escriba nuestro propio código: si alguien construye o “arregla” un manifiesto a mano, el modelo lo rechaza. Un hash inventado es peor que ninguno, porque miente con autoridad. Y datetime.now(UTC) lleva zona horaria porque el fichero se va a comparar entre máquinas.

Un manifiesto es la ficha de catálogo de la biblioteca: no es el libro, pero dice de dónde vino, cuándo entró y cuántas páginas tiene, y con ella puedes comprobar que el ejemplar de la estantería es el que dice ser.

fetch.py: la consulta, el plan y la descarga

Antes del código, el recorrido completo: el subcomando valida la configuración, construye un plan con la consulta dentro y, si es un ensayo, se para ahí. Si no, ejecuta ese mismo plan mes a mes y escribe el manifiesto al final.

"""Fetch a filtered slice of ``Lichess/standard-chess-games`` with DuckDB over ``hf://``.

Only the games that pass the filters are materialized locally (predicate pushdown on the
remote parquet files). The ply count (``min_plies``) is not applied here: it is enforced in P1
once ``movetext`` has been converted to UCI with python-chess, because counting plies from SAN
with comments (``%clk``, ``%eval``) in SQL is unreliable. The manifest therefore records it as
``min_plies_deferred``. Variant exclusion by ``Event`` is a coarse first pass that P1 refines.
"""

from __future__ import annotations

import hashlib
import re
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field, field_validator

from rukh import paths
from rukh.config import BaseConfig
from rukh.data.manifest import FileHash, Manifest

MONTH_RE = re.compile(r"^(\d{4})-(0[1-9]|1[0-2])$")

↗src/rukh/data/fetch.pylíneas 1-22 · p0

En el docstring vive la decisión más importante del fichero: min_plies no se aplica aquí. Contar medias jugadas (pliesPly (media jugada)Una jugada de un solo bando. 1. e4 e5 son dos plies y una jugada completa. Los filtros del recorte y las longitudes de secuencia del modelo se cuentan en plies porque es lo que ve el modelo: un token por ply.) en SQL es poco fiable, porque el movetext trae comentarios (%clk, %eval) y una notación SANSANNotación algebraica estándar, la de los libros: Nf3, O-O, exd5+. Es compacta para humanos pero ambigua sin el tablero (hay que saber qué caballo puede ir a f3). Lichess publica las partidas en SAN; Rukh las convierte a UCI con python-chess. que hay que parsear. Se aplica en M1, con python-chess, después de convertir a UCIUCIDos cosas con el mismo nombre. La notación UCI escribe una jugada como casilla de origen, casilla de destino y promoción opcional (e7e8q): no depende del contexto, por eso es la tokenización por defecto. El protocolo UCI es la forma en que hablamos con Stockfish desde python-chess.. Mientras tanto, el manifiesto guarda el filtro como min_plies_deferred, para que nadie crea que ya se aplicó.

MONTH_RE valida YYYY-MM con los meses reales: (0[1-9]|1[0-2]) rechaza 2025-13 y 2025-00, que un \d{2} aceptaría.

class FetchConfig(BaseConfig):
    """What to fetch and how to filter it."""

    dataset: str = "Lichess/standard-chess-games"
    months: list[str] = Field(min_length=1)
    min_elo: int = Field(default=1800, ge=0)
    min_base_seconds: int = Field(default=180, ge=0)
    terminations: list[str] = Field(default=["Normal", "Time forfeit"], min_length=1)
    min_plies: int = Field(default=20, ge=0)
    exclude_variants: bool = True
    out_dir: str = "data/raw"
    limit: int | None = Field(default=None, ge=1)

    @field_validator("months")
    @classmethod
    def _check_months(cls, months: list[str]) -> list[str]:
        for month in months:
            if not MONTH_RE.match(month):
                raise ValueError(f"month {month!r} must look like YYYY-MM")
        return months

↗src/rukh/data/fetch.pylíneas 23-44 · p0

Hereda de BaseConfig, así que una clave que no esté aquí es un error, y cada campo lleva una restricción que documenta qué valores tienen sentido. terminations vacía produciría Termination IN (), SQL inválido; LIMIT 0 no devuelve nada y casi siempre es una errata. La lista mutable como valor por defecto es segura en pydantic, que la copia por instancia (en una dataclass sería un error clásico).

La misma regex que valida los meses los parte después en año y mes: validar y parsear con el mismo patrón evita que se separen.

class FetchPlan(BaseModel):
    """Everything ``run`` would do, without doing it.

    ``out_dir`` is the absolute output directory; ``out_paths`` and ``manifest_path`` are
    relative to it, in POSIX form, so plans and manifests do not leak machine paths.
    """

    model_config = ConfigDict(extra="forbid")

    query: str
    months: list[str]
    out_dir: str
    out_paths: list[str]
    manifest_path: str

↗src/rukh/data/fetch.pylíneas 45-60 · p0

La decisión de diseño del fichero es que este modelo exista: el plan es un objeto. run(cfg, dry_run=True) devuelve el mismo FetchPlan que run(cfg, dry_run=False) va a ejecutar, así que lo que imprime la ejecución en secoEjecución en seco (dry run)Recorrer entero el camino de un comando sin el efecto que la hace irreversible —sin red, sin subir nada, sin escribir—; en inglés, dry run, y en Rukh el flag --dry-run. Vale la pena porque ejercita justo lo que suele romperse (qué se iba a subir, con qué nombre y a qué repositorio) sin publicar nada, y los tests lo usan para probar publicadores sin tocar el Hub. no puede diferir de lo que hace la descarga. Un --dry-run que redacta su propia descripción acaba mintiendo el día que alguien cambia una parte y no la otra. Pasa lo mismo con un agente que te enseña su plan antes de actuar: lo que apruebas tiene que ser, objeto por objeto, lo que luego se ejecuta.

Las rutas van relativas y en POSIX porque el plan y el manifiesto se publican, y una ruta absoluta de Windows en un fichero de Hugging Face es ruido que además delata la máquina de quien lo generó.

def _split_month(month: str) -> tuple[str, str]:
    match = MONTH_RE.match(month)
    if match is None:
        raise ValueError(f"month {month!r} must look like YYYY-MM")
    return match.group(1), match.group(2)


def _source(cfg: FetchConfig, month: str) -> str:
    year, mm = _split_month(month)
    return f"hf://datasets/{cfg.dataset}/data/year={year}/month={mm}/*.parquet"


def _sql_string(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"

↗src/rukh/data/fetch.pylíneas 61-76 · p0

_source construye la URL hf:// con la que DuckDB lee parquet remoto del Hub sin descargarlo entero; la ruta refleja el particionado del dataset (data/year=2025/month=01/). Que sea una función tiene un efecto que se ve en los tests: se puede sustituir por una que apunte a un parquet local, y el pipeline entero se prueba sin red.

_sql_string duplica las comillas simples. Los valores vienen de una configuración, no de un usuario anónimo, pero un nombre de dataset con un apóstrofo rompería la consulta de una forma que cuesta diagnosticar.

def build_query(cfg: FetchConfig) -> str:
    """Build the DuckDB SQL that selects the filtered games, one ``read_parquet`` per month.

    ``TRY_CAST`` drops rows whose base time is not an integer (correspondence games export
    ``TimeControl = '-'``). ``hive_partitioning = false`` stops DuckDB from injecting ``year``
    and ``month`` columns from the path, which would collide with the explicit ``month`` alias.
    """
    conditions = [
        f"WhiteElo >= {cfg.min_elo} AND BlackElo >= {cfg.min_elo}",
        f"TRY_CAST(split_part(TimeControl, '+', 1) AS INTEGER) >= {cfg.min_base_seconds}",
        "Termination IN (" + ", ".join(_sql_string(t) for t in cfg.terminations) + ")",
    ]
    if cfg.exclude_variants:
        conditions.append("Event NOT ILIKE '%variant%'")
    where = "\n    AND ".join(conditions)
    selects = [
        "SELECT *, "
        + _sql_string(month)
        + " AS month\n"
        + f"FROM read_parquet({_sql_string(_source(cfg, month))}, hive_partitioning = false)\n"
        + f"WHERE {where}"
        for month in cfg.months
    ]
    sql = "\nUNION ALL\n".join(selects)
    if cfg.limit is not None:
        sql += f"\nLIMIT {cfg.limit}"
    return sql

↗src/rukh/data/fetch.pylíneas 77-105 · p0

Los dos detalles del docstring cuestan una tarde cuando no se conocen.

TRY_CAST y no CAST: las partidas por correspondencia llegan con TimeControl = '-', y un CAST('-' AS INTEGER) aborta la consulta entera por una sola fila. Con TRY_CAST esa fila da NULL, NULL >= 180 es falso y la partida se descarta.

hive_partitioning = false: DuckDB reconoce las carpetas year=2025/month=01 de la ruta y, si no se lo impides, las inyecta como columnas year y month. Como la consulta ya añade su propia month ('2025-01' AS month, porque el split es por mes y tras el UNION ALL hay que saber de dónde viene cada partida), habría dos columnas con el mismo nombre: DuckDB renombra una a month_1, y cuál vale 01 y cuál 2025-01 depende de la versión.

El LIMIT final es opcional y sirve para explorar: con limit: 1000 la misma consulta baja mil partidas en segundos.

def _out_dir(cfg: FetchConfig) -> Path:
    out = Path(cfg.out_dir)
    return out if out.is_absolute() else paths.root() / out

↗src/rukh/data/fetch.pylíneas 106-110 · p0

Una ruta relativa se resuelve contra la raíz del proyecto, no contra el directorio desde el que lanzaste el comando: rukh data fetch da lo mismo desde src/ que desde la raíz.

def plan(cfg: FetchConfig) -> FetchPlan:
    """Describe the query, months, output directory, relative output files and manifest."""
    out_dir = _out_dir(cfg)
    out_paths = []
    for month in cfg.months:
        year, mm = _split_month(month)
        target = out_dir / f"year={year}" / f"month={mm}" / "games.parquet"
        out_paths.append(target.relative_to(out_dir).as_posix())
    return FetchPlan(
        query=build_query(cfg),
        months=list(cfg.months),
        out_dir=out_dir.as_posix(),
        out_paths=out_paths,
        manifest_path=(out_dir / "manifest.json").relative_to(out_dir).as_posix(),
    )

↗src/rukh/data/fetch.pylíneas 111-127 · p0

La salida reproduce el particionado del origen (year=2025/month=01/games.parquet), que DuckDB, pandas y polars leen como una tabla particionada: M1 puede consultar los dos meses juntos o uno solo con la misma expresión.

def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()

↗src/rukh/data/fetch.pylíneas 128-135 · p0

Por trozos de un mega (1 << 20), porque estos parquet pesan gigabytes. iter(callable, sentinel) es el modismo de la biblioteca estándar para “llama a esto hasta que devuelva b''”.

def run(cfg: FetchConfig, dry_run: bool = False) -> FetchPlan:
    """Plan the fetch and, unless ``dry_run``, execute it month by month.

    With ``dry_run=True`` nothing touches the network or the disk. Otherwise each month is
    materialized as ``<out_dir>/year=YYYY/month=MM/games.parquet`` (so ``limit`` applies per
    month) and a ``manifest.json`` with filters, counts and file hashes is written last. The
    manifest stores file paths relative to ``out_dir`` and ``min_plies`` as
    ``min_plies_deferred`` because plies are only filtered in P1, after UCI conversion.
    """
    fetch_plan = plan(cfg)
    if dry_run:
        return fetch_plan

    import duckdb

    out_dir = Path(fetch_plan.out_dir)
    counts: dict[str, int] = {}
    files: list[FileHash] = []
    con = duckdb.connect()
    try:
        for month, out_path in zip(cfg.months, fetch_plan.out_paths, strict=True):
            target = out_dir / out_path
            target.parent.mkdir(parents=True, exist_ok=True)
            month_query = build_query(cfg.model_copy(update={"months": [month]}))
            target_sql = _sql_string(target.as_posix())
            con.execute(f"COPY ({month_query}) TO {target_sql} (FORMAT PARQUET)")
            row = con.execute(
                f"SELECT count(*) FROM read_parquet({target_sql}, hive_partitioning = false)"
            ).fetchone()
            counts[month] = int(row[0]) if row else 0
            files.append(
                FileHash(path=out_path, sha256=_sha256(target), bytes=target.stat().st_size)
            )
    finally:
        con.close()

    filters = cfg.model_dump(exclude={"dataset", "months", "out_dir", "min_plies"})
    filters["min_plies_deferred"] = cfg.min_plies
    manifest = Manifest(
        dataset=cfg.dataset,
        months=list(cfg.months),
        filters=filters,
        counts=counts,
        files=files,
    )
    manifest_path = out_dir / fetch_plan.manifest_path
    manifest_path.parent.mkdir(parents=True, exist_ok=True)
    manifest_path.write_text(manifest.model_dump_json(indent=2) + "\n", encoding="utf-8")
    return fetch_plan

↗src/rukh/data/fetch.pylíneas 136-186 · p0

La función que descarga lleva varias decisiones dentro:

• if dry_run: return fetch_plan en la tercera línea: el ensayo es el mismo camino, cortado antes del primer efecto.
• Un COPY por mes, para que limit se aplique por mes y no al total. La consulta de cada mes se reconstruye con cfg.model_copy(update={"months": [month]}), que devuelve una copia con un campo cambiado.
• zip(..., strict=True): si alguien cambiara plan() y las dos listas dejaran de medir lo mismo, un zip normal recortaría en silencio y bajarían menos meses de los pedidos. Con strict=True es un ValueError.
• El conteo se lee del fichero ya escrito, así que un COPY que escribiera mal daría un número incoherente en vez de pasar inadvertido.
• El try/finally cierra la conexión aunque la red se caiga a mitad de un mes. No se limpian ficheros a medias, pero el manifiesto se escribe al final: su presencia significa «esto terminó», y un directorio con parquet y sin manifest.json es una descarga interrumpida.
• exclude= saca de filters lo que no es un filtro y min_plies, que se renombra en la línea siguiente.

uv run rukh data fetch --config configs/data/lichess-2025-01-02.yaml --dry-run

Salida real en la máquina de referencia (RTX 5090, 2026-09-18); <repo> es la carpeta donde hayas clonado rukh:

$ uv run rukh data fetch --config configs/data/lichess-2025-01-02.yaml --dry-run
dataset:  Lichess/standard-chess-games
months:   2025-01, 2025-02
mode:     dry-run (nothing written)
query:
  SELECT *, '2025-01' AS month
  FROM read_parquet('hf://datasets/Lichess/standard-chess-games/data/year=2025/month=01/*.parquet', hive_partitioning = false)
  WHERE WhiteElo >= 1800 AND BlackElo >= 1800
      AND TRY_CAST(split_part(TimeControl, '+', 1) AS INTEGER) >= 180
      AND Termination IN ('Normal', 'Time forfeit')
      AND Event NOT ILIKE '%variant%'
  UNION ALL
  SELECT *, '2025-02' AS month
  FROM read_parquet('hf://datasets/Lichess/standard-chess-games/data/year=2025/month=02/*.parquet', hive_partitioning = false)
  WHERE WhiteElo >= 1800 AND BlackElo >= 1800
      AND TRY_CAST(split_part(TimeControl, '+', 1) AS INTEGER) >= 180
      AND Termination IN ('Normal', 'Time forfeit')
      AND Event NOT ILIKE '%variant%'
out_dir:  <repo>/data/raw
outputs:
  year=2025/month=01/games.parquet
  year=2025/month=02/games.parquet
manifest: manifest.json

Los cuatro filtros están ahí, legibles, antes de que nada se descargue: la tabla de arriba se puede comprobar.

Los tests: el pipeline entero, sin red

tests/unit/test_fetch.py es el test más instructivo de M0: prueba de verdad un código cuyo trabajo es hablar con un servidor remoto, sin red y en milisegundos.

Puedes dejarlos plegados en la primera lectura, pero el «Ejecútalo» de abajo los nombra uno a uno: si no los tecleas, pytest dice file or directory not found. Tráelos antes con git checkout p0 -- tests/unit/test_fetch.py tests/unit/test_manifest.py.

"""Tests for rukh.data.fetch: query builder, dry-run plan and local execution."""

from __future__ import annotations

import json
from pathlib import Path

import pytest
from pydantic import ValidationError
from typer.testing import CliRunner

from rukh.cli import app
from rukh.config import load_yaml
from rukh.data.fetch import FetchConfig, FetchPlan, build_query, plan, run

pytestmark = pytest.mark.unit

MONTHS = ["2025-01", "2025-02"]

# Columns: Event, WhiteElo, BlackElo, TimeControl, Termination. Exactly one row passes the
# default filters; the '-' time control is what Lichess exports for correspondence games.
FAKE_ROWS = """
    SELECT * FROM (VALUES
      ('Rated Blitz game', 1900, 1850, '300+0', 'Normal'),
      ('Rated Bullet game', 1900, 1850, '60+0', 'Normal'),
      ('Rated Blitz game', 1500, 1850, '300+0', 'Normal'),
      ('Rated Blitz game', 1900, 1850, '300+0', 'Abandoned'),
      ('Rated Atomic variant game', 1900, 1850, '300+0', 'Normal'),
      ('Rated Correspondence game', 1900, 1850, '-', 'Normal')
    ) AS t(Event, WhiteElo, BlackElo, TimeControl, Termination)
"""

def _cfg(**overrides: object) -> FetchConfig:
    return FetchConfig(months=MONTHS, **overrides)  # type: ignore[arg-type]


def _write_hive_source(rukh_home: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """Write the fake rows under the real ``year=YYYY/month=MM`` layout; point ``_source`` at it."""
    import duckdb

    from rukh.data import fetch as fetch_module

    source_dir = rukh_home / "source" / "year=2025" / "month=01"
    source_dir.mkdir(parents=True)
    duckdb.sql(FAKE_ROWS).write_parquet((source_dir / "part-0.parquet").as_posix())
    monkeypatch.setattr(
        fetch_module, "_source", lambda cfg, month: (source_dir / "*.parquet").as_posix()
    )
    return source_dir


def _read_output(path: Path) -> tuple[list[str], list[tuple[object, ...]]]:
    import duckdb

    rel = duckdb.sql(f"SELECT * FROM read_parquet('{path.as_posix()}', hive_partitioning = false)")
    return list(rel.columns), rel.fetchall()

def test_fetch_config_defaults_match_spec() -> None:
    cfg = _cfg()
    assert cfg.dataset == "Lichess/standard-chess-games"
    assert cfg.min_elo == 1800
    assert cfg.min_base_seconds == 180
    assert cfg.terminations == ["Normal", "Time forfeit"]
    assert cfg.min_plies == 20
    assert cfg.exclude_variants is True
    assert cfg.out_dir == "data/raw"
    assert cfg.limit is None


def test_fetch_config_rejects_unknown_key() -> None:
    with pytest.raises(ValidationError):
        FetchConfig(months=MONTHS, min_rating=1800)  # type: ignore[call-arg]


@pytest.mark.parametrize("bad", ["2025-1", "2025/01", "202501", "2025-13", ""])
def test_fetch_config_rejects_bad_month(bad: str) -> None:
    with pytest.raises(ValidationError):
        FetchConfig(months=[bad])


def test_fetch_config_requires_at_least_one_month() -> None:
    with pytest.raises(ValidationError):
        FetchConfig(months=[])


@pytest.mark.parametrize(
    "overrides",
    [
        {"terminations": []},
        {"limit": 0},
        {"min_elo": -1},
        {"min_base_seconds": -1},
        {"min_plies": -1},
    ],
    ids=["empty-terminations", "limit-zero", "negative-elo", "negative-base", "negative-plies"],
)
def test_fetch_config_rejects_out_of_range_values(overrides: dict[str, object]) -> None:
    with pytest.raises(ValidationError):
        FetchConfig(months=MONTHS, **overrides)  # type: ignore[arg-type]

def test_build_query_has_filters_and_both_months() -> None:
    sql = build_query(_cfg())
    for month in ("month=01", "month=02"):
        assert f"hf://datasets/Lichess/standard-chess-games/data/year=2025/{month}/*.parquet" in sql
    assert sql.count("read_parquet(") == 2
    assert sql.count("hive_partitioning = false") == 2
    assert "UNION ALL" in sql
    assert "WhiteElo >= 1800 AND BlackElo >= 1800" in sql
    assert "TRY_CAST(split_part(TimeControl, '+', 1) AS INTEGER) >= 180" in sql
    assert "Termination IN ('Normal', 'Time forfeit')" in sql
    assert "Event NOT ILIKE '%variant%'" in sql
    assert "LIMIT" not in sql


def test_build_query_optional_parts() -> None:
    sql = build_query(_cfg(exclude_variants=False, limit=1000))
    assert "ILIKE" not in sql
    assert sql.rstrip().endswith("LIMIT 1000")

def test_plan_lists_relative_output_paths_and_manifest(rukh_home: Path) -> None:
    fetch_plan = plan(_cfg())
    assert isinstance(fetch_plan, FetchPlan)
    assert fetch_plan.months == MONTHS
    assert fetch_plan.out_dir == (rukh_home / "data" / "raw").as_posix()
    assert fetch_plan.out_paths == [
        "year=2025/month=01/games.parquet",
        "year=2025/month=02/games.parquet",
    ]
    assert fetch_plan.manifest_path == "manifest.json"
    assert fetch_plan.query == build_query(_cfg())


def test_run_dry_does_not_touch_disk(rukh_home: Path) -> None:
    result = run(_cfg(), dry_run=True)
    assert result == plan(_cfg())
    assert not (rukh_home / "data").exists()


def test_shipped_config_loads(repo_root: Path) -> None:
    cfg = load_yaml(repo_root / "configs" / "data" / "lichess-2025-01-02.yaml", FetchConfig)
    assert cfg.months == MONTHS

def test_cli_data_fetch_dry_run_prints_plan(rukh_home: Path, repo_root: Path) -> None:
    config = repo_root / "configs" / "data" / "lichess-2025-01-02.yaml"
    result = CliRunner().invoke(app, ["data", "fetch", "--config", str(config), "--dry-run"])
    assert result.exit_code == 0, result.output
    assert "read_parquet(" in result.output
    assert "2025-01" in result.output and "2025-02" in result.output
    assert (rukh_home / "data" / "raw").as_posix() in result.output
    assert "  year=2025/month=01/games.parquet" in result.output
    assert "manifest.json" in result.output
    assert not (rukh_home / "data").exists()


def test_cli_data_fetch_dry_run_json(rukh_home: Path, repo_root: Path) -> None:
    config = repo_root / "configs" / "data" / "lichess-2025-01-02.yaml"
    result = CliRunner().invoke(
        app, ["data", "fetch", "--config", str(config), "--dry-run", "--json"]
    )
    assert result.exit_code == 0, result.output
    payload = json.loads(result.output)
    assert payload["months"] == MONTHS
    assert payload["out_paths"] == [
        "year=2025/month=01/games.parquet",
        "year=2025/month=02/games.parquet",
    ]

def test_run_writes_parquet_and_manifest_from_local_source(
    rukh_home: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from rukh.data.manifest import Manifest

    _write_hive_source(rukh_home, monkeypatch)

    cfg = FetchConfig(months=["2025-01"])
    fetch_plan = run(cfg, dry_run=False)

    out_dir = rukh_home / "data" / "raw"
    assert fetch_plan.out_paths == ["year=2025/month=01/games.parquet"]
    out = out_dir / fetch_plan.out_paths[0]
    assert out.is_file()
    columns, rows = _read_output(out)
    assert len(rows) == 1
    assert columns.count("month") == 1
    assert "month_1" not in columns
    assert "year" not in columns

    manifest = Manifest.model_validate_json((out_dir / "manifest.json").read_text("utf-8"))
    assert manifest.counts == {"2025-01": 1}
    assert manifest.files[0].path == "year=2025/month=01/games.parquet"
    assert manifest.filters["min_elo"] == 1800
    assert manifest.filters["min_plies_deferred"] == 20
    assert "min_plies" not in manifest.filters


def test_run_drops_correspondence_games_instead_of_aborting(
    rukh_home: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _write_hive_source(rukh_home, monkeypatch)

    fetch_plan = run(FetchConfig(months=["2025-01"], min_elo=0), dry_run=False)

    columns, rows = _read_output(rukh_home / "data" / "raw" / fetch_plan.out_paths[0])
    events = [row[columns.index("Event")] for row in rows]
    assert events == ["Rated Blitz game", "Rated Blitz game"]

"""Tests for rukh.data.manifest."""

from __future__ import annotations

from datetime import UTC, datetime

import pytest
from pydantic import ValidationError

from rukh import __version__
from rukh.data.manifest import FileHash, Manifest

pytestmark = pytest.mark.unit

def _manifest() -> Manifest:
    return Manifest(
        dataset="Lichess/standard-chess-games",
        months=["2025-01", "2025-02"],
        filters={"min_elo": 1800, "terminations": ["Normal", "Time forfeit"]},
        counts={"2025-01": 10, "2025-02": 12},
        files=[
            FileHash(path="data/raw/year=2025/month=01/games.parquet", sha256="ab" * 32, bytes=1)
        ],
        created_at=datetime(2026, 9, 18, 12, 0, tzinfo=UTC),
    )


def test_manifest_json_roundtrip_is_idempotent() -> None:
    m = _manifest()
    again = Manifest.model_validate_json(m.model_dump_json())
    assert again == m
    assert again.model_dump_json() == m.model_dump_json()


def test_manifest_defaults_rukh_version() -> None:
    assert _manifest().rukh_version == __version__


def test_manifest_forbids_unknown_keys() -> None:
    with pytest.raises(ValidationError):
        Manifest.model_validate({**_manifest().model_dump(), "extra": 1})


def test_file_hash_validates_sha256() -> None:
    with pytest.raises(ValidationError):
        FileHash(path="x", sha256="not-a-hash", bytes=0)

Antes de descargar: el parche y una prueba de mil partidas

La descarga completa tarda una o dos horas, y en p0 tiene un problema conocido. Antes de lanzarla hay dos pasos baratos: aplicar un parche a run y probar la tubería entera con mil partidas.

Aplica el parche, pasa otra vez uv run pytest -m unit -q (los tests no tocan la red, así que siguen en verde) y haz la prueba barata. Crea una configuración de prueba, igual que la de arriba pero con un solo mes, limit: 1000 y otra carpeta de salida para no mezclarla con la de verdad. Va dentro de data/, que git ignora, porque es de usar y tirar:

dataset: Lichess/standard-chess-games
months: ["2025-01"]
min_elo: 1800
min_base_seconds: 180
terminations: ["Normal", "Time forfeit"]
min_plies: 20
exclude_variants: true
out_dir: data/prueba
limit: 1000

movetext mezcla dos de las cuatro notaciones de ajedrez que usa el curso, y conviene separarlas ya. PGNPGNFormato de fichero de partidas: cabeceras entre corchetes ([White "..."], [Result "1-0"]) y las jugadas en SAN, con comentarios opcionales como %clk o %eval. La demo exporta PGN; los volcados de Lichess lo usan como origen. es el formato de una partida entera (cabeceras más jugadas), y lo que hay en movetext es su parte de jugadas. SANSANNotación algebraica estándar, la de los libros: Nf3, O-O, exd5+. Es compacta para humanos pero ambigua sin el tablero (hay que saber qué caballo puede ir a f3). Lichess publica las partidas en SAN; Rukh las convierte a UCI con python-chess. es cómo escribe una persona cada jugada, Nf3 o exd5: corta, pero para entenderla hay que saber la posición. UCIUCIDos cosas con el mismo nombre. La notación UCI escribe una jugada como casilla de origen, casilla de destino y promoción opcional (e7e8q): no depende del contexto, por eso es la tokenización por defecto. El protocolo UCI es la forma en que hablamos con Stockfish desde python-chess. es cómo la escribe un motor, casilla de origen y de destino (g1f3): no depende de la posición, y por eso es el token de Rukh desde M1. FENFENCadena de texto que describe una posición completa: piezas por fila, turno, derechos de enroque, casilla al paso y contadores de jugadas. Es la clave con la que se cruzan las posiciones de las partidas con las evaluaciones públicas de Stockfish. no es una jugada sino una posición, una línea de texto con dónde está cada pieza y a quién le toca. M1 convierte movetext de SAN a UCI con python-chess.

La descarga de verdad

uv run rukh data fetch --config configs/data/lichess-2025-01-02.yaml

Con el parche aplicado, tarda entre una y dos horas en la máquina de referencia, sobre todo por la conexión. Si aun así se corta con un 429, relanza el mismo comando. Se lanza al terminar esta lección y se recoge en M1; mientras corre, la lección siguiente (el tablero de la demo) no necesita datos.

Qué has aprendido

El recorte de datos y, sobre todo, la disciplina que lo acompaña: los filtros se leen antes de descargar, se aplican desde un único sitio, y lo que se aplicó de verdad queda en un manifiesto con conteos y hashes. El filtro que no se aplica se llama distinto en el manifiesto para que nadie crea que sí.

Cómo se mide: el --dry-run imprime la SQL con los cuatro filtros; uv run pytest -m unit -q pasa los veintitantos tests de fetch y manifest, incluidos los dos que ejecutan el pipeline completo sobre un parquet local; y cuando la descarga termine, data/raw/manifest.json tendrá counts por mes, un sha256 por fichero y min_plies_deferred: 20.

Lo siguiente es el tablero: la demo donde acabarás jugando contra tu propio modelo, que en este hito ya sabe las reglas aunque no haya modelo ninguno.