Bartypes en Aggregatiemethoden voor Algoritmische Handel

Elke candlestick-grafiek die u ooit op Binance, TradingView of een andere exchange-UI hebt gezien, wordt op dezelfde manier opgebouwd: trades aggregeren binnen een vast tijdvenster — 1 minuut, 5 minuten, 1 uur — en daaruit een OHLCV-bar produceren. Dit is zo alomtegenwoordig dat de meeste traders het nooit ter discussie stellen. Maar voor algoritmische handel zijn de keuze van het bartype en de aggregatiemethode twee onafhankelijke beslissingen — en de meeste systemen vermengen ze.

Dit artikel scheidt de twee assen van candle-constructie: welk type bar u bouwt (17 types) en hoe u ze aggregeert naar hogere timeframes (3 methoden). De combinatie levert 51 mogelijke configuraties op, elk met andere eigenschappen voor backtesting, live trading en signaalgeneratie.

Voor een introductie over hoe ruwe trades standaard candles worden, zie Trading Candles Demystified.

TL;DR

• Candle-constructie kent twee onafhankelijke assen: bartype en aggregatiemethode
• 17 basis-bartypes: tijd, tick, volume, dollar, Renko, range, volatiliteit, Heikin-Ashi, Kagi, Line Break, P&F, tick imbalance (TIB), volume imbalance (VIB), run, CUSUM, entropie, delta
• 3 aggregatiemethoden: kalendergericht, rolling window, adaptief rolling
• 17 × 3 = 51 mogelijke combinaties, elk met andere eigenschappen
• De meeste systemen gebruiken slechts één combinatie: kalendergerichte tijd-bars. De overige 50 blijven onbenut.
• Praktisch advies: gebruik meerdere combinaties in lagen — rolling tijd-bars voor signalen, kalender-tijd-bars voor marktstructuur, informatiegedreven bars voor microstructuur

Twee Assen van Candle-Constructie

De traditionele visie zet alle bartypes op een platte lijst: tijd-bars, tick-bars, volume-bars, Renko, enzovoort. Dit is misleidend. Er zijn eigenlijk twee orthogonale keuzes:

As 1 — Basis-Bartype (17 types): Hoe bepaalt u wanneer een nieuwe bar sluit? Na een vast tijdsinterval? Na N trades? Na een prijsbeweging? Wanneer de informatie-inhoud verandert? Dit bepaalt wat "één bar" betekent.

As 2 — Aggregatiemethode (3 methoden): Hoe stelt u basis-bars samen tot candles op een hoger timeframe? Uitlijnen op kalendergrenzen (00:00, 01:00, ...)? Een rolling window van de laatste N bars gebruiken? De venstergrootte aanpassen aan de volatiliteit?

Deze twee assen zijn onafhankelijk. U kunt bijvoorbeeld hebben:

• Kalendergerichte tick-bars — tick-bars aggregeren die tussen 14:00 en 14:59 sloten tot één enkele uur-candle
• Rolling volume-bars — de laatste 24 volume-bars nemen, ongeacht wanneer ze sloten
• Adaptieve delta-bars — een volatiliteitsgedreven venster over delta-bars gebruiken

De standaard "1-uur-candle" is slechts één punt in deze 17×3-matrix: tijd-bars + kalenderuitlijning. Elke andere combinatie is een alternatief dat het overwegen waard is.

1. Tijd-Bars (Standaard)

Ongelijke informatiedichtheid: starre tijdgrenzen behandelen een rustig uur met 200 trades hetzelfde als een aankondigingsuur met 50.000 trades.

De standaard. Een nieuwe bar vormt zich na een vast tijdsinterval: 1 minuut, 5 minuten, 1 uur. Elke exchange levert deze standaard.

Eigenschappen:

• Tijdens de Aziatische sessie (00:00–08:00 UTC) bevat een 1-uur-candle mogelijk 200 trades. Tijdens een Binance-listingaankondiging kan datzelfde venster 50.000 trades bevatten. Tijd-bars behandelen beide als gelijkwaardig. Het detecteren van dergelijke activiteitspieken is cruciaal voor botbescherming — zie Anomaly Detection for Trading Bots.
• Alle marktdeelnemers zien dezelfde candle-grenzen — een Schelling-punt. Dit maakt tijd-bars essentieel voor het analyseren van kuddegedrag.
• Indicatoren berekend op gedeeltelijke candles (na een herstart) leveren onbruikbare waarden op.

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. Activiteitsgebaseerde Bars

Tick-, volume- en dollar-bars: drie manieren om marktdeelname — niet de klok — de bar-grenzen te laten bepalen.

In plaats van te samplen op vaste tijdsintervallen, wordt gesampled na een vaste hoeveelheid marktactiviteit. Dit levert bars op met ongeveer gelijke "informatie-inhoud", ongeacht het tijdstip van de dag.

2. Tick-Bars

Een nieuwe bar vormt zich na elke N trades (ticks). Bij hoge activiteit vormen bars zich snel. In rustige periodes kan een enkele bar zich over uren uitstrekken.

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

Voordelen: Past zich van nature aan de marktactiviteit aan. Rendementen uit tick-bars liggen doorgaans dichter bij een normale verdeling dan rendementen uit tijd-bars — een eigenschap die de prestaties van veel statistische modellen verbetert.

Nadelen: Vereist een ruwe trade-stream (niet bij alle dataproviders beschikbaar voor historische data). De timing van bars is onvoorspelbaar — u kunt niet zeggen "de volgende bar sluit op X".

3. Volume-Bars

Een nieuwe bar vormt zich nadat N contracten (of coins, in crypto) zijn verhandeld. Vergelijkbaar met tick-bars, maar gewogen naar transactiegrootte — één enkele trade van 100 BTC draagt 100x meer bij dan een trade van 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. Dollar-Bars

Een nieuwe bar vormt zich nadat een vaste nominale waarde (in USD/USDT) is verhandeld. De meest robuuste van de activiteitsgebaseerde bars omdat deze zowel voor het aantal trades als voor het prijsniveau normaliseert.

Bedenk: als ETH van $1.000 naar$4.000 stijgt, vereist het verkopen van $10.000 aan ETH 2,5 ETH bij$4.000 maar 10 ETH bij $1.000. Volume-bars zouden deze verschillend behandelen; dollar-bars behandelen ze hetzelfde.

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

De Drempel Kiezen

De drempel voor activiteitsgebaseerde bars moet ongeveer hetzelfde aantal bars per dag opleveren als de tijd-bars die u vervangt. Voor BTCUSDT op Binance:

Deze cijfers zijn bij benadering en verschuiven drastisch met het marktregime. Tijdens een rally of crash produceren activiteitsgebaseerde bars 5-10x meer bars dan gebruikelijk — wat precies de bedoeling is.

5–7. Prijsgebaseerde Bars

Renko-bricks, range-bars en volatiliteits-bars: alleen samplen wanneer de prijs genoeg beweegt om ertoe te doen.

Prijsgebaseerde bars negeren zowel tijd als activiteit. Een nieuwe bar vormt zich alleen wanneer de prijs met een gespecificeerd bedrag beweegt. Dit filtert op natuurlijke wijze zijwaartse ruis en benadrukt trends.

5. Renko-Bars

Een nieuwe Renko-"brick" vormt zich wanneer de slotkoers met minstens N eenheden beweegt ten opzichte van de slotkoers van de vorige brick. Bricks hebben altijd dezelfde grootte, wat een heldere visuele weergave van de trendrichting oplevert.

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

Dynamische Renko gebruikt ATR (Average True Range) in plaats van een vaste brick-grootte, en past zich zo automatisch aan de volatiliteit aan.

6. Range-Bars

Elke bar heeft een vaste high-low-range. Wanneer de range wordt overschreden, sluit de bar en begint een nieuwe. In tegenstelling tot Renko bevatten range-bars staarten en kunnen ze de intra-bar-volatiliteit tonen.

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

