Types de Barres et Méthodes d'Agrégation pour le Trading Algorithmique

Chaque graphique en chandeliers que vous avez déjà vu sur Binance, TradingView ou l'interface de n'importe quelle plateforme est construit de la même manière : agréger les transactions dans une fenêtre temporelle fixe — 1 minute, 5 minutes, 1 heure — et produire une barre OHLCV. C'est si répandu que la plupart des traders ne le remettent jamais en question. Mais pour le trading algorithmique, le choix du type de barre et la méthode d'agrégation sont deux décisions indépendantes — et la plupart des systèmes les confondent.

Cet article sépare les deux axes de la construction des bougies : quel type de barre vous construisez (17 types) et comment vous les agrégez en unités de temps supérieures (3 méthodes). La combinaison donne 51 configurations possibles, chacune avec des propriétés différentes pour le backtesting, le trading en direct et la génération de signaux.

Pour une introduction sur la façon dont les transactions brutes deviennent des bougies standard, voir Trading Candles Demystified.

TL;DR

• La construction des bougies possède deux axes indépendants : type de barre et méthode d'agrégation
• 17 types de barres de base : temps, tick, volume, dollar, Renko, range, volatilité, Heikin-Ashi, Kagi, Line Break, P&F, tick imbalance (TIB), volume imbalance (VIB), run, CUSUM, entropie, delta
• 3 méthodes d'agrégation : alignée sur le calendrier, fenêtre glissante, glissante adaptative
• 17 × 3 = 51 combinaisons possibles, chacune avec des propriétés différentes
• La plupart des systèmes n'utilisent qu'une seule combinaison : les barres de temps alignées sur le calendrier. Les 50 autres restent inexploitées.
• Recommandation pratique : utiliser plusieurs combinaisons en couches — barres de temps glissantes pour les signaux, barres de temps calendaires pour la structure de marché, barres pilotées par l'information pour la microstructure

Deux Axes de la Construction des Bougies

La vision traditionnelle place tous les types de barres dans une liste plate : barres de temps, barres de tick, barres de volume, Renko, etc. C'est trompeur. Il existe en réalité deux choix orthogonaux :

Axe 1 — Type de Barre de Base (17 types) : Comment décidez-vous du moment où une nouvelle barre se clôture ? Après un intervalle de temps fixe ? Après N transactions ? Après un mouvement de prix ? Lorsque le contenu informationnel change ? Cela détermine ce que signifie « une barre ».

Axe 2 — Méthode d'Agrégation (3 méthodes) : Comment composez-vous les barres de base en bougies d'unité de temps supérieure ? Alignement sur les limites du calendrier (00:00, 01:00, …) ? Utilisation d'une fenêtre glissante des N dernières barres ? Adaptation de la taille de la fenêtre à la volatilité ?

Ces deux axes sont indépendants. Vous pouvez avoir :

• Barres de tick alignées sur le calendrier — agréger les barres de tick clôturées entre 14h00 et 14h59 en une seule bougie horaire
• Barres de volume glissantes — prendre les 24 dernières barres de volume, quel que soit le moment de leur clôture
• Barres delta adaptatives — utiliser une fenêtre pilotée par la volatilité sur des barres delta

La « bougie 1 heure » standard n'est qu'un point dans cette matrice 17×3 : barres de temps + alignement calendaire. Toute autre combinaison est une alternative qui mérite d'être envisagée.

1. Barres de Temps (Standard)

Densité d'information inégale : les limites temporelles rigides traitent une heure calme de 200 transactions de la même manière qu'une heure d'annonce de 50 000 transactions.

Le mode par défaut. Une nouvelle barre se forme après un intervalle de temps fixe : 1 minute, 5 minutes, 1 heure. Chaque plateforme les fournit nativement.

Propriétés :

