Jenis Bar dan Kaedah Pengagregatan untuk Dagangan Algoritma

Setiap carta lilin yang pernah anda lihat di Binance, TradingView, atau mana-mana antara muka bursa dibina dengan cara yang sama: kumpulkan dagangan dalam tetingkap masa tetap — 1 minit, 5 minit, 1 jam — dan hasilkan bar OHLCV. Perkara ini begitu lazim sehingga kebanyakan peniaga tidak pernah mempersoalkannya. Namun bagi dagangan algoritma, pemilihan jenis bar dan kaedah pengagregatan adalah dua keputusan yang bebas — dan kebanyakan sistem menggabungkan kedua-duanya tanpa sedar.

Artikel ini memisahkan dua paksi pembinaan lilin: jenis bar yang anda bina (17 jenis) dan bagaimana anda mengagregasinya ke dalam tempoh masa yang lebih tinggi (3 kaedah). Kombinasinya menghasilkan 51 konfigurasi yang mungkin, masing-masing dengan sifat berbeza untuk backtesting, dagangan langsung, dan penjanaan isyarat.

Untuk pengenalan tentang bagaimana dagangan mentah menjadi lilin standard, lihat Trading Candles Demystified.

TL;DR

• Pembinaan lilin mempunyai dua paksi bebas: jenis bar dan kaedah pengagregatan
• 17 jenis bar asas: masa, tik, isipadu, dolar, Renko, julat, kemeruapan, Heikin-Ashi, Kagi, Line Break, P&F, ketakseimbangan tik (TIB), ketakseimbangan isipadu (VIB), larian, CUSUM, entropi, delta
• 3 kaedah pengagregatan: sejajar kalender, tetingkap gelinding, gelinding penyesuaian
• 17 × 3 = 51 kombinasi yang mungkin, masing-masing dengan sifat berbeza
• Kebanyakan sistem hanya menggunakan satu kombinasi: bar masa sejajar kalender. 50 yang lain belum diterokai.
• Cadangan praktikal: gunakan pelbagai kombinasi secara berlapis — bar masa gelinding untuk isyarat, bar masa kalender untuk struktur pasaran, bar berpandukan maklumat untuk mikrostruktur

Dua Paksi Pembinaan Lilin

Pandangan tradisional meletakkan semua jenis bar dalam senarai rata: bar masa, bar tik, bar isipadu, Renko, dsb. Ini mengelirukan. Sebenarnya terdapat dua pilihan berortogon:

Paksi 1 — Jenis Bar Asas (17 jenis): Bagaimana anda memutuskan bila bar baru ditutup? Selepas selang masa tetap? Selepas N dagangan? Selepas pergerakan harga? Apabila kandungan maklumat berubah? Ini menentukan erti "satu bar."

Paksi 2 — Kaedah Pengagregatan (3 kaedah): Bagaimana anda menyusun bar asas menjadi lilin tempoh masa lebih tinggi? Sejajarkan dengan sempadan kalender (00:00, 01:00, ...)? Gunakan tetingkap gelinding N bar terakhir? Sesuaikan saiz tetingkap mengikut kemeruapan?

Kedua-dua paksi ini adalah bebas. Anda boleh mempunyai:

• Bar tik sejajar kalender — kumpulkan bar tik yang ditutup antara 14:00 dan 14:59 menjadi satu lilin jam
• Bar isipadu gelinding — ambil 24 bar isipadu terakhir tanpa mengira bila ia ditutup
• Bar delta penyesuaian — gunakan tetingkap didorong kemeruapan ke atas bar delta

"Lilin 1 jam" standard hanyalah satu titik dalam matriks 17×3 ini: bar masa + sejajar kalender. Setiap kombinasi lain adalah alternatif yang wajar dipertimbangkan.

1. Bar Masa (Standard)

Ketumpatan maklumat tidak seragam: sempadan masa rigid melayan jam sunyi dengan 200 dagangan sama seperti jam pengumuman dengan 50,000 dagangan.

Lalai. Bar baru terbentuk selepas selang masa tetap: 1 minit, 5 minit, 1 jam. Setiap bursa menyediakannya secara natif.

Sifat:

