DuckDB y SQL esencial: SELECT, JOIN, GROUP BY y funciones de ventana sobre Parquet.

SQL sin servidor

SQL (Structured Query Language) describe qué datos se quieren, no cómo obtenerlos: el motor decide el orden de lectura, los índices y los algoritmos. Es el lenguaje de las bases de datos de órdenes y posiciones de los brókers, de los almacenes de datos de mercado y de buena parte de las herramientas de research en la nube. Saber leerlo y escribirlo es tan básico como pandas.

Para usarlo hace falta un motor. Las bases de datos clásicas (PostgreSQL, MySQL) son un servidor aparte al que el programa se conecta. DuckDB es otra cosa: una base de datos analítica que se instala como una librería (uv add duckdb) y se ejecuta dentro del proceso de Python, como SQLite, pero diseñada para consultas que leen y agregan millones de filas por columnas. Y lee directamente los ficheros: un Parquet o un CSV se consultan por su ruta, sin importarlos antes.

La versión del curso es DuckDB 1.5.6 (fijada en curso-code/uv.lock).

import duckdb

con = duckdb.connect()   # base de datos en memoria
print(duckdb.__version__)
con.sql("""
    SELECT fecha, cierre, volumen
    FROM 'sint_diario.parquet'
    ORDER BY fecha
    LIMIT 3
""")

1.5.6
┌─────────────────────┬────────┬─────────┐
│        fecha        │ cierre │ volumen │
│      timestamp      │ double │  int64  │
├─────────────────────┼────────┼─────────┤
│ 2021-01-04 00:00:00 │  99.14 │ 2322609 │
│ 2021-01-05 00:00:00 │  99.25 │ 1517954 │
│ 2021-01-06 00:00:00 │  97.38 │ 3245658 │
└─────────────────────┴────────┴─────────┘

duckdb.connect() sin argumentos crea una base de datos en memoria, que desaparece al terminar; con una ruta (duckdb.connect("research.duckdb")) la guarda en un fichero. con.sql devuelve una relación: el resultado de la consulta, que se imprime como tabla y se puede convertir a otros formatos. Debajo de cada columna aparece su tipo SQL: timestamp (fecha con hora), double (número de coma flotante) y int64.

SELECT, WHERE y ORDER BY

Una consulta tiene siempre la misma estructura. Lo que cambia con la experiencia es saber que el orden en que se escribe no es el orden en que se ejecuta:

flowchart TB
  A[FROM y JOIN] --> B[WHERE]
  B --> C[GROUP BY y HAVING]
  C --> D[SELECT y ventanas]
  D --> E[ORDER BY y LIMIT]

Primero se leen y combinan las tablas, después se descartan filas con WHERE, se agrupan, se calculan las columnas del SELECT y al final se ordena y se recorta. Ese orden explica errores frecuentes: WHERE no puede usar un alias definido en el SELECT ni el resultado de una función de ventana, porque todavía no existen cuando se evalúa.

con.sql("""
    SELECT fecha::DATE AS dia, cierre, volumen
    FROM 'sint_diario.parquet'
    WHERE year(fecha) = 2022 AND volumen > 3500000
    ORDER BY volumen DESC
    LIMIT 5
""")

┌────────────┬────────┬─────────┐
│    dia     │ cierre │ volumen │
│    date    │ double │  int64  │
├────────────┼────────┼─────────┤
│ 2022-04-22 │  99.38 │ 5400080 │
│ 2022-08-26 │ 112.63 │ 5010487 │
│ 2022-11-28 │  96.98 │ 4662485 │
│ 2022-05-23 │ 109.61 │ 4342762 │
│ 2022-12-05 │  99.44 │ 3676322 │
└────────────┴────────┴─────────┘

fecha::DATE convierte la marca de tiempo en fecha (el Parquet la guardó con hora, como se vio en la lección 0.5.4) y AS dia da nombre a la columna resultante. year() es una de las funciones de fecha de DuckDB. Las condiciones se combinan con AND, OR y NOT, sin los paréntesis obligatorios de pandas.

