Tipos de Barras y Métodos de Agregación para Trading Algorítmico

Todos los gráficos de velas que ha visto en Binance, TradingView o en cualquier interfaz de exchange se construyen de la misma manera: agregar operaciones dentro de una ventana temporal fija — 1 minuto, 5 minutos, 1 hora — y producir una barra OHLCV. Esto es tan omnipresente que la mayoría de los traders nunca lo cuestionan. Pero para el trading algorítmico, la elección del tipo de barra y el método de agregación son dos decisiones independientes — y la mayoría de los sistemas las confunden.

Este artículo separa los dos ejes de la construcción de velas: qué tipo de barra construye (17 tipos) y cómo las agrega en marcos temporales superiores (3 métodos). La combinación da 51 configuraciones posibles, cada una con propiedades diferentes para backtesting, trading en vivo y generación de señales.

Para una introducción sobre cómo las operaciones en bruto se convierten en velas estándar, vea Trading Candles Demystified.

TL;DR

• La construcción de velas tiene dos ejes independientes: tipo de barra y método de agregación
• 17 tipos base de barras: tiempo, tick, volumen, dólar, Renko, rango, volatilidad, Heikin-Ashi, Kagi, Line Break, P&F, tick imbalance (TIB), volume imbalance (VIB), run, CUSUM, entropía, delta
• 3 métodos de agregación: alineado al calendario, ventana rolling, rolling adaptativo
• 17 × 3 = 51 combinaciones posibles, cada una con propiedades diferentes
• La mayoría de los sistemas usan solo una combinación: barras de tiempo alineadas al calendario. Las otras 50 están sin explotar.
• Recomendación práctica: usar múltiples combinaciones en capas — barras de tiempo rolling para señales, barras de tiempo de calendario para estructura de mercado, barras impulsadas por información para microestructura

Dos Ejes de la Construcción de Velas

La visión tradicional coloca todos los tipos de barras en una lista plana: barras de tiempo, barras de tick, barras de volumen, Renko, etc. Esto es engañoso. En realidad hay dos decisiones ortogonales:

Eje 1 — Tipo Base de Barra (17 tipos): ¿Cómo decide cuándo cierra una nueva barra? ¿Después de un intervalo de tiempo fijo? ¿Después de N operaciones? ¿Después de un movimiento de precio? ¿Cuando cambia el contenido de información? Esto determina qué significa "una barra".

Eje 2 — Método de Agregación (3 métodos): ¿Cómo compone las barras base en velas de marco temporal superior? ¿Alinear con los límites del calendario (00:00, 01:00, ...)? ¿Usar una ventana rolling de las últimas N barras? ¿Adaptar el tamaño de la ventana a la volatilidad?

Estos dos ejes son independientes. Puede tener:

• Barras de tick alineadas al calendario — agregar barras de tick que cerraron entre las 14:00 y las 14:59 en una única vela horaria
• Barras de volumen rolling — tomar las últimas 24 barras de volumen sin importar cuándo cerraron
• Barras delta adaptativas — usar una ventana impulsada por la volatilidad sobre barras delta

La "vela de 1 hora" estándar es solo un punto en esta matriz de 17×3: barras de tiempo + alineación al calendario. Cualquier otra combinación es una alternativa que vale la pena considerar.

1. Barras de Tiempo (Estándar)

Densidad de información desigual: los límites rígidos de tiempo tratan las horas tranquilas de 200 operaciones igual que las horas de anuncios con 50.000 operaciones.

El predeterminado. Se forma una nueva barra después de un intervalo de tiempo fijo: 1 minuto, 5 minutos, 1 hora. Cada exchange las proporciona de forma nativa.

Propiedades:

• Durante la sesión asiática (00:00–08:00 UTC), una vela de 1 hora podría contener 200 operaciones. Durante un anuncio de listado en Binance, esa misma ventana podría contener 50.000 operaciones. Las barras de tiempo tratan ambas como equivalentes. Detectar tales picos de actividad es fundamental para la protección de bots — vea Anomaly Detection for Trading Bots.
• Todos los participantes del mercado ven los mismos límites de vela — un punto Schelling. Esto hace que las barras de tiempo sean esenciales para analizar el comportamiento de la multitud.
• Los indicadores calculados sobre velas parciales (tras un reinicio) producen valores inservibles.

from datetime import datetime

def time_until_valid_hourly_candle():
    """How long until the first complete hourly candle after restart."""
    now = datetime.utcnow()
    minutes_into_hour = now.minute
    seconds_into_minute = now.second

    wait_seconds = (60 - minutes_into_hour) * 60 - seconds_into_minute
    wait_seconds += 3600

    return wait_seconds

2–4. Barras Basadas en Actividad

Barras de tick, volumen y dólar: tres formas de dejar que la participación del mercado — no el reloj — determine los límites de las barras.

En lugar de muestrear en intervalos de tiempo fijos, se muestrea después de una cantidad fija de actividad de mercado. Esto produce barras con un "contenido de información" aproximadamente igual, sin importar la hora del día.

2. Barras de Tick

Se forma una nueva barra después de cada N operaciones (ticks). Durante alta actividad, las barras se forman rápidamente. Durante períodos tranquilos, una sola barra puede abarcar horas.

from collections import deque
from dataclasses import dataclass

@dataclass
class OHLCV:
    timestamp: int
    open: float
    high: float
    low: float
    close: float
    volume: float

class TickBarGenerator:
    """
    Generates a new bar every `threshold` trades.
    Each bar contains equal number of market "opinions".
    """

    def __init__(self, threshold: int = 1000):
        self.threshold = threshold
        self.trades: list[tuple[float, float]] = []  # (price, qty)
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        self.trades.append((price, qty))

        if len(self.trades) >= self.threshold:
            self._close_bar(timestamp)

    def _close_bar(self, timestamp: int):
        prices = [t[0] for t in self.trades]
        volumes = [t[1] for t in self.trades]

        bar = OHLCV(
            timestamp=timestamp,
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)
        self.trades = []
        return bar

Ventajas: Se adapta naturalmente a la actividad del mercado. Los retornos de las barras de tick tienden a estar más cerca de una distribución normal que los retornos de las barras de tiempo — una propiedad que mejora el rendimiento de muchos modelos estadísticos.

Desventajas: Requiere un flujo de operaciones en bruto (no disponible en todos los proveedores de datos para datos históricos). El tiempo de las barras es impredecible — no se puede decir "la próxima barra cerrará en X".

3. Barras de Volumen

Se forma una nueva barra después de que se hayan negociado N contratos (o monedas, en cripto). Similar a las barras de tick, pero ponderada por el tamaño de la operación — una única operación de 100 BTC contribuye 100 veces más que una operación de 1 BTC.

class VolumeBarGenerator:
    """
    Generates a new bar every `threshold` units of volume.
    Normalizes for trade size: one large order ≠ one small order.
    """

    def __init__(self, threshold: float = 100.0):
        self.threshold = threshold
        self.accumulated_volume = 0.0
        self.trades: list[tuple[int, float, float]] = []  # (ts, price, qty)
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        self.trades.append((timestamp, price, qty))
        self.accumulated_volume += qty

        if self.accumulated_volume >= self.threshold:
            self._close_bar()

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)
        self.accumulated_volume = 0.0
        self.trades = []
        return bar

4. Barras de Dólar