• Semasa sesi Asia (00:00–08:00 UTC), lilin 1 jam mungkin mengandungi 200 dagangan. Semasa pengumuman penyenaraian Binance, tetingkap yang sama boleh mengandungi 50,000 dagangan. Bar masa melayan kedua-duanya sebagai setara. Mengesan lonjakan aktiviti sedemikian adalah kritikal untuk perlindungan bot — lihat Anomaly Detection for Trading Bots.
• Semua peserta pasaran melihat sempadan lilin yang sama — titik Schelling. Ini menjadikan bar masa penting untuk menganalisis tingkah laku orang ramai.
• Penunjuk yang dikira pada lilin separa (selepas mula semula) menghasilkan nilai sampah.

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. Bar Berasaskan Aktiviti

Bar tik, isipadu, dan dolar: tiga cara membiarkan penyertaan pasaran — bukan jam — menentukan sempadan bar.

Daripada persampelan pada selang masa tetap, persampelan dilakukan selepas jumlah aktiviti pasaran yang tetap. Ini menghasilkan bar dengan "kandungan maklumat" yang lebih seragam tanpa mengira masa dalam hari.

2. Bar Tik

Bar baru terbentuk selepas setiap N dagangan (tik). Semasa aktiviti tinggi, bar terbentuk dengan cepat. Semasa tempoh tenang, satu bar boleh merentasi berjam-jam.

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

Kelebihan: Menyesuaikan diri dengan aktiviti pasaran secara semulajadi. Pulangan dari bar tik cenderung lebih hampir kepada taburan normal berbanding pulangan bar masa — sifat yang meningkatkan prestasi banyak model statistik.

Kelemahan: Memerlukan strim dagangan mentah (tidak tersedia dari semua pembekal data untuk data sejarah). Masa bar tidak dapat diramal — anda tidak boleh mengatakan "bar berikutnya akan ditutup pada X."

3. Bar Isipadu

Bar baru terbentuk selepas N kontrak (atau syiling, dalam kripto) didagangkan. Serupa dengan bar tik, tetapi berwajaran mengikut saiz dagangan — satu dagangan 100-BTC menyumbang 100x lebih banyak daripada dagangan 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. Bar Dolar

Bar baru terbentuk selepas nilai nosional tetap (dalam USD/USDT) ditukarkan. Yang paling kukuh dalam bar berasaskan aktiviti kerana ia menormalisasi untuk bilangan dagangan dan tahap harga.

Pertimbangkan: jika ETH bergerak dari $1,000 ke$4,000, menjual ETH bernilai $10,000 memerlukan 2.5 ETH pada$4,000 tetapi 10 ETH pada $1,000. Bar isipadu akan melayan kedua-duanya secara berbeza; bar dolar melayan kedua-duanya sama.

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

Memilih Ambang

Ambang untuk bar berasaskan aktiviti seharusnya menghasilkan lebih kurang bilangan bar yang sama setiap hari seperti bar masa yang digantikan. Untuk BTCUSDT di Binance:

Angka-angka ini adalah anggaran dan berubah secara dramatik mengikut rejim pasaran. Semasa rali atau jatuhnya harga, bar berasaskan aktiviti akan menghasilkan 5-10x lebih banyak bar daripada biasa — itulah tujuannya.

5–7. Bar Berasaskan Harga

Bata Renko, bar julat, dan bar kemeruapan: persampelan hanya apabila harga bergerak cukup bermakna.

Bar berasaskan harga mengabaikan masa dan aktiviti. Bar baru terbentuk hanya apabila harga bergerak sebanyak nilai tertentu. Ini menapis bunyi sisi secara semulajadi dan menonjolkan trend.

5. Bar Renko

"Bata" Renko baru terbentuk apabila harga penutup bergerak sekurang-kurangnya N unit dari penutup bata sebelumnya. Bata sentiasa bersaiz sama, mencipta perwakilan visual yang jelas tentang arah trend.

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

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

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

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

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

        if num_bricks == 0:
            return []

        direction = 1 if diff > 0 else -1

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

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

        self.bricks.extend(new_bricks)
        return new_bricks

Renko Dinamik menggunakan ATR (Purata Julat Sebenar) sebagai ganti saiz bata tetap, menyesuaikan diri dengan kemeruapan secara automatik.

6. Bar Julat

Setiap bar mempunyai julat tinggi-rendah yang tetap. Apabila julat melebihi had, bar ditutup dan yang baru bermula. Tidak seperti Renko, bar julat merangkumi bayang dan boleh menunjukkan kemeruapan dalam bar.

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