• Pendant la session asiatique (00:00–08:00 UTC), une bougie de 1 heure pourrait contenir 200 transactions. Lors d'une annonce de listing sur Binance, cette même fenêtre pourrait en contenir 50 000. Les barres de temps traitent les deux cas comme équivalents. Détecter de tels pics d'activité est essentiel pour la protection contre les bots — voir Anomaly Detection for Trading Bots.
• Tous les participants du marché voient les mêmes limites de bougie — un point de Schelling. Cela rend les barres de temps essentielles pour analyser le comportement de foule.
• Les indicateurs calculés sur des bougies partielles (après un redémarrage) produisent des valeurs inutilisables.

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. Barres Basées sur l'Activité

Barres de tick, de volume et de dollar : trois façons de laisser la participation du marché — et non l'horloge — déterminer les limites des barres.

Plutôt que d'échantillonner à intervalles de temps fixes, on échantillonne après une quantité fixe d'activité de marché. Cela produit des barres avec un « contenu informationnel » à peu près égal, quel que soit le moment de la journée.

2. Barres de Tick

Une nouvelle barre se forme après chaque N transactions (ticks). En période de forte activité, les barres se forment rapidement. En période calme, une seule barre peut s'étendre sur plusieurs heures.

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

Avantages : S'adapte naturellement à l'activité du marché. Les rendements issus des barres de tick tendent à se rapprocher davantage d'une distribution normale que les rendements des barres de temps — une propriété qui améliore la performance de nombreux modèles statistiques.

Inconvénients : Nécessite un flux de transactions brutes (non disponible chez tous les fournisseurs de données historiques). Le timing des barres est imprévisible — impossible de dire « la prochaine barre se clôturera à X ».

3. Barres de Volume

Une nouvelle barre se forme après que N contrats (ou pièces, en crypto) ont été échangés. Similaire aux barres de tick, mais pondérée par la taille de la transaction — une seule transaction de 100 BTC contribue 100 fois plus qu'une transaction 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. Barres de Dollar

Une nouvelle barre se forme après qu'une valeur nominale fixe (en USD/USDT) a été échangée. La plus robuste des barres basées sur l'activité car elle normalise à la fois le nombre de transactions et le niveau de prix.

Considérez ceci : si l'ETH passe de 1 000 $à 4 000$, vendre 10 000 $d'ETH nécessite 2,5 ETH à 4 000$ mais 10 ETH à 1 000 $. Les barres de volume traiteraient ces deux cas différemment ; les barres de dollar les traitent de la même façon.

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

Choisir le Seuil

Le seuil des barres basées sur l'activité devrait produire à peu près le même nombre de barres par jour que les barres de temps qu'il remplace. Pour BTCUSDT sur Binance :

Ces chiffres sont approximatifs et varient fortement selon le régime de marché. Lors d'un rally ou d'un crash, les barres basées sur l'activité produiront 5 à 10 fois plus de barres qu'à l'accoutumée — c'est exactement le but recherché.

5–7. Barres Basées sur le Prix

Briques Renko, barres de range et barres de volatilité : n'échantillonner que lorsque le prix bouge suffisamment pour être significatif.

Les barres basées sur le prix ignorent à la fois le temps et l'activité. Une nouvelle barre ne se forme que lorsque le prix évolue d'un montant spécifié. Cela filtre naturellement le bruit latéral et met en évidence les tendances.

5. Barres Renko

Une nouvelle « brique » Renko se forme lorsque le prix de clôture évolue d'au moins N unités par rapport à la clôture de la brique précédente. Les briques ont toujours la même taille, ce qui crée une représentation visuelle nette de la direction de la tendance.

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

Le Renko dynamique utilise l'ATR (Average True Range) au lieu d'une taille de brique fixe, s'adaptant automatiquement à la volatilité.

6. Barres de Range

Chaque barre possède une plage haut-bas fixe. Lorsque la plage est dépassée, la barre se clôture et une nouvelle commence. Contrairement au Renko, les barres de range incluent des mèches et peuvent montrer la volatilité intra-barre.

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

Différence clé entre les barres Renko et Range : Renko ne suit que les prix de clôture et montre la direction ; les barres de range suivent toute la plage de prix et montrent la structure à l'intérieur de la barre. Les barres de range sont généralement plus utiles pour le trading algorithmique car elles préservent l'information haut-bas nécessaire à la simulation du stop-loss et du take-profit.