DuckDB lee del Parquet solo las columnas y los bloques que la consulta necesita, igual que el modo lazy de Polars: EXPLAIN antes de una consulta muestra el plan, con las columnas proyectadas (Projections) y los filtros empujados al lector (Filters).

GROUP BY: resúmenes por año

GROUP BY agrupa las filas con el mismo valor de la clave y las funciones de agregación (count, min, max, avg, sum…) reducen cada grupo a una fila. Toda columna del SELECT debe estar en el GROUP BY o dentro de una agregación:

con.sql("""
    SELECT year(fecha)               AS ano,
           count(*)                  AS sesiones,
           round(min(cierre), 2)     AS minimo,
           round(max(cierre), 2)     AS maximo,
           arg_max(cierre, fecha)    AS ultimo,
           round(avg(volumen) / 1e6, 2) AS vol_medio_m
    FROM 'sint_diario.parquet'
    GROUP BY ano
    ORDER BY ano
""")

┌───────┬──────────┬────────┬────────┬────────┬─────────────┐
│  ano  │ sesiones │ minimo │ maximo │ ultimo │ vol_medio_m │
│ int64 │  int64   │ double │ double │ double │   double    │
├───────┼──────────┼────────┼────────┼────────┼─────────────┤
│  2021 │      260 │  95.38 │ 115.78 │ 114.21 │        1.89 │
│  2022 │      260 │  94.45 │ 125.98 │  104.5 │         1.9 │
│  2023 │      260 │ 101.72 │  121.8 │ 119.35 │         1.8 │
│  2024 │      262 │ 118.36 │ 137.15 │ 136.44 │        1.87 │
│  2025 │      261 │ 122.38 │ 149.74 │ 142.22 │        1.74 │
└───────┴──────────┴────────┴────────┴────────┴─────────────┘

arg_max(cierre, fecha) devuelve el cierre de la fila con la fecha máxima del grupo: el último cierre de cada año. Es una función de DuckDB muy útil con series temporales, y evita una subconsulta. 2022 tiene el máximo más alto de los tres primeros años (125,98) y también el mínimo más bajo de todo el periodo (94,45): el año de la caída. DuckDB permite usar el alias ano en el GROUP BY, una comodidad que no todos los motores SQL admiten.

HAVING filtra después de agrupar, sobre los resultados de las agregaciones. Para quedarse con los años cuyo mínimo bajó de 100, la condición va en HAVING min(cierre) < 100, no en WHERE.

JOIN: operaciones y precios

JOIN combina filas de dos tablas que cumplen una condición, como merge en pandas (lección 0.5.3). Las dos tablas pueden venir de ficheros distintos, incluso de formatos distintos: el libro de operaciones está en CSV y los precios en Parquet.

con.sql("""
    SELECT o.id, o.fecha_entrada, o.lado,
           o.precio_entrada, p.cierre, p.volumen
    FROM 'operaciones_sint.csv' AS o
    JOIN 'sint_diario.parquet' AS p
      ON p.fecha = o.fecha_entrada
    WHERE o.simbolo = 'SINT'
    ORDER BY o.id
    LIMIT 3
""")

┌───────┬───────────────┬─────────┬────────────────┬────────┬─────────┐
│  id   │ fecha_entrada │  lado   │ precio_entrada │ cierre │ volumen │
│ int64 │     date      │ varchar │     double     │ double │  int64  │
├───────┼───────────────┼─────────┼────────────────┼────────┼─────────┤
│     2 │ 2021-09-02    │ largo   │         108.64 │ 108.64 │ 1454984 │
│     3 │ 2021-10-27    │ corto   │         112.07 │ 112.07 │ 1644654 │
│     4 │ 2021-11-10    │ corto   │          104.3 │  104.3 │ 2575784 │
└───────┴───────────────┴─────────┴────────────────┴────────┴─────────┘

AS o y AS p son alias para no repetir los nombres de los ficheros. La condición ON compara una fecha del CSV (DATE) con una marca de tiempo del Parquet (TIMESTAMP); DuckDB convierte la primera a medianoche y la comparación funciona. Con marcas de tiempo con hora, la misma unión dejaría de casar sin dar error: comprueba siempre los tipos de las claves.