Perbezaan utama antara Renko dan bar Julat: Renko hanya menjejak harga penutup dan menunjukkan arah; bar julat menjejak julat harga penuh dan menunjukkan struktur dalam bar. Bar julat umumnya lebih berguna untuk dagangan algoritma kerana ia memelihara maklumat tinggi-rendah yang diperlukan untuk simulasi henti rugi dan ambil untung.

7. Bar Kemeruapan

Bar baru terbentuk apabila kemeruapan dalam bar mencapai ambang dinamik — contohnya, gandaan ATR terkini. Tidak seperti bar julat (ambang tetap), bar kemeruapan menyesuaikan diri dengan keadaan pasaran.

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 (Transformasi Halus)

Heikin-Ashi: purata mengubah lilin bising menjadi isyarat trend yang halus — tetapi dengan kos maklumat harga tepat.

Heikin-Ashi (bahasa Jepun bermaksud "bar purata") bukan satu jenis bar — ia adalah transformasi yang boleh digunakan pada mana-mana jenis bar asas. Ia menghaluskan lilin dengan mengambil purata nilai bar semasa dan sebelumnya:

• HA Tutup = (Buka + Tinggi + Rendah + Tutup) / 4
• HA Buka = (HA Buka Sebelum + HA Tutup Sebelum) / 2
• HA Tinggi = maks(Tinggi, HA Buka, HA Tutup)
• HA Rendah = min(Rendah, HA Buka, HA Tutup)

Trend muncul sebagai urutan lilin berwarna sama tanpa bayang bawah (tren naik) atau tanpa bayang atas (tren turun).

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

Amaran kritikal untuk backtesting: Harga Heikin-Ashi adalah sintetik. Jika backtesting anda menggunakan penutup HA sebagai harga masuk, hasilnya akan salah. Sentiasa gunakan HA untuk penjanaan isyarat sahaja dan laksanakan pada harga OHLC sebenar.

Bila HA berguna: Strategi mengikut trend yang memerlukan isyarat "kekal dalam" yang bersih. Gunakan HA pada mana-mana jenis bar asas — bar masa, bar isipadu, bar dolar — untuk menapis persilangan palsu.

Bila HA berbahaya: Sebarang strategi yang memerlukan tahap harga tepat — sokongan/rintangan, analisis buku pesanan, PIQ (Position In Queue). Purata merosakkan maklumat harga tepat.

9–11. Carta Pembalikan Jepun

Kagi, Line Break, dan Point & Figure: kaedah carta tanpa masa yang fokus semata-mata pada struktur harga.

Ini adalah kaedah carta Jepun tradisional (bersama Renko) yang membuang masa sepenuhnya dan memberi tumpuan kepada struktur harga.

9. Carta Kagi

Carta Kagi terdiri daripada garis menegak yang bertukar arah apabila harga berbalik sebanyak nilai tertentu. Garisan bertukar ketebalan apabila harga memecah tinggi sebelumnya (tebal = "yang" = permintaan) atau rendah sebelumnya (nipis = "yin" = bekalan).

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

Carta Line Break melukis garisan baru (kotak) hanya apabila harga penutup melebihi tinggi atau rendah N garisan sebelumnya (biasanya 3). Tiada garisan baru dilukis jika harga kekal dalam julat.

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

Carta Point & Figure (P&F) menggunakan lajur X (harga naik) dan O (harga turun). Pertukaran lajur memerlukan pembalikan biasanya 3 saiz kotak. Salah satu kaedah tertua menapis bunyi dan mengenal pasti sokongan/rintangan.

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, dan P&F dalam dagangan algoritma: Digunakan terutamanya untuk pengesanan trend jangka panjang dan pengenalpastian sokongan/rintangan. Sebagai lapisan penapis — "jangan ambil isyarat panjang apabila carta Kagi dalam mod yin" — mereka memberi nilai dengan menyelaraskan dagangan dengan struktur makro.

12–14. Bar Berpandukan Maklumat

Bar ketakseimbangan, bar larian, penapis CUSUM, dan bar entropi: persampelan apabila pasaran memberitahu kita sesuatu telah berubah.

Pendekatan paling canggih, dari buku Marcos Lopez de Prado Advances in Financial Machine Learning (2018). Pandangan teras: persampelan apabila maklumat baru tiba ke pasaran, bukan pada selang tetap.

12. Bar Ketakseimbangan Tik (TIB)