Se forma una nueva barra después de que se haya intercambiado un valor nocional fijo (en USD/USDT). La más robusta de las barras basadas en actividad porque normaliza tanto el número de operaciones como el nivel de precio.

Considere: si ETH pasa de $1.000 a$4.000, vender ETH por valor de $10.000 requiere 2,5 ETH a$4.000 pero 10 ETH a $1.000. Las barras de volumen tratarían esto de forma diferente; las barras de dólar las tratan igual.

class DollarBarGenerator:
    """
    Generates a new bar every `threshold` dollars (USDT) of notional volume.
    Most robust normalization: independent of price level.

    Lopez de Prado (2018) recommends dollar bars as the default
    for most quantitative applications.
    """

    def __init__(self, threshold: float = 1_000_000.0):
        self.threshold = threshold
        self.accumulated_dollars = 0.0
        self.trades: list[tuple[int, float, float]] = []
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        self.trades.append((timestamp, price, qty))
        self.accumulated_dollars += price * qty

        if self.accumulated_dollars >= self.threshold:
            self._close_bar()

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)
        self.accumulated_dollars = 0.0
        self.trades = []
        return bar

Elegir el Umbral

El umbral para las barras basadas en actividad debería producir aproximadamente el mismo número de barras por día que las barras de tiempo que está reemplazando. Para BTCUSDT en Binance:

Estos números son aproximados y varían drásticamente con el régimen de mercado. Durante un rally o un crash, las barras basadas en actividad producirán entre 5 y 10 veces más barras de lo habitual — que es exactamente el punto.

5–7. Barras Basadas en Precio

Ladrillos Renko, barras de rango y barras de volatilidad: muestrear solo cuando el precio se mueve lo suficiente como para importar.

Las barras basadas en precio ignoran tanto el tiempo como la actividad. Una nueva barra se forma solo cuando el precio se mueve una cantidad especificada. Esto filtra naturalmente el ruido lateral y resalta las tendencias.

5. Barras Renko

Se forma un nuevo "ladrillo" Renko cuando el precio de cierre se mueve al menos N unidades desde el cierre del ladrillo anterior. Los ladrillos siempre tienen el mismo tamaño, creando una representación visual limpia de la dirección de la tendencia.

class RenkoBarGenerator:
    """
    Generates Renko bricks based on price movement.

    Key property: during sideways movement, no new bricks form.
    During strong trends, bricks form rapidly.
    """

    def __init__(self, brick_size: float = 10.0):
        self.brick_size = brick_size
        self.bricks: list[dict] = []
        self.last_close: float | None = None

    def on_price(self, timestamp: int, price: float, volume: float = 0.0):
        if self.last_close is None:
            self.last_close = price
            return []

        new_bricks = []
        diff = price - self.last_close
        num_bricks = int(abs(diff) / self.brick_size)

        if num_bricks == 0:
            return []

        direction = 1 if diff > 0 else -1

        for i in range(num_bricks):
            brick_open = self.last_close
            brick_close = self.last_close + direction * self.brick_size

            brick = {
                'timestamp': timestamp,
                'open': brick_open,
                'high': max(brick_open, brick_close),
                'low': min(brick_open, brick_close),
                'close': brick_close,
                'volume': volume / num_bricks if num_bricks > 0 else 0,
                'direction': direction,
            }
            new_bricks.append(brick)
            self.last_close = brick_close

        self.bricks.extend(new_bricks)
        return new_bricks

Renko dinámico usa ATR (Average True Range) en lugar de un tamaño de ladrillo fijo, adaptándose automáticamente a la volatilidad.

6. Barras de Rango

Cada barra tiene un rango máximo-mínimo fijo. Cuando se supera el rango, la barra cierra y comienza una nueva. A diferencia de Renko, las barras de rango incluyen mechas y pueden mostrar la volatilidad intra-barra.

class RangeBarGenerator:
    """
    Generates bars with a fixed high-low range.

    Difference from Renko: range bars show the full OHLC within
    the range, not just brick direction. More information-rich.
    """

    def __init__(self, range_size: float = 20.0):
        self.range_size = range_size
        self.current_high: float | None = None
        self.current_low: float | None = None
        self.current_open: float | None = None
        self.current_volume: float = 0.0
        self.current_start_ts: int = 0
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        if self.current_open is None:
            self.current_open = price
            self.current_high = price
            self.current_low = price
            self.current_start_ts = timestamp

        self.current_high = max(self.current_high, price)
        self.current_low = min(self.current_low, price)
        self.current_volume += qty

        if self.current_high - self.current_low >= self.range_size:
            bar = OHLCV(
                timestamp=timestamp,
                open=self.current_open,
                high=self.current_high,
                low=self.current_low,
                close=price,
                volume=self.current_volume,
            )
            self.bars.append(bar)

            self.current_open = price
            self.current_high = price
            self.current_low = price
            self.current_volume = 0.0
            self.current_start_ts = timestamp

            return bar

        return None

Diferencia clave entre Renko y barras de rango: Renko rastrea solo los precios de cierre y muestra la dirección; las barras de rango rastrean el rango de precio completo y muestran la estructura dentro de la barra. Las barras de rango son generalmente más útiles para el trading algorítmico porque preservan la información de máximo-mínimo necesaria para la simulación de stop-loss y take-profit.

7. Barras de Volatilidad

Se forma una nueva barra cuando la volatilidad intra-barra alcanza un umbral dinámico — por ejemplo, un múltiplo del ATR reciente. A diferencia de las barras de rango (umbral fijo), las barras de volatilidad se adaptan a las condiciones del mercado.

class VolatilityBarGenerator:
    """
    Generates bars when intra-bar volatility reaches a threshold.

    Similar to range bars, but the threshold adapts to market conditions
    using a rolling ATR measure. In calm markets, bars need less
    absolute movement to close; in volatile markets, more.
    """

    def __init__(
        self,
        atr_period: int = 14,
        atr_multiplier: float = 1.0,
        initial_threshold: float = 20.0,
    ):
        self.atr_period = atr_period
        self.atr_multiplier = atr_multiplier
        self.threshold = initial_threshold

        self.recent_ranges: list[float] = []
        self.current_open: float | None = None
        self.current_high: float | None = None
        self.current_low: float | None = None
        self.current_volume: float = 0.0
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        if self.current_open is None:
            self.current_open = price
            self.current_high = price
            self.current_low = price

        self.current_high = max(self.current_high, price)
        self.current_low = min(self.current_low, price)
        self.current_volume += qty

        intra_bar_range = self.current_high - self.current_low

        if intra_bar_range >= self.threshold:
            bar = OHLCV(
                timestamp=timestamp,
                open=self.current_open,
                high=self.current_high,
                low=self.current_low,
                close=price,
                volume=self.current_volume,
            )
            self.bars.append(bar)

            self.recent_ranges.append(intra_bar_range)
            if len(self.recent_ranges) > self.atr_period:
                self.recent_ranges = self.recent_ranges[-self.atr_period:]
            if len(self.recent_ranges) >= self.atr_period:
                avg_range = sum(self.recent_ranges) / len(self.recent_ranges)
                self.threshold = avg_range * self.atr_multiplier

            self.current_open = price
            self.current_high = price
            self.current_low = price
            self.current_volume = 0.0
            return bar

        return None

8. Heikin-Ashi (Transformación Suavizada)

Heikin-Ashi: el promediado transforma velas ruidosas en señales de tendencia suaves — a costa de la información exacta de precio.

