Dask, Spark y Ray: de una máquina a un clúster.

Cuándo hace falta más de una máquina

La lección 6.6.2 llevó una sola máquina lejos: 21 millones de ticks con Polars en streaming y DuckDB con 300 MB de memoria. Hay tres situaciones en las que eso deja de bastar:

• Los datos no caben ni en el disco de una máquina, o leerlos con un solo disco y una sola tarjeta de red tarda demasiado: años de libros L3 de un universo global, cadenas de opciones completas (lección 6.6.1).
• El cálculo es enorme y se parte bien: miles de backtests, un barrido de parámetros, una validación cruzada con muchos pliegues, cada uno independiente de los demás.
• Varias personas o procesos comparten los mismos recursos: un equipo que lanza trabajos sobre el mismo lago.

Para la primera, los marcos de datos distribuidos (Dask DataFrame, Spark) parten una tabla en trozos y reparten los trozos entre máquinas. Para la segunda, los marcos de tareas distribuidas (Dask delayed, Ray) reparten funciones de Python. Los dos comparten la misma arquitectura: un proceso que planifica y varios que trabajan.

flowchart TB
  C[Cliente: tu código] --> S[Planificador]
  S --> W1[Trabajador 1]
  S --> W2[Trabajador 2]
  S --> W3[Trabajador 3]
  W1 --> L[(Lago Parquet)]
  W2 --> L
  W3 --> L

En Dask el planificador se llama scheduler; en Spark, driver, y los trabajadores, executors; en Ray, cada nodo tiene un raylet que planifica localmente y un servicio central (GCS) guarda el estado del clúster. Los nombres cambian; la idea, no. Y tiene un coste que conviene tener presente desde el principio: mover datos entre procesos o máquinas exige serializarlos y enviarlos por la red, que es mucho más lento que leerlos de la memoria propia. La documentación de Dask lo dice sin rodeos: el paralelismo añade complejidad y sobrecarga, a veces es necesario para problemas grandes, «pero a menudo no lo es»11Nota 1 · Dask, «Dask Best Practices»: https://docs.dask.org/en/stable/best-practices.html, consultado el 23 de octubre de 2026..

Esta lección ejecuta Dask en una máquina (su planificador local con hilos), que es lo mismo que haría en un clúster con otro planificador, y enseña el código de Spark y Ray sin ejecutarlo, por las razones que se explican en cada sección. Los datos son ticks sintéticos generados con el mismo método que en la lección 6.6.2, más pequeños: las 39 sesiones de la NYSE de enero y febrero de 2025 (el 9 de enero fue un cierre extraordinario, día de luto nacional por el expresidente Carter, que el calendario XNYS de exchange_calendars recoge como festivo) para 200 acciones, un fichero por sesión.

from pathlib import Path

import exchange_calendars as xcals
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

tickers = np.array(
    sorted(pd.read_csv("panel5_activos.csv")["ticker"])[:200])
cal = xcals.get_calendar("XNYS")
sesiones = cal.sessions_in_range("2025-01-02", "2025-02-28")
rng = np.random.default_rng(663)
actividad = np.exp(rng.normal(np.log(1000), 0.8, 200))
ultimo = np.exp(rng.uniform(np.log(10), np.log(400), 200))

def ticks_de_una_sesion(sesion):
    global ultimo
    abre = cal.session_open(sesion).value // 1000   # µs, UTC
    cierra = cal.session_close(sesion).value // 1000
    n = rng.poisson(actividad)
    sim = np.repeat(np.arange(200), n)
    N = len(sim)
    inicio = np.repeat(np.cumsum(n) - n, n)
    ts = rng.integers(abre, cierra - 100_000, N)
    ts = ts[np.lexsort((ts, sim))] + (np.arange(N) - inicio)
    paso = rng.normal(0, 1, N) * np.repeat(
        0.02 / np.sqrt(np.maximum(n, 1)), n)
    c = np.concatenate([[0.0], np.cumsum(paso)])
    camino = c[1:] - c[np.cumsum(n) - n].repeat(n)
    precio = np.round(np.repeat(ultimo, n) * np.exp(camino), 2)
    ultimo = np.where(n > 0, precio[np.cumsum(n) - 1], ultimo)
    return pa.table({
        "ts": pa.array(ts, pa.timestamp("us", tz="UTC")),
        "simbolo": pa.array(tickers).take(sim),
        "precio": precio,
        "tamano": rng.geometric(0.02, N).astype(np.int32)})