Jika pasaran dalam keseimbangan, dagangan yang dimulakan pembeli dan penjual seharusnya lebih kurang seimbang. Apabila ketakseimbangan melebihi jangkaan kita, sesuatu telah berubah. Persampelan bar pada ketika itu.

Setiap dagangan diklasifikasikan sebagai dimulakan pembeli (+1) atau dimulakan penjual (-1) menggunakan peraturan tik. Kami menjejak ketakseimbangan kumulatif θ dan persampelan apabila |θ| melebihi ambang dinamik.

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. Bar Ketakseimbangan Isipadu (VIB)

Lanjutan TIB: daripada mengira setiap dagangan sebagai ±1, berat mengikut isipadu bertanda. Pembelian 100-BTC menyumbang +100, jualan 1-BTC menyumbang -1. Menangkap pesanan besar yang bermaklumat yang mungkin dibahagikan kepada banyak dagangan kecil.

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

Masalah Letupan

Isu yang diketahui dengan bar ketakseimbangan: ambang berasaskan EWMA boleh memasuki gelung maklum balas positif. Penyelesaiannya: jepit dengan had min_ticks dan max_ticks.

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

14. Bar Larian

Bar larian menjejak panjang larian berarah semasa — urutan berturutan terpanjang pembelian atau penjualan. Apabila pedagang besar yang bermaklumat membahagikan pesanan kepada banyak dagangan kecil, urutan menjadi panjang luar biasa. Bar larian mengesannya.

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

Bar larian boleh dilanjutkan kepada larian isipadu dan larian dolar.

15. Bar Penapis CUSUM

Penapis CUSUM (Cumulative Sum) menentukan bila untuk persampelan dengan menjejak pulangan kumulatif. Tidak seperti bar ketakseimbangan (yang berfungsi pada dagangan mentah), CUSUM boleh digunakan pada data OHLCV 1m sedia ada — tiada data tik diperlukan.

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 + Kaedah Triple Barrier: Dalam rangka kerja Lopez de Prado, peristiwa CUSUM digunakan sebagai titik masuk untuk kaedah Triple Barrier — di mana setiap peristiwa mencetuskan dagangan dengan halangan henti rugi, ambil untung, dan tamat tempoh. Untuk pengesahan teguh strategi berasaskan peristiwa sedemikian, lihat Walk-Forward Optimization dan Monte Carlo Bootstrap for Backtesting.

16. Bar Entropi

Pendekatan paling elegan secara teori: persampelan apabila kandungan maklumat (entropi Shannon) siri harga dalam bar melebihi ambang.

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

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

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

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

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

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

        entropy = self._compute_entropy()

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

        return None

    def _compute_entropy(self) -> float:
        import math

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

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

        if not returns:
            return 0.0

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

        if max_r == min_r:
            return 0.0

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

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

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

        return entropy

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

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

Nota praktikal: Bar entropi mahal dari segi pengiraan dan terutamanya diminati dalam penyelidikan — tetapi untuk strategi berasaskan ML, ia menghasilkan ciri dengan sifat statistik yang lebih baik kerana setiap bar mengandungi lebih kurang "maklumat" yang sama.

17. Bar Delta (Aliran Pesanan)

Delta kumulatif: mengukur daya bersih pembeli agresif berbanding penjual dalam masa nyata.

Bar delta melakukan persampelan berdasarkan delta kumulatif — perbezaan berterusan antara isipadu beli dan isipadu jual. Tidak seperti bar ketakseimbangan (yang menggunakan tanda tik ±1), bar delta menggunakan aliran pesanan berwajaran isipadu sebenar.

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

Divergens delta: Salah satu isyarat paling berkuasa — harga naik sementara delta kumulatif negatif (penjual agresif tetapi harga masih naik, menunjukkan penyerapan had beli). Berkaitan langsung dengan pendekatan cap jari tingkah laku yang diterangkan dalam artikel Digital Fingerprint: Trader Identification. Untuk pembuat pasaran yang menggunakan model Avellaneda-Stoikov, bar delta memberikan pandangan masa nyata tentang risiko inventori dan tekanan agresor.

Penimbal pekeliling bar asas: data baru masuk, data lama keluar, dan lilin yang diagregatkan sentiasa sah.

Kaedah pengagregatan menentukan bagaimana bar asas digubah menjadi lilin tempoh masa lebih tinggi (HTF). Ia bebas daripada jenis bar — anda boleh menggunakan mana-mana kaedah pengagregatan pada mana-mana jenis bar asas.