Heikin-Ashi (japonés para "barra promedio") no es un tipo de barra — es una transformación que puede aplicarse sobre cualquier tipo base de barra. Suaviza las velas promediando los valores de la barra actual y anterior:

• HA Close = (Open + High + Low + Close) / 4
• HA Open = (HA Open anterior + HA Close anterior) / 2
• HA High = max(High, HA Open, HA Close)
• HA Low = min(Low, HA Open, HA Close)

Las tendencias aparecen como secuencias de velas del mismo color sin mechas inferiores (tendencia alcista) o sin mechas superiores (tendencia bajista).

class HeikinAshiTransformer:
    """
    Transforms standard OHLCV candles into Heikin-Ashi candles.

    Can be applied on top of ANY bar type: time bars, volume bars,
    rolling bars, etc. It's a transformation, not a sampling method.

    WARNING: HA prices are synthetic — they don't represent real
    traded prices. Never use HA close for order placement or
    PnL calculation. Use HA only for signal generation, then
    execute at real prices.
    """

    def __init__(self):
        self.prev_ha_open: float | None = None
        self.prev_ha_close: float | None = None

    def transform(self, candle: OHLCV) -> OHLCV:
        ha_close = (candle.open + candle.high + candle.low + candle.close) / 4

        if self.prev_ha_open is None:
            ha_open = (candle.open + candle.close) / 2
        else:
            ha_open = (self.prev_ha_open + self.prev_ha_close) / 2

        ha_high = max(candle.high, ha_open, ha_close)
        ha_low = min(candle.low, ha_open, ha_close)

        self.prev_ha_open = ha_open
        self.prev_ha_close = ha_close

        return OHLCV(
            timestamp=candle.timestamp,
            open=ha_open,
            high=ha_high,
            low=ha_low,
            close=ha_close,
            volume=candle.volume,
        )

    def transform_series(self, candles: list[OHLCV]) -> list[OHLCV]:
        """Transform an entire series. Resets state first."""
        self.prev_ha_open = None
        self.prev_ha_close = None
        return [self.transform(c) for c in candles]


def ha_trend_signal(ha_candles: list[OHLCV], lookback: int = 3) -> int:
    """
    Simple HA trend signal.

    Returns:
        +1: bullish (N consecutive green HA candles with no lower wick)
        -1: bearish (N consecutive red HA candles with no upper wick)
         0: no clear trend
    """
    if len(ha_candles) < lookback:
        return 0

    recent = ha_candles[-lookback:]

    all_bullish = all(
        c.close > c.open and abs(c.low - min(c.open, c.close)) < 1e-10
        for c in recent
    )

    all_bearish = all(
        c.close < c.open and abs(c.high - max(c.open, c.close)) < 1e-10
        for c in recent
    )

    if all_bullish:
        return 1
    elif all_bearish:
        return -1
    return 0

Advertencia crítica para el backtesting: Los precios de Heikin-Ashi son sintéticos. Si su backtest usa el cierre HA como precio de entrada, los resultados serán incorrectos. Use siempre HA solo para la generación de señales y ejecute a precios OHLC reales.

Cuándo es útil HA: Estrategias de seguimiento de tendencia que necesitan señales limpias de "mantenerse dentro". Aplique HA sobre cualquier tipo base de barra — barras de tiempo, barras de volumen, barras de dólar — para filtrar cruces falsos.

Cuándo es perjudicial HA: Cualquier estrategia que necesite niveles de precio precisos — soporte/resistencia, análisis del libro de órdenes, PIQ (Position In Queue). El promediado destruye la información exacta de precio.

9–11. Gráficos de Reversión Japoneses

Kagi, Line Break y Point & Figure: métodos de gráficos libres de tiempo que se centran puramente en la estructura del precio.

Estos son métodos tradicionales de gráficos japoneses (junto con Renko) que descartan por completo el tiempo y se centran en la estructura del precio.

9. Gráficos Kagi

Los gráficos Kagi consisten en líneas verticales que cambian de dirección cuando el precio se revierte una cantidad especificada. Las líneas cambian de grosor cuando el precio rompe un máximo anterior (grueso = "yang" = demanda) o un mínimo anterior (delgado = "yin" = oferta).

class KagiChartGenerator:
    """
    Generates Kagi chart lines based on price reversals.

    Unlike Renko (fixed brick size), Kagi tracks the actual magnitude
    of each move and changes line thickness at breakout points.

    Useful for identifying support/resistance breaks and
    supply/demand shifts without time noise.
    """

    def __init__(self, reversal_amount: float = 10.0):
        self.reversal_amount = reversal_amount
        self.lines: list[dict] = []
        self.current_direction: int = 0  # 1=up, -1=down
        self.current_price: float | None = None
        self.extreme_price: float | None = None
        self.prev_high: float | None = None
        self.prev_low: float | None = None
        self.line_type: str = 'yang'  # 'yang' (thick) or 'yin' (thin)

    def on_price(self, timestamp: int, price: float):
        if self.current_price is None:
            self.current_price = price
            self.extreme_price = price
            return None

        if self.current_direction == 0:
            if price - self.current_price >= self.reversal_amount:
                self.current_direction = 1
                self.extreme_price = price
            elif self.current_price - price >= self.reversal_amount:
                self.current_direction = -1
                self.extreme_price = price
            return None

        if self.current_direction == 1:
            if price > self.extreme_price:
                self.extreme_price = price
                if self.prev_high is not None and price > self.prev_high:
                    self.line_type = 'yang'
            elif self.extreme_price - price >= self.reversal_amount:
                line = {
                    'timestamp': timestamp,
                    'start': self.current_price,
                    'end': self.extreme_price,
                    'direction': 'up',
                    'type': self.line_type,
                }
                self.lines.append(line)
                self.prev_high = self.extreme_price
                self.current_price = self.extreme_price
                self.extreme_price = price
                self.current_direction = -1
                if self.prev_low is not None and price < self.prev_low:
                    self.line_type = 'yin'
                return line
        else:
            if price < self.extreme_price:
                self.extreme_price = price
                if self.prev_low is not None and price < self.prev_low:
                    self.line_type = 'yin'
            elif price - self.extreme_price >= self.reversal_amount:
                line = {
                    'timestamp': timestamp,
                    'start': self.current_price,
                    'end': self.extreme_price,
                    'direction': 'down',
                    'type': self.line_type,
                }
                self.lines.append(line)
                self.prev_low = self.extreme_price
                self.current_price = self.extreme_price
                self.extreme_price = price
                self.current_direction = 1
                if self.prev_high is not None and price > self.prev_high:
                    self.line_type = 'yang'
                return line

        return None

10. Gráficos Line Break

Los gráficos Line Break dibujan una nueva línea (caja) solo cuando el precio de cierre supera el máximo o el mínimo de las N líneas anteriores (típicamente 3). No se dibuja ninguna línea nueva si el precio permanece dentro del rango.