La agregación funciona sobre el resultado de la unión como sobre cualquier tabla. El resultado neto de cada símbolo, con el signo de la operación escrito con CASE:

con.sql("""
    SELECT simbolo,
           count(*) AS operaciones,
           round(sum(
               CASE WHEN lado = 'largo' THEN 1 ELSE -1 END
               * (precio_salida - precio_entrada) * cantidad
               - comision), 2) AS total
    FROM 'operaciones_sint.csv'
    GROUP BY simbolo
    ORDER BY total DESC
""")

┌─────────┬─────────────┬──────────┐
│ simbolo │ operaciones │  total   │
│ varchar │    int64    │  double  │
├─────────┼─────────────┼──────────┤
│ SINT    │          30 │  4405.08 │
│ RVAR    │           4 │    17.09 │
│ TECH    │           3 │   -35.02 │
│ ORO     │           3 │ -3237.96 │
└─────────┴─────────────┴──────────┘

Los totales son los mismos que con pandas y Polars en las lecciones anteriores: SINT gana 4.405,08 y ORO pierde 3.237,96. Tres herramientas distintas, el mismo número: cuando se reescribe un cálculo en otra herramienta, comparar el resultado con el de la original es la prueba más barata de que no se ha colado un error.

Funciones de ventana

Una función de ventana calcula, para cada fila, un valor a partir de otras filas relacionadas, sin agruparlas: la tabla conserva todas sus filas. Se escribe con OVER (...), que define tres cosas:

• PARTITION BY: los grupos independientes (por ejemplo, un símbolo). Sin él, toda la tabla es una partición.
• ORDER BY: el orden de las filas dentro de la partición. En series temporales, siempre la fecha.
• El marco (ROWS BETWEEN … AND …): qué filas, relativas a la actual, entran en el cálculo.

La rentabilidad diaria necesita el cierre anterior. LAG(cierre) lo trae a la fila actual:

con.sql("""
    CREATE TABLE sint AS
    SELECT fecha::DATE AS fecha, cierre,
           cierre / LAG(cierre) OVER (ORDER BY fecha) - 1 AS r
    FROM 'sint_diario.parquet'
""")
con.sql("""
    SELECT fecha, cierre, round(r, 6) AS r
    FROM sint ORDER BY fecha LIMIT 3
""")

┌────────────┬────────┬───────────┐
│   fecha    │ cierre │     r     │
│    date    │ double │  double   │
├────────────┼────────┼───────────┤
│ 2021-01-04 │  99.14 │      NULL │
│ 2021-01-05 │  99.25 │   0.00111 │
│ 2021-01-06 │  97.38 │ -0.018841 │
└────────────┴────────┴───────────┘

CREATE TABLE ... AS guarda el resultado en una tabla de la base de datos en memoria, para no repetir la lectura en las consultas siguientes. La primera fila no tiene cierre anterior y su rentabilidad es NULL, como el NaN de pandas. LAG solo mira hacia atrás; su pareja, LEAD, mira hacia delante y es la forma más directa de introducir anticipación en una señal.

Medias móviles con marco explícito

Una media móvil de 50 sesiones es la media de la fila actual y las 49 anteriores. Se escribe con un marco ROWS explícito, que cuenta filas (sesiones) y no fechas:

con.sql("""
    SELECT row_number() OVER (ORDER BY fecha) AS fila,
           fecha, cierre,
           round(AVG(cierre) OVER w, 2) AS media_50,
           COUNT(cierre) OVER w AS n
    FROM sint
    WINDOW w AS (ORDER BY fecha
                 ROWS BETWEEN 49 PRECEDING AND CURRENT ROW)
    QUALIFY fila IN (1, 2, 49, 50, 51)
    ORDER BY fecha
""")

┌───────┬────────────┬────────┬──────────┬───────┐
│ fila  │   fecha    │ cierre │ media_50 │   n   │
│ int64 │    date    │ double │  double  │ int64 │
├───────┼────────────┼────────┼──────────┼───────┤
│     1 │ 2021-01-04 │  99.14 │    99.14 │     1 │
│     2 │ 2021-01-05 │  99.25 │     99.2 │     2 │
│    49 │ 2021-03-11 │ 101.76 │     97.6 │    49 │
│    50 │ 2021-03-12 │ 104.72 │    97.74 │    50 │
│    51 │ 2021-03-15 │ 106.21 │    97.88 │    50 │
└───────┴────────────┴────────┴──────────┴───────┘