Belangrijk verschil tussen Renko en Range-bars: Renko volgt alleen slotkoersen en toont de richting; range-bars volgen de volledige prijsspreiding en tonen de structuur binnen de bar. Range-bars zijn over het algemeen bruikbaarder voor algoritmische handel omdat ze de high-low-informatie behouden die nodig is voor stop-loss- en take-profit-simulatie.

7. Volatiliteits-Bars

Een nieuwe bar vormt zich wanneer de intra-bar-volatiliteit een dynamische drempel bereikt — bijvoorbeeld een veelvoud van de recente ATR. In tegenstelling tot range-bars (vaste drempel) passen volatiliteits-bars zich aan de marktomstandigheden aan.

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 (Gladgestreken Transformatie)

Heikin-Ashi: middeling zet ruizige candles om in gladde trendsignalen — ten koste van exacte prijsinformatie.

Heikin-Ashi (Japans voor "gemiddelde bar") is geen bartype — het is een transformatie die kan worden toegepast bovenop elk basis-bartype. Het strijkt candles glad door de huidige en vorige bar-waarden te middelen:

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

Trends verschijnen als reeksen candles van dezelfde kleur zonder onderstaarten (opwaartse trend) of zonder bovenstaarten (neerwaartse trend).

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

Kritieke waarschuwing voor backtesting: Heikin-Ashi-prijzen zijn synthetisch. Als uw backtest de HA-close als instapprijs gebruikt, zullen de resultaten onjuist zijn. Gebruik HA altijd alleen voor signaalgeneratie en voer uit tegen echte OHLC-prijzen.

Wanneer HA nuttig is: Trendvolgende strategieën die schone "blijf erin"-signalen nodig hebben. Pas HA toe bovenop elk basis-bartype — tijd-bars, volume-bars, dollar-bars — om valse crossovers te filteren.

Wanneer HA schadelijk is: Elke strategie die precieze prijsniveaus nodig heeft — support/resistance, orderboekanalyse, PIQ (Position In Queue). De middeling vernietigt exacte prijsinformatie.

9–11. Japanse Omkeergrafieken

Kagi, Line Break en Point & Figure: tijdvrije chartingmethoden die zich puur richten op de prijsstructuur.

Dit zijn traditionele Japanse chartingmethoden (naast Renko) die de tijd volledig negeren en zich richten op de prijsstructuur.

9. Kagi-Grafieken

Kagi-grafieken bestaan uit verticale lijnen die van richting veranderen wanneer de prijs met een gespecificeerd bedrag omkeert. Lijnen veranderen van dikte wanneer de prijs een vorig hoog doorbreekt (dik = "yang" = vraag) of een vorig laag (dun = "yin" = aanbod).

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

Line-Break-grafieken tekenen een nieuwe lijn (box) alleen wanneer de slotkoers het hoog of laag van de vorige N lijnen (doorgaans 3) overschrijdt. Er wordt geen nieuwe lijn getekend als de prijs binnen de range blijft.

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

Point-&-Figure- (P&F-)grafieken gebruiken kolommen van X'en (stijgende prijzen) en O's (dalende prijzen). Kolomwissels vereisen een omkering van doorgaans 3 boxgroottes. Een van de oudste methoden om ruis te filteren en support/resistance te identificeren.

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 en P&F in algoritmische handel: Vooral gebruikt voor het detecteren van langetermijntrends en het identificeren van support/resistance. Als filterlaag — "geen long-signalen aannemen als de Kagi-grafiek in yin-modus staat" — leveren ze waarde door trades af te stemmen op de macrostructuur.

12–14. Informatiegedreven Bars

Imbalance-bars, run-bars, CUSUM-filters en entropie-bars: samplen wanneer de markt ons vertelt dat er iets is veranderd.

De meest geavanceerde aanpak, uit Marcos Lopez de Prado's Advances in Financial Machine Learning (2018). Het kerninzicht: samplen wanneer nieuwe informatie de markt bereikt, niet op vaste intervallen.

