Cómo alimentamos velas OHLC a TradingView desde ClickHouse® DB

Un trader abre un gráfico. Por la pipeline están pasando unas 4.000 eventos de precio por segundo. El gráfico necesita meses de velas a la resolución que elija, y las necesita ya, no dentro de dos segundos. Y encima tiene que mostrar ticks en vivo por encima, sin que se note un hueco entre la última vela almacenada y el precio actual.

Lo construimos usando tablas AggregatingMergeTree en cascada que pre-computan velas a cuatro resoluciones (1m, 5m, 1h, 1d), servidas a través de la API UDF de TradingView, con precios en vivo en paralelo a través de Cloudflare Durable Objects. Una carga completa del gráfico vuelve en menos de 200ms porque ClickHouse nunca toca un evento raw.

Este post forma parte de la serie ClickHouse Deep Dive. El anterior cubría AggregatingMergeTree como referencia: qué es, cómo funciona y cuándo usarlo. Este muestra cómo lo usamos en producción bajo cargas masivas. Con queries reales y la pipeline completa.

Velas OHLC: qué son y por qué son difíciles

Cada vela cubre una ventana temporal y captura cuatro precios (apertura, máximo, mínimo, cierre) más el volumen. Ese es el formato que espera cualquier librería de gráficos, TradingView incluida.

El problema de ingeniería está en calcularlas a partir de un stream de eventos raw a varias resoluciones a la vez (1m, 5m, 1h, 1d), mantener todas actualizadas mientras llegan eventos a unas 4.000/segundo, y servirlas lo bastante rápido como para que la carga del gráfico se sienta instantánea aunque pidas meses de histórico. Recalcular desde eventos raw en cada petición ni siquiera es una opción con este throughput.

Decisiones de diseño

Los dos pipelines comparten cuatro decisiones de diseño que merece la pena explicar antes de entrar en detalle.

Por qué OHLC y volumen son familias de tablas separadas

El primer instinto es meter OHLCV en una sola tabla. Una fila por vela, los cinco campos juntos. Así es como lo consume TradingView. Si tus precios vienen de trades (el caso habitual en mercados spot), puedes hacerlo exactamente así: una MV desde la tabla de trades escribe las cinco columnas. Pero en este exchange, los precios vienen de un feed de oráculos, y el volumen de trades ejecutados. Dos fuentes distintas. Podrías pensar: dos MVs escribiendo en el mismo AggregatingMergeTree, una desde raw_prices (para OHLC), otra desde raw_trades (para volumen). El patrón "varias MVs en un mismo destino" está bien documentado. Por qué no aquí?

El problema son las columnas para las que no tienes datos. La MV de precios puede escribir high, low, open y close, y en ese caso, sumSimpleState(0) suma cero para el volumen, bien. Pero al revés, la MV de trades escribiendo volumen tendría que escribir algo para high y low. minSimpleState(0) corrompe tu precio mínimo (0 es menor que cualquier precio real). maxSimpleState(0) está mal para high. Y argMinState/argMaxState para open/close necesitan un par (precio, timestamp) real. No existe un estado "vacío" válido que puedas escribir para esas columnas.

Así que la arquitectura son dos cascadas paralelas que se unen en tiempo de consulta. No es óptimo en tiempo de lectura, y son dos pipelines que mantener. No es la solución que preferíamos, pero sí la arquitectónicamente viable, dado que los campos Nullable no son óptimos en ClickHouse.

SimpleAggregateFunction vs AggregateFunction

SimpleAggregateFunction almacena un escalar plano. Funciona para operaciones asociativas: max(max(a), max(b)) = max(a, b). El máximo de dos máximos siempre es el máximo. Lo mismo para min, lo mismo para sum. ClickHouse simplemente compara dos números durante los merges de fondo. Menos almacenamiento, merges más rápidos, y puedes leer el valor directamente en un SELECT sin necesidad de un combinador -Merge.