filas = 0
for s in sesiones:
    t = ticks_de_una_sesion(s)
    ruta = Path(f"ticks/fecha={s.date()}/parte-0.parquet")
    ruta.parent.mkdir(parents=True)
    pq.write_table(t, ruta, compression="zstd")
    filas += t.num_rows
print(f"{filas:,} ticks, {len(sesiones)} ficheros")

9,800,152 ticks, 39 ficheros

Dask DataFrame: pandas por particiones

Un DataFrame de Dask es una colección de DataFrame de pandas, las particiones, ordenadas a lo largo del índice y repartidas por filas22Nota 2 · Dask, «Dask DataFrame» y «Scheduling» (el planificador de hilos es el predeterminado para Dask DataFrame): https://docs.dask.org/en/stable/dataframe.html, consultado el 23 de octubre de 2026.. Cada operación sobre él se traduce en una operación de pandas sobre cada partición más, cuando hace falta, un paso que combina los resultados. Con read_parquet sobre un lago particionado, cada fichero es una partición:

import dask
import dask.dataframe as dd

ddf = dd.read_parquet("ticks", columns=["simbolo", "precio",
                                        "tamano", "fecha"])
print(type(ddf).__name__, "con", ddf.npartitions, "particiones")
print(ddf.dtypes)
ddf.partitions[0].compute().head(3)

DataFrame con 39 particiones
simbolo      string
precio      float64
tamano        int32
fecha      category
dtype: object
  simbolo  precio  tamano       fecha
0    BAGO   34.29     164  2025-01-02
1    BAGO   34.27      20  2025-01-02
2    BAGO   34.25      46  2025-01-02

Crear ddf no ha leído ningún dato: solo los metadatos para conocer el esquema y el número de particiones. La columna fecha sale de los nombres de carpeta como category, igual que con pandas (lección 1.4.7). ddf.partitions[0] selecciona la primera partición y .compute() la convierte en un DataFrame de pandas: la sesión del 2 de enero.

Una consulta perezosa

Como Polars lazy, Dask construye un plan antes de ejecutar nada. El VWAP de cada símbolo en los dos meses:

c = ddf.assign(importe=ddf.precio * ddf.tamano)
por_simbolo = c.groupby("simbolo")
vwap = por_simbolo.importe.sum() / por_simbolo.tamano.sum()
vwap.pprint()

OpAlignPartitions: op='__truediv__'
  Sum: observed=True chunk_kwargs={'numeric_only': False} aggregate_kwargs={'numeric_only': False} _slice='importe''simbolo'
    Projection: columns=['simbolo', 'importe']
      Assign: importe=
        ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
        Mul:
          Projection: columns='precio'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
          Projection: columns='tamano'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
  Sum: observed=True chunk_kwargs={'numeric_only': False} aggregate_kwargs={'numeric_only': False} _slice='tamano''simbolo'
    Projection: columns=['simbolo', 'tamano']
      Assign: importe=
        ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
        Mul:
          Projection: columns='precio'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
          Projection: columns='tamano'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']

Es el árbol tal como se escribió, de arriba abajo: una división (__truediv__) de dos sumas por grupo, cada una sobre una proyección de la tabla con la columna importe añadida, que a su vez sale de multiplicar precio por tamano leídas del Parquet. Antes de ejecutarlo, Dask lo optimiza como hace Polars: reconoce que las dos ramas leen lo mismo y las une, empuja la proyección al lector (fecha no hace falta) y fusiona las operaciones que se hacen partición a partición en una sola tarea. Después lo convierte en un grafo de tareas: una tarea por partición y operación, más las de combinación.