12. Tick Imbalance Bars (TIB)

Als de markt in evenwicht is, zouden koper-geïnitieerde en verkoper-geïnitieerde trades ongeveer in balans moeten zijn. Wanneer de onbalans onze verwachting overschrijdt, is er iets veranderd. Op dat moment wordt een bar gesampled.

Elke trade wordt geclassificeerd als koper-geïnitieerd (+1) of verkoper-geïnitieerd (-1) via de tick-regel. We volgen de cumulatieve onbalans θ en samplen wanneer |θ| een dynamische drempel overschrijdt.

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)

Uitbreiding van TIB's: in plaats van elke trade als ±1 te tellen, wordt gewogen naar ondertekend volume. Een koop van 100 BTC draagt +100 bij, een verkoop van 1 BTC -1. Vangt grote geïnformeerde orders op die mogelijk zijn opgesplitst in veel kleine trades.

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

Het Explosieprobleem

Een bekend probleem bij imbalance-bars: de EWMA-gebaseerde drempel kan in een positieve terugkoppelingslus terechtkomen. De oplossing: begrenzen met min_ticks- en max_ticks-grenzen.

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. Run-Bars

Run-bars volgen de lengte van de huidige directionele run — de langste opeenvolgende reeks van kopen of verkopen. Wanneer een grote geïnformeerde trader een order opsplitst in veel kleine trades, wordt de reeks ongewoon lang. Run-bars detecteren dit.

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

Run-bars kunnen worden uitgebreid tot volume-runs en dollar-runs.

15. CUSUM-Filter-Bars

Het CUSUM- (Cumulative Sum-)filter bepaalt wanneer er gesampled wordt door cumulatieve rendementen te volgen. In tegenstelling tot imbalance-bars (die werken op ruwe trades), kan CUSUM worden toegepast op bestaande 1m-OHLCV-data — er zijn geen tick-data vereist.

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 + Triple-Barrier-Methode: In het framework van Lopez de Prado worden CUSUM-events gebruikt als instappunten voor de Triple-Barrier-methode — waarbij elk event een trade activeert met stop-loss-, take-profit- en verlooptriggers. Voor robuuste validatie van dergelijke event-gedreven strategieën, zie Walk-Forward Optimization en Monte Carlo Bootstrap for Backtesting.

16. Entropie-Bars

De theoretisch meest elegante aanpak: samplen wanneer de informatie-inhoud (Shannon-entropie) van de intra-bar-prijsreeks een drempel overschrijdt.

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

Praktische opmerking: Entropie-bars zijn rekenintensief en vooral van onderzoeksmatig belang — maar voor ML-gebaseerde strategieën leveren ze features op met betere statistische eigenschappen, omdat elke bar ongeveer evenveel "informatie" bevat.

17. Delta-Bars (Order Flow)

Cumulatieve delta: het meten van de netto kracht van agressieve kopers versus verkopers in real time.

Delta-bars samplen op basis van de cumulatieve delta — het lopende verschil tussen koopvolume en verkoopvolume. In tegenstelling tot imbalance-bars (die tick-tekens ±1 gebruiken), gebruiken delta-bars daadwerkelijke volumegewogen order flow.

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

Delta-divergentie: Een van de krachtigste signalen — de prijs stijgt terwijl de cumulatieve delta negatief is (verkopers zijn agressief, maar de prijs stijgt toch, wat wijst op absorptie door limietkopen). Direct relevant voor de behavioral-fingerprinting-aanpak beschreven in het artikel Digital Fingerprint: Trader Identification. Voor market makers die het Avellaneda-Stoikov-model gebruiken, bieden delta-bars een real-time beeld van voorraadrisico en agressordruk.

Een circulaire buffer van basis-bars: nieuwe data komt binnen, oude data verlaat, en de geaggregeerde candle is altijd geldig.

Aggregatiemethoden bepalen hoe basis-bars worden samengesteld tot candles op een hoger timeframe (HTF). Ze zijn onafhankelijk van het bartype — u kunt elke aggregatiemethode toepassen op elk basis-bartype.