AggregateFunction almacena un estado intermedio binario. Lo necesitas cuando la operación de merge requiere más que el valor resultado. argMin rastrea tanto un valor COMO el argumento que lo produjo. Para determinar qué precio fue el primero en un intervalo, necesitas comparar timestamps entre parts. Eso requiere el par (precio, timestamp) completo almacenado como estado binario.

En el esquema OHLC, high y low usan SimpleAggregateFunction. open y close usan AggregateFunction con argMin/argMax. En el esquema de volumen, todo usa SimpleAggregateFunction porque sum es asociativa.

El patrón en cascada

Los dos pipelines usan la misma estructura: 1m -> 5m -> 1h -> 1d. Cuatro tablas AggregatingMergeTree, cada una alimentada por una materialized view que lee del nivel inferior. La tabla de 1 día nunca ve un evento raw.

En el nivel 1, las funciones -State agregan valores raw en un estado intermedio. En los niveles 2-4, las funciones -MergeState combinan estados existentes del nivel inferior. Nunca re-agregas datos raw a resoluciones superiores. Combinas estados pre-computados. Eso es lo que hace eficiente la cascada. Las consultas finales tampoco tocan datos raw. Siempre leen de un estado pre-agregado, que es la razón por la que la capa de servicio se mantiene rápida bajo carga.

División entera para bucketing temporal

El primer nivel de la cascada usa toStartOfMinute(event_timestamp) para la agregación de 1 minuto. Los niveles superiores usan intDiv(time, N) * N: intDiv(time, 300) * 300 redondea un timestamp unix al límite de 5 minutos más cercano. intDiv(time, 3600) * 3600 para horas. intDiv(time, 86400) * 86400 para días. Aritmética entera pura, sin overhead de funciones de fecha.

Ingesta: deduplicar antes de ClickHouse

Los eventos llegan a través de una cola de mensajes. Un consumer recoge cada batch y lo escribe en las tablas fuente de ClickHouse. Si el consumer falla a mitad de batch, el mensaje se queda en la cola y se reintenta. Nada nuevo hasta aquí.

El problema son los duplicados. Reintentos de cola, reinicios de pipeline, replays del upstream. Para OHLC, los duplicados no hacen daño: max sigue siendo max, argMin sobre el mismo par precio-timestamp da el mismo resultado. Idempotente por naturaleza.

El volumen es otra historia. sum no es idempotente. Un trade duplicado significa volumen contado dos veces, y para cuando ReplacingMergeTree elimina el duplicado en un merge de fondo, la materialized view ya se disparó y sumó la cantidad extra al agregado. No puedes deshacer eso.

Así que deduplicamos los trades antes de que lleguen a ClickHouse. La pipeline de ingestión usa una cache en Redis: cada evento entrante se comprueba contra eventos procesados recientemente usando SET NX. Tres estados: listo (procesarlo), ya-procesado (ignorearlo), en-proceso (otro worker lo tiene). La ventana de dedup es de un día. Generosa, pero preferimos gastar memoria de Redis a duplicar el volumen de un trades.

Datos raw: dos tablas ReplacingMergeTree como fuente

Los datos raw aterrizan en dos tablas ReplacingMergeTree separadas. La primera almacena los eventos de precio:

CREATE TABLE raw_prices
(
    event_timestamp DateTime64(3),
    asset_id        String,
    price_usd       Float64,
    ingestion_ts    DateTime DEFAULT now()
)
ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMM(event_timestamp)
ORDER BY (asset_id, event_timestamp)

La segunda almacena los eventos individuales de trade:

CREATE TABLE raw_trades
(
    event_timestamp DateTime64(3),
    market_id       String,
    asset_in        String,
    amount_in       Float64,
    asset_out       String,
    amount_out      Float64,
    ingestion_ts    DateTime DEFAULT now()
)
ENGINE = ReplacingMergeTree(ingestion_ts)
PARTITION BY toYYYYMM(event_timestamp)
ORDER BY (event_timestamp, asset_in, asset_out)

