De PGN a UCI: reproducir seis millones de partidas · Rukh

Qué vas a construir

El primer paso del pipeline: rukh data uci. Entra el parquet que M0 descargó, con las jugadas escritas como en los libros y llenas de comentarios de reloj, y sale un parquet por mes con una columna uci que un tokenizador puede leer sin saber nada de ajedrez. En medio, seis millones de partidas se reproducen jugada a jugada sobre un tablero de verdad, en un proceso por núcleo, con una conexión de DuckDB configurada para que no te llene el disco del sistema.

Son tres ficheros: src/rukh/data/uci.py (la conversión), src/rukh/data/parallel.py (el reparto entre procesos) y src/rukh/data/db.py (la conexión), más dos ficheros de test que fijan el contrato.

`uci.py`: el fichero que convierte

Empieza por la cabecera: los esquemas y las seis expresiones regulares de la limpieza.

"""SAN movetext to UCI: cleaning, legality check and ply filter, one parquet per month.

Every game is replayed with python-chess; a game with an unparsable or illegal move is
dropped (``illegal``), as are games shorter than ``min_plies`` (``short``) or longer than
``max_plies`` (``long``). Batches of ``BATCH_GAMES`` rows are converted in a ``spawn``
process pool (Windows-safe: every worker entry point is a module-level function).
"""

from __future__ import annotations

import hashlib
import os
import re
from collections.abc import Iterator
from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq
from pydantic import Field

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

BATCH_GAMES = 20_000
RAW_COLUMNS = [
    "Site",
    "movetext",
    "WhiteElo",
    "BlackElo",
    "Result",
    "TimeControl",
    "UTCDate",
    "ECO",
    "month",
]
OUT_SCHEMA = pa.schema(
    [
        ("game_id", pa.int64()),
        ("uci", pa.string()),
        ("n_plies", pa.int16()),
        ("white_elo", pa.int16()),
        ("black_elo", pa.int16()),
        ("result", pa.string()),
        ("time_control", pa.string()),
        ("utc_date", pa.date32()),
        ("eco", pa.string()),
        ("month", pa.string()),
    ]
)
RESULTS = ("1-0", "0-1", "1/2-1/2")

_COMMENT_RE = re.compile(r"\{[^}]*\}")
_VARIATION_RE = re.compile(r"\([^()]*\)")
_NAG_RE = re.compile(r"\$\d+")
_MOVE_NUMBER_RE = re.compile(r"\b\d+\.(\.\.)?")
_RESULT_RE = re.compile(r"(1-0|0-1|1/2-1/2|\*)\s*$")
_ANNOTATION_RE = re.compile(r"[!?]+$")

src/rukh/data/uci.pylíneas 1-59 · p1

El OUT_SCHEMA se declara a mano en vez de dejar que pyarrow lo deduzca. Así los tipos son los justos (int16 para plies y Elo, que no pasan de 32 767) y el orden de las columnas es parte del contrato: un test comprueba la lista completa, y añadir una columna en medio falla en la CI y no tres módulos después.

RESULTS tiene tres valores y no cuatro. Las partidas sin resultado (*) se descartan antes de tocar el tablero, porque el tokenizador necesita un token de resultado al final y no hay ninguno para “no se sabe”.

Cada regex corresponde a una parte del movetext de Lichess, que es más sucio de lo que parece:

1. e4 { [%eval 0.3] [%clk 0:05:00] } e5 { [%clk 0:05:00] } 2. Nf3 $1 1... Nc6 1/2-1/2

Comentarios entre llaves con la evaluación y el reloj, anotaciones numéricas ($1), números de jugada en dos formas (2. y 1...), el resultado y, en partidas exportadas de un análisis, variantes entre paréntesis.

class UciConfig(BaseConfig):
    """Input/output directories, ply bounds and parallelism (``0`` = ``cpu_count() - 1``)."""

    raw_dir: str = "data/raw"
    out_dir: str = "data/uci"
    min_plies: int = Field(default=20, ge=0)
    max_plies: int = Field(default=300, ge=1)
    workers: int = Field(default=0, ge=0)

src/rukh/data/uci.pylíneas 62-69 · p1

workers: 0 significa “uno menos que los núcleos” (lo resuelve resolve_workers más abajo), para que la misma configuración valga en un portátil de cuatro núcleos y en la máquina de referencia.

Limpiar, reproducir, identificar

def clean_movetext(movetext: str) -> str:
    """Strip ``{...}`` comments (``%clk``, ``%eval``), variations, NAGs, move numbers, result."""
    text = _COMMENT_RE.sub(" ", movetext)
    while _VARIATION_RE.search(text):
        text = _VARIATION_RE.sub(" ", text)
    text = _NAG_RE.sub(" ", text)
    text = _RESULT_RE.sub(" ", text.strip())
    text = _MOVE_NUMBER_RE.sub(" ", text)
    return " ".join(text.split())


def san_to_uci(movetext: str) -> tuple[str, int] | None:
    """Replay the cleaned SAN with python-chess; ``None`` when any move is illegal."""
    import chess

    board = chess.Board()
    moves: list[str] = []
    for san in clean_movetext(movetext).split():
        san = _ANNOTATION_RE.sub("", san)
        if not san:
            continue
        try:
            move = board.parse_san(san)
        except ValueError:
            return None
        moves.append(move.uci())
        board.push(move)
    return " ".join(moves), len(moves)


def game_id(site: str) -> int:
    """Signed 64-bit id: the first 8 bytes (little-endian) of ``sha256(Site)``."""
    return int.from_bytes(hashlib.sha256(site.encode("utf-8")).digest()[:8], "little", signed=True)

src/rukh/data/uci.pylíneas 72-104 · p1