Methode A: Kalendergerichte Aggregatie

Aggregeer alle basis-bars die binnen een vaste kalendergrens vallen. De "1-uur"-candle bestrijkt alle bars van 14:00:00 tot 14:59:59.

Eigenschappen:

• Alle marktdeelnemers zien dezelfde grenzen — essentieel voor marktstructuuranalyse, support/resistance, PIQ-triggers
• Cold-start-probleem: gedeeltelijke candle na herstart
• Natuurlijk voor tijd-bars (dit is wat exchanges standaard leveren)
• Werkt ook voor niet-tijd-bars: "alle volume-bars die tussen 14:00 en 15:00 sloten" = een kalendergerichte uur-candle uit volume-bars

Methode B: Rolling-Window-Aggregatie

Aggregeer de laatste N gesloten basis-bars, herberekend bij elke nieuwe bar. Een "1-uur"-rolling-candle = de laatste 60 gesloten 1-minuut-tijd-bars, elke minuut bijgewerkt.

De atomaire eenheid is de gesloten basis-bar. Deze ontwerpkeuze levert op:

1. Geen cold start. Na N bars is de candle geldig. Geen ruis door gedeeltelijke candles.
2. Backtest-pariteit. Als live trading dezelfde atomaire eenheid gebruikt als de backtest-engine, zijn de signalen identiek.
3. Eenvoudige validatie. Eén regel: 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

Fasedeviatie-afweging: Rolling candles sluiten op :37 als u om :37 bent begonnen, niet op :00 zoals bij iedereen. Dit is van belang voor strategieën die afhankelijk zijn van niveaus die zichtbaar zijn voor de menigte. De oplossing: gebruik beide — kalender voor marktstructuur, rolling voor signalen.

Methode C: Adaptieve Rolling-Aggregatie

Zoals rolling, maar de venstergrootte past zich aan de huidige volatiliteit aan. Rustige markten → breder venster (meer gladstrijken). Volatiele markten → smaller venster (snellere reactie).

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

Elk basis-bartype kan worden gecombineerd met elke aggregatiemethode. Sommige combinaties zijn standaard (kalendergerichte tijd-bars = wat exchanges leveren), andere zijn exotisch maar krachtig.

Combinatievoorbeelden

Hybride Engine: Kalender + Rolling

In de praktijk wilt u zowel kalender- als rolling-aggregatie tegelijkertijd. De geheugenoverhead is verwaarloosbaar — twee deque-buffers per timeframe per symbool.

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

Time-Volume-Hybride: Kalender met Volume-Splitsingen

Een speciale aggregatievariant: kalendergerichte candles die vroegtijdig sluiten wanneer het volume een drempel overschrijdt. Behoudt de tijdsynchronisatie terwijl aangepast wordt aan activiteitspieken.

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

Praktische Aggregatie: Cascaderend Preloaden

Cascaderend preloaden: dagcandles samenstellen uit uurcandles, en uurcandles uit minuutcandles — waarbij API-limieten worden omzeild.

Exchanges beperken hoeveel historische data ze leveren. Binance geeft ~1000 candles per REST-verzoek, OKX begrenst op 300. Als u een rolling-1D-candle (1440 minuten) nodig hebt, kunt u niet altijd genoeg 1m-historie krijgen. Voor real-time streaming van trades en orderboeken via WebSocket, zie CCXT Pro WebSocket Methods.

De oplossing: cascaderende aggregatie — hogere timeframes bouwen vanuit de hoogst beschikbare resolutie op elk niveau, en deze vervolgens aan elkaar knopen.

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

Dit werkt omdat OHLCV-aggregatie componeerbaar is: de high van een 1D-candle is het maximum van 24 1h-highs, wat het maximum is van 1440 1m-highs.

Multi-Exchange-Limieten

Aggregatie-Consistentiecontrole

De 1h-candle van een REST-API komt mogelijk niet overeen met wat u zou berekenen uit 60 1m-candles. Altijd valideren:

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