7. Barres de Volatilité

Une nouvelle barre se forme lorsque la volatilité intra-barre atteint un seuil dynamique — par exemple, un multiple de l'ATR récent. Contrairement aux barres de range (seuil fixe), les barres de volatilité s'adaptent aux conditions de marché.

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 (Transformation Lissée)

Heikin-Ashi : le moyennage transforme des bougies bruitées en signaux de tendance lisses — au prix d'une information de prix exacte.

Heikin-Ashi (japonais pour « barre moyenne ») n'est pas un type de barre — c'est une transformation qui peut être appliquée par-dessus n'importe quel type de barre de base. Elle lisse les bougies en moyennant les valeurs de la barre actuelle et précédente :

• HA Close = (Open + High + Low + Close) / 4
• HA Open = (HA Open précédent + HA Close précédent) / 2
• HA High = max(High, HA Open, HA Close)
• HA Low = min(Low, HA Open, HA Close)

Les tendances apparaissent comme des séquences de bougies de même couleur sans mèches inférieures (tendance haussière) ou sans mèches supérieures (tendance baissière).

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

Mise en garde essentielle pour le backtesting : Les prix Heikin-Ashi sont synthétiques. Si votre backtest utilise la clôture HA comme prix d'entrée, les résultats seront faux. Utilisez toujours HA uniquement pour la génération de signaux et exécutez aux prix OHLC réels.

Quand HA est utile : Les stratégies de suivi de tendance qui ont besoin de signaux clairs de « rester dans la position ». Appliquez HA par-dessus n'importe quel type de barre de base — barres de temps, barres de volume, barres de dollar — pour filtrer les faux croisements.

Quand HA est nuisible : Toute stratégie nécessitant des niveaux de prix précis — support/résistance, analyse du carnet d'ordres, PIQ (Position In Queue). Le moyennage détruit l'information de prix exacte.

9–11. Graphiques de Renversement Japonais

Kagi, Line Break et Point & Figure : des méthodes de charting sans temps qui se concentrent purement sur la structure du prix.

Ce sont des méthodes traditionnelles de charting japonais (aux côtés du Renko) qui écartent complètement le temps et se concentrent sur la structure du prix.

9. Graphiques Kagi

Les graphiques Kagi sont constitués de lignes verticales qui changent de direction lorsque le prix se retourne d'un montant spécifié. Les lignes changent d'épaisseur lorsque le prix franchit un plus haut précédent (épaisse = « yang » = demande) ou un plus bas précédent (fine = « yin » = offre).

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. Graphiques Line Break

Les graphiques Line Break dessinent une nouvelle ligne (boîte) uniquement lorsque le prix de clôture dépasse le plus haut ou le plus bas des N lignes précédentes (typiquement 3). Aucune nouvelle ligne n'est dessinée si le prix reste dans la plage.

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. Graphiques Point & Figure

Les graphiques Point & Figure (P&F) utilisent des colonnes de X (prix en hausse) et de O (prix en baisse). Un changement de colonne nécessite un renversement typiquement de 3 tailles de boîte. C'est l'une des plus anciennes méthodes de filtrage du bruit et d'identification du support/résistance.

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 et P&F dans le trading algorithmique : Principalement utilisés pour la détection de tendance à long terme et l'identification du support/résistance. En tant que couche de filtre — « ne pas prendre de signaux longs quand le graphique Kagi est en mode yin » — ils apportent de la valeur en alignant les transactions sur la structure macro.

12–14. Barres Pilotées par l'Information

Barres d'imbalance, barres de run, filtres CUSUM et barres d'entropie : échantillonner lorsque le marché nous indique que quelque chose a changé.

L'approche la plus sophistiquée, tirée de Advances in Financial Machine Learning (2018) de Marcos Lopez de Prado. L'idée centrale : échantillonner lorsque de nouvelles informations arrivent sur le marché, et non à intervalles fixes.