El while de clean_movetext está porque los paréntesis anidan. _VARIATION_RE solo casa los paréntesis que no contienen otros, así que en (1. e4 e5 (1... c5)) la primera pasada se lleva el de dentro y la segunda el de fuera. Con una sola quedaría un paréntesis huérfano y parse_san moriría en él.

El orden de las sustituciones importa: los comentarios se quitan antes que nada porque dentro llevan cosas que se parecen a otras partes del movetext. [%eval 0.3] contiene un 0. que _MOVE_NUMBER_RE tomaría por un número de jugada.

san_to_uci reproduce la partida sobre un chess.Board() de verdad en vez de traducir cadenas. Nf3 no dice de qué casilla viene el caballo; solo la posición de ese instante lo sabe. O-O es e1g1 (el movimiento del rey) y e8=Q+ es e7e8q. Una traducción con expresiones regulares produce ficheros que parecen bien hasta que el modelo emite jugadas imposibles.

_ANNOTATION_RE se aplica jugada a jugada: Bb5!? es una jugada legal con una opinión pegada. Y la función devuelve None en vez de lanzar porque corre dentro de un pool, y una excepción por una partida corrupta se llevaría por delante el lote de veinte mil en el que iba.

El id de una partida son los primeros ocho bytes de sha256(Site) como entero de 64 bits con signo (con signo porque la columna es int64 de Arrow). Podría ser hash(site), y sería un error: el hash() de Python se siembra al azar en cada proceso, así que dos workers darían ids distintos para la misma partida. Todo lo que después se muestree por este id necesita que sea estable. En cualquier pipeline, un identificador derivado se calcula con una función determinista, nunca con la que el lenguaje usa para sus diccionarios.

El lote: de filas crudas a columnas

def convert_rows(
    rows: list[dict[str, object]], min_plies: int, max_plies: int
) -> dict[str, object]:
    """Convert one batch of raw rows (dicts with ``RAW_COLUMNS``) into output columns + counts.

    Module-level so ``multiprocessing`` can pickle it under the ``spawn`` start method.
    """
    out: dict[str, list[object]] = {name: [] for name in OUT_SCHEMA.names}
    counts = {"rows": 0, "kept": 0, "illegal": 0, "short": 0, "long": 0}
    for row in rows:
        counts["rows"] += 1
        result = str(row["Result"])
        converted = san_to_uci(str(row["movetext"])) if result in RESULTS else None
        if converted is None:
            counts["illegal"] += 1
            continue
        uci, n_plies = converted
        if n_plies < min_plies:
            counts["short"] += 1
            continue
        if n_plies > max_plies:
            counts["long"] += 1
            continue
        counts["kept"] += 1
        out["game_id"].append(game_id(str(row["Site"])))
        out["uci"].append(uci)
        out["n_plies"].append(n_plies)
        out["white_elo"].append(_parse_elo(row["WhiteElo"]))
        out["black_elo"].append(_parse_elo(row["BlackElo"]))
        out["result"].append(result)
        out["time_control"].append(row["TimeControl"])
        out["utc_date"].append(_parse_date(row["UTCDate"]))
        out["eco"].append(row["ECO"])
        out["month"].append(row["month"])
    return {"columns": out, "counts": counts}

src/rukh/data/uci.pylíneas 107-141 · p1

Aquí se aplica el filtro de longitud que M0 dejó pendiente (min_plies_deferred en su manifiesto): contar plies exige parsear el movetext, y eso no se hace en SQL.

Los cinco contadores van al manifiesto y son excluyentes (cuenta el primer motivo que aparece), así que suman las filas leídas y cualquier fila que desaparezca tiene una casilla que lo explica. En la ejecución de referencia, de 6 000 000 de partidas quedaron 5 896 388, con 103 251 cortas, 361 largas y cero ilegales. Ese cero es el número que había que mirar: Lichess valida las partidas al guardarlas, así que un solo ilegal habría delatado una limpieza que rompe algo.

La función está a nivel de módulo a propósito, y es la restricción que gobierna todo el fichero.

// Error que enseña`spawn` no serializa un `lambda`, ni un método, ni una clausura

En Linux, multiprocessing arranca los procesos con fork: el hijo hereda la memoria del padre y cualquier función vale como punto de entrada. En Windows y en macOS arranca con spawn: el hijo es un intérprete nuevo que reimporta el módulo y recibe la función por pickle, es decir, por su nombre completo. Una función anidada, un lambda o un método de instancia no tienen nombre importable y fallan con AttributeError: Can't pickle local object.

Por eso convert_rows y convert_batch están al nivel del módulo y reciben sus parámetros en la tupla del trabajo. Esta restricción no se descubre leyendo: aparece cuando el código que funcionaba en la CI de Linux no arranca en otra máquina. Te la encontrarás en cualquier trabajo en paralelo con Python, sea un pipeline de datos, un servidor con workers o un agente que lanza herramientas en procesos aparte.

def _parse_elo(value: object) -> int | None:
    """A missing or unparsable rating is ``None``; the column stays nullable int16."""
    if value is None:
        return None
    try:
        return int(value)  # type: ignore[call-overload]
    except (TypeError, ValueError):
        return None


def _parse_date(value: object) -> object:
    from datetime import date

    if value is None or isinstance(value, date):
        return value
    text = str(value).replace(".", "-")
    try:
        return date.fromisoformat(text)
    except ValueError:
        return None


def convert_batch(args: tuple[list[dict[str, object]], int, int]) -> dict[str, object]:
    """Worker entry point: ``(rows, min_plies, max_plies)`` into columns and counts."""
    rows, min_plies, max_plies = args
    return convert_rows(rows, min_plies, max_plies)

src/rukh/data/uci.pylíneas 144-169 · p1