La suma por grupo es el patrón más común de un cálculo distribuido: cada partición calcula su suma parcial por símbolo (una tabla pequeña de 200 filas) y después se combinan en árbol, de ocho en ocho por defecto (split_every=8), hasta un único resultado. Solo viajan las sumas parciales, nunca los ticks:

flowchart TB
  P1[Partición 1] --> S1[Suma parcial]
  P2[Partición 2] --> S2[Suma parcial]
  P3[Partición 39] --> S3[Suma parcial]
  S1 --> C[Combinar]
  S2 --> C
  S3 --> C
  C --> R[VWAP por símbolo]

compute, dask.compute y persist

compute() ejecuta el grafo y devuelve un objeto de pandas. Si después se pide otro resultado sobre los mismos datos, se vuelve a ejecutar todo, lectura incluida. Hay dos formas de evitarlo: pedir los dos resultados a la vez con dask.compute(a, b), que comparte las tareas comunes, o materializar los datos en la memoria de los trabajadores con persist(), que devuelve otro DataFrame de Dask cuyas particiones ya están calculadas. Para ver la diferencia sin cronómetro, la celda vuelve a usar un sistema de ficheros que cuenta lo que se lee (como en la lección 6.6.2) y calcula dos cosas: el VWAP por símbolo y el número de ticks por símbolo.

import os

from fsspec.implementations.local import LocalFileSystem

class Contador(LocalFileSystem):
    """Disco local que anota bytes leídos y ficheros abiertos."""
    bytes_leidos = 0
    aperturas = 0

    def _open(self, path, mode="rb", **kw):
        f = super()._open(path, mode, **kw)
        self.aperturas += 1
        leer = f.read
        def read(n=-1):
            datos = leer(n)
            self.bytes_leidos += len(datos)
            return datos
        f.read = read
        return f

disco = Contador()
lago = dd.read_parquet(os.path.abspath("ticks"), filesystem=disco,
                       columns=["simbolo", "precio", "tamano"])
c = lago.assign(importe=lago.precio * lago.tamano)

def consultas(df):
    g = df.groupby("simbolo")
    return g.importe.sum() / g.tamano.sum(), df.simbolo.value_counts()

def medir(nombre, accion):
    disco.bytes_leidos = disco.aperturas = 0
    resultado = accion()
    filas.append({"estrategia": nombre, "aperturas": disco.aperturas,
                  "MB_leidos": round(disco.bytes_leidos / 1e6, 1)})
    return resultado

filas = []
vwap, n = consultas(c)
medir("compute por separado", lambda: (vwap.compute(), n.compute()))
r1 = medir("dask.compute juntas", lambda: dask.compute(vwap, n))
p = medir("persist", lambda: c.persist())
r2 = medir("después de persist",
           lambda: dask.compute(*consultas(p)))
print("mismos resultados:",
      np.allclose(r1[0], r2[0].reindex(r1[0].index), rtol=1e-12),
      r1[1].sort_index().equals(r2[1].sort_index()))
pd.DataFrame(filas)

mismos resultados: True True
             estrategia  aperturas  MB_leidos
0  compute por separado        156       39.1
1   dask.compute juntas         78       33.9
2               persist         78       33.9
3    después de persist          0        0.0

La figura de la cabecera lo resume. Con compute() por separado, cada consulta lee el lago por su cuenta: 156 aperturas de fichero (dos por fichero y pasada, cuatro pasadas en total contando las dos ramas que el optimizador no une entre consultas distintas) y 39,1 MB. No son el doble de una pasada porque el optimizador empuja la proyección: value_counts solo lee la columna simbolo. Con dask.compute(vwap, n), Dask optimiza los dos grafos juntos y lee cada fichero en una sola pasada: 33,9 MB. persist() lee lo mismo una vez y guarda las particiones en memoria; las dos consultas siguientes no tocan el disco (0,0 MB) y devuelven lo mismo (True True; el VWAP se compara con np.allclose porque las sumas parciales se combinan en otro orden y pueden diferir en el último decimal).

