Bar-Typen und Aggregationsmethoden für algorithmischen Handel

Jedes Candlestick-Chart, das Sie je auf Binance, TradingView oder in irgendeiner Exchange-UI gesehen haben, wird auf dieselbe Weise gebaut: Trades innerhalb eines festen Zeitfensters aggregieren — 1 Minute, 5 Minuten, 1 Stunde — und daraus eine OHLCV-Bar erzeugen. Das ist so allgegenwärtig, dass die meisten Trader es nie hinterfragen. Für algorithmischen Handel sind die Wahl des Bar-Typs und die Aggregationsmethode jedoch zwei unabhängige Entscheidungen — und die meisten Systeme vermischen sie.

Dieser Artikel trennt die beiden Achsen der Candle-Konstruktion: welche Art von Bar Sie bauen (17 Typen) und wie Sie diese aggregieren zu höheren Zeitrahmen (3 Methoden). Die Kombination ergibt 51 mögliche Konfigurationen, jede mit unterschiedlichen Eigenschaften für Backtesting, Live-Handel und Signalgenerierung.

Für eine Einführung, wie rohe Trades zu Standard-Candles werden, siehe Trading Candles Demystified.

TL;DR

• Die Candle-Konstruktion hat zwei unabhängige Achsen: Bar-Typ und Aggregationsmethode
• 17 Basis-Bar-Typen: Zeit, Tick, Volumen, Dollar, Renko, Range, Volatilität, Heikin-Ashi, Kagi, Line Break, P&F, Tick Imbalance (TIB), Volume Imbalance (VIB), Run, CUSUM, Entropie, Delta
• 3 Aggregationsmethoden: kalenderausgerichtet, Rolling Window, adaptives Rolling
• 17 × 3 = 51 mögliche Kombinationen, jede mit unterschiedlichen Eigenschaften
• Die meisten Systeme nutzen nur eine Kombination: kalenderausgerichtete Zeit-Bars. Die anderen 50 sind ungenutzt.
• Praktische Empfehlung: mehrere Kombinationen in Schichten verwenden — Rolling-Zeit-Bars für Signale, Kalender-Zeit-Bars für Marktstruktur, informationsgetriebene Bars für Mikrostruktur

Zwei Achsen der Candle-Konstruktion

Die traditionelle Sichtweise stellt alle Bar-Typen in eine flache Liste: Zeit-Bars, Tick-Bars, Volumen-Bars, Renko usw. Das ist irreführend. Tatsächlich gibt es zwei orthogonale Entscheidungen:

Achse 1 — Basis-Bar-Typ (17 Typen): Wie entscheiden Sie, wann eine neue Bar schließt? Nach einem festen Zeitintervall? Nach N Trades? Nach einer Preisbewegung? Wenn sich der Informationsgehalt ändert? Das bestimmt, was "eine Bar" bedeutet.

Achse 2 — Aggregationsmethode (3 Methoden): Wie komponieren Sie Basis-Bars zu Candles mit höherem Zeitrahmen? An Kalendergrenzen ausrichten (00:00, 01:00, ...)? Ein Rolling Window der letzten N Bars verwenden? Die Fenstergröße an die Volatilität anpassen?

Diese beiden Achsen sind unabhängig. Sie können haben:

• Kalenderausgerichtete Tick-Bars — Tick-Bars aggregieren, die zwischen 14:00 und 14:59 geschlossen haben, zu einer einzigen Stunden-Candle
• Rolling-Volumen-Bars — die letzten 24 Volumen-Bars nehmen, unabhängig davon, wann sie geschlossen haben
• Adaptive Delta-Bars — ein volatilitätsgetriebenes Fenster über Delta-Bars verwenden

Die Standard-"1-Stunden-Candle" ist nur ein Punkt in dieser 17×3-Matrix: Zeit-Bars + Kalenderausrichtung. Jede andere Kombination ist eine Alternative, die es zu erwägen lohnt.

1. Zeit-Bars (Standard)

Ungleichmäßige Informationsdichte: starre Zeitgrenzen behandeln 200-Trade-Ruhestunden genauso wie 50.000-Trade-Ankündigungsstunden.

Der Standard. Eine neue Bar bildet sich nach einem festen Zeitintervall: 1 Minute, 5 Minuten, 1 Stunde. Jede Exchange liefert diese nativ.

Eigenschaften:

• Während der asiatischen Session (00:00–08:00 UTC) könnte eine 1-Stunden-Candle 200 Trades enthalten. Bei einer Binance-Listing-Ankündigung könnte dasselbe Fenster 50.000 Trades enthalten. Zeit-Bars behandeln beides als gleichwertig. Das Erkennen solcher Aktivitätsspitzen ist entscheidend für den Bot-Schutz — siehe Anomaly Detection for Trading Bots.
• Alle Marktteilnehmer sehen dieselben Candle-Grenzen — ein Schelling-Punkt. Das macht Zeit-Bars essenziell für die Analyse von Herdenverhalten.
• Indikatoren, die auf unvollständigen Candles berechnet werden (nach einem Neustart), erzeugen unbrauchbare Werte.

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. Aktivitätsbasierte Bars

Tick-, Volumen- und Dollar-Bars: drei Wege, die Marktbeteiligung — nicht die Uhr — die Bar-Grenzen bestimmen zu lassen.

Statt in festen Zeitintervallen zu sampeln, wird nach einer festen Menge an Marktaktivität gesampelt. Das erzeugt Bars mit annähernd gleichem "Informationsgehalt" unabhängig von der Tageszeit.

2. Tick-Bars

Eine neue Bar bildet sich nach jeweils N Trades (Ticks). Bei hoher Aktivität bilden sich Bars schnell. In ruhigen Phasen kann eine einzelne Bar sich über Stunden erstrecken.

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

Vorteile: Passt sich natürlich der Marktaktivität an. Renditen aus Tick-Bars tendieren dazu, näher an einer Normalverteilung zu liegen als Zeit-Bar-Renditen — eine Eigenschaft, die die Performance vieler statistischer Modelle verbessert.

Nachteile: Erfordert einen Roh-Trade-Stream (nicht bei allen Datenanbietern für historische Daten verfügbar). Das Timing der Bars ist unvorhersehbar — man kann nicht sagen "die nächste Bar wird bei X schließen."

3. Volumen-Bars

Eine neue Bar bildet sich, nachdem N Kontrakte (oder Coins, im Krypto-Bereich) gehandelt wurden. Ähnlich wie Tick-Bars, aber gewichtet nach Trade-Größe — ein einzelner 100-BTC-Trade trägt 100x mehr bei als ein 1-BTC-Trade.

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

Eine neue Bar bildet sich, nachdem ein fester Nominalwert (in USD/USDT) gehandelt wurde. Die robusteste der aktivitätsbasierten Bars, weil sie sowohl für die Trade-Anzahl als auch für das Preisniveau normalisiert.

Man bedenke: Wenn ETH von 1.000 $auf 4.000$ steigt, erfordert der Verkauf von ETH im Wert von 10.000 $2,5 ETH bei 4.000$, aber 10 ETH bei 1.000 $. Volumen-Bars würden diese unterschiedlich behandeln; Dollar-Bars behandeln sie gleich.

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

Die Wahl der Schwelle

Die Schwelle für aktivitätsbasierte Bars sollte etwa dieselbe Anzahl an Bars pro Tag erzeugen wie die Zeit-Bars, die Sie ersetzen. Für BTCUSDT auf Binance:

Diese Zahlen sind Näherungswerte und verschieben sich dramatisch je nach Marktregime. Während einer Rally oder eines Crashs erzeugen aktivitätsbasierte Bars 5-10x mehr Bars als üblich — was genau der Sinn der Sache ist.

5–7. Preisbasierte Bars

Renko-Ziegel, Range-Bars und Volatilitäts-Bars: nur dann sampeln, wenn sich der Preis genug bewegt, um relevant zu sein.

Preisbasierte Bars ignorieren sowohl Zeit als auch Aktivität. Eine neue Bar bildet sich nur, wenn sich der Preis um einen festgelegten Betrag bewegt. Das filtert auf natürliche Weise Seitwärtsrauschen heraus und hebt Trends hervor.

5. Renko-Bars

Ein neuer Renko-"Ziegel" bildet sich, wenn sich der Schlusskurs um mindestens N Einheiten vom Schlusskurs des vorherigen Ziegels bewegt. Ziegel haben immer dieselbe Größe, was eine klare visuelle Darstellung der Trendrichtung ergibt.

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

Dynamisches Renko verwendet ATR (Average True Range) statt einer festen Ziegelgröße und passt sich so automatisch an die Volatilität an.

6. Range-Bars

Jede Bar hat eine feste High-Low-Range. Wenn die Range überschritten wird, schließt die Bar und eine neue beginnt. Anders als Renko enthalten Range-Bars Dochte und können die Intra-Bar-Volatilität zeigen.

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

