---
title: "6.6.3 Dask, Spark y Ray"
description: "Dask sobre un lago de ticks (plan, compute, persist, particiones con sesgo), Spark y Ray con su código real, y un barrido de parámetros en paralelo."
url: "https://pythonparatrading.com/curso-python-trading/nivel-6/dask-spark-y-ray/"
lang: "es"
image: "https://pythonparatrading.com/og/curso-python-trading/nivel-6/dask-spark-y-ray.png"
---
# 6.6.3 · Dask, Spark y Ray: de una máquina a un clúster
> Un clúster reparte datos o reparte cálculos, y en los dos casos paga por mover datos entre procesos. Esta lección ejecuta Dask sobre diez millones de ticks y mide en bytes la diferencia entre `compute`, `dask.compute` y `persist`, enseña por qué partir por símbolo desequilibra el trabajo, muestra el código de Spark y Ray (que no se ejecutan, y se explica por qué) y reparte un barrido de parámetros cuyo mejor resultado es puro azar.
- Curso: [Nivel 6: IA generativa, LLMs, agentes, big data y datos alternativos](https://pythonparatrading.com/curso-python-trading/nivel-6/)
- Módulo: 6.6 · Big data para trading
- Duración: 65 min
- Requisitos: [6.6.2 · Parquet, Arrow, Polars en streaming y DuckDB fuera de memoria](https://pythonparatrading.com/curso-python-trading/nivel-6/parquet-arrow-polars-streaming-y-duckdb/), [2.5.4 · Deflated Sharpe Ratio y Probability of Backtest Overfitting (PBO)](https://pythonparatrading.com/curso-python-trading/nivel-2/deflated-sharpe-ratio-y-pbo/)
## Al terminar esta lección
- Explicar la arquitectura común de Dask, Spark y Ray (cliente, planificador, trabajadores) y el coste de mover datos entre ellos.
- Construir consultas perezosas con Dask DataFrame, leer su árbol de expresiones y elegir entre `compute`, `dask.compute` y `persist` midiendo lo que se lee.
- Distinguir los cálculos que no cruzan particiones (`map_partitions`) de los que exigen redistribuir datos, y detectar particiones desequilibradas.
- Escribir el equivalente en PySpark y en Ray de un cálculo de Dask, y saber cuándo compensa cada marco.
- Paralelizar un barrido de parámetros con `dask.delayed` y corregir el mejor resultado por pruebas múltiples.
## 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.
```mermaid
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»[^1].
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.
```python
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")
```
```output
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 filas[^2]. 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:
```python
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)
```
```output
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:
```python
c = ddf.assign(importe=ddf.precio * ddf.tamano)
por_simbolo = c.groupby("simbolo")
vwap = por_simbolo.importe.sum() / por_simbolo.tamano.sum()
vwap.pprint()
```
```output
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:
```mermaid
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.
```python
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)
```
```output
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:
```python
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}")
```
```output
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[^3]. 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:
```python
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"])
```
```output
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
```
*Fig. — Filas por partición del lago de 9,8 millones de ticks si se parte por fecha o por símbolo.* Dos paneles con las particiones ordenadas de mayor a menor. Por fecha, 39 particiones casi iguales, entre 250 155 y 252 509 filas. Por símbolo, 200 particiones que caen de 238 330 filas (CELE, destacada) a 2 575, con una mediana de 39 607.
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`
configurado[^4]. 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:
```python skip title="vwap_spark.py"
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[^4]:
```python skip title="pandas_en_spark.py"
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 copiarlos[^5]. 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í:
```python skip title="barrido_ray.py"
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`[^5].
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.
```python
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)
```
```output
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
| Marco | Unidad de trabajo | Brilla en | Coste |
|---|---|---|---|
| Polars, DuckDB (lección 6.6.2) | Consulta sobre una máquina | Casi todo lo que cabe en un disco | Ninguno nuevo |
| Dask DataFrame | Partición de pandas | Código de pandas que ya no cabe; escalar del portátil al clúster | Optimizador más joven; `persist` con aristas |
| Dask `delayed` | Función de Python | Barridos y tareas con dependencias entre sí | El planificador añade sobrecarga por tarea |
| Spark (PySpark) | Partición en la JVM | SQL a escala de terabytes, clústeres ya montados, Delta e Iceberg | Java, un clúster que mantener, serializar entre Python y la JVM |
| Ray | Tarea o actor | Experimentos de ML, búsqueda de hiperparámetros, servir modelos | Servicios propios (GCS, raylets), red obligatoria |
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.
[^1]: Dask, «Dask Best Practices»: , consultado el 23 de octubre
de 2026.
[^2]: Dask, «Dask DataFrame» y «Scheduling» (el planificador de hilos es el predeterminado para Dask DataFrame):
, consultado el 23 de octubre de 2026.
[^3]: Dask, «Dask DataFrames Best Practices»: ,
consultado el 23 de octubre de 2026.
[^4]: Apache Spark 4.2.0, «Overview» e «Installation» de PySpark: y
; noticias del proyecto
(). Consultados el 23 de octubre de 2026.
[^5]: Ray 2.59.0, «What's Ray Core?», «Usage Stats Collection» y el glosario (raylet, GCS):
y ,
consultados el 23 de octubre de 2026.
## 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.
**Solución.**
```python
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()))
```
```output
200 VWAP y 7800 volatilidades; 33.9 MB leídos
igual que antes: True
```
Una sola pasada: 33,9 MB, lo mismo que leer para el VWAP solo, porque la volatilidad necesita las mismas columnas
más `fecha`, que no está en los ficheros sino en las rutas. Las volatilidades coinciden con las de la celda de
`map_partitions`. Es la regla práctica de Dask: juntar en un `dask.compute` todo lo que se va a pedir a los mismos
datos.
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.
**Solución.**
```python
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"])
```
```output
particiones mediana_MB max_MB
un fichero por partición 39 6.3 6.3
100 MB 3 124.4 124.4
```
Cada fichero ocupa 6,3 MB en memoria con pandas (frente a menos de 1 MB comprimido en disco): particiones muy
pequeñas para Dask, en el que cada tarea tiene un coste fijo de planificación. `repartition(partition_size="100MB")`
las junta en 3 particiones de unos 124 MB: Dask estima los tamaños y agrupa de forma aproximada. En un clúster, 3
particiones dejarían ociosos a casi todos los trabajadores; el tamaño razonable es un compromiso entre la sobrecarga
por tarea (particiones pequeñas) y la memoria y el paralelismo (particiones grandes). La última celda borra los datos
temporales.
```python
import shutil
shutil.rmtree("ticks")
print(sorted(p.name for p in Path(".").glob("ticks*")))
```
```output
[]
```
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?
**Solución.**
(a) Los factores diarios: 60 GB en Parquet, particionados por fecha y ordenados por símbolo dentro de cada día, caben
en el disco de una sola máquina, y un cálculo nocturno que solo procesa el día nuevo lee unos pocos megabytes. Polars
lazy con streaming o DuckDB con límite de memoria (lección 6.6.2) bastan, también para recalcular la historia
completa de vez en cuando. Dask solo si el código ya está escrito en pandas y no cabe en memoria.
(b) Los barridos: 500 combinaciones independientes es un problema de tareas, no de datos. En una máquina, `dask.delayed`
o `concurrent.futures` con procesos (lección 0.4.6) aprovechan todos los núcleos; si un barrido tarda demasiado,
Ray o un clúster de Dask en la nube reparten las mismas funciones entre varias máquinas sin cambiar la lógica. Spark
no aporta nada aquí y añade Java y un clúster que mantener. Y con 500 combinaciones, el mejor resultado hay que
corregirlo por pruebas múltiples (Sharpe deflactado, lección 2.5.4) antes de usarlo.
## Autoevaluación
1. Se necesitan el VWAP por símbolo y el número de ticks por símbolo del mismo lago. Con dos `compute()` se leen 39,1 MB; ¿qué hace `dask.compute(vwap, n)`?
- A. Lo mismo que dos `compute()`: cada resultado tiene su grafo
- B. Optimiza los dos grafos juntos y lee cada fichero una sola vez: 33,9 MB
- C. Guarda los datos en memoria para siempre
- D. Calcula los resultados de forma aproximada
*Respuesta B: Optimiza los dos grafos juntos y lee cada fichero una sola vez: 33,9 MB.* Las tareas comunes (la lectura de cada partición) se comparten. `persist()` también lee una vez, pero además deja las particiones en memoria: las consultas posteriores leen 0 bytes a cambio de ocupar memoria.
2. La volatilidad diaria de cada símbolo se calcula con `map_partitions` sin ninguna comunicación entre trabajadores. ¿Qué lo permite?
- A. Que Dask reordena los datos automáticamente
- B. Que la volatilidad es una suma
- C. Que `meta` describe el resultado
- D. Que cada partición es un fichero y cada fichero una sesión completa: todas las filas que necesita cada grupo símbolo-día están en la misma partición
*Respuesta D: Que cada partición es un fichero y cada fichero una sesión completa: todas las filas que necesita cada grupo símbolo-día están en la misma partición.* Si el cálculo necesitara filas de varias particiones (por ejemplo, rendimientos de un símbolo a través de los días), habría que redistribuirlas por símbolo (shuffle), la operación más cara de un sistema distribuido.
3. Por fecha, las 39 particiones tienen entre 250.155 y 252.509 filas; por símbolo, entre 2.575 y 238.330. ¿Qué problema tiene partir por símbolo?
- A. El sesgo de particiones: el trabajo termina cuando termina la partición más grande, y los demás trabajadores esperan
- B. Ninguno, hay más particiones y por tanto más paralelismo
- C. Que Dask no admite particiones por texto
- D. Que las particiones pequeñas no caben en memoria
*Respuesta A: El sesgo de particiones: el trabajo termina cuando termina la partición más grande, y los demás trabajadores esperan.* La actividad de las acciones es muy desigual. La partición más grande tiene seis veces la mediana. Las salidas: partir por fecha y ordenar por símbolo dentro, agrupar símbolos pequeños en cubos parecidos o dividir los grandes.
4. ¿Por qué no se ejecuta Ray en el curso, aunque está instalado en el entorno `ml`?
- A. Porque Ray no funciona con Python 3.14
- B. Porque Ray necesita una GPU
- C. Porque la lección corre en el entorno principal y, en la prueba, `ray.init()` buscó la IP local hacia `8.8.8.8`, levantó sus servicios en la red local y sin red se quedó esperando a su servicio de control
- D. Porque Ray es de pago
*Respuesta C: Porque la lección corre en el entorno principal y, en la prueba, `ray.init()` buscó la IP local hacia `8.8.8.8`, levantó sus servicios en la red local y sin red se quedó esperando a su servicio de control.* Con red, el mismo código terminó sin problemas. El curso exige ejecuciones sin red y deterministas; el barrido se ejecuta con `dask.delayed`, que reparte funciones de Python de la misma forma, y el código de Ray se muestra sin ejecutar.
5. Un barrido de 20 combinaciones de un cruce de medias sobre 200 paseos aleatorios sin deriva da un Sharpe medio máximo de 0,028, pero la mejor de las 4.000 estrategias tiene un Sharpe de 0,98 en diez años. ¿Qué conclusión es correcta?
- A. La combinación (40, 250) funciona en esa serie y hay que usarla
- B. El máximo de muchas pruebas sin señal es alto por azar: hay que corregirlo por pruebas múltiples (Sharpe deflactado) antes de creerlo
- C. Diez años no bastan para estimar un Sharpe
- D. El paralelismo introduce errores en el cálculo
*Respuesta B: El máximo de muchas pruebas sin señal es alto por azar: hay que corregirlo por pruebas múltiples (Sharpe deflactado) antes de creerlo.* Las series no tienen ninguna señal por construcción. Paralelizar abarata probar más combinaciones, y cuantas más se prueban, mayor es el mejor resultado por azar (lección 2.5.4).
## Ideas clave
- Dask, Spark y Ray comparten arquitectura (cliente, planificador, trabajadores) y coste: mover datos entre procesos o máquinas es mucho más lento que leer la memoria propia. Antes de distribuir, una máquina bien usada (lección 6.6.2).
- Dask DataFrame es una colección de `DataFrame` de pandas perezosa y optimizada: dos `compute()` leyeron 39,1 MB, `dask.compute` de los dos 33,9 MB y, tras `persist()`, 0 MB a cambio de memoria.
- Los cálculos dentro de una partición (`map_partitions`) no comunican nada; los que cruzan particiones exigen un shuffle. Partir por fecha dio particiones casi iguales; por símbolo, la mayor seis veces la mediana.
- Spark 4.2 necesita Java 17, 21 o 25 y brilla con SQL a escala de terabytes y clústeres ya montados; Ray reparte tareas y actores para experimentos de ML; ninguno se ejecuta en el curso, por Java y por la red.
- Paralelizar un barrido lo abarata y multiplica las pruebas: la mejor de 4.000 estrategias sin señal tuvo un Sharpe de 0,98. El Sharpe deflactado no es opcional.
## Recursos
### Documentación
- [Dask DataFrame (documentación oficial)](https://docs.dask.org/en/stable/dataframe.html) · docs.dask.org
- [Dask DataFrames: buenas prácticas (tamaño de partición, persist)](https://docs.dask.org/en/stable/dataframe-best-practices.html) · docs.dask.org
- [Apache Spark 4.2.0: visión general y requisitos](https://spark.apache.org/docs/latest/) · spark.apache.org
- [PySpark: instalación (Java 17 o superior)](https://spark.apache.org/docs/latest/api/python/getting_started/install.html) · spark.apache.org
- [Ray Core: tareas, actores y objetos](https://docs.ray.io/en/latest/ray-core/walkthrough.html) · docs.ray.io
### Datos
- [Datos del nivel 5: tickers del panel de acciones (CSV)](https://pythonparatrading.com/curso/datos/nivel-5/panel5_activos.csv) · Descarga · panel5_activos.csv
Lección anterior: [6.6.2 · Parquet, Arrow, Polars en streaming y DuckDB fuera de memoria](https://pythonparatrading.com/curso-python-trading/nivel-6/parquet-arrow-polars-streaming-y-duckdb/)
Lección siguiente: [6.6.4 · Bases de series temporales: TimescaleDB, QuestDB, ClickHouse, ArcticDB y kdb+](https://pythonparatrading.com/curso-python-trading/nivel-6/bases-de-series-temporales-y-arcticdb/)