class LineBreakGenerator:
    """
    Generates Line Break bars (Three Line Break by default).

    A new bar is drawn only when the close exceeds the high or low
    of the last N bars. Filters out minor noise by requiring price
    to break through a multi-bar range.

    The 'N' parameter (line_count) controls sensitivity:
    - N=2: more sensitive, more bars, more noise
    - N=3: standard (Three Line Break)
    - N=4+: less sensitive, fewer bars, stronger signals
    """

    def __init__(self, line_count: int = 3):
        self.line_count = line_count
        self.lines: list[dict] = []

    def on_close(self, timestamp: int, close: float) -> dict | None:
        if not self.lines:
            self.lines.append({
                'timestamp': timestamp,
                'open': close,
                'close': close,
                'high': close,
                'low': close,
                'direction': 0,
            })
            return None

        lookback = self.lines[-self.line_count:] if len(self.lines) >= self.line_count else self.lines

        highest = max(l['high'] for l in lookback)
        lowest = min(l['low'] for l in lookback)
        last = self.lines[-1]

        new_line = None

        if close > highest:
            new_line = {
                'timestamp': timestamp,
                'open': last['close'],
                'close': close,
                'high': close,
                'low': last['close'],
                'direction': 1,
            }
        elif close < lowest:
            new_line = {
                'timestamp': timestamp,
                'open': last['close'],
                'close': close,
                'high': last['close'],
                'low': close,
                'direction': -1,
            }

        if new_line:
            self.lines.append(new_line)
            return new_line

        return None

11. Gráficos Point & Figure

Los gráficos Point & Figure (P&F) usan columnas de X (precios en alza) y O (precios en baja). Los cambios de columna requieren una reversión de típicamente 3 tamaños de caja. Uno de los métodos más antiguos para filtrar ruido e identificar soporte/resistencia.

class PointAndFigureGenerator:
    """
    Generates Point & Figure chart data.

    X column: price rising by box_size increments.
    O column: price falling by box_size increments.
    Column switch: requires reversal_boxes * box_size movement
    in the opposite direction.

    Classic setting: box_size based on ATR, reversal_boxes = 3.
    """

    def __init__(self, box_size: float = 10.0, reversal_boxes: int = 3):
        self.box_size = box_size
        self.reversal_boxes = reversal_boxes
        self.reversal_amount = box_size * reversal_boxes

        self.columns: list[dict] = []
        self.current_direction: int = 0
        self.current_top: float | None = None
        self.current_bottom: float | None = None

    def on_price(self, timestamp: int, price: float):
        if self.current_top is None:
            box_price = self._round_to_box(price)
            self.current_top = box_price
            self.current_bottom = box_price
            self.current_direction = 1
            return None

        events = []

        if self.current_direction == 1:
            while price >= self.current_top + self.box_size:
                self.current_top += self.box_size
                events.append(('X', self.current_top, timestamp))

            if price <= self.current_top - self.reversal_amount:
                col = {
                    'type': 'X',
                    'top': self.current_top,
                    'bottom': self.current_bottom,
                    'boxes': int((self.current_top - self.current_bottom) / self.box_size) + 1,
                    'timestamp': timestamp,
                }
                self.columns.append(col)
                self.current_direction = -1
                self.current_top = self.current_top - self.box_size
                self.current_bottom = self._round_to_box(price)
                events.append(('new_column', 'O', timestamp))

        else:
            while price <= self.current_bottom - self.box_size:
                self.current_bottom -= self.box_size
                events.append(('O', self.current_bottom, timestamp))

            if price >= self.current_bottom + self.reversal_amount:
                col = {
                    'type': 'O',
                    'top': self.current_top,
                    'bottom': self.current_bottom,
                    'boxes': int((self.current_top - self.current_bottom) / self.box_size) + 1,
                    'timestamp': timestamp,
                }
                self.columns.append(col)
                self.current_direction = 1
                self.current_bottom = self.current_bottom + self.box_size
                self.current_top = self._round_to_box(price)
                events.append(('new_column', 'X', timestamp))

        return events if events else None

    def _round_to_box(self, price: float) -> float:
        return round(price / self.box_size) * self.box_size

Kagi, Line Break y P&F en el trading algorítmico: Se usan principalmente para la detección de tendencias a largo plazo y la identificación de soporte/resistencia. Como capa de filtro — "no tomar señales largas cuando el gráfico Kagi está en modo yin" — aportan valor al alinear las operaciones con la estructura macro.

12–14. Barras Impulsadas por Información

Barras de imbalance, barras de run, filtros CUSUM y barras de entropía: muestrear cuando el mercado nos dice que algo ha cambiado.

El enfoque más sofisticado, de Advances in Financial Machine Learning (2018) de Marcos Lopez de Prado. La idea central: muestrear cuando llega nueva información al mercado, no en intervalos fijos.

12. Tick Imbalance Bars (TIB)

Si el mercado está en equilibrio, las operaciones iniciadas por compradores y las iniciadas por vendedores deberían balancearse aproximadamente. Cuando el desequilibrio supera nuestra expectativa, algo ha cambiado. Se muestrea una barra en ese momento.

Cada operación se clasifica como iniciada por comprador (+1) o iniciada por vendedor (-1) usando la regla del tick. Rastreamos el desequilibrio acumulado θ y muestreamos cuando |θ| supera un umbral dinámico.

class TickImbalanceBarGenerator:
    """
    Generates bars when the cumulative tick imbalance exceeds
    expected levels — i.e., when "new information" arrives.

    Based on Lopez de Prado (2018), Chapter 2.
    """

    def __init__(
        self,
        expected_ticks_init: int = 1000,
        ewma_window: int = 100,
        min_ticks: int = 100,
        max_ticks: int = 50000,
    ):
        self.expected_ticks_init = expected_ticks_init
        self.ewma_window = ewma_window
        self.min_ticks = min_ticks
        self.max_ticks = max_ticks

        self.theta = 0.0
        self.prev_price: float | None = None
        self.prev_sign = 1
        self.trades: list[tuple[int, float, float]] = []

        self.bar_lengths: list[int] = []
        self.imbalances: list[float] = []
        self.expected_ticks = float(expected_ticks_init)
        self.expected_imbalance = 0.0

        self.bars: list[OHLCV] = []

    def _tick_sign(self, price: float) -> int:
        """Classify trade as buy (+1) or sell (-1) using tick rule."""
        if self.prev_price is None:
            self.prev_price = price
            return 1

        if price > self.prev_price:
            sign = 1
        elif price < self.prev_price:
            sign = -1
        else:
            sign = self.prev_sign

        self.prev_price = price
        self.prev_sign = sign
        return sign

    def on_trade(self, timestamp: int, price: float, qty: float):
        sign = self._tick_sign(price)
        self.theta += sign
        self.trades.append((timestamp, price, qty))

        threshold = self.expected_ticks * abs(self.expected_imbalance)
        if threshold == 0:
            threshold = self.expected_ticks_init * 0.5

        if abs(self.theta) >= threshold and len(self.trades) >= self.min_ticks:
            return self._close_bar()

        if len(self.trades) >= self.max_ticks:
            return self._close_bar()

        return None

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)

        self.bar_lengths.append(len(self.trades))
        self.imbalances.append(self.theta / len(self.trades))

        if len(self.bar_lengths) >= 2:
            alpha = 2.0 / (self.ewma_window + 1)
            self.expected_ticks = (
                alpha * self.bar_lengths[-1]
                + (1 - alpha) * self.expected_ticks
            )
            self.expected_ticks = max(
                self.min_ticks,
                min(self.max_ticks, self.expected_ticks)
            )
            self.expected_imbalance = (
                alpha * self.imbalances[-1]
                + (1 - alpha) * self.expected_imbalance
            )

        self.theta = 0.0
        self.trades = []
        return bar