ReplacingMergeTree(ingestion_ts) usa el timestamp de ingesta como columna de versión. Si el mismo trade aterriza dos veces, los merges de fondo conservan la fila con el ingestion_ts más reciente para cada clave ORDER BY. Que (asset_id, event_timestamp) y (event_timestamp, asset_in, asset_out) sean suficientemente únicos depende de la granularidad de trades de tu plataforma. Si dos trades distintos pueden compartir el mismo timestamp y par de assets, necesitarías un ID de transacción o número de secuencia en el ORDER BY.

Esta es la segunda capa de defensa contra duplicados. Si algo se cuela pasado Redis, los merges de fondo acabarán limpiándolo. Pero "acabarán" es la palabra clave. Las materialized views se disparan en el insert, no en el merge, así que el volumen en las MVs estará duplicado de todas formas. Redis lo atrapa antes de que eso importe. Esta capa asegura la consistencia de los datos históricos.

La pipeline OHLC

Los eventos de precio fluyen desde raw_prices a través de cuatro tablas AggregatingMergeTree. Cada nivel enrolla el anterior. Los esquemas comparten las mismas definiciones de columnas, pero las funciones de estado cambian a partir del nivel 2.

La tabla de 1 minuto

Una materialized view observa raw_prices y alimenta el primer nivel de agregación:

CREATE TABLE ohlc_1m
(
    asset_id   String,
    time       DateTime,
    high       SimpleAggregateFunction(max, Float64),
    low        SimpleAggregateFunction(min, Float64),
    open       AggregateFunction(argMin, Float64, DateTime64(3)),
    close      AggregateFunction(argMax, Float64, DateTime64(3))
)
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(time)
ORDER BY (asset_id, time)

CREATE MATERIALIZED VIEW ohlc_1m_mv TO ohlc_1m AS
SELECT
    asset_id,
    toStartOfMinute(event_timestamp) AS time,
    maxSimpleState(price_usd)        AS high,
    minSimpleState(price_usd)        AS low,
    argMinState(price_usd, event_timestamp) AS open,
    argMaxState(price_usd, event_timestamp) AS close
FROM raw_prices
WHERE price_usd > 0
GROUP BY asset_id, time

high/low almacenan Float64s planos. open/close almacenan blobs binarios. Dos tipos de columna en la misma tabla, cada uno elegido por su comportamiento durante el merge.

La cascada

Este es el nivel de 5 minutos:

CREATE MATERIALIZED VIEW ohlc_5m_mv TO ohlc_5m AS
SELECT
    asset_id,
    intDiv(toUnixTimestamp(time), 300) * 300 AS time,
    maxSimpleState(high)        AS high,
    minSimpleState(low)         AS low,
    argMinMergeState(open)      AS open,
    argMaxMergeState(close)     AS close
FROM ohlc_1m
GROUP BY asset_id, time

Fíjate en lo que cambió. high y low siguen usando maxSimpleState/minSimpleState. Como almacenan escalares planos, sacar el máximo de máximos de 1 minuto es la misma operación que sacar el máximo de valores raw.

open y close pasaron de argMinState a argMinMergeState. En el nivel 1, -State crea un estado binario a partir de pares (precio, timestamp) raw. A partir del nivel 2, -MergeState combina estados binarios existentes para determinar qué precio tuvo el timestamp más temprano en el intervalo más amplio.

La cascada OHLC completa:

Tabla	Fuente	Bucketing temporal	high/low	open/close
ohlc_1m	raw_prices	toStartOfMinute(event_timestamp)	maxSimpleState, minSimpleState	argMinState, argMaxState
ohlc_5m	ohlc_1m	intDiv(time, 300) * 300	maxSimpleState, minSimpleState	argMinMergeState, argMaxMergeState
ohlc_1h	ohlc_5m	intDiv(time, 3600) * 3600	maxSimpleState, minSimpleState	argMinMergeState, argMaxMergeState
ohlc_1d	ohlc_1h	intDiv(time, 86400) * 86400	maxSimpleState, minSimpleState	argMinMergeState, argMaxMergeState

La pipeline de volumen

El volumen usa un agregado más sencillo (sum), pero la ruta de datos es más compleja.

La tabla de 1 minuto con dos MVs

Cada trade tiene dos lados: un asset que entra y un asset que sale. Calculamos el volumen para ambos lados con materialized views separadas que escriben en el mismo AggregatingMergeTree:

CREATE TABLE volume_1m
(
    asset_id   String,
    time       DateTime,
    volume     SimpleAggregateFunction(sum, Float64),
    volume_usd SimpleAggregateFunction(sum, Float64)
)
ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(time)
ORDER BY (asset_id, time)

CREATE MATERIALIZED VIEW volume_in_1m_mv TO volume_1m AS
SELECT
    asset_in                              AS asset_id,
    toStartOfMinute(event_timestamp)      AS time,
    sumSimpleState(amount_in)             AS volume,
    sumSimpleState(amount_in * price_usd) AS volume_usd
FROM raw_trades
INNER JOIN last_prices USING (asset_in)
GROUP BY asset_id, time

CREATE MATERIALIZED VIEW volume_out_1m_mv TO volume_1m AS
SELECT
    asset_out                              AS asset_id,
    toStartOfMinute(event_timestamp)       AS time,
    sumSimpleState(amount_out)             AS volume,
    sumSimpleState(amount_out * price_usd) AS volume_usd
FROM raw_trades
INNER JOIN last_prices USING (asset_out)
GROUP BY asset_id, time

Dos MVs, una tabla destino. La MV del lado IN agrega amount_in, la del lado OUT agrega amount_out. Ambas hacen JOIN contra una tabla de último precio para convertir los montos de trade a USD en tiempo de inserción. El engine combina los valores sumSimpleState automáticamente durante los merges de fondo. Mismo tipo de agregado desde ambas fuentes, sin conflicto.

La cascada

Mismo patrón que OHLC. Este es el nivel de 5 minutos:

CREATE MATERIALIZED VIEW volume_5m_mv TO volume_5m AS
SELECT
    asset_id,
    intDiv(toUnixTimestamp(time), 300) * 300 AS time,
    sumSimpleState(volume)     AS volume,
    sumSimpleState(volume_usd) AS volume_usd
FROM volume_1m
GROUP BY asset_id, time

Nada de -MergeState por ningún lado. Como sum es asociativa y volume es un SimpleAggregateFunction que almacena un Float64 plano, sumar sumas de 1 minuto es lo mismo que sumar valores raw. sumSimpleState funciona en todos los niveles.

Tabla	Fuente	Bucketing temporal	Funciones
volume_1m	raw_trades	toStartOfMinute(event_timestamp)	sumSimpleState
volume_5m	volume_1m	intDiv(time, 300) * 300	sumSimpleState
volume_1h	volume_5m	intDiv(time, 3600) * 3600	sumSimpleState
volume_1d	volume_1h	intDiv(time, 86400) * 86400	sumSimpleState

Comparada con la de OHLC, la cascada de volumen es directa. Sin estados binarios, sin combinadores -MergeState, sin lógica especial de merge. Sumas de arriba abajo.

Servicio: consultas + streaming

Dos caminos alimentan el gráfico. Velas históricas desde ClickHouse. Ticks en vivo desde la pipeline.

Consultas: el protocolo UDF de TradingView

La librería de gráficos de TradingView habla UDF. El endpoint que importa es /history: recibe un símbolo, una resolución y un rango temporal, y espera OHLCV de vuelta como arrays paralelos.

El backend elige el nivel de cascada más cercano que sea menor o igual a la resolución pedida. Un gráfico de 2 horas lee de ohlc_1h. Uno de 15 minutos lee de ohlc_5m. La clave: las resoluciones que no tienen su propia tabla funcionan igualmente, porque la query re-agrega estados pre-computados del nivel inferior. Una vela de 2 horas son dos estados de 1 hora combinados en tiempo de lectura usando combinadores -Merge:

SELECT
    asset_id,
    intDiv(time, 7200) * 7200 AS time,  -- buckets de 2 horas desde datos de 1h
    max(high)                  AS high,  -- SimpleAggregateFunction: escalar plano
    min(low)                   AS low,
    argMinMerge(open)          AS open,  -- AggregateFunction: combinar estados binarios
    argMaxMerge(close)         AS close