_parse_elo devuelve None cuando falta el Elo, no un cero: un cero sería un Elo, y el muestreo por tramos de M4 crearía un tramo <w0600> lleno de partidas sin dato. El replace(".", "-") de _parse_date está porque Lichess escribe 2025.01.01. convert_batch solo desempaqueta: existe para tener un punto de entrada con nombre y un único argumento, que es lo que apply_async manda.

Leer, repartir, escribir

def iter_batches(src: Path) -> Iterator[list[dict[str, object]]]:
    reader = pq.ParquetFile(src)
    columns = [c for c in RAW_COLUMNS if c in reader.schema_arrow.names]
    for batch in reader.iter_batches(batch_size=BATCH_GAMES, columns=columns):
        rows = batch.to_pylist()
        for row in rows:
            for name in RAW_COLUMNS:
                row.setdefault(name, None)
        yield rows


def resolve_workers(workers: int) -> int:
    return workers if workers > 0 else max(1, (os.cpu_count() or 2) - 1)


def convert_month(src: Path, dst: Path, cfg: UciConfig) -> dict[str, int]:
    """Convert ``src`` (raw parquet) into ``dst`` (UCI parquet); returns the counts."""
    dst = Path(dst)
    dst.parent.mkdir(parents=True, exist_ok=True)
    totals = {"rows": 0, "kept": 0, "illegal": 0, "short": 0, "long": 0}
    workers = resolve_workers(cfg.workers)
    jobs = ((rows, cfg.min_plies, cfg.max_plies) for rows in iter_batches(Path(src)))
    with pq.ParquetWriter(dst, OUT_SCHEMA, compression="zstd") as writer:
        write_results(writer, run_batches(convert_batch, jobs, workers), totals)
    return totals

src/rukh/data/uci.pylíneas 172-196 · p1

iter_batches lee el parquet por grupos de 20 000 filas en vez de con un read_table: un mes entero con su movetext son varios gigas en RAM antes de convertir la primera partida. El setdefault rellena las columnas que falten, para que el paso lea tanto el parquet de fetch (con month) como uno bajado de Lichess sin ella.

En convert_month, jobs es un generador: los lotes se construyen a medida que el pool los pide. Que eso baste para acotar la memoria depende del fichero siguiente.

def write_results(
    writer: pq.ParquetWriter, results: Iterator[dict[str, object]], totals: dict[str, int]
) -> None:
    for result in results:
        counts: dict[str, int] = result["counts"]  # type: ignore[assignment]
        for key, value in counts.items():
            totals[key] += value
        columns: dict[str, list[object]] = result["columns"]  # type: ignore[assignment]
        if columns["game_id"]:
            writer.write_table(pa.table(columns, schema=OUT_SCHEMA))


def month_dirs(raw_dir: Path) -> list[tuple[str, Path]]:
    """``(YYYY-MM, games.parquet)`` for every ``year=YYYY/month=MM`` under ``raw_dir``."""
    found: list[tuple[str, Path]] = []
    for path in sorted(Path(raw_dir).glob("year=*/month=*/games.parquet")):
        year = path.parents[1].name.split("=", 1)[1]
        mm = path.parent.name.split("=", 1)[1]
        found.append((f"{year}-{mm}", path))
    return found


def sha256_file(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/uci.pylíneas 199-226 · p1

El if columns["game_id"] de write_results evita escribir un grupo de cero filas cuando ninguna partida del lote pasó los filtros. month_dirs saca el mes de la ruta (year=2025/month=01), como hacía M0 con hive_partitioning = false.

def run(cfg: UciConfig) -> Manifest:
    """Convert every month under ``raw_dir`` and write ``out_dir/manifest.json``."""
    raw_dir = resolve(cfg.raw_dir)
    out_dir = resolve(cfg.out_dir)
    months = month_dirs(raw_dir)
    if not months:
        raise FileNotFoundError(f"no year=*/month=*/games.parquet under {raw_dir}")
    raw_manifest_path = raw_dir / "manifest.json"
    raw_manifest = (
        Manifest.model_validate_json(raw_manifest_path.read_text("utf-8"))
        if raw_manifest_path.is_file()
        else None
    )
    counts: dict[str, int] = {}
    details: dict[str, dict[str, int]] = {}
    files: list[FileHash] = []
    for month, src in months:
        rel = src.relative_to(raw_dir)
        dst = out_dir / rel
        totals = convert_month(src, dst, cfg)
        counts[month] = totals["kept"]
        details[month] = totals
        files.append(
            FileHash(path=rel.as_posix(), sha256=sha256_file(dst), bytes=dst.stat().st_size)
        )
    filters: dict[str, object] = {
        "min_plies": cfg.min_plies,
        "max_plies": cfg.max_plies,
        "legal_only": True,
        "conversion": details,
    }
    if raw_manifest is not None:
        filters["raw"] = raw_manifest.filters
    manifest = Manifest(
        dataset=raw_manifest.dataset if raw_manifest else "Lichess/standard-chess-games",
        months=[m for m, _ in months],
        filters=filters,
        counts=counts,
        files=files,
    )
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "manifest.json").write_text(manifest.model_dump_json(indent=2) + "\n", "utf-8")
    return manifest

src/rukh/data/uci.pylíneas 229-271 · p1

El run final tiene una idea dentro: el manifiesto nuevo se cuelga del viejo. Los filtros de data/raw/manifest.json entran en el de data/uci/ bajo la clave raw, y el nombre del dataset de origen se hereda. Así el último manifiesto de la cadena lleva la procedencia completa: de qué volcado salió, con qué filtros se descargó y se convirtió, y el sha256 de cada fichero. Es lo que irá a la dataset card de Hugging Face. sha256_file, por su parte, lee en trozos de 1 MiB para no cargar un parquet de 480 MB entero.