13. Volume Imbalance Bars (VIB)

Extensión de las TIB: en lugar de contar cada operación como ±1, se pondera por volumen con signo. Una compra de 100 BTC contribuye +100, una venta de 1 BTC contribuye -1. Captura grandes órdenes informadas que podrían haberse dividido en muchas operaciones pequeñas.

class VolumeImbalanceBarGenerator:
    """
    Like TIBs, but uses signed volume instead of signed ticks.

    Captures the insight that a 100-BTC buy signal is 100x more
    informative than a 1-BTC buy signal.
    """

    def __init__(
        self,
        expected_ticks_init: int = 1000,
        ewma_window: int = 100,
    ):
        self.expected_ticks_init = expected_ticks_init
        self.ewma_window = ewma_window

        self.theta = 0.0
        self.prev_price: float | None = None
        self.prev_sign = 1
        self.trades: list[tuple[int, float, float]] = []

        self.bar_lengths: list[int] = []
        self.volume_imbalances: list[float] = []
        self.expected_ticks = float(expected_ticks_init)
        self.expected_vol_imbalance = 0.0

        self.bars: list[OHLCV] = []

    def _tick_sign(self, price: float) -> int:
        if self.prev_price is None:
            self.prev_price = price
            return 1
        if price > self.prev_price:
            sign = 1
        elif price < self.prev_price:
            sign = -1
        else:
            sign = self.prev_sign
        self.prev_price = price
        self.prev_sign = sign
        return sign

    def on_trade(self, timestamp: int, price: float, qty: float):
        sign = self._tick_sign(price)
        self.theta += sign * qty
        self.trades.append((timestamp, price, qty))

        threshold = self.expected_ticks * abs(self.expected_vol_imbalance)
        if threshold == 0:
            threshold = self.expected_ticks_init * 0.5

        if abs(self.theta) >= threshold and len(self.trades) >= 10:
            return self._close_bar()
        return None

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)

        self.bar_lengths.append(len(self.trades))
        self.volume_imbalances.append(self.theta / len(self.trades))

        alpha = 2.0 / (self.ewma_window + 1)
        if len(self.bar_lengths) >= 2:
            self.expected_ticks = (
                alpha * self.bar_lengths[-1] + (1 - alpha) * self.expected_ticks
            )
            self.expected_vol_imbalance = (
                alpha * self.volume_imbalances[-1]
                + (1 - alpha) * self.expected_vol_imbalance
            )

        self.theta = 0.0
        self.trades = []
        return bar

El Problema de la Explosión

Un problema conocido de las barras de imbalance: el umbral basado en EWMA puede entrar en un bucle de retroalimentación positiva. La solución: acotar con límites min_ticks y max_ticks.

self.expected_ticks = max(
    self.min_ticks,    # Floor: never less than 100 ticks
    min(
        self.max_ticks,  # Ceiling: never more than 50000 ticks
        new_expected_ticks
    )
)

14. Barras de Run

Las barras de run rastrean la longitud de la racha direccional actual — la secuencia consecutiva más larga de compras o ventas. Cuando un gran trader informado divide una orden en muchas operaciones pequeñas, la secuencia se vuelve inusualmente larga. Las barras de run detectan esto.

class TickRunBarGenerator:
    """
    Generates bars when the length of a directional run exceeds expectations.

    Based on Lopez de Prado (2018), Chapter 2.

    Difference from imbalance bars:
    - Imbalance bars track NET imbalance (buys minus sells)
    - Run bars track the MAXIMUM run length (consecutive buys OR sells)
    """

    def __init__(
        self,
        expected_ticks_init: int = 1000,
        ewma_window: int = 100,
        min_ticks: int = 100,
        max_ticks: int = 50000,
    ):
        self.expected_ticks_init = expected_ticks_init
        self.ewma_window = ewma_window
        self.min_ticks = min_ticks
        self.max_ticks = max_ticks

        self.prev_price: float | None = None
        self.prev_sign = 1
        self.trades: list[tuple[int, float, float]] = []

        self.buy_run = 0
        self.sell_run = 0
        self.max_buy_run = 0
        self.max_sell_run = 0

        self.bar_lengths: list[int] = []
        self.max_runs: list[float] = []
        self.expected_ticks = float(expected_ticks_init)
        self.expected_max_run = 0.0

        self.bars: list[OHLCV] = []

    def _tick_sign(self, price: float) -> int:
        if self.prev_price is None:
            self.prev_price = price
            return 1
        if price > self.prev_price:
            sign = 1
        elif price < self.prev_price:
            sign = -1
        else:
            sign = self.prev_sign
        self.prev_price = price
        self.prev_sign = sign
        return sign

    def on_trade(self, timestamp: int, price: float, qty: float):
        sign = self._tick_sign(price)
        self.trades.append((timestamp, price, qty))

        if sign == 1:
            self.buy_run += 1
            self.sell_run = 0
        else:
            self.sell_run += 1
            self.buy_run = 0

        self.max_buy_run = max(self.max_buy_run, self.buy_run)
        self.max_sell_run = max(self.max_sell_run, self.sell_run)

        theta = max(self.max_buy_run, self.max_sell_run)
        threshold = self.expected_ticks * self.expected_max_run if self.expected_max_run > 0 else self.expected_ticks_init * 0.3

        if theta >= threshold and len(self.trades) >= self.min_ticks:
            return self._close_bar()

        if len(self.trades) >= self.max_ticks:
            return self._close_bar()

        return None

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)

        max_run = max(self.max_buy_run, self.max_sell_run) / len(self.trades)
        self.bar_lengths.append(len(self.trades))
        self.max_runs.append(max_run)

        alpha = 2.0 / (self.ewma_window + 1)
        if len(self.bar_lengths) >= 2:
            self.expected_ticks = alpha * self.bar_lengths[-1] + (1 - alpha) * self.expected_ticks
            self.expected_ticks = max(self.min_ticks, min(self.max_ticks, self.expected_ticks))
            self.expected_max_run = alpha * self.max_runs[-1] + (1 - alpha) * self.expected_max_run

        self.trades = []
        self.buy_run = 0
        self.sell_run = 0
        self.max_buy_run = 0
        self.max_sell_run = 0

        return bar

Las barras de run pueden extenderse a runs de volumen y runs de dólar.

15. Barras de Filtro CUSUM

El filtro CUSUM (Cumulative Sum) determina cuándo muestrear rastreando retornos acumulados. A diferencia de las barras de imbalance (que funcionan con operaciones en bruto), CUSUM puede aplicarse a datos OHLCV de 1m existentes — no se requieren datos de ticks.