Kaedah A: Pengagregatan Sejajar Kalender

Kumpulkan semua bar asas yang jatuh dalam sempadan kalender tetap. Lilin "1 jam" merangkumi semua bar dari 14:00:00 hingga 14:59:59.

Sifat:

• Semua peserta pasaran melihat sempadan yang sama — penting untuk analisis struktur pasaran, sokongan/rintangan, pencetus PIQ
• Masalah mula sejuk: lilin separa selepas mula semula
• Semulajadi untuk bar masa (ini yang bursa sediakan secara natif)
• Juga berfungsi untuk bar bukan masa: "semua bar isipadu yang ditutup antara 14:00 dan 15:00" = lilin jam sejajar kalender dari bar isipadu

Kaedah B: Pengagregatan Tetingkap Gelinding

Kumpulkan N bar asas tertutup terakhir, dikira semula pada setiap bar baru. Lilin gelinding "1 jam" = 60 bar masa 1-minit tertutup terakhir, dikemas kini setiap minit.

Unit atom adalah bar asas tertutup. Pilihan reka bentuk ini memberikan:

1. Tiada mula sejuk. Selepas N bar, lilin adalah sah. Tiada bunyi lilin separa.
2. Pariti backtesting. Jika dagangan langsung menggunakan unit atom yang sama seperti enjin backtesting, isyarat adalah identik.
3. Pengesahan mudah. Satu peraturan: 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

Pertukaran anjakan fasa: Lilin gelinding ditutup pada :37 jika anda bermula pada :37, bukan pada :00 seperti milik semua orang lain. Ini penting untuk strategi yang bergantung pada tahap yang kelihatan oleh orang ramai. Penyelesaiannya: gunakan kedua-duanya — kalender untuk struktur pasaran, gelinding untuk isyarat.

Kaedah C: Pengagregatan Gelinding Penyesuaian

Seperti gelinding, tetapi saiz tetingkap menyesuaikan diri dengan kemeruapan semasa. Pasaran tenang → tetingkap lebih lebar (lebih banyak penghalusan). Pasaran meruap → tetingkap lebih sempit (reaksi lebih pantas).

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

Setiap jenis bar asas boleh digabungkan dengan setiap kaedah pengagregatan. Sesetengah kombinasi adalah standard (bar masa kalender = apa yang bursa berikan kepada anda), yang lain adalah eksotik tetapi berkuasa.

Contoh Kombinasi

Enjin Hibrid: Kalender + Gelinding

Dalam amalan, anda mahukan pengagregatan kalender dan gelinding secara serentak. Overhead ingatan adalah boleh diabaikan — dua penimbal deque setiap tempoh masa setiap simbol.

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

Hibrid Masa-Isipadu: Kalender dengan Pemisahan Isipadu

Varian pengagregatan khas: lilin sejajar kalender yang menutup awal apabila isipadu melebihi ambang. Mengekalkan penyegerakan masa sambil menyesuaikan diri dengan lonjakan aktiviti.

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

Pengagregatan Praktikal: Pramuat Berantai

Pramuat berantai: menyusun lilin harian dari jam, dan jam dari minit — memintas had API.

Bursa mengehadkan berapa banyak data sejarah yang mereka sajikan. Binance memberikan ~1000 lilin setiap permintaan REST, OKX mengehadkan pada 300. Jika anda memerlukan lilin gelinding 1H (1440 minit), anda tidak selalu dapat mendapatkan sejarah 1m yang cukup. Untuk penstriman masa nyata dagangan dan buku pesanan melalui WebSocket, lihat CCXT Pro WebSocket Methods.

Penyelesaiannya: pengagregatan berantai — bina tempoh masa lebih tinggi dari resolusi tertinggi yang tersedia pada setiap kedalaman, kemudian gabungkannya.

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

Ini berfungsi kerana pengagregatan OHLCV boleh digubah: tinggi lilin 1H adalah maks 24 tinggi 1j, yang merupakan maks 1440 tinggi 1m.

Had Berbilang Bursa

Semakan Konsistensi Pengagregatan

Lilin 1j dari API REST mungkin tidak sepadan dengan apa yang anda kira dari 60 lilin 1m. Sentiasa sahkan:

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

Jika pengesahan gagal secara konsisten, sentiasa agregat dari 1m sendiri — jangan percaya lilin HTF bursa untuk pariti backtesting.