Wichtiger Unterschied zwischen Renko- und Range-Bars: Renko verfolgt nur Schlusskurse und zeigt die Richtung; Range-Bars verfolgen die gesamte Preisspanne und zeigen die Struktur innerhalb der Bar. Range-Bars sind für algorithmischen Handel im Allgemeinen nützlicher, weil sie High-Low-Informationen erhalten, die für die Stop-Loss- und Take-Profit-Simulation benötigt werden.

7. Volatilitäts-Bars

Eine neue Bar bildet sich, wenn die Intra-Bar-Volatilität eine dynamische Schwelle erreicht — zum Beispiel ein Vielfaches des aktuellen ATR. Anders als Range-Bars (feste Schwelle) passen sich Volatilitäts-Bars an die Marktbedingungen an.

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 (geglättete Transformation)

Heikin-Ashi: Mittelwertbildung verwandelt verrauschte Candles in glatte Trendsignale — auf Kosten exakter Preisinformationen.

Heikin-Ashi (japanisch für "durchschnittliche Bar") ist kein Bar-Typ — es ist eine Transformation, die auf jeden beliebigen Basis-Bar-Typ angewendet werden kann. Sie glättet Candles, indem sie aktuelle und vorherige Bar-Werte mittelt:

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

Trends erscheinen als Sequenzen gleichfarbiger Candles ohne untere Dochte (Aufwärtstrend) oder ohne obere Dochte (Abwärtstrend).

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

Kritischer Vorbehalt für das Backtesting: Heikin-Ashi-Preise sind synthetisch. Wenn Ihr Backtest den HA-Close als Einstiegspreis verwendet, werden die Ergebnisse falsch sein. Verwenden Sie HA immer nur zur Signalgenerierung und führen Sie zu echten OHLC-Preisen aus.

Wann HA nützlich ist: Trendfolgestrategien, die saubere "Bleib drin"-Signale benötigen. HA über jeden beliebigen Basis-Bar-Typ anwenden — Zeit-Bars, Volumen-Bars, Dollar-Bars —, um falsche Crossover herauszufiltern.

Wann HA schädlich ist: Jede Strategie, die präzise Preisniveaus benötigt — Support/Resistance, Orderbuch-Analyse, PIQ (Position In Queue). Die Mittelwertbildung zerstört exakte Preisinformationen.

9–11. Japanische Umkehr-Charts

Kagi, Line Break und Point & Figure: zeitfreie Charting-Methoden, die sich rein auf die Preisstruktur konzentrieren.

Dies sind traditionelle japanische Charting-Methoden (neben Renko), die die Zeit vollständig verwerfen und sich auf die Preisstruktur konzentrieren.

9. Kagi-Charts

Kagi-Charts bestehen aus vertikalen Linien, die die Richtung ändern, wenn sich der Preis um einen festgelegten Betrag umkehrt. Linien ändern ihre Dicke, wenn der Preis ein vorheriges Hoch durchbricht (dick = "Yang" = Nachfrage) oder ein vorheriges Tief (dünn = "Yin" = Angebot).

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-Charts

Line-Break-Charts zeichnen eine neue Linie (Box) nur dann, wenn der Schlusskurs das Hoch oder Tief der vorherigen N Linien (typischerweise 3) übersteigt. Es wird keine neue Linie gezeichnet, wenn der Preis innerhalb der Range bleibt.

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-Charts

Point-&-Figure- (P&F-)Charts verwenden Spalten aus X's (steigende Preise) und O's (fallende Preise). Spaltenwechsel erfordern eine Umkehr von typischerweise 3 Box-Größen. Eine der ältesten Methoden zur Rauschfilterung und Identifikation von Support/Resistance.

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 und P&F im algorithmischen Handel: Werden vor allem zur langfristigen Trenderkennung und Identifikation von Support/Resistance verwendet. Als Filter-Layer — "keine Long-Signale annehmen, wenn der Kagi-Chart im Yin-Modus ist" — schaffen sie Mehrwert, indem sie Trades an der Makrostruktur ausrichten.

12–14. Informationsgetriebene Bars

Imbalance-Bars, Run-Bars, CUSUM-Filter und Entropie-Bars: sampeln, wenn der Markt uns mitteilt, dass sich etwas geändert hat.

Der ausgefeilteste Ansatz, aus Marcos Lopez de Prados Advances in Financial Machine Learning (2018). Die Kernidee: sampeln, wenn neue Informationen am Markt ankommen, nicht in festen Intervallen.