WINDOW w AS (...) define el marco una vez para varias columnas, y row_number() numera las filas para mostrar solo unas cuantas con QUALIFY (el filtro que se aplica después de las ventanas; se explica más abajo). La columna n revela la diferencia más importante con rolling(50) de pandas: SQL calcula con las filas que haya en el marco. La primera fila tiene una «media de 50 sesiones» que es su propio cierre, y la 49 promedia 49 cierres; pandas devolvería NaN en las 49 primeras. Desde la fila 50 (97,74) los dos coinciden. Para obtener un nulo con ventanas incompletas, se envuelve en CASE WHEN COUNT(cierre) OVER w = 50 THEN AVG(cierre) OVER w END, o se filtra por n como en el ejercicio 3.

La diferencia entre un marco que mira al pasado y uno que usa toda la tabla se ve en una sola fila:

con.sql("""
    SELECT fecha, cierre,
           round(AVG(cierre) OVER (), 2) AS media_toda,
           round(AVG(cierre) OVER (
               ORDER BY fecha
               ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
           ), 2) AS media_hasta_hoy
    FROM sint
    QUALIFY fecha = DATE '2021-06-01'
""")

┌────────────┬────────┬────────────┬─────────────────┐
│   fecha    │ cierre │ media_toda │ media_hasta_hoy │
│    date    │ double │   double   │     double      │
├────────────┼────────┼────────────┼─────────────────┤
│ 2021-06-01 │  99.56 │     117.26 │           100.2 │
└────────────┴────────┴────────────┴─────────────────┘

El 1 de junio de 2021, la media de toda la serie (117,26) incluye cuatro años y medio de precios futuros y está muy por encima de la media real hasta ese día (100,2). Una regla como «comprar si el cierre está por debajo de su media» parecería excelente con la primera columna en un activo que sube, y no se podría operar. QUALIFY filtra después de calcular las ventanas: con WHERE, el filtro se aplicaría antes y la media se calcularía sobre una sola fila.

Máximo acumulado y drawdown

El drawdown necesita el máximo de todos los cierres hasta hoy: un marco que empieza en UNBOUNDED PRECEDING. Y la volatilidad móvil, la desviación típica de las rentabilidades en un marco de 21 filas:

riesgo = con.sql("""
    SELECT fecha, cierre,
           cierre / MAX(cierre) OVER (
               ORDER BY fecha
               ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
           ) - 1 AS dd,
           STDDEV_SAMP(r) OVER (
               ORDER BY fecha
               ROWS BETWEEN 20 PRECEDING AND CURRENT ROW
           ) * sqrt(252) AS vol_21
    FROM sint
""")
con.sql("""
    SELECT round(min(dd), 4) AS peor_dd,
           arg_min(fecha, dd) AS fondo,
           round(max(vol_21), 4) AS vol_max
    FROM riesgo
""")

┌─────────┬────────────┬─────────┐
│ peor_dd │   fondo    │ vol_max │
│ double  │    date    │ double  │
├─────────┼────────────┼─────────┤
│ -0.2503 │ 2022-10-20 │  0.3395 │
└─────────┴────────────┴─────────┘

El mismo −25,0 % con fondo el 20 de octubre de 2022 y la misma volatilidad máxima del 33,9 % que en pandas y Polars. Una relación se puede consultar por el nombre de su variable de Python (FROM riesgo): DuckDB encadena las dos consultas sin materializar la primera. Como en la media móvil, las primeras filas tienen una volatilidad calculada con menos de 21 rentabilidades (STDDEV_SAMP ignora el NULL inicial), donde pandas daba NaN; el máximo, lejos del principio, es el mismo.

De SQL a pandas y Polars, y vuelta

Una consulta de DuckDB termina normalmente en un DataFrame. Los métodos de la relación lo hacen directamente:

import pandas as pd

df_pd = riesgo.df()
df_pl = riesgo.pl()
print(type(df_pd).__name__, df_pd.shape, df_pd["fecha"].dtype)
print(type(df_pl).__name__, df_pl.shape, df_pl.schema["fecha"])

precios = pd.read_csv("sint_diario.csv", parse_dates=["fecha"])
con.sql("SELECT count(*) AS sesiones FROM precios WHERE cierre > 120")

DataFrame (1303, 4) datetime64[us]
DataFrame (1303, 4) Date
┌──────────┐
│ sesiones │
│  int64   │
├──────────┤
│      523 │
└──────────┘

.df() devuelve pandas y .pl() devuelve Polars (también hay .arrow(), que devuelve datos de Arrow, y .fetchall(), que devuelve tuplas de Python). La última consulta va en sentido contrario: precios no es una tabla de la base de datos, sino un DataFrame de pandas en una variable de Python, y DuckDB lo encuentra por su nombre y lo consulta en el sitio, sin copiarlo. Las 523 sesiones por encima de 120 son las de la lección 0.5.2.

Las tres herramientas se combinan sin fricción porque comparten Apache Arrow como formato en memoria. Un flujo habitual: DuckDB filtra y agrega los ficheros grandes, y pandas recibe la tabla ya pequeña para el análisis.

Ejercicios

1. Con una consulta sobre sint_diario.parquet, calcula la rentabilidad media diaria y el número de sesiones de cada día de la semana (usa dayname(fecha) y LAG). Compara con la tabla de la lección 0.5.3.

con.sql("""
    SELECT dayname(fecha) AS dia,
           count(r) AS sesiones,
           round(avg(r) * 100, 3) AS media_pct
    FROM sint
    GROUP BY dia, isodow(fecha)
    ORDER BY isodow(fecha)
""")

┌───────────┬──────────┬───────────┐
│    dia    │ sesiones │ media_pct │
│  varchar  │  int64   │  double   │
├───────────┼──────────┼───────────┤
│ Monday    │      260 │     0.045 │
│ Tuesday   │      261 │    -0.052 │
│ Wednesday │      261 │     0.019 │
│ Thursday  │      260 │     0.051 │
│ Friday    │      260 │     0.108 │
└───────────┴──────────┴───────────┘

2. Une cada operación de SINT con el cierre del día de salida y comprueba que el precio de salida es ese cierre en todas. Usa LEFT JOIN y cuenta las operaciones sin pareja.

con.sql("""
    SELECT count(*) AS operaciones,
           count(p.cierre) AS con_precio,
           sum(CASE WHEN o.precio_salida = p.cierre
                    THEN 1 ELSE 0 END) AS iguales
    FROM 'operaciones_sint.csv' AS o
    LEFT JOIN sint AS p ON p.fecha = o.fecha_salida
    WHERE o.simbolo = 'SINT'
""")

┌─────────────┬────────────┬─────────┐
│ operaciones │ con_precio │ iguales │
│    int64    │   int64    │ int128  │
├─────────────┼────────────┼─────────┤
│          30 │         30 │      30 │
└─────────────┴────────────┴─────────┘

3. Escribe una señal de cruce sin anticipación: 1 si el cierre está por encima de su media de 50 sesiones (con la ventana completa) y 0 si no. ¿En qué porcentaje de las sesiones con ventana completa estuvo SINT por encima?

con.sql("""
    WITH m AS (
        SELECT fecha, cierre,
               AVG(cierre) OVER w AS media,
               COUNT(cierre) OVER w AS n
        FROM sint
        WINDOW w AS (ORDER BY fecha
                     ROWS BETWEEN 49 PRECEDING AND CURRENT ROW)
    )
    SELECT count(*) AS sesiones,
           round(avg(CASE WHEN cierre > media THEN 1 ELSE 0 END), 3)
               AS por_encima
    FROM m
    WHERE n = 50
""")

┌──────────┬────────────┐
│ sesiones │ por_encima │
│  int64   │   double   │
├──────────┼────────────┤
│     1254 │      0.649 │
└──────────┴────────────┘