Matriks Perbandingan

Paksi 1: Jenis Bar Asas

Paksi 2: Kaedah Pengagregatan

Cadangan Praktikal

Seni bina lilin empat lapisan: isyarat gelinding, struktur kalender, aliran mikrostruktur, dan penapis trend.

Jika enjin backtesting anda berjalan pada data OHLCV 1m:

1. Bar masa gelinding — peningkatan paling mudah. Tiada data tambahan. Menghapuskan mula sejuk.
2. Bar masa hibrid (gelinding + kalender) — kalender untuk struktur pasaran, gelinding untuk isyarat.
3. Penapis CUSUM — berfungsi pada penutupan 1m, tiada data tik. "Sesuatu telah bergerak cukup untuk menarik perhatian."

Jika anda mempunyai data tik/dagangan:

4. Bar dolar + gelinding — lalai yang disyorkan dari literatur kewangan kuantitatif.
5. Bar ketakseimbangan isipadu + gelinding — mengesan aliran bermaklumat, persampelan lebih banyak semasa peristiwa penting.
6. Bar delta + kalender — jika anda mempunyai klasifikasi sisi agresor, pandangan paling langsung tentang siapa yang mendorong pasaran.

Sebagai penapis (gunakan Heikin-Ashi atau Line Break di atas mana-mana kombinasi asas+pengagregatan):

7. Heikin-Ashi ke atas bar isipadu gelinding — isyarat trend bersih pada data dinormalisasi aktiviti.
8. Line Break / Kagi ke atas bar kalender harian — penapis trend makro.

Untuk Marketmaker.cc khususnya — pendekatan berlapis:

• Lapisan 1 (isyarat): Pengagregatan gelinding bar masa untuk penunjuk dan isyarat masuk/keluar. Tiada mula sejuk, pariti backtesting sempurna.
• Lapisan 2 (struktur pasaran): Bar masa sejajar kalender untuk sokongan/rintangan, analisis penutupan jam, dan pencetus PIQ.
• Lapisan 3 (mikrostruktur): Bar ketakseimbangan isipadu + bar delta dari strim dagangan mentah untuk mengesan aliran bermaklumat, pemisahan pesanan, dan menjangka pergerakan besar. Lihat juga Digital Fingerprint: Trader Identification untuk pengecaman corak tingkah laku pada data aliran pesanan.
• Lapisan 4 (penapis trend): Transformasi Heikin-Ashi pada bar gelinding, atau Line Break pada penutupan kalender 4j, untuk menyelaraskan isyarat dengan arah makro.

Kesimpulan

Pembinaan lilin bukan satu pilihan tunggal — ia adalah dua keputusan bebas:

1. Jenis bar apa? Masa menangkap selang jam. Aktiviti (tik, isipadu, dolar) menangkap penyertaan pasaran. Harga (Renko, julat, kemeruapan) menangkap pergerakan. Maklumat (ketakseimbangan, larian, CUSUM, entropi) menangkap ketibaan maklumat baru. Aliran pesanan (delta) menangkap tekanan agresif.

2. Bagaimana untuk mengagregat ke tempoh masa lebih tinggi? Kalender sejajar dengan orang ramai. Gelinding menghapuskan mula sejuk. Penyesuaian bertindak balas terhadap kemeruapan.

"Lilin 1 jam standard dari Binance" hanyalah satu sel dalam matriks 17×3. 50 kombinasi lain tersedia bagi sesiapa yang sanggup melaksanakannya. Untuk sistem pengeluaran, jawapannya adalah "pilih kombinasi yang betul untuk setiap lapisan enjin keputusan anda."

Unit atom — bar asas tertutup — kekal sebagai asas. Segala-galanya adalah pengagregatan.

Untuk maklumat lanjut tentang ketepatan backtesting dengan data berbutiran halus, lihat Adaptive Drill-Down: Backtest with Variable Granularity. Untuk kesan pra-pengiraan penunjuk pada strategi berbilang tempoh masa, lihat Aggregated Parquet Cache.

Pautan Berguna

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 library implementing information-driven bars
4. Binance — Historical Market Data
5. Apache Parquet — columnar storage format

Petikan

@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 = {Klasifikasi dua paksi pembinaan lilin: 17 jenis bar asas × 3 kaedah pengagregatan = 51 kombinasi, dengan kod pelaksanaan dan cadangan praktikal untuk dagangan algo kripto.}
}