12. Tick Imbalance Bars (TIB)

Wenn sich der Markt im Gleichgewicht befindet, sollten käuferinitiierte und verkäuferinitiierte Trades sich in etwa ausgleichen. Wenn die Imbalance unsere Erwartung übersteigt, hat sich etwas geändert. In diesem Moment wird eine Bar gesampelt.

Jeder Trade wird mittels der Tick-Regel als käuferinitiiert (+1) oder verkäuferinitiiert (-1) klassifiziert. Wir verfolgen die kumulative Imbalance θ und sampeln, wenn |θ| eine dynamische Schwelle übersteigt.

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)

Erweiterung der TIBs: Statt jeden Trade als ±1 zu zählen, wird nach vorzeichenbehaftetem Volumen gewichtet. Ein 100-BTC-Kauf trägt +100 bei, ein 1-BTC-Verkauf -1. Erfasst große informierte Orders, die möglicherweise in viele kleine Trades aufgeteilt wurden.

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

Das Explosionsproblem

Ein bekanntes Problem bei Imbalance-Bars: Die EWMA-basierte Schwelle kann in eine positive Rückkopplungsschleife geraten. Die Lösung: Begrenzung mit min_ticks- und 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 verfolgen die Länge des aktuellen gerichteten Runs — die längste aufeinanderfolgende Sequenz von Käufen oder Verkäufen. Wenn ein großer informierter Trader eine Order in viele kleine Trades aufteilt, wird die Sequenz ungewöhnlich lang. Run-Bars erkennen dies.

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 können zu Volumen-Runs und Dollar-Runs erweitert werden.

15. CUSUM-Filter-Bars

Der CUSUM- (Cumulative Sum-)Filter bestimmt wann gesampelt wird, indem er kumulative Renditen verfolgt. Anders als Imbalance-Bars (die auf Rohdaten von Trades arbeiten) kann CUSUM auf bestehende 1m-OHLCV-Daten angewendet werden — keine Tick-Daten erforderlich.

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 Lopez de Prados Framework werden CUSUM-Events als Einstiegspunkte für die Triple-Barrier-Methode verwendet — bei der jedes Event einen Trade mit Stop-Loss-, Take-Profit- und Ablauf-Barrieren auslöst. Für eine robuste Validierung solcher ereignisgetriebener Strategien siehe Walk-Forward Optimization und Monte Carlo Bootstrap for Backtesting.

16. Entropie-Bars

Der theoretisch eleganteste Ansatz: sampeln, wenn der Informationsgehalt (Shannon-Entropie) der Intra-Bar-Preisreihe eine Schwelle übersteigt.

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

Praxishinweis: Entropie-Bars sind rechenintensiv und primär von forschungstechnischem Interesse — aber für ML-basierte Strategien erzeugen sie Features mit besseren statistischen Eigenschaften, weil jede Bar annähernd gleich viel "Information" enthält.

17. Delta-Bars (Order Flow)

Kumulatives Delta: Messung der Nettokraft aggressiver Käufer gegenüber Verkäufern in Echtzeit.

Delta-Bars sampeln basierend auf dem kumulativen Delta — der laufenden Differenz zwischen Kauf- und Verkaufsvolumen. Anders als Imbalance-Bars (die Tick-Vorzeichen ±1 verwenden), verwenden Delta-Bars tatsächlichen volumengewichteten 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-Divergenz: Eines der stärksten Signale — der Preis steigt, während das kumulative Delta negativ ist (Verkäufer sind aggressiv, aber der Preis steigt trotzdem, was auf Absorption durch Limit-Käufe hindeutet). Direkt relevant für den im Artikel Digital Fingerprint: Trader Identification beschriebenen Ansatz des Behavioral Fingerprinting. Für Market Maker, die das Avellaneda-Stoikov-Modell verwenden, bieten Delta-Bars eine Echtzeitsicht auf Inventarrisiko und Aggressor-Druck.

Ein Ringpuffer aus Basis-Bars: neue Daten kommen hinzu, alte gehen heraus, und die aggregierte Candle ist immer gültig.

Aggregationsmethoden bestimmen, wie Basis-Bars zu Candles mit höherem Zeitrahmen (HTF) komponiert werden. Sie sind unabhängig vom Bar-Typ — Sie können jede Aggregationsmethode auf jeden Basis-Bar-Typ anwenden.

Methode A: Kalenderausgerichtete Aggregation