FROM ohlc_1h
WHERE asset_id = 'USD'
  AND time >= 1710000000
  AND time <  1710086400
GROUP BY asset_id, time
ORDER BY time ASC

El backend lanza una consulta separada contra volume_1h para el volumen (SimpleAggregateFunction plano, sin combinador necesario), luego fusiona ambos resultados en el formato de respuesta UDF antes de devolvérselo al cliente.

Una cache KV se sienta delante de ClickHouse. En cada petición a /history, el backend consulta primero el KV. Cache miss o datos caducados? Trae solo el delta desde ClickHouse a partir del último timestamp cacheado, lo fusiona con el histórico almacenado y actualiza la cache. ClickHouse solo recibe consultas por velas nuevas, no por el rango completo.

Si el KV se cae, cada carga de gráfico pega directamente contra ClickHouse. Una petición típica de 6 meses de velas horarias devuelve unas 4.300 filas de ohlc_1h en menos de 200ms. Sin la cascada, esa misma petición escanearía millones de eventos raw. Es tanto la optimización de rendimiento como la red de seguridad.

Streaming: Durable Objects como fan-out de WebSockets

Las velas históricas están pre-computadas y se escanean rápido. Pero hay una latencia de 2-5 segundos entre que ocurre un evento de precio y es consultable en ClickHouse: ingesta, procesamiento de la materialized view, merges de fondo. Para traders mirando mercados en vivo, ese hueco importa.

Así que usamos una ruta dual. Cuando la pipeline procesa nuevos eventos, mapea cada precio al formato de streaming de TradingView y lo envía por POST a un endpoint /broadcast:

// Lado de la pipeline: mapear precio al formato de streaming de TradingView
const payload = prices
    .map((p) =>
        JSON.stringify({
            f: 't', // formato: trade
            id: marketByAsset[p.assetId], // "USD"
            p: p.priceUsd, // precio
            t: Math.floor(p.eventTime / 1000), // timestamp unix
            s: 0, // tamaño
        })
    )
    .join('\n') // NDJSON

await fetch(`${apiUrl}/tradingview/broadcast`, {
    method: 'POST',
    headers: { Authorization: `Bearer ${apiKey}` },
    body: payload,
})

Una única instancia de Durable Object de Cloudflare mantiene las conexiones WebSocket con todos los clientes del gráfico. Cuando llega un broadcast, lo distribuye a todos los sockets conectados.

Las conexiones usan hibernatable WebSockets: this.ctx.acceptWebSocket(server) en lugar de server.accept(). El DO se desaloja de memoria mientras las conexiones permanecen abiertas. Cuando llega un mensaje, el runtime lo recrea. El uso de memoria escala con la actividad, no con el número de conexiones.

Los gráficos muestran velas históricas de ClickHouse y ticks en vivo de la pipeline al mismo tiempo. Sin hueco entre la última vela almacenada y el movimiento en tiempo real.

Arquitectura de extremo a extremo

Así es como queda el sistema completo. Un evento de precio entra por una cola de mensajes, se deduplica en Redis, y se divide en dos caminos. Uno escribe en ClickHouse, donde las materialized views van en cascada a través de cuatro niveles de resolución tanto para OHLC como para volumen. El otro lo emite directamente a los gráficos conectados a través de un Durable Object.

La estructura de costes sigue la arquitectura. ClickHouse se lleva el grueso del almacenamiento y el cómputo de ambos pipelines. Ahí es donde va la mayor parte del gasto en infraestructura. La capa de servicio es ligera en comparación: una cache KV que absorbe la mayoría de lecturas, un único Durable Object para el fan-out de WebSockets, y un API Worker que fusiona dos resultados de query en una sola respuesta. Sin cache, la historia cambia; estuvimos ahí, y no fue agradable.

La ruta dual de servicio (histórico desde ClickHouse, en vivo desde la pipeline) añade complejidad arquitectónica pero elimina el hueco de latencia que de otro modo existiría entre ingesta y consultabilidad. Para un producto de gráficos, ese hueco es la diferencia entre una herramienta usable y una frustrante.