12. Tick Imbalance Bars (TIB)

Si le marché est en équilibre, les transactions initiées par les acheteurs et celles initiées par les vendeurs devraient à peu près s'équilibrer. Lorsque le déséquilibre dépasse notre attente, quelque chose a changé. Une barre est échantillonnée à ce moment-là.

Chaque transaction est classée comme initiée par l'acheteur (+1) ou par le vendeur (-1) à l'aide de la règle du tick. Nous suivons le déséquilibre cumulé θ et échantillonnons lorsque |θ| dépasse un seuil dynamique.

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)

Extension des TIB : au lieu de compter chaque transaction comme ±1, on pondère par le volume signé. Un achat de 100 BTC contribue à +100, une vente de 1 BTC à -1. Capture les grands ordres informés qui pourraient être fractionnés en de nombreuses petites transactions.

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

Le Problème de l'Explosion

Un problème connu des barres d'imbalance : le seuil basé sur l'EWMA peut entrer dans une boucle de rétroaction positive. La solution : limiter avec des bornes min_ticks et 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. Barres de Run

Les barres de run suivent la longueur de la série directionnelle en cours — la plus longue séquence consécutive d'achats ou de ventes. Lorsqu'un grand trader informé fractionne un ordre en de nombreuses petites transactions, la séquence devient anormalement longue. Les barres de run détectent cela.

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

Les barres de run peuvent être étendues aux runs de volume et aux runs de dollar.

15. Barres à Filtre CUSUM

Le filtre CUSUM (Cumulative Sum) détermine quand échantillonner en suivant les rendements cumulés. Contrairement aux barres d'imbalance (qui fonctionnent sur des transactions brutes), CUSUM peut être appliqué à des données OHLCV de 1m existantes — aucune donnée de tick n'est requise.

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éthode Triple Barrière : Dans le cadre de Lopez de Prado, les événements CUSUM sont utilisés comme points d'entrée pour la méthode Triple Barrière — où chaque événement déclenche une transaction avec des barrières de stop-loss, de take-profit et d'expiration. Pour une validation robuste de ces stratégies pilotées par événements, voir Walk-Forward Optimization et Monte Carlo Bootstrap for Backtesting.

16. Barres d'Entropie

L'approche la plus élégante sur le plan théorique : échantillonner lorsque le contenu informationnel (entropie de Shannon) de la série de prix intra-barre dépasse un seuil.

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

Note pratique : Les barres d'entropie sont coûteuses en calcul et principalement d'intérêt pour la recherche — mais pour les stratégies basées sur le ML, elles produisent des caractéristiques avec de meilleures propriétés statistiques car chaque barre contient approximativement la même quantité d'« information ».

17. Barres Delta (Order Flow)

Delta cumulé : mesurer la force nette des acheteurs et vendeurs agressifs en temps réel.

Les barres delta échantillonnent en fonction du delta cumulé — la différence courante entre le volume acheteur et le volume vendeur. Contrairement aux barres d'imbalance (qui utilisent des signes de tick ±1), les barres delta utilisent un order flow réel pondéré par le volume.

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

Divergence de delta : L'un des signaux les plus puissants — le prix monte alors que le delta cumulé est négatif (les vendeurs sont agressifs, mais le prix monte quand même, ce qui indique une absorption par des achats à cours limité). Directement pertinent pour l'approche de behavioral fingerprinting décrite dans l'article Digital Fingerprint: Trader Identification. Pour les market makers utilisant le modèle Avellaneda-Stoikov, les barres delta offrent une vue en temps réel du risque d'inventaire et de la pression de l'agresseur.

Un tampon circulaire de barres de base : les nouvelles données entrent, les anciennes sortent, et la bougie agrégée est toujours valide.

Les méthodes d'agrégation déterminent comment les barres de base sont composées en bougies d'unité de temps supérieure (HTF). Elles sont indépendantes du type de barre — vous pouvez appliquer n'importe quelle méthode d'agrégation à n'importe quel type de barre de base.