Alle Basis-Bars aggregieren, die in eine feste Kalendergrenze fallen. Die "1-Stunden"-Candle deckt alle Bars von 14:00:00 bis 14:59:59 ab.

Eigenschaften:

• Alle Marktteilnehmer sehen dieselben Grenzen — essenziell für die Marktstrukturanalyse, Support/Resistance, PIQ-Trigger
• Cold-Start-Problem: unvollständige Candle nach Neustart
• Natürlich für Zeit-Bars (das liefern Exchanges nativ)
• Funktioniert auch für Nicht-Zeit-Bars: "alle Volumen-Bars, die zwischen 14:00 und 15:00 geschlossen haben" = eine kalenderausgerichtete Stunden-Candle aus Volumen-Bars

Methode B: Rolling-Window-Aggregation

Die letzten N geschlossenen Basis-Bars aggregieren, bei jeder neuen Bar neu berechnet. Eine "1-Stunden"-Rolling-Candle = die letzten 60 geschlossenen 1-Minuten-Zeit-Bars, jede Minute aktualisiert.

Die atomare Einheit ist die geschlossene Basis-Bar. Diese Designentscheidung ergibt:

1. Kein Cold Start. Nach N Bars ist die Candle gültig. Kein Rauschen durch unvollständige Candles.
2. Backtest-Parität. Wenn der Live-Handel dieselbe atomare Einheit wie die Backtest-Engine verwendet, sind die Signale identisch.
3. Einfache Validierung. Eine 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

Phasenverschiebungs-Kompromiss: Rolling-Candles schließen um :37, wenn Sie um :37 gestartet sind, nicht um :00 wie bei allen anderen. Das ist relevant für Strategien, die von Herden-sichtbaren Niveaus abhängen. Die Lösung: beides verwenden — Kalender für Marktstruktur, Rolling für Signale.

Methode C: Adaptive Rolling-Aggregation

Wie Rolling, aber die Fenstergröße passt sich an die aktuelle Volatilität an. Ruhige Märkte → breiteres Fenster (mehr Glättung). Volatile Märkte → engeres Fenster (schnellere Reaktion).

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

Jeder Basis-Bar-Typ kann mit jeder Aggregationsmethode kombiniert werden. Manche Kombinationen sind Standard (kalenderausgerichtete Zeit-Bars = das, was Exchanges liefern), andere sind exotisch, aber leistungsfähig.

Kombinationsbeispiele

Hybrid-Engine: Kalender + Rolling

In der Praxis möchte man gleichzeitig sowohl kalenderausgerichtete als auch Rolling-Aggregation. Der Speicher-Overhead ist vernachlässigbar — zwei deque-Puffer pro Zeitrahmen pro Symbol.

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-Hybrid: Kalender mit Volumen-Splits

Eine spezielle Aggregationsvariante: kalenderausgerichtete Candles, die frühzeitig schließen, wenn das Volumen eine Schwelle überschreitet. Erhält die Zeitsynchronisation, während sie sich an Aktivitätsspitzen anpasst.

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 Aggregation: Kaskadierendes Preloading

Kaskadierendes Preloading: Tages-Candles aus Stunden-Candles zusammensetzen, und Stunden-Candles aus Minuten-Candles — unter Umgehung von API-Limits.

Exchanges begrenzen, wie viele historische Daten sie liefern. Binance gibt ~1000 Candles pro REST-Request, OKX begrenzt auf 300. Wenn Sie eine Rolling-1D-Candle (1440 Minuten) benötigen, können Sie nicht immer genug 1m-Historie bekommen. Für das Echtzeit-Streaming von Trades und Orderbüchern über WebSocket siehe CCXT Pro WebSocket Methods.

Die Lösung: kaskadierende Aggregation — höhere Zeitrahmen aus der auf jeder Ebene verfügbaren höchsten Auflösung aufbauen und dann zusammenfügen.

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

Das funktioniert, weil OHLCV-Aggregation komponierbar ist: das High einer 1D-Candle ist das Maximum von 24 1h-Highs, was das Maximum von 1440 1m-Highs ist.

Limits über mehrere Exchanges

Aggregations-Konsistenzprüfung

Die 1h-Candle aus einer REST-API stimmt möglicherweise nicht mit dem überein, was Sie aus 60 1m-Candles berechnen würden. Immer validieren:

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

Wenn die Validierung konsistent fehlschlägt, immer selbst aus 1m aggregieren — niemals der HTF-Candle der Exchange für Backtest-Parität vertrauen.