class CUSUMFilterBarGenerator:
    """
    Symmetric CUSUM filter for event-based sampling.

    Based on Lopez de Prado (2018), Chapter 2.5.

    Key advantage over Bollinger Bands: CUSUM requires a FULL
    run of threshold magnitude before triggering. Bollinger Bands
    trigger repeatedly when price hovers near the band.

    Can be applied to 1m OHLCV data — no tick data required.
    """

    def __init__(self, threshold: float = 0.01):
        self.threshold = threshold
        self.s_pos = 0.0
        self.s_neg = 0.0
        self.prev_price: float | None = None
        self.buffer: list[OHLCV] = []
        self.bars: list[OHLCV] = []

    def on_candle_1m(self, candle: OHLCV) -> OHLCV | None:
        self.buffer.append(candle)

        if self.prev_price is None:
            self.prev_price = candle.close
            return None

        import math
        log_ret = math.log(candle.close / self.prev_price)
        self.prev_price = candle.close

        self.s_pos = max(0.0, self.s_pos + log_ret)
        self.s_neg = min(0.0, self.s_neg + log_ret)

        triggered = False

        if self.s_pos > self.threshold:
            self.s_pos = 0.0
            triggered = True

        if self.s_neg < -self.threshold:
            self.s_neg = 0.0
            triggered = True

        if triggered and len(self.buffer) >= 2:
            bars = self.buffer
            bar = OHLCV(
                timestamp=bars[-1].timestamp,
                open=bars[0].open,
                high=max(b.high for b in bars),
                low=min(b.low for b in bars),
                close=bars[-1].close,
                volume=sum(b.volume for b in bars),
            )
            self.bars.append(bar)
            self.buffer = []
            return bar

        return None

CUSUM + Método Triple Barrier: En el marco de Lopez de Prado, los eventos CUSUM se usan como puntos de entrada para el método Triple Barrier — donde cada evento dispara una operación con barreras de stop-loss, take-profit y expiración. Para una validación robusta de tales estrategias impulsadas por eventos, vea Walk-Forward Optimization y Monte Carlo Bootstrap for Backtesting.

16. Barras de Entropía

El enfoque más elegante desde el punto de vista teórico: muestrear cuando el contenido de información (entropía de Shannon) de la serie de precios intra-barra supera un umbral.

class EntropyBarGenerator:
    """
    Generates bars when the entropy of intra-bar returns exceeds
    a threshold.

    Based on Shannon's information theory: bars are sampled when
    "new information" arrives, measured as the entropy of the
    return distribution within the current bar.

    This is the most theoretically "pure" information-driven bar.
    """

    def __init__(
        self,
        entropy_threshold: float = 2.0,
        min_trades: int = 50,
        n_bins: int = 10,
    ):
        self.entropy_threshold = entropy_threshold
        self.min_trades = min_trades
        self.n_bins = n_bins
        self.trades: list[tuple[int, float, float]] = []
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float):
        self.trades.append((timestamp, price, qty))

        if len(self.trades) < self.min_trades:
            return None

        entropy = self._compute_entropy()

        if entropy >= self.entropy_threshold:
            return self._close_bar()

        return None

    def _compute_entropy(self) -> float:
        import math

        prices = [t[1] for t in self.trades]
        if len(prices) < 2:
            return 0.0

        returns = [
            math.log(prices[i] / prices[i-1])
            for i in range(1, len(prices))
            if prices[i-1] > 0
        ]

        if not returns:
            return 0.0

        min_r = min(returns)
        max_r = max(returns)

        if max_r == min_r:
            return 0.0

        bin_width = (max_r - min_r) / self.n_bins
        bins = [0] * self.n_bins

        for r in returns:
            idx = min(int((r - min_r) / bin_width), self.n_bins - 1)
            bins[idx] += 1

        total = sum(bins)
        entropy = 0.0
        for count in bins:
            if count > 0:
                p = count / total
                entropy -= p * math.log2(p)

        return entropy

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        self.bars.append(bar)
        self.trades = []
        return bar

Nota práctica: Las barras de entropía son computacionalmente costosas y principalmente de interés investigativo — pero para estrategias basadas en ML, producen características con mejores propiedades estadísticas porque cada barra contiene aproximadamente la misma cantidad de "información".

17. Barras Delta (Order Flow)

Delta acumulado: medir la fuerza neta de compradores y vendedores agresivos en tiempo real.

Las barras delta muestrean en función del delta acumulado — la diferencia continua entre el volumen de compra y el de venta. A diferencia de las barras de imbalance (que usan signos de tick ±1), las barras delta usan order flow real ponderado por volumen.

class DeltaBarGenerator:
    """
    Generates bars based on cumulative order flow delta.

    Delta = Buy Volume - Sell Volume (classified by aggressor side).

    Requires trade-level data with side classification
    (available from Binance aggTrades, Bybit trades, etc.)
    """

    def __init__(self, threshold: float = 500.0):
        self.threshold = threshold
        self.cumulative_delta = 0.0
        self.trades: list[tuple[int, float, float, int]] = []
        self.bars: list[OHLCV] = []

    def on_trade(self, timestamp: int, price: float, qty: float, is_buyer_maker: bool):
        side = -1 if is_buyer_maker else 1
        signed_qty = side * qty

        self.cumulative_delta += signed_qty
        self.trades.append((timestamp, price, qty, side))

        if abs(self.cumulative_delta) >= self.threshold:
            return self._close_bar()

        return None

    def _close_bar(self):
        prices = [t[1] for t in self.trades]
        volumes = [t[2] for t in self.trades]

        bar = OHLCV(
            timestamp=self.trades[-1][0],
            open=prices[0],
            high=max(prices),
            low=min(prices),
            close=prices[-1],
            volume=sum(volumes),
        )
        bar.delta = self.cumulative_delta  # type: ignore
        bar.buy_volume = sum(t[2] for t in self.trades if t[3] == 1)  # type: ignore
        bar.sell_volume = sum(t[2] for t in self.trades if t[3] == -1)  # type: ignore

        self.bars.append(bar)
        self.cumulative_delta = 0.0
        self.trades = []
        return bar

Divergencia de delta: Una de las señales más poderosas — el precio sube mientras el delta acumulado es negativo (los vendedores son agresivos, pero el precio aún sube, lo que indica absorción por compras límite). Directamente relevante para el enfoque de behavioral fingerprinting descrito en el artículo Digital Fingerprint: Trader Identification. Para los market makers que usan el modelo Avellaneda-Stoikov, las barras delta proporcionan una vista en tiempo real del riesgo de inventario y la presión del agresor.

Un búfer circular de barras base: entran datos nuevos, salen datos antiguos, y la vela agregada siempre es válida.

Los métodos de agregación determinan cómo se componen las barras base en velas de marco temporal superior (HTF). Son independientes del tipo de barra — puede aplicar cualquier método de agregación a cualquier tipo base de barra.

Método A: Agregación Alineada al Calendario

Agregar todas las barras base que caen dentro de un límite de calendario fijo. La vela de "1 hora" cubre todas las barras desde las 14:00:00 hasta las 14:59:59.

Propiedades:

• Todos los participantes del mercado ven los mismos límites — esencial para el análisis de la estructura de mercado, soporte/resistencia, triggers de PIQ
• Problema de cold start: vela parcial tras un reinicio
• Natural para barras de tiempo (esto es lo que los exchanges proporcionan de forma nativa)
• También funciona para barras que no son de tiempo: "todas las barras de volumen que cerraron entre las 14:00 y las 15:00" = una vela horaria alineada al calendario a partir de barras de volumen

Método B: Agregación por Ventana Rolling

Agregar las últimas N barras base cerradas, recalculadas en cada nueva barra. Una vela rolling de "1 hora" = las últimas 60 barras de tiempo de 1 minuto cerradas, actualizadas cada minuto.

La unidad atómica es la barra base cerrada. Esta decisión de diseño ofrece:

1. Sin cold start. Después de N barras, la vela es válida. Sin ruido de velas parciales.
2. Paridad de backtest. Si el trading en vivo usa la misma unidad atómica que el motor de backtest, las señales son idénticas.
3. Validación simple. Una regla: if buffer not full: skip.