Méthode A : Agrégation Alignée sur le Calendrier

Agréger toutes les barres de base qui tombent dans une limite calendaire fixe. La bougie « 1 heure » couvre toutes les barres de 14:00:00 à 14:59:59.

Propriétés :

• Tous les participants du marché voient les mêmes limites — essentiel pour l'analyse de la structure de marché, le support/résistance, les déclencheurs PIQ
• Problème de cold start : bougie partielle après un redémarrage
• Naturel pour les barres de temps (c'est ce que les plateformes fournissent nativement)
• Fonctionne aussi pour les barres autres que temporelles : « toutes les barres de volume clôturées entre 14h00 et 15h00 » = une bougie horaire alignée sur le calendrier à partir de barres de volume

Méthode B : Agrégation par Fenêtre Glissante

Agréger les N dernières barres de base clôturées, recalculées à chaque nouvelle barre. Une bougie glissante « 1 heure » = les 60 dernières barres de temps de 1 minute clôturées, mise à jour chaque minute.

L'unité atomique est la barre de base clôturée. Ce choix de conception apporte :

1. Aucun cold start. Après N barres, la bougie est valide. Aucun bruit dû aux bougies partielles.
2. Parité de backtest. Si le trading en direct utilise la même unité atomique que le moteur de backtest, les signaux sont identiques.
3. Validation simple. Une seule règle : 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

Compromis du décalage de phase : Les bougies glissantes se clôturent à :37 si vous avez démarré à :37, et non à :00 comme tout le monde. Cela compte pour les stratégies qui dépendent de niveaux visibles par la foule. La solution : utiliser les deux — calendrier pour la structure de marché, glissant pour les signaux.

Méthode C : Agrégation Glissante Adaptative

Comme le glissant, mais la taille de la fenêtre s'adapte à la volatilité actuelle. Marchés calmes → fenêtre plus large (plus de lissage). Marchés volatils → fenêtre plus étroite (réaction plus rapide).

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),
        )

Chaque type de barre de base peut être combiné avec chaque méthode d'agrégation. Certaines combinaisons sont standard (barres de temps alignées sur le calendrier = ce que fournissent les plateformes), d'autres sont exotiques mais puissantes.

Exemples de Combinaisons

Moteur Hybride : Calendrier + Glissant

En pratique, vous voudrez à la fois une agrégation calendaire et une agrégation glissante simultanément. La surcharge mémoire est négligeable — deux tampons deque par unité de temps par symbole.

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] = []

Hybride Temps-Volume : Calendrier avec Divisions de Volume

Une variante d'agrégation spéciale : des bougies alignées sur le calendrier qui se clôturent prématurément lorsque le volume dépasse un seuil. Maintient la synchronisation temporelle tout en s'adaptant aux pics d'activité.

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

Agrégation Pratique : Préchargement en Cascade

Préchargement en cascade : composer des bougies journalières à partir des horaires, et les horaires à partir des minutes — en contournant les limites d'API.

Les plateformes limitent la quantité de données historiques qu'elles fournissent. Binance donne ~1000 bougies par requête REST, OKX plafonne à 300. Si vous avez besoin d'une bougie glissante 1J (1440 minutes), vous ne pouvez pas toujours obtenir suffisamment d'historique 1m. Pour le streaming en temps réel des transactions et des carnets d'ordres via WebSocket, voir CCXT Pro WebSocket Methods.

La solution : l'agrégation en cascade — construire des unités de temps supérieures à partir de la résolution la plus élevée disponible à chaque niveau, puis les assembler.

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

Cela fonctionne parce que l'agrégation OHLCV est composable : le plus haut d'une bougie 1J est le maximum de 24 plus hauts 1h, qui est lui-même le maximum de 1440 plus hauts 1m.

Limites Multi-Plateformes

Vérification de Cohérence de l'Agrégation