`parallel.py`: el pool que no se adelanta

Este fichero existe por un fallo concreto: el pool de la biblioteca estándar se adelanta.

Lo que sigue es ingeniería de procesos, no de modelos: en la primera lectura basta con saber que run_batches reparte los lotes entre núcleos sin adelantarse. Despliégalo cuando vayas a escribirlo. Sin él, uci.py no convierte nada; si no lo tecleas, tráelo con git checkout p1 -- src/rukh/data/parallel.py.

// Fontaneríaparallel.py: un pool de spawn que no lee más lotes de los que puede procesar

"""Bounded fan-out over a ``spawn`` process pool.

``multiprocessing.Pool.imap`` drains its input iterable as fast as it can, so feeding it a
generator of 20 000-game batches pulls a whole month into memory before the first result comes
back. ``run_batches`` keeps at most ``2 * workers`` batches in flight: a new job is submitted
only when an older one has been consumed, so the feeder advances at the speed of the writer.
"""

from __future__ import annotations

import multiprocessing
from collections import deque
from collections.abc import Callable, Iterable, Iterator
from typing import Protocol


class AsyncResult[R](Protocol):
    """What ``imap_bounded`` needs from ``Pool.apply_async``."""

    def get(self) -> R: ...


class Pool(Protocol):
    """What ``imap_bounded`` needs from ``multiprocessing.Pool`` (tests use a fake)."""

    def apply_async[R](
        self, func: Callable[..., R], args: tuple[object, ...]
    ) -> AsyncResult[R]: ...

src/rukh/data/parallel.pylíneas 1-28 · p1

Los dos Protocol declaran los únicos dos métodos del pool que hacen falta, y así el test puede probar la cola con un pool falso, sin lanzar un solo proceso.

def imap_bounded[J, R](
    pool: Pool,
    func: Callable[[J], R],
    jobs: Iterable[J],
    max_in_flight: int,
) -> Iterator[R]:
    """``pool.imap`` with backpressure: never more than ``max_in_flight`` jobs are queued."""
    if max_in_flight < 1:
        raise ValueError("max_in_flight must be at least 1")
    pending: deque[AsyncResult[R]] = deque()
    for job in jobs:
        pending.append(pool.apply_async(func, (job,)))
        if len(pending) >= max_in_flight:
            yield pending.popleft().get()
    while pending:
        yield pending.popleft().get()

src/rukh/data/parallel.pylíneas 31-46 · p1

Pool.imap vacía su iterable de entrada tan rápido como puede, sin esperar a que haya un worker libre. Con un generador de lotes de 20 000 partidas, el mes entero acaba en memoria antes de que vuelva el primer resultado, y el proceso muere mucho antes de escribir una fila.

imap_bounded invierte quién manda: en cuanto hay max_in_flight trabajos pendientes, espera al más antiguo con popleft().get(). Mientras ese get() bloquea, el for no avanza y el generador deja de producir. El alimentador va al ritmo del escritor.

def run_batches[J, R](func: Callable[[J], R], jobs: Iterable[J], workers: int) -> Iterator[R]:
    """Map ``func`` over ``jobs``; serially with one worker, else in a bounded spawn pool.

    ``func`` and the jobs must be picklable: every worker entry point is a module-level
    function (Windows starts processes with ``spawn``).
    """
    if workers <= 1:
        yield from map(func, jobs)
        return
    ctx = multiprocessing.get_context("spawn")
    with ctx.Pool(workers) as pool:
        yield from imap_bounded(pool, func, jobs, 2 * workers)

src/rukh/data/parallel.pylíneas 49-60 · p1

Con workers <= 1 no hay pool, solo map. Es lo que hace rápidos los tests: un pool de spawn en Windows paga segundos de arranque por caso, y la ruta serie prueba la conversión en milisegundos. La ruta paralela la comprueba un único test aparte.

get_context("spawn") fija el método de arranque en todas las plataformas, así que una función que no se puede serializar falla también en la CI de Linux.

`db.py`: una conexión configurada, no la de fábrica

La conexión configurada es fontanería del pipeline, y puedes saltarla en la primera lectura; el «Ejecútalo» de abajo comprueba que la escribiste bien. Si no la tecleas, db.py es nuevo y se trae entero con git checkout p1 -- src/rukh/data/db.py. Los otros tres cambios del pliegue tocan ficheros que ya tecleaste en M0 (.gitignore, fetch.py y paths.py), así que esos no los traigas, que pisarías lo tuyo: git diff p1 -- .gitignore src/rukh/paths.py src/rukh/data/fetch.py te marca con - las líneas que faltan.

// Fontaneríadb.py, la ruta de descarga y resolve(): una conexión de DuckDB con memoria, hilos y temporales fijados

