typestar

pl_pipeline.py en Python

Un pipeline perezoso de Polars: escanear, limpiar, ventanas, agregar, unir y escribir.

from pathlib import Path

import polars as pl

CSV_INTENTOS = """day,lang,chars,seconds,errors
2026-07-20,python,1820,60.0,4
2026-07-20,rust,1510,60.0,9
2026-07-21,python,1980,60.0,2
2026-07-21,sql,1240,60.0,1
2026-07-22,python,2050,60.0,3
2026-07-22,rust,1660,60.0,5
2026-07-24,css,980,45.0,7
2026-07-29,rust,1740,60.0,2
2026-07-29,,900,60.0,0
"""

TOURS = pl.DataFrame({
    "lang": ["python", "rust", "sql", "css"],
    "tours": [11, 12, 3, 3],
})


def armar_plan(ruta: Path) -> pl.LazyFrame:
    """Un plan: limpia, deriva, clasifica por lenguaje y une los tours."""
    return (
        pl.scan_csv(ruta, try_parse_dates=True)
        .filter(pl.col("lang").is_not_null() & (pl.col("chars") > 0))
        .with_columns(
            ((pl.col("chars") / 5) / (pl.col("seconds") / 60))
            .round(1).alias("wpm"),
            ((1 - pl.col("errors") / pl.col("chars")) * 100)
            .round(2).alias("accuracy"),
        )
        .with_columns(
            pl.col("wpm").mean().over("lang").round(1).alias("lang_mean"),
            pl.col("wpm").rank(descending=True).over("lang")
            .cast(pl.Int32).alias("rank_in_lang"),
            pl.when(pl.col("accuracy") >= 99.5).then(pl.lit("limpio"))
            .when(pl.col("accuracy") >= 99.0).then(pl.lit("bueno"))
            .otherwise(pl.lit("flojo")).alias("grade"),
        )
        .join(TOURS.lazy(), on="lang", how="left")
        .sort(["lang", "rank_in_lang"])
    )


def resumir(intentos: pl.DataFrame) -> pl.DataFrame:
    return (
        intentos.group_by("lang")
        .agg(
            pl.len().alias("runs"),
            pl.col("wpm").max().alias("best_wpm"),
            pl.col("wpm").mean().round(1).alias("mean_wpm"),
            pl.col("accuracy").mean().round(2).alias("mean_accuracy"),
            (pl.col("grade") == "limpio").sum().alias("clean_runs"),
            pl.col("tours").first().alias("tours"),
        )
        .sort("mean_wpm", descending=True)
    )


def diario(intentos: pl.DataFrame) -> pl.DataFrame:
    return (
        intentos.sort("day")
        .group_by_dynamic("day", every="3d")
        .agg(pl.col("wpm").mean().round(1).alias("mean_wpm"),
             pl.len().alias("runs"))
    )


def principal() -> None:
    fuente = Path("intentos.csv")
    fuente.write_text(CSV_INTENTOS)
    plan = armar_plan(fuente)

    print("--- plan ---")
    print(plan.explain(optimized=True).splitlines()[0])

    intentos = plan.collect()
    print(f"\n{intentos.height} intentos conservados de 9 filas leídas")

    print("\n--- por lenguaje ---")
    print(resumir(intentos))

    print("\n--- cada tres días ---")
    print(diario(intentos))

    mejor = intentos.sort("wpm", descending=True).row(0, named=True)
    print(f"\nmejor intento: {mejor['lang']} a {mejor['wpm']} wpm "
          f"el {mejor['day']}")

    intentos.write_parquet("intentos.parquet")
    print("escrito", pl.read_parquet("intentos.parquet").shape)


if __name__ == "__main__":
    principal()

Cómo funciona

  1. Todo es un solo plan de LazyFrame hasta el único collect del final.
  2. El optimizador empuja filtros y proyecciones hacia abajo por ti.
  3. El plan se imprime primero, así ves qué va a correr en realidad.

Palabras clave y builtins usados aquí

El intento, en números

Líneas
97
Caracteres a escribir
2548
Tokens
833
Ritmo de tres estrellas
115 tpm

Al ritmo de tres estrellas de 115 tokens por minuto, este intento toma unos 435 segundos.

Escribe este fragmento

Paso 2 de 2 en Bis; paso 32 de 32 en Polars.

← Anterior