La bougie 1h issue d'une API REST pourrait ne pas correspondre à ce que vous calculeriez à partir de 60 bougies 1m. Validez toujours :

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 validation échoue de façon récurrente, toujours agréger vous-même à partir de 1m — ne jamais faire confiance à la bougie HTF de la plateforme pour la parité backtest-live.

Matrice Comparative

Axe 1 : Types de Barres de Base

Axe 2 : Méthodes d'Agrégation

Recommandations Pratiques

Architecture de bougies à quatre couches : signaux glissants, structure calendaire, flux de microstructure et filtres de tendance.

Si votre moteur de backtest fonctionne sur des données OHLCV 1m :

1. Barres de temps glissantes — la mise à niveau la plus simple. Aucune donnée supplémentaire. Élimine le cold start.
2. Barres de temps hybrides (glissant + calendrier) — calendrier pour la structure de marché, glissant pour les signaux.
3. Filtre CUSUM — fonctionne sur des clôtures 1m, sans données de tick. « Quelque chose a assez bougé pour être intéressant. »

Si vous disposez de données de tick/transaction :

4. Barres de dollar + glissant — le standard recommandé par la littérature de finance quantitative.
5. Barres de volume imbalance + glissant — détecte le flux informé, échantillonne plus souvent lors d'événements significatifs.
6. Barres delta + calendrier — si vous disposez d'une classification du côté agresseur, la vue la plus directe de qui pousse le marché.

En tant que filtres (appliquer Heikin-Ashi ou Line Break sur toute combinaison base+agrégation) :

7. Heikin-Ashi sur des barres de volume glissantes — signaux de tendance propres sur des données normalisées par l'activité.
8. Line Break/Kagi sur des barres calendaires journalières — filtre de tendance macro.

Pour Marketmaker.cc en particulier — une approche en couches :

• Couche 1 (signaux) : Agrégation glissante de barres de temps pour les indicateurs et les signaux d'entrée/sortie. Pas de cold start, parité de backtest parfaite.
• Couche 2 (structure de marché) : Barres de temps alignées sur le calendrier pour le support/résistance, l'analyse des clôtures horaires et les déclencheurs PIQ.
• Couche 3 (microstructure) : Barres de volume imbalance + barres delta à partir du flux de transactions brutes pour détecter le flux informé, le fractionnement d'ordres et anticiper les grands mouvements. Voir aussi Digital Fingerprint: Trader Identification pour la reconnaissance de motifs comportementaux dans les données d'order flow.
• Couche 4 (filtre de tendance) : Transformation Heikin-Ashi sur des barres glissantes, ou Line Break sur des clôtures calendaires 4h, pour garder les signaux alignés sur la direction macro.

Conclusion

La construction des bougies n'est pas un choix unique — ce sont deux décisions indépendantes :

1. Quel type de barre ? Le temps capture les intervalles d'horloge. L'activité (tick, volume, dollar) capture la participation du marché. Le prix (Renko, range, volatilité) capture les mouvements. L'information (imbalance, runs, CUSUM, entropie) capture l'arrivée de nouvelles informations. L'order flow (delta) capture la pression agressive.

2. Comment agréger en unités de temps supérieures ? Le calendrier s'aligne sur la foule. Le glissant élimine le cold start. L'adaptatif réagit à la volatilité.

La « bougie 1 heure de Binance » standard n'est qu'une cellule dans une matrice 17×3. Les 50 autres combinaisons sont à la disposition de quiconque est prêt à les implémenter. Pour un système en production, la réponse est « choisir la bonne combinaison pour chaque couche de votre moteur de décision ».

L'unité atomique — la barre de base clôturée — reste le fondement. Tout le reste est agrégation.

Pour en savoir plus sur la précision du backtest avec des données à granularité fine, voir Adaptive Drill-Down: Backtest with Variable Granularity. Sur l'impact du précalcul des indicateurs pour les stratégies multi-unités-de-temps, voir Aggregated Parquet Cache.

Liens Utiles

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 — bibliothèque Python implémentant des barres pilotées par l'information
4. Binance — Données de Marché Historiques
5. Apache Parquet — format de stockage colonnaire

Citation

@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.}
}