Cuándo usar cada uno:

• dask.compute(a, b, …) cuando se conocen de antemano todos los resultados que se quieren: una sola pasada, sin ocupar memoria con los datos.
• persist() cuando se va a explorar de forma interactiva sobre un subconjunto filtrado que cabe en la memoria del clúster: se paga la lectura una vez. Los datos persistidos ocupan memoria hasta que se borra la variable.
• compute() al final, para traer a la máquina local un resultado pequeño. Llamar a compute() sobre un DataFrame grande intenta meterlo entero en la memoria del cliente, que es justo lo que se quería evitar.

Cálculos que no cruzan particiones

Si un cálculo solo necesita las filas de una partición, Dask lo reparte sin ninguna comunicación entre trabajadores. map_partitions aplica una función de pandas a cada partición tal cual. Aquí, la volatilidad realizada de cada símbolo en cada sesión, que solo depende de los ticks de esa sesión:

def vol_diaria(df):
    r = np.log(df["precio"]).groupby(
        [df["simbolo"], df["fecha"]], observed=True).diff()
    return (r.pow(2).groupby([df["simbolo"], df["fecha"]],
                             observed=True).sum() ** 0.5
            ).rename("vol").reset_index()

meta = pd.DataFrame({"simbolo": pd.Series(dtype="str"),
                     "fecha": pd.Series(dtype=ddf.fecha.dtype),
                     "vol": pd.Series(dtype="float64")})
vol = ddf.map_partitions(vol_diaria, meta=meta).compute()
print(len(vol), "pares símbolo-día;",
      f"volatilidad diaria mediana: {vol['vol'].median():.4f}")

7800 pares símbolo-día; volatilidad diaria mediana: 0.0203

7.800 pares símbolo-día (200 × 39) y una volatilidad diaria mediana del 2,03 %: la volatilidad con la que se generaron los precios (2 %), recuperada desde los ticks. meta le dice a Dask qué columnas y tipos devolverá la función, para poder seguir construyendo el plan sin ejecutarla.

El cálculo es correcto solo porque cada sesión está entera dentro de una partición: las particiones coinciden con los ficheros, y los ficheros, con los días. Si se quisiera la volatilidad de cada símbolo en todo el periodo con el rendimiento entre el último tick de un día y el primero del siguiente, las filas de un símbolo estarían repartidas en 39 particiones y habría que reorganizarlas por símbolo: una redistribución (shuffle), en la que cada trabajador envía a cada otro las filas que le tocan. Es la operación más cara de cualquier sistema distribuido, y la razón por la que la forma de particionar importa más que el número de máquinas.

Particiones equilibradas

El tamaño de las particiones también importa. La guía de Dask recomienda particiones que quepan holgadamente en la memoria de un trabajador, de unos 100 MB cada una, y avisa de que con particiones de 1 GB y diez núcleos el proceso puede usar al menos 10 GB³. Igual de importante es que sean parecidas entre sí: el cálculo termina cuando termina la partición más lenta. La celda compara las dos formas naturales de partir este lago, por fecha o por símbolo, contando las filas de cada partición:

por_fecha = ddf.fecha.value_counts().compute()
por_simbolo = r1[1]

def resumen(s):
    return {"particiones": len(s), "mínimo": s.min(),
            "mediana": int(s.median()), "máximo": s.max(),
            "máx/mediana": round(s.max() / s.median(), 1)}

pd.DataFrame([resumen(por_fecha), resumen(por_simbolo)],
             index=["por fecha", "por símbolo"])

             particiones  mínimo  mediana  máximo  máx/mediana
por fecha             39  250155   251314  252509          1.0
por símbolo          200    2575    39607  238330          6.0