import numpy as np

class RollingCandleAggregator:
    """
    Produces rolling higher-timeframe candles from closed base bars.

    Works with ANY bar type: time bars, tick bars, volume bars,
    dollar bars, delta bars — anything that produces OHLCV output.

    Example: RollingCandleAggregator(window=60) with 1m time bars
    produces a "1h" candle updated every minute.

    Example: RollingCandleAggregator(window=24) with volume bars
    produces a candle spanning the last 24 volume bars.
    """

    def __init__(self, window: int):
        self.window = window
        self.buffer: deque[OHLCV] = deque(maxlen=window)

    def push(self, bar: OHLCV) -> OHLCV | None:
        """
        Add a closed base bar. Returns aggregated candle
        only when buffer is full (= candle is valid).
        """
        self.buffer.append(bar)

        if len(self.buffer) < self.window:
            return None

        return self._aggregate()

    def _aggregate(self) -> OHLCV:
        bars = list(self.buffer)
        return OHLCV(
            timestamp=bars[-1].timestamp,
            open=bars[0].open,
            high=max(b.high for b in bars),
            low=min(b.low for b in bars),
            close=bars[-1].close,
            volume=sum(b.volume for b in bars),
        )

    @property
    def is_valid(self) -> bool:
        return len(self.buffer) == self.window

Compensación de desfase de fase: Las velas rolling cierran a las :37 si empezó a las :37, no a las :00 como todas las demás. Esto importa para estrategias que dependen de niveles visibles para la multitud. La solución: usar ambas — calendario para la estructura de mercado, rolling para las señales.

Método C: Agregación Rolling Adaptativa

Como rolling, pero el tamaño de la ventana se adapta a la volatilidad actual. Mercados tranquilos → ventana más ancha (más suavizado). Mercados volátiles → ventana más estrecha (reacción más rápida).

class AdaptiveRollingAggregator:
    """
    Rolling window where the window size adapts to volatility.

    Works with any base bar type. Uses ATR of recent bars
    as the volatility measure.

    Low volatility → wider window (more smoothing, fewer signals)
    High volatility → narrower window (faster reaction)
    """

    def __init__(
        self,
        base_window: int = 60,
        min_window: int = 15,
        max_window: int = 240,
        atr_period: int = 14,
        atr_base: float | None = None,
    ):
        self.base_window = base_window
        self.min_window = min_window
        self.max_window = max_window
        self.atr_period = atr_period
        self.atr_base = atr_base

        self.all_candles: deque[OHLCV] = deque(maxlen=max_window)
        self.atr_values: deque[float] = deque(maxlen=atr_period * 2)
        self.current_window = base_window

    def push(self, bar: OHLCV) -> OHLCV | None:
        self.all_candles.append(bar)

        tr = bar.high - bar.low
        self.atr_values.append(tr)

        if len(self.atr_values) < self.atr_period:
            return None

        current_atr = sum(list(self.atr_values)[-self.atr_period:]) / self.atr_period

        if self.atr_base is None and len(self.atr_values) >= self.atr_period * 2:
            self.atr_base = sum(self.atr_values) / len(self.atr_values)

        if self.atr_base is None or self.atr_base == 0:
            return None

        vol_ratio = current_atr / self.atr_base
        self.current_window = int(self.base_window / vol_ratio)
        self.current_window = max(self.min_window, min(self.max_window, self.current_window))

        if len(self.all_candles) < self.current_window:
            return None

        bars = list(self.all_candles)[-self.current_window:]
        return OHLCV(
            timestamp=bars[-1].timestamp,
            open=bars[0].open,
            high=max(b.high for b in bars),
            low=min(b.low for b in bars),
            close=bars[-1].close,
            volume=sum(b.volume for b in bars),
        )

Cada tipo base de barra puede combinarse con cada método de agregación. Algunas combinaciones son estándar (barras de tiempo alineadas al calendario = lo que dan los exchanges), otras son exóticas pero poderosas.

Ejemplos de Combinaciones

Motor Híbrido: Calendario + Rolling

En la práctica, querrá tener simultáneamente agregación de calendario y de rolling. La sobrecarga de memoria es insignificante — dos búferes deque por marco temporal por símbolo.

class HybridCandleEngine:
    """
    Maintains both calendar-aligned and rolling candles
    for any base bar type.

    Calendar candles: for market structure, support/resistance, PIQ.
    Rolling candles: for indicators, signal generation, entries/exits.
    """

    def __init__(self):
        self.rolling = {
            '1h': RollingCandleAggregator(60),
            '4h': RollingCandleAggregator(240),
        }
        self.calendar: dict[str, list[OHLCV]] = {
            '1h': [],
            '4h': [],
        }
        self._calendar_buffer: dict[str, list[OHLCV]] = {
            '1h': [],
            '4h': [],
        }

    def on_bar(self, bar: OHLCV):
        """Process any base bar type — time, volume, tick, delta, etc."""
        rolling_results = {}
        for tf, agg in self.rolling.items():
            rolling_results[tf] = agg.push(bar)

        self._update_calendar(bar)

        return rolling_results

    def _update_calendar(self, bar: OHLCV):
        from datetime import datetime
        ts = datetime.utcfromtimestamp(bar.timestamp)

        for tf, minutes in [('1h', 60), ('4h', 240)]:
            self._calendar_buffer[tf].append(bar)

            total_minutes = ts.hour * 60 + ts.minute
            if (total_minutes + 1) % minutes == 0:
                bars = self._calendar_buffer[tf]
                if bars:
                    agg = OHLCV(
                        timestamp=bars[-1].timestamp,
                        open=bars[0].open,
                        high=max(b.high for b in bars),
                        low=min(b.low for b in bars),
                        close=bars[-1].close,
                        volume=sum(b.volume for b in bars),
                    )
                    self.calendar[tf].append(agg)
                    self._calendar_buffer[tf] = []

Híbrido Tiempo-Volumen: Calendario con Divisiones por Volumen

Una variante especial de agregación: velas alineadas al calendario que se cierran anticipadamente cuando el volumen supera un umbral. Mantiene la sincronización temporal mientras se adapta a los picos de actividad.

class TimeVolumeHybridGenerator:
    """
    Calendar-aligned candles that split when volume spikes.

    Rule: close the candle at the calendar boundary OR when
    accumulated volume exceeds vol_threshold, whichever comes first.

    Works with any base bar type — the volume trigger adds an
    extra split dimension on top of calendar alignment.
    """

    def __init__(
        self,
        interval_minutes: int = 60,
        vol_threshold: float = 5000.0,
    ):
        self.interval_minutes = interval_minutes
        self.vol_threshold = vol_threshold

        self.buffer: list[OHLCV] = []
        self.accumulated_volume = 0.0
        self.bars: list[OHLCV] = []

    def on_bar(self, bar: OHLCV) -> OHLCV | None:
        self.buffer.append(bar)
        self.accumulated_volume += bar.volume

        from datetime import datetime
        ts = datetime.utcfromtimestamp(bar.timestamp)
        total_minutes = ts.hour * 60 + ts.minute
        at_boundary = (total_minutes + 1) % self.interval_minutes == 0

        vol_spike = self.accumulated_volume >= self.vol_threshold

        if at_boundary or vol_spike:
            return self._close_bar(split_reason='volume' if vol_spike else 'time')

        return None

    def _close_bar(self, split_reason: str) -> OHLCV:
        bars = self.buffer
        bar = OHLCV(
            timestamp=bars[-1].timestamp,
            open=bars[0].open,
            high=max(b.high for b in bars),
            low=min(b.low for b in bars),
            close=bars[-1].close,
            volume=sum(b.volume for b in bars),
        )
        bar.split_reason = split_reason  # type: ignore
        bar.num_bars = len(bars)  # type: ignore

        self.bars.append(bar)
        self.buffer = []
        self.accumulated_volume = 0.0
        return bar