Vergleichsmatrix

Achse 1: Basis-Bar-Typen

Achse 2: Aggregationsmethoden

Praktische Empfehlungen

Vier-Schichten-Candle-Architektur: Rolling-Signale, Kalenderstruktur, Mikrostruktur-Flow und Trendfilter.

Wenn Ihre Backtest-Engine auf 1m-OHLCV-Daten läuft:

1. Rolling-Zeit-Bars — einfachstes Upgrade. Keine zusätzlichen Daten. Beseitigt Cold Start.
2. Hybrid (Rolling + Kalender) Zeit-Bars — Kalender für Marktstruktur, Rolling für Signale.
3. CUSUM-Filter — funktioniert auf 1m-Schlusskursen, keine Tick-Daten. "Etwas hat sich genug bewegt, um interessant zu sein."

Wenn Sie Tick-/Trade-Daten haben:

4. Dollar-Bars + Rolling — empfohlener Standard aus der Quant-Finance-Literatur.
5. Volume-Imbalance-Bars + Rolling — erkennt informierten Flow, sampelt bei bedeutsamen Ereignissen häufiger.
6. Delta-Bars + Kalender — wenn Sie eine Aggressor-Seiten-Klassifizierung haben, die direkteste Sicht darauf, wer den Markt bewegt.

Als Filter (Heikin-Ashi oder Line Break auf jede beliebige Basis-+Aggregations-Kombination anwenden):

7. Heikin-Ashi über Rolling-Volumen-Bars — saubere Trendsignale auf aktivitätsnormalisierten Daten.
8. Line Break/Kagi über tägliche Kalender-Bars — Makro-Trendfilter.

Für Marketmaker.cc speziell — ein geschichteter Ansatz:

• Schicht 1 (Signale): Rolling-Aggregation von Zeit-Bars für Indikatoren und Ein-/Ausstiegssignale. Kein Cold Start, perfekte Backtest-Parität.
• Schicht 2 (Marktstruktur): Kalenderausgerichtete Zeit-Bars für Support/Resistance, Stunden-Schlusskurs-Analyse und PIQ-Trigger.
• Schicht 3 (Mikrostruktur): Volume-Imbalance-Bars + Delta-Bars aus dem Roh-Trade-Stream zur Erkennung von informiertem Flow, Order-Splitting und zur Antizipation großer Bewegungen. Siehe auch Digital Fingerprint: Trader Identification für Mustererkennung im Verhalten anhand von Order-Flow-Daten.
• Schicht 4 (Trendfilter): Heikin-Ashi-Transformation auf Rolling-Bars, oder Line Break auf 4h-Kalender-Schlusskursen, um Signale mit der Makrorichtung ausgerichtet zu halten.

Fazit

Die Candle-Konstruktion ist keine einzelne Entscheidung — es sind zwei unabhängige Entscheidungen:

1. Welche Art von Bar? Zeit erfasst Uhrintervalle. Aktivität (Tick, Volumen, Dollar) erfasst Marktbeteiligung. Preis (Renko, Range, Volatilität) erfasst Bewegungen. Information (Imbalance, Runs, CUSUM, Entropie) erfasst das Eintreffen neuer Informationen. Order Flow (Delta) erfasst aggressiven Druck.

2. Wie zu höheren Zeitrahmen aggregieren? Kalender richtet sich an der Herde aus. Rolling beseitigt Cold Start. Adaptiv reagiert auf Volatilität.

Die Standard-"1-Stunden-Candle von Binance" ist nur eine Zelle in einer 17×3-Matrix. Die anderen 50 Kombinationen stehen jedem zur Verfügung, der bereit ist, sie zu implementieren. Für ein Produktivsystem lautet die Antwort: "Für jede Schicht Ihrer Entscheidungs-Engine die richtige Kombination wählen."

Die atomare Einheit — die geschlossene Basis-Bar — bleibt das Fundament. Alles andere ist Aggregation.

Mehr zur Backtest-Genauigkeit mit feingranularen Daten finden Sie unter Adaptive Drill-Down: Backtest with Variable Granularity. Zur Auswirkung der Indikator-Vorberechnung auf Multi-Timeframe-Strategien siehe Aggregated Parquet Cache.

Nützliche 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-Bibliothek zur Implementierung informationsgetriebener Bars
4. Binance — Historische Marktdaten
5. Apache Parquet — spaltenorientiertes Speicherformat

Zitierung

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