Por fecha, las 39 particiones tienen entre 250.155 y 252.509 filas: casi idénticas, porque cada día se generan ticks de todos los símbolos. Por símbolo, de 2.575 a 238.330: la más grande tiene seis veces la mediana y noventa veces la más pequeña, porque la actividad de las acciones es muy desigual (en los datos, una lognormal; en el mercado, igual o peor). Con 200 particiones por símbolo repartidas entre ocho trabajadores, el que reciba CELE y otras dos de las grandes trabajará mucho más que el resto mientras los demás esperan. Es el sesgo de particiones (skew), y en datos de mercado aparece en cuanto se parte por activo. Las salidas habituales: partir por fecha y ordenar por símbolo dentro (como en la lección 6.6.2), agrupar los símbolos pequeños en cubos de tamaño parecido o dividir los grandes.

Spark: el estándar de los clústeres

Apache Spark es el marco de datos distribuidos más extendido: un motor escrito en Scala que se ejecuta en la máquina virtual de Java, con API para Python (PySpark), SQL, Scala, Java y R. Su modelo es el mismo que el de Dask: un DataFrame perezoso partido en particiones, un optimizador de consultas y un planificador que reparte tareas entre executors. La última versión publicada es la 4.2.0, del 14 de julio de 2026; según su documentación, se ejecuta con Java 17, 21 o 25 y Python 3.10 o superior, y PySpark necesita Java 17 o superior con JAVA_HOME configurado44Nota 4 · Apache Spark 4.2.0, «Overview» e «Installation» de PySpark: https://spark.apache.org/docs/latest/ y https://spark.apache.org/docs/latest/api/python/getting_started/install.html; noticias del proyecto (https://spark.apache.org/news/index.html). Consultados el 23 de octubre de 2026.. El entorno del curso no tiene Java (y pyspark no se resolvió en la instalación), así que el código no se ejecuta. Es el equivalente del VWAP por símbolo y de la volatilidad tick a tick con una ventana:

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("ticks")
         .master("local[4]")            # o la URL del clúster
         .getOrCreate())

ticks = spark.read.parquet("ticks")     # descubre fecha=… como columna
vwap = (ticks.groupBy("simbolo")
        .agg((F.sum(F.col("precio") * F.col("tamano"))
              / F.sum("tamano")).alias("vwap"),
             F.count("*").alias("ticks")))

w = Window.partitionBy("simbolo").orderBy("ts")
ret = ticks.withColumn(
    "ret", F.log(F.col("precio") / F.lag("precio").over(w)))
resumen = ret.agg(F.count("ret"), F.sum(F.col("ret") ** 2))

vwap.write.mode("overwrite").parquet("vwap_spark")  # se ejecuta aquí
print(resumen.collect())
spark.stop()

Como en Dask y Polars, nada se calcula hasta una acción (write, collect, count); hasta entonces Spark solo construye el plan, que se puede ver con vwap.explain(). Window.partitionBy("simbolo") obliga a redistribuir los ticks por símbolo entre los executors (el shuffle de la sección anterior). Spark ofrece además una API con forma de pandas, pyspark.pandas, que sigue la especificación de la última versión de pandas⁴:

import pyspark.pandas as ps

t = ps.read_parquet("ticks")
t["importe"] = t["precio"] * t["tamano"]
g = t.groupby("simbolo")
vwap = g["importe"].sum() / g["tamano"].sum()
print(vwap.sort_index().head(3))

Cuándo Spark y no Dask: cuando ya hay un clúster de Spark (Databricks, EMR, Dataproc o uno propio) y un equipo que lo mantiene, cuando el trabajo es sobre todo SQL a escala de terabytes o cuando se necesita su ecosistema (lectura y escritura de Delta Lake e Iceberg, lección 6.6.5). Dask es Python de principio a fin: las particiones son DataFrame de pandas, las funciones propias se ejecutan tal cual y el mismo código corre en un portátil y en un clúster; a cambio, su optimizador es más joven y su ecosistema de clúster, más pequeño.

Ray y dask.delayed: paralelizar experimentos

Muchos problemas de un investigador no son de datos grandes, sino de muchos cálculos independientes: probar cada combinación de parámetros, cada pliegue de una validación cruzada purgada (lección 5.4.2) o cada semilla de un modelo. Para eso no hace falta un DataFrame distribuido; basta con repartir llamadas a funciones.