Agregación Práctica: Precarga en Cascada

Precarga en cascada: componer velas diarias a partir de las horarias, y las horarias a partir de las de minuto — sorteando los límites de la API.

Los exchanges limitan la cantidad de datos históricos que sirven. Binance da ~1000 velas por solicitud REST, OKX limita a 300. Si necesita una vela rolling 1D (1440 minutos), no siempre puede obtener suficiente historial de 1m. Para el streaming en tiempo real de operaciones y libros de órdenes vía WebSocket, vea CCXT Pro WebSocket Methods.

La solución: agregación en cascada — construir marcos temporales superiores a partir de la resolución más alta disponible en cada nivel, y luego unirlos.

Rolling 1W candle:
├── 6 completed 1D candles ← fetch from REST /klines?interval=1d
├── 1 partial day:
│   ├── 23 completed 1h candles ← fetch from REST /klines?interval=1h
│   └── 1 partial hour:
│       └── N completed 1m candles ← fetch from REST /klines?interval=1m
└── Live: each new closed 1m candle updates the entire chain

Esto funciona porque la agregación OHLCV es componible: el máximo de una vela 1D es el máximo de 24 máximos de 1h, que a su vez es el máximo de 1440 máximos de 1m.

Límites Multi-Exchange

Verificación de Consistencia de la Agregación

La vela de 1h de una API REST podría no coincidir con lo que calcularía a partir de 60 velas de 1m. Siempre valide:

def validate_aggregation(
    candle_htf: OHLCV,
    candles_ltf: list[OHLCV],
    tolerance_pct: float = 0.001,
) -> dict[str, bool]:
    agg = OHLCV(
        timestamp=candles_ltf[-1].timestamp,
        open=candles_ltf[0].open,
        high=max(c.high for c in candles_ltf),
        low=min(c.low for c in candles_ltf),
        close=candles_ltf[-1].close,
        volume=sum(c.volume for c in candles_ltf),
    )

    def close_enough(a: float, b: float) -> bool:
        if a == 0 and b == 0:
            return True
        return abs(a - b) / max(abs(a), abs(b)) < tolerance_pct

    return {
        'open': close_enough(candle_htf.open, agg.open),
        'high': close_enough(candle_htf.high, agg.high),
        'low': close_enough(candle_htf.low, agg.low),
        'close': close_enough(candle_htf.close, agg.close),
        'volume': close_enough(candle_htf.volume, agg.volume),
    }

Si la validación falla de forma consistente, siempre agregue usted mismo a partir de 1m — nunca confíe en la vela HTF del exchange para la paridad de backtest.

Matriz Comparativa

Eje 1: Tipos Base de Barras

Eje 2: Métodos de Agregación

Recomendaciones Prácticas

Arquitectura de velas de cuatro capas: señales rolling, estructura de calendario, flujo de microestructura y filtros de tendencia.

Si su motor de backtest funciona con datos OHLCV de 1m:

1. Barras de tiempo rolling — la actualización más simple. Sin datos adicionales. Elimina el cold start.
2. Barras de tiempo híbridas (rolling + calendario) — calendario para la estructura de mercado, rolling para las señales.
3. Filtro CUSUM — funciona con cierres de 1m, sin datos de tick. "Algo se movió lo suficiente como para ser interesante."

Si tiene datos de tick/operaciones:

4. Barras de dólar + rolling — el estándar recomendado por la literatura de finanzas cuantitativas.
5. Barras de volume imbalance + rolling — detecta flujo informado, muestrea con más frecuencia durante eventos significativos.
6. Barras delta + calendario — si tiene clasificación del lado agresor, la vista más directa de quién está empujando el mercado.

Como filtros (aplicar Heikin-Ashi o Line Break sobre cualquier combinación de base+agregación):

7. Heikin-Ashi sobre barras de volumen rolling — señales de tendencia limpias sobre datos normalizados por actividad.
8. Line Break/Kagi sobre barras de calendario diarias — filtro de tendencia macro.

Para Marketmaker.cc específicamente — un enfoque por capas:

• Capa 1 (señales): Agregación rolling de barras de tiempo para indicadores y señales de entrada/salida. Sin cold start, paridad de backtest perfecta.
• Capa 2 (estructura de mercado): Barras de tiempo alineadas al calendario para soporte/resistencia, análisis de cierres horarios y triggers de PIQ.
• Capa 3 (microestructura): Barras de volume imbalance + barras delta a partir del flujo de operaciones en bruto para detectar flujo informado, división de órdenes y anticipar grandes movimientos. Vea también Digital Fingerprint: Trader Identification para el reconocimiento de patrones de comportamiento en datos de order flow.
• Capa 4 (filtro de tendencia): Transformación Heikin-Ashi sobre barras rolling, o Line Break sobre cierres de calendario de 4h, para mantener las señales alineadas con la dirección macro.

Conclusión

La construcción de velas no es una única elección — son dos decisiones independientes:

1. ¿Qué tipo de barra? El tiempo captura intervalos de reloj. La actividad (tick, volumen, dólar) captura la participación del mercado. El precio (Renko, rango, volatilidad) captura movimientos. La información (imbalance, runs, CUSUM, entropía) captura la llegada de nueva información. El order flow (delta) captura la presión agresiva.

2. ¿Cómo agregar en marcos temporales superiores? El calendario se alinea con la multitud. Rolling elimina el cold start. Adaptativo reacciona a la volatilidad.

La "vela de 1 hora de Binance" estándar es solo una celda en una matriz de 17×3. Las otras 50 combinaciones están disponibles para cualquiera dispuesto a implementarlas. Para un sistema en producción, la respuesta es "elegir la combinación correcta para cada capa de su motor de decisión".

La unidad atómica — la barra base cerrada — sigue siendo el fundamento. Todo lo demás es agregación.

Para más sobre la precisión del backtest con datos de granularidad fina, vea Adaptive Drill-Down: Backtest with Variable Granularity. Para el impacto del precálculo de indicadores en estrategias multi-marco-temporal, vea Aggregated Parquet Cache.

Enlaces Útiles

1. Lopez de Prado — Advances in Financial Machine Learning (2018)
2. Easley, Lopez de Prado, O'Hara — The Volume Clock: Insights into the High Frequency Paradigm (2012)
3. mlfinlab — biblioteca Python que implementa barras impulsadas por información
4. Binance — Datos Históricos de Mercado
5. Apache Parquet — formato de almacenamiento columnar

Cita

@article{soloviov2026bartypes,
  author = {Soloviov, Eugen},
  title = {Bar Types and Aggregation Methods for Algorithmic Trading},
  year = {2026},
  url = {https://marketmaker.cc/en/blog/post/beyond-time-bars-candle-construction},
  description = {Two-axis classification of candle construction: 17 base bar types × 3 aggregation methods = 51 combinations, with implementation code and practical recommendations for crypto algotrading.}
}