Als de validatie consistent mislukt, altijd zelf aggregeren vanaf 1m — vertrouw nooit op de HTF-candle van de exchange voor backtest-pariteit.

Vergelijkingsmatrix

As 1: Basis-Bartypes

As 2: Aggregatiemethoden

Praktische Aanbevelingen

Vierlaagse candle-architectuur: rolling signalen, kalenderstructuur, microstructuur-flow en trendfilters.

Als uw backtest-engine draait op 1m-OHLCV-data:

1. Rolling tijd-bars — de eenvoudigste upgrade. Geen extra data. Elimineert cold start.
2. Hybride (rolling + kalender) tijd-bars — kalender voor marktstructuur, rolling voor signalen.
3. CUSUM-filter — werkt op 1m-closes, geen tick-data. "Er is genoeg bewogen om interessant te zijn."

Als u tick-/trade-data hebt:

4. Dollar-bars + rolling — de aanbevolen standaard uit de quant-finance-literatuur.
5. Volume-imbalance-bars + rolling — detecteert geïnformeerde flow, sampelt vaker tijdens belangrijke gebeurtenissen.
6. Delta-bars + kalender — als u een classificatie van de agressorzijde hebt, het meest directe beeld van wie de markt duwt.

Als filters (Heikin-Ashi of Line Break toepassen op elke basis+aggregatie-combinatie):

7. Heikin-Ashi over rolling volume-bars — schone trendsignalen op activiteitsgenormaliseerde data.
8. Line Break/Kagi over dagelijkse kalender-bars — macro-trendfilter.

Specifiek voor Marketmaker.cc — een gelaagde aanpak:

• Laag 1 (signalen): Rolling-aggregatie van tijd-bars voor indicatoren en instap-/uitstapsignalen. Geen cold start, perfecte backtest-pariteit.
• Laag 2 (marktstructuur): Kalendergerichte tijd-bars voor support/resistance, analyse van uurlijkse closes en PIQ-triggers.
• Laag 3 (microstructuur): Volume-imbalance-bars + delta-bars uit de ruwe trade-stream voor het detecteren van geïnformeerde flow, order-splitsing en het anticiperen op grote bewegingen. Zie ook Digital Fingerprint: Trader Identification voor patroonherkenning van gedrag in order-flow-data.
• Laag 4 (trendfilter): Heikin-Ashi-transformatie op rolling bars, of Line Break op 4u-kalender-closes, om signalen afgestemd te houden op de macrorichting.

Conclusie

Candle-constructie is geen enkele keuze — het zijn twee onafhankelijke beslissingen:

1. Welk type bar? Tijd legt kloktussenpozen vast. Activiteit (tick, volume, dollar) legt marktdeelname vast. Prijs (Renko, range, volatiliteit) legt bewegingen vast. Informatie (imbalance, runs, CUSUM, entropie) legt de aankomst van nieuwe informatie vast. Order flow (delta) legt agressieve druk vast.

2. Hoe aggregeren naar hogere timeframes? Kalender lijnt uit met de menigte. Rolling elimineert cold start. Adaptief reageert op volatiliteit.

De standaard "1-uur-candle van Binance" is slechts één cel in een 17×3-matrix. De overige 50 combinaties zijn beschikbaar voor iedereen die bereid is ze te implementeren. Voor een productiesysteem is het antwoord: "kies de juiste combinatie voor elke laag van uw beslissingsengine."

De atomaire eenheid — de gesloten basis-bar — blijft de fundering. Al het andere is aggregatie.

Voor meer over backtest-nauwkeurigheid met fijnmazige data, zie Adaptive Drill-Down: Backtest with Variable Granularity. Voor de impact van indicator-voorberekening op multi-timeframe-strategieën, zie Aggregated Parquet Cache.

Nuttige Links

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 — Python-bibliotheek die informatiegedreven bars implementeert
4. Binance — Historische Marktdata
5. Apache Parquet — kolomgeoriënteerd opslagformaat

Citatie

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