"""One configured DuckDB connection for the whole pipeline.

Every step joins or groups files far larger than RAM, so a default connection is a foot-gun:
DuckDB spills to the system temp directory (often a small system drive) and, without a memory
limit, competes with the worker processes. ``connect`` sets ``temp_directory`` to
``<data>/duckdb-tmp`` (created on demand, gitignored), ``memory_limit`` and ``threads`` on
every connection.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import TYPE_CHECKING

from pydantic import Field

from rukh import paths
from rukh.config import BaseConfig

if TYPE_CHECKING:
    import duckdb

TMP_DIRNAME = "duckdb-tmp"


class DuckDbConfig(BaseConfig):
    """Settings applied to every DuckDB connection (``duckdb:`` in ``pipeline.yaml``)."""

    memory_limit: str = "32GB"
    threads: int = Field(default=0, ge=0)
    """``0`` = ``cpu_count()``."""

src/rukh/data/db.pylíneas 1-32 · p1

Lo que parece una línea (duckdb.connect()) ocupa un fichero: tres ajustes, cada uno contra un fallo concreto.

def temp_directory() -> Path:
    """``<data_dir>/duckdb-tmp``, created if missing: spills stay on the data drive."""
    tmp = paths.data_dir() / TMP_DIRNAME
    tmp.mkdir(parents=True, exist_ok=True)
    return tmp


def connect(cfg: DuckDbConfig | None = None) -> duckdb.DuckDBPyConnection:
    """An in-memory DuckDB connection with the spill directory, memory limit and threads set."""
    import duckdb

    settings = cfg or DuckDbConfig()
    con = duckdb.connect()
    con.execute("SET temp_directory = ?", [temp_directory().as_posix()])
    con.execute("SET memory_limit = ?", [settings.memory_limit])
    con.execute("SET threads = ?", [settings.threads or (os.cpu_count() or 4)])
    return con

src/rukh/data/db.pylíneas 35-51 · p1

temp_directory es el importante. Cuando una consulta no cabe en memoria, DuckDB vuelca a disco, y por defecto al temporal del sistema: en Windows, AppData\Local\Temp, que suele estar en el disco pequeño. El semi join de las evaluaciones puede volcar decenas de gigas. Con el volcado en <data>/duckdb-tmp, el que se llena, si se llena, es el disco donde ya decidiste que caben los datos.

memory_limit deja 32 de los 64 GB de la máquina de referencia para los procesos del pool, que corren a la vez; sin límite, DuckDB se reserva el 80 % de la RAM y los dos se pelean. Los valores se pasan con parámetros (SET memory_limit = ?) para que un valor mal escrito dé un error sobre el valor y no uno de sintaxis SQL.

Ese directorio va al .gitignore sin barra inicial, a diferencia de /data/, porque un volcado de DuckDB puede aparecer en cualquier subdirectorio desde el que se lance una consulta.

# Project data and runs (never in git)
/data/
.tmp/
duckdb-tmp/
/mlruns/
/tools/
/checkpoints/

.gitignorelíneas 13-19 · p1

DuckDbConfig vive aquí porque todos los pasos lo comparten: es una sección del pipeline.yaml que cada paso hereda (lo monta la lección siguiente). El campo duckdb de FetchConfig es el único cambio que M1 le hace al fetch.py de M0:

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)
    duckdb: DuckDbConfig = Field(default_factory=DuckDbConfig)

src/rukh/data/fetch.pylíneas 26-38 · p1

La última línea es la nueva. Con ella, el run de fetch pasa de duckdb.connect() a connect(cfg.duckdb) y hereda los tres ajustes. No hace falta repetir la descarga.

Si en M0 aplicaste el parche contra el HTTP 429, borra sus líneas al hacer este cambio, porque connect ya fija los hilos. Solo si tuvieras que relanzar la descarga, pon threads: 3 bajo duckdb: en el YAML y conserva el resto del parche (todo menos SET threads) justo después de con = connect(cfg.duckdb).

La otra línea que cambia excluye duckdb de los filtros del manifiesto: es configuración de la máquina, no de los datos.

    filters = cfg.model_dump(exclude={"dataset", "months", "out_dir", "min_plies", "duckdb"})
    filters["min_plies_deferred"] = cfg.min_plies

src/rukh/data/fetch.pylíneas 174-175 · p1

Y el paths.py de M0 gana una función, al final, después de tools_dir:

def resolve(rel: str | Path) -> Path:
    """Absolute path for a config entry: as is if absolute, otherwise under ``root()``."""
    path = Path(rel)
    return path if path.is_absolute() else root() / path

src/rukh/paths.pylíneas 41-44 · p1

Convierte out_dir: data/uci en una ruta absoluta bajo root(), que es lo que RUKH_HOME mueve. Todas las rutas del pipeline pasan por aquí, y por eso un test puede apuntar el proyecto entero a un directorio temporal con una variable de entorno.

// EjecútaloLa conexión, configurada

Con db.py y el resolve de paths.py escritos, abre una conexión y pregúntale a DuckDB qué ajustes tiene:

uv run python -c "from rukh.data.db import connect; con = connect(); print(con.execute('SELECT current_setting(?), current_setting(?), current_setting(?)', ['temp_directory', 'memory_limit', 'threads']).fetchone())"

El primer valor tiene que ser la ruta de data/duckdb-tmp dentro de tu repositorio (y el directorio aparece en disco), no el temporal del sistema. El límite de memoria sale en la unidad de DuckDB (unos 29,8 GiB para los 32 GB por defecto) y los hilos, tantos como núcleos.

Los tests: el contrato de la conversión

En tests/unit/test_uci.py, cada una de las cinco partidas del bloque siguiente cubre un caso que rompe implementaciones.

Los tests fijan el contrato de uci.py; léelos al teclearlos, no hacen falta para seguir la lección. El «Ejecútalo» de después del test del pool los pasa: si no los tecleas, tráelos antes con git checkout p1 -- tests/unit/test_uci.py.

// Teststests/unit/test_uci.py: las cinco partidas que rompen una conversión ingenua

"""Tests for rukh.data.uci: movetext cleaning, SAN to UCI, ply filters and month conversion."""

from __future__ import annotations

import json
from pathlib import Path

import polars as pl
import pytest
from typer.testing import CliRunner

from rukh.cli import app
from rukh.data.manifest import Manifest
from rukh.data.uci import (
    UciConfig,
    clean_movetext,
    convert_month,
    convert_rows,
    game_id,
    run,
    san_to_uci,
)

pytestmark = pytest.mark.unit

LICHESS = (
    "1. e4 { [%eval 0.3] [%clk 0:05:00] } e5 { [%clk 0:05:00] } 2. Nf3 { [%clk 0:04:58] } "
    "Nc6 3. Bb5 a6 4. Ba4 Nf6 5. O-O Be7 6. Re1 b5 7. Bb3 d6 8. c3 O-O 9. h3 Nb8 10. d4 Nbd7 "
    "1/2-1/2"
)
ILLEGAL = "1. e4 e5 2. Nf3 Nc6 3. Bb5 Nxe4 4. Qh5 Qh4 1-0"  # 3...Nxe4: c6 knight cannot go there
PROMO = (
    "1. e4 d5 2. exd5 c6 3. dxc6 Nf6 4. cxb7 Bd7 5. bxa8=Q Nc6 6. Qxc6 Bxc6 7. d4 e5 "
    "8. dxe5 Qxd1+ 9. Kxd1 Nd7 10. f4 Nb6 1-0"
)
SHORT = "1. e4 e5 2. Qh5 Nc6 3. Bc4 Nf6 4. Qxf7# 1-0"
CASTLES = (
    "1. d4 d5 2. Nf3 Nf6 3. Nc3 Nc6 4. Bf4 Bf5 5. Qd2 Qd7 6. O-O-O O-O-O 7. e3 e6 8. Bd3 Bd6 "
    "9. Ne5 Ne4 10. Bxe4 Bxe5 1-0"
)

tests/unit/test_uci.pylíneas 1-40 · p1

LICHESS trae comentarios de reloj y de evaluación para que la limpieza tenga trabajo. ILLEGAL lleva su motivo en un comentario, porque un test que descarta una partida sin decir por qué es imposible de mantener. PROMO promociona con captura (b7a8q). CASTLES enroca largo por los dos bandos, el caso que una traducción por cadenas siempre falla (O-O-O es e1c1, no e1a1). SHORT es un mate del pastor, para el filtro de longitud.

def test_clean_movetext_strips_comments_numbers_and_result() -> None:
    cleaned = clean_movetext("1. e4 { [%clk 0:05:00] } e5 { [%eval 0.2] } 2. Nf3 $1 1... Nc6 1-0")
    assert cleaned == "e4 e5 Nf3 Nc6"
    assert clean_movetext("1. d4 (1. e4 e5) d5 *") == "d4 d5"
    assert clean_movetext("") == ""


def test_san_to_uci_castling_and_promotion() -> None:
    uci, n = san_to_uci(LICHESS)
    assert n == 20
    assert uci.split()[8] == "e1g1"  # 5. O-O
    assert uci.split()[15] == "e8g8"  # 8... O-O
    uci, _ = san_to_uci(PROMO)
    assert "b7a8q" in uci.split()
    uci, _ = san_to_uci(CASTLES)
    assert uci.split()[10] == "e1c1" and uci.split()[11] == "e8c8"
    annotated = san_to_uci("1. e4 e5 2. Nf3 Nc6 3. Bb5!? a6 1-0")
    assert annotated == ("e2e4 e7e5 g1f3 b8c6 f1b5 a7a6", 6)


def test_san_to_uci_rejects_illegal_or_garbage() -> None:
    assert san_to_uci(ILLEGAL) is None
    assert san_to_uci("1. e4 e5 2. Zz9") is None
    assert san_to_uci("1. e5") is None


def test_game_id_is_stable_int64() -> None:
    value = game_id("https://lichess.org/abcd1234")
    assert value == game_id("https://lichess.org/abcd1234")
    assert -(2**63) <= value < 2**63
    assert value != game_id("https://lichess.org/abcd1235")

tests/unit/test_uci.pylíneas 43-73 · p1

Los índices de test_san_to_uci_castling_and_promotion (split()[8], split()[15]) son posiciones concretas: el noveno medio movimiento es 5. O-O. Buscar "e1g1" en la lista sería más legible, pero seguiría pasando si la conversión se desplazara una jugada.

test_san_to_uci_rejects_illegal_or_garbage separa los tres modos de fallo: una jugada legal solo en otra posición (Nxe4), una cadena que no es jugada (Zz9) y una bien escrita pero imposible desde la posición inicial (1. e5).

def _raw_rows() -> list[dict[str, object]]:
    def row(site: str, movetext: str, result: str) -> dict[str, object]:
        return {
            "Site": f"https://lichess.org/{site}",
            "movetext": movetext,
            "WhiteElo": 1900,
            "BlackElo": 1850,
            "Result": result,
            "TimeControl": "300+0",
            "UTCDate": "2025.01.01",
            "ECO": "C60",
            "month": "2025-01",
        }

    return [
        row("g1", LICHESS, "1/2-1/2"),
        row("g2", PROMO, "1-0"),
        row("g3", CASTLES, "1-0"),
        row("g4", ILLEGAL, "1-0"),
        row("g5", LICHESS.replace("1/2-1/2", "0-1"), "0-1"),
    ]


def test_convert_rows_counts() -> None:
    out = convert_rows(_raw_rows(), min_plies=20, max_plies=300)
    assert out["counts"] == {"rows": 5, "kept": 4, "illegal": 1, "short": 0, "long": 0}
    assert out["columns"]["n_plies"] == [20, 20, 20, 20]
    out = convert_rows(_raw_rows() + [_raw_rows()[0] | {"movetext": SHORT}], 20, 300)
    assert out["counts"]["short"] == 1
    out = convert_rows(_raw_rows(), min_plies=0, max_plies=10)
    assert out["counts"]["long"] == 4 and out["counts"]["kept"] == 0

tests/unit/test_uci.pylíneas 76-106 · p1

test_convert_rows_counts comprueba los contadores con los cinco casos tal cual, con una partida corta añadida (_raw_rows()[0] | {"movetext": SHORT}, la misma fila con un campo cambiado) y con un max_plies absurdo que descarta las cuatro.

def _write_raw(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    pl.DataFrame(_raw_rows()).write_parquet(path.as_posix())


def test_convert_month_keeps_four_of_five(tmp_path: Path) -> None:
    src = tmp_path / "raw.parquet"
    _write_raw(src)
    dst = tmp_path / "out" / "games.parquet"
    counts = convert_month(src, dst, UciConfig(workers=1))
    assert counts["rows"] == 5 and counts["kept"] == 4 and counts["illegal"] == 1
    frame = pl.read_parquet(dst.as_posix())
    assert frame.columns == [
        "game_id",
        "uci",
        "n_plies",
        "white_elo",
        "black_elo",
        "result",
        "time_control",
        "utc_date",
        "eco",
        "month",
    ]
    assert frame.schema["game_id"] == pl.Int64
    assert frame.schema["n_plies"] == pl.Int16
    assert frame.schema["white_elo"] == pl.Int16
    assert frame.schema["utc_date"] == pl.Date
    assert frame["result"].to_list() == ["1/2-1/2", "1-0", "1-0", "0-1"]
    assert frame["uci"][0].startswith("e2e4 e7e5 g1f3 b8c6 f1b5")
    assert str(frame["utc_date"][0]) == "2025-01-01"


def test_convert_month_with_two_workers(tmp_path: Path) -> None:
    """The spawn pool path: on Windows every worker re-imports the entry point module."""
    src = tmp_path / "raw.parquet"
    _write_raw(src)
    dst = tmp_path / "out" / "games.parquet"
    counts = convert_month(src, dst, UciConfig(workers=2))
    assert counts == {"rows": 5, "kept": 4, "illegal": 1, "short": 0, "long": 0}
    assert pl.read_parquet(dst.as_posix()).height == 4

tests/unit/test_uci.pylíneas 109-149 · p1

Este es el test que fija el esquema: las diez columnas en orden y los tipos. Parece frágil, y es lo que se quiere: el parquet lo leen cuatro pasos y dos módulos posteriores, así que cambiarlo tiene que costar tocar un test y pensar un momento. test_convert_month_with_two_workers lanza un pool de verdad y es el que falla si alguien convierte convert_batch en un lambda.

def test_run_writes_months_and_manifest(rukh_home: Path) -> None:
    for month in ("01", "02"):
        _write_raw(rukh_home / "data" / "raw" / "year=2025" / f"month={month}" / "games.parquet")
    manifest = run(UciConfig(workers=1))
    assert manifest.months == ["2025-01", "2025-02"]
    assert manifest.counts == {"2025-01": 4, "2025-02": 4}
    assert manifest.filters["min_plies"] == 20 and manifest.filters["max_plies"] == 300
    assert manifest.filters["conversion"]["2025-01"]["illegal"] == 1
    assert [f.path for f in manifest.files] == [
        "year=2025/month=01/games.parquet",
        "year=2025/month=02/games.parquet",
    ]
    on_disk = Manifest.model_validate_json(
        (rukh_home / "data" / "uci" / "manifest.json").read_text("utf-8")
    )
    assert on_disk.counts == manifest.counts


def test_cli_data_uci(rukh_home: Path) -> None:
    _write_raw(rukh_home / "data" / "raw" / "year=2025" / "month=01" / "games.parquet")
    cfg = rukh_home / "pipeline.yaml"
    cfg.write_text("uci:\n  workers: 1\n", encoding="utf-8")
    result = CliRunner().invoke(app, ["data", "uci", "--config", str(cfg), "--json"])
    assert result.exit_code == 0, result.output
    payload = json.loads(result.output)
    assert payload["counts"] == {"2025-01": 4}

tests/unit/test_uci.pylíneas 152-177 · p1

Los dos últimos usan la fixture rukh_home de M0 para escribir un data/raw/ falso en un directorio temporal. El último invoca la CLI de verdad con CliRunner y --json: garantiza que el comando existe, que acepta --config y que devuelve el manifiesto.

El test del pool: cómo se prueba la contrapresión

Igual que los de la conversión: si no lo tecleas, git checkout p1 -- tests/unit/test_parallel.py.

// Teststests/unit/test_parallel.py: comprobar que el pool no se adelanta

"""Tests for rukh.data.parallel: the feeder never runs ahead of the consumer."""

from __future__ import annotations

from collections.abc import Callable, Iterator

import pytest

from rukh.data.parallel import imap_bounded, run_batches
from rukh.data.uci import convert_batch

pytestmark = pytest.mark.unit


class _Immediate:
    def __init__(self, value: object) -> None:
        self.value = value

    def get(self) -> object:
        return self.value


class _FakePool:
    """Runs the job at submission time, so only the queue bound is under test."""

    def __init__(self) -> None:
        self.submitted = 0

    def apply_async(self, func: Callable[..., object], args: tuple[object, ...]) -> _Immediate:
        self.submitted += 1
        return _Immediate(func(*args))


def _counted(n: int, seen: list[int]) -> Iterator[int]:
    for i in range(n):
        seen.append(i)
        yield i


def test_imap_bounded_keeps_the_queue_short() -> None:
    seen: list[int] = []
    pool = _FakePool()
    produced: list[int] = []
    for value in imap_bounded(pool, lambda x: x * 2, _counted(50, seen), max_in_flight=4):
        # The feeder is at most ``max_in_flight`` batches ahead of the consumer.
        assert len(seen) - len(produced) <= 4
        produced.append(value)
    assert produced == [i * 2 for i in range(50)]
    assert pool.submitted == 50
    with pytest.raises(ValueError):
        next(imap_bounded(pool, lambda x: x, [1], 0))

tests/unit/test_parallel.pylíneas 1-51 · p1

_FakePool ejecuta el trabajo al enviarlo, así que lo único bajo prueba es el tamaño de la cola. El truco está en _counted: el generador apunta cada valor que produce, y comparar len(seen) con len(produced) dentro del bucle mide cuánto se ha adelantado el alimentador en cada iteración.

def _row(movetext: str) -> dict[str, object]:
    return {
        "Site": "https://lichess.org/x",
        "movetext": movetext,
        "WhiteElo": 1900,
        "BlackElo": 1850,
        "Result": "1-0",
        "TimeControl": "300+0",
        "UTCDate": "2025.01.01",
        "ECO": "C60",
        "month": "2025-01",
    }


def test_run_batches_bounds_a_real_spawn_pool() -> None:
    """With ``workers`` processes the generator stays within ``2 * workers + 1`` batches."""
    workers = 2
    materialised: list[int] = []

    def jobs() -> Iterator[tuple[list[dict[str, object]], int, int]]:
        for index in range(12):
            materialised.append(index)
            yield [_row("1. e4 e5 2. Nf3 Nc6")], 0, 300

    consumed = 0
    for result in run_batches(convert_batch, jobs(), workers):
        consumed += 1
        assert len(materialised) - consumed <= 2 * workers + 1
        assert result["counts"]["kept"] == 1  # type: ignore[index]
    assert consumed == 12


def test_run_batches_is_serial_with_one_worker() -> None:
    assert list(run_batches(lambda x: x + 1, [1, 2, 3], workers=1)) == [2, 3, 4]

tests/unit/test_parallel.pylíneas 54-87 · p1

Y este comprueba lo mismo sobre un pool de spawn real con dos procesos, con margen de uno porque el trabajo que se está enviando cuenta. Cuesta segundos, y paga: que doce lotes no se materialicen los doce no se puede comprobar con un pool falso, donde el orden es trivialmente correcto.

// EjecútaloLos tests de la conversión y del pool

Con los tres ficheros y sus dos tests escritos, pasa los dos ficheros. El último test de test_uci.py invoca rukh data uci, un comando que carga el pipeline.py de la lección siguiente, así que de momento se deja fuera:

uv run pytest -m unit -q tests/unit/test_uci.py tests/unit/test_parallel.py -k "not cli"

Once tests en verde y uno deseleccionado. Los dos que lanzan un pool de spawn de verdad son los que tardan un par de segundos; si alguno falla con Can't pickle local object, hay un punto de entrada que no está a nivel de módulo. Sin -k, test_cli_data_uci falla en su exit_code == 0: se pasa en la lección siguiente, en cuanto escribas la CLI y el pipeline.py.

// EjecútaloLa conversión de verdad, sobre los seis millones

Con los tests en verde ya puedes convertir los dos meses que dejó M0. El comando rukh data uci llega en la lección siguiente, pero no hace falta esperarlo: el paso entero es la función run, y su configuración por defecto es la misma que tendrá el pipeline.yaml.

uv run python -c "from rukh.data.uci import UciConfig, run; print(run(UciConfig()).counts)"

Tarda entre diez y veinte minutos con todos los núcleos ocupados y la memoria estable (es la contrapresión de parallel.py haciendo su trabajo). Al terminar tienen que existir data/uci/year=2025/month=01/games.parquet, su gemelo de febrero y data/uci/manifest.json; en la ejecución de referencia los conteos fueron 2 949 514 y 2 946 874, con cero ilegales en filters.conversion del manifiesto.

// Ejercicio 01Quítale el límite al pool y mira la memoria

En una copia del repositorio, cambia run_batches para que use pool.imap(func, jobs) de la biblioteca estándar en vez de imap_bounded, y ejecuta uv run rukh data uci --workers 4 (el comando de la lección siguiente) sobre un mes de verdad mirando el consumo de memoria del proceso. ¿Cuánto tarda en subir y cuánto sube? Después ejecuta uv run pytest -m unit -q tests/unit/test_parallel.py: ¿qué test falla y en qué aserción?

// SoluciónVer la solución

La memoria sube de forma sostenida desde el primer segundo, porque el alimentador recorre el mes entero construyendo lotes de 20 000 filas con su movetext; en la máquina de referencia pasa de varios gigas en menos de un minuto, y con un mes completo acaba en un MemoryError o en el sistema paginando.

Falla test_run_batches_bounds_a_real_spawn_pool, en el assert len(materialised) - consumed <= 2 * workers + 1. Con doce lotes el número real es 12: imap los materializa todos antes de que el primero vuelva. Y fíjate en que test_imap_bounded_keeps_the_queue_short sigue pasando, porque prueba la función que ya no se usa: un test verde sobre código muerto es lo que el segundo test existe para evitar.

Qué has aprendido

Has escrito el primer paso del pipeline: la limpieza del movetext en el orden correcto, la conversión a UCI sobre un tablero de verdad, un id estable que no depende de hash() y un pool de procesos que no se traga un mes en memoria antes de escribir una fila.

Cómo se mide: uv run rukh data uci deja data/uci/year=2025/month=01/games.parquet y su gemelo de febrero, con 2 949 514 y 2 946 874 partidas, y un manifest.json cuyos contadores suman las seis millones leídas con cero ilegales. uv run pytest -m unit -q tests/unit/test_uci.py tests/unit/test_parallel.py pasa los doce casos en menos de diez segundos, el último en cuanto escribas la CLI y el pipeline.py en la lección siguiente.

Lo siguiente es el orquestador: nueve pasos como este comparten un único YAML y una sección de DuckDB, y la CLI gana siete subcomandos nuevos.