Ray es un marco pensado para eso. Su núcleo tiene tres piezas: una tarea es una función decorada con @ray.remote que se lanza con .remote() y devuelve una referencia a un resultado futuro; un actor es una clase con estado que vive en un proceso; y un almacén de objetos compartido guarda los resultados para que otras tareas los usen sin copiarlos55Nota 5 · Ray 2.59.0, «What's Ray Core?», «Usage Stats Collection» y el glosario (raylet, GCS): https://docs.ray.io/en/latest/ray-core/walkthrough.html y https://docs.ray.io/en/latest/cluster/usage-stats.html, consultados el 23 de octubre de 2026.. Sobre él están Ray Tune (búsqueda de hiperparámetros), Ray Train y Ray Data. El barrido de abajo, escrito con Ray, sería así:

import ray

ray.init(num_cpus=4)          # local; ray.init("auto") en un clúster

@ray.remote
def sharpe_cruce_remoto(rapida, lenta):
    return sharpe_cruce(rapida, lenta)

refs = [sharpe_cruce_remoto.remote(a, b)
        for a in (5, 10, 20, 40) for b in (50, 100, 150, 200, 250)]
res = pd.DataFrame(ray.get(refs))   # espera a las 20 tareas
ray.shutdown()

Ray está instalado en el entorno ml del curso (llega con NeuralForecast, nivel 5), pero no se ejecuta en esta lección por dos razones. La primera, que esta lección corre en el entorno principal, donde está Dask y no Ray. La segunda, que su arranque no es compatible con la ejecución sin red del curso: en la prueba hecha para esta lección (Ray 2.59.0), ray.init() intentó averiguar la IP local abriendo una conexión hacia 8.8.8.8:53, levantó sus servicios en la dirección de la red local y no en 127.0.0.1, y sin red el arranque se quedó esperando a su propio servicio de control (GCS). También escribe sus ficheros de sesión en /tmp/ray, fuera de la carpeta de la lección. Con red, el mismo código terminó sin problemas. Ray recoge estadísticas de uso anónimas en algunos casos; su documentación explica cómo desactivarlo con RAY_USAGE_STATS_ENABLED=0⁵.

Dask tiene su propia forma de repartir funciones, dask.delayed, que se ejecuta aquí. El experimento: un cruce de medias móviles (comprado si la rápida está por encima de la lenta, vendido si no) sobre 200 paseos aleatorios sin deriva de 10 años, con 20 combinaciones de parámetros. La respuesta conocida es que ninguna combinación tiene ventaja: en un paseo aleatorio, el pasado no dice nada del futuro.

from dask import delayed

def sharpe_cruce(rapida, lenta, n_series=200, dias=2520):
    """Cruce de medias sobre 200 paseos aleatorios sin deriva."""
    rng = np.random.default_rng(2026)   # las mismas series siempre
    r = rng.normal(0.0, 0.01, (dias, n_series))
    p = pd.DataFrame(np.exp(np.cumsum(r, axis=0)))
    senal = np.where(p.rolling(rapida).mean()
                     > p.rolling(lenta).mean(), 1.0, -1.0)
    pos = pd.DataFrame(senal).shift(1)      # se opera al día siguiente
    rend = (pos * np.expm1(r)).iloc[lenta:]
    sr = rend.mean() / rend.std() * np.sqrt(252)
    return {"rapida": rapida, "lenta": lenta,
            "sharpe_medio": sr.mean(), "sharpe_max": sr.max()}

tareas = [delayed(sharpe_cruce)(a, b)
          for a in (5, 10, 20, 40) for b in (50, 100, 150, 200, 250)]
res = pd.DataFrame(dask.compute(*tareas, scheduler="threads"))
print(len(tareas), "tareas")
mejor = res.loc[res["sharpe_max"].idxmax()]
print(f"mejor serie de las {20 * 200:,}: Sharpe"
      f" {mejor['sharpe_max']:.2f} con ({mejor['rapida']:.0f},"
      f" {mejor['lenta']:.0f})")
res.sort_values("sharpe_medio", ascending=False).round(3).head(4)

20 tareas
mejor serie de las 4,000: Sharpe 0.98 con (40, 250)
    rapida  lenta  sharpe_medio  sharpe_max
3        5    200         0.028       0.808
13      20    200         0.024       0.930
10      20     50         0.024       0.736
4        5    250         0.019       0.917

Las 20 tareas se ejecutan en paralelo con el planificador de hilos (NumPy y pandas sueltan el GIL en buena parte de sus operaciones; para funciones de Python puro conviene scheduler="processes"). Los resultados no dependen del planificador: cada tarea genera las mismas series con su propia semilla.

Lo que dicen los números es una lección del nivel 2 en otro contexto. El mejor Sharpe medio de las 20 combinaciones es 0,028, compatible con cero, como debe ser. Pero entre las 4.000 estrategias probadas (200 series × 20 combinaciones), la mejor (rápida de 40 sesiones, lenta de 250) tiene un Sharpe de 0,98 en diez años, sin ninguna señal detrás. Paralelizar un barrido lo hace barato y por eso tienta a probar más: cuantas más pruebas, mayor el máximo por azar. El Sharpe deflactado (lección 2.5.4) es la corrección obligatoria antes de creerse el mejor resultado de un barrido, se ejecute en un portátil o en un clúster.

Qué elegir

El orden de decisión que sigue de la lección: primero, una máquina bien usada (formato, plan, streaming); después, Dask si el código ya es pandas y hace falta más memoria o más núcleos; Spark si la organización ya lo tiene; Ray o delayed para repartir experimentos. En los tres casos, la forma de partir los datos (por fecha, sin sesgo, con particiones de un tamaño razonable) decide más que el número de máquinas.

Ejercicios

1. Calcula con una sola pasada por el lago el VWAP por símbolo y la volatilidad diaria de la celda de map_partitions, y comprueba con el contador cuántos bytes se leen.

lago_f = dd.read_parquet(
    os.path.abspath("ticks"), filesystem=disco,
    columns=["simbolo", "precio", "tamano", "fecha"])
c_f = lago_f.assign(importe=lago_f.precio * lago_f.tamano)
disco.bytes_leidos = 0
v, vol2 = dask.compute(consultas(c_f)[0],
                       c_f.map_partitions(vol_diaria, meta=meta))
print(len(v), "VWAP y", len(vol2), "volatilidades;",
      f"{disco.bytes_leidos / 1e6:.1f} MB leídos")
print("igual que antes:", np.allclose(
    vol2.set_index(["simbolo", "fecha"])["vol"].sort_index(),
    vol.set_index(["simbolo", "fecha"])["vol"].sort_index()))

200 VWAP y 7800 volatilidades; 33.9 MB leídos
igual que antes: True

2. ¿Cuánto ocupa en memoria cada partición de ddf? Reparte el lago en particiones de unos 100 MB con repartition(partition_size="100MB") y compara.

def mb_por_particion(df):
    m = df.memory_usage_per_partition(deep=True).compute() / 1e6
    return {"particiones": len(m), "mediana_MB": round(m.median(), 1),
            "max_MB": round(m.max(), 1)}

grandes = ddf.repartition(partition_size="100MB")
pd.DataFrame([mb_por_particion(ddf), mb_por_particion(grandes)],
             index=["un fichero por partición", "100 MB"])

                          particiones  mediana_MB  max_MB
un fichero por partición           39         6.3     6.3
100 MB                              3       124.4   124.4

import shutil

shutil.rmtree("ticks")
print(sorted(p.name for p in Path(".").glob("ticks*")))

[]

3. Un equipo de tres personas tiene diez años de barras de 1 minuto de 3.000 acciones (unos 3.000 millones de filas, del orden de 60 GB en Parquet) y quiere (a) calcular factores diarios cada noche y (b) lanzar barridos de 500 combinaciones de parámetros de un modelo. No tienen clúster. ¿Qué herramientas propones para cada cosa?