Jenis Bar dan Metode Agregasi untuk Trading Algoritmik
Setiap grafik candlestick yang pernah Anda lihat di Binance, TradingView, atau antarmuka bursa mana pun dibangun dengan cara yang sama: agregat transaksi dalam jendela waktu tetap — 1 menit, 5 menit, 1 jam — dan hasilkan bar OHLCV. Ini begitu umum sehingga kebanyakan trader tidak pernah mempertanyakannya. Namun untuk trading algoritmik, pilihan jenis bar dan metode agregasi adalah dua keputusan independen — dan sebagian besar sistem mencampurkan keduanya.
Artikel ini memisahkan dua sumbu konstruksi candle: jenis bar apa yang Anda bangun (17 jenis) dan bagaimana Anda mengagregasikannya ke timeframe yang lebih tinggi (3 metode). Kombinasinya menghasilkan 51 konfigurasi yang mungkin, masing-masing dengan properti berbeda untuk backtesting, live trading, dan pembuatan sinyal.
Untuk pengenalan tentang bagaimana transaksi mentah menjadi candle standar, lihat Trading Candles Demystified.
TL;DR
- Konstruksi candle memiliki dua sumbu independen: jenis bar dan metode agregasi
- 17 jenis bar dasar: waktu, tick, volume, dolar, Renko, range, volatilitas, Heikin-Ashi, Kagi, Line Break, P&F, tick imbalance (TIB), volume imbalance (VIB), run, CUSUM, entropi, delta
- 3 metode agregasi: selaras kalender, rolling window, rolling adaptif
- 17 × 3 = 51 kombinasi yang mungkin, masing-masing dengan properti berbeda
- Sebagian besar sistem hanya menggunakan satu kombinasi: bar waktu selaras kalender. Lima puluh kombinasi lainnya belum dimanfaatkan.
- Rekomendasi praktis: gunakan beberapa kombinasi secara berlapis — bar waktu rolling untuk sinyal, bar waktu kalender untuk struktur pasar, bar berbasis informasi untuk mikrostruktur
Dua Sumbu Konstruksi Candle
Pandangan tradisional menempatkan semua jenis bar dalam daftar datar: bar waktu, bar tick, bar volume, Renko, dll. Ini menyesatkan. Sebenarnya ada dua pilihan yang ortogonal:
Sumbu 1 — Jenis Bar Dasar (17 jenis): Bagaimana Anda memutuskan kapan bar baru ditutup? Setelah interval waktu tetap? Setelah N transaksi? Setelah pergerakan harga? Ketika konten informasi berubah? Ini menentukan apa arti "satu bar."
Sumbu 2 — Metode Agregasi (3 metode): Bagaimana Anda menyusun bar dasar menjadi candle timeframe lebih tinggi? Selaraskan ke batas kalender (00:00, 01:00, ...)? Gunakan rolling window dari N bar terakhir? Sesuaikan ukuran window dengan volatilitas?
Kedua sumbu ini independen. Anda bisa memiliki:
- Bar tick selaras kalender — agregat bar tick yang ditutup antara pukul 14:00 dan 14:59 menjadi satu candle per jam
- Bar volume rolling — ambil 24 bar volume terakhir terlepas dari kapan ditutup
- Bar delta adaptif — gunakan window berbasis volatilitas pada bar delta
Candle "1 jam" standar hanyalah satu titik dalam matriks 17×3 ini: bar waktu + selaras kalender. Setiap kombinasi lainnya adalah alternatif yang patut dipertimbangkan.
1. Bar Waktu (Standar)
Kepadatan informasi yang tidak merata: batas waktu kaku memperlakukan jam sepi dengan 200 transaksi sama seperti jam pengumuman dengan 50.000 transaksi.
Default. Bar baru terbentuk setelah interval waktu tetap: 1 menit, 5 menit, 1 jam. Setiap bursa menyediakan ini secara native.
Properti:
- Selama sesi Asia (00:00–08:00 UTC), candle 1 jam mungkin berisi 200 transaksi. Selama pengumuman listing Binance, jendela yang sama bisa berisi 50.000 transaksi. Bar waktu memperlakukan keduanya sebagai setara. Mendeteksi lonjakan aktivitas seperti itu sangat penting untuk perlindungan bot — lihat Anomaly Detection for Trading Bots.
- Semua peserta pasar melihat batas candle yang sama — sebuah titik Schelling. Ini membuat bar waktu penting untuk menganalisis perilaku massa.
- Indikator yang dihitung pada candle parsial (setelah restart) menghasilkan nilai yang salah.
from datetime import datetime
def time_until_valid_hourly_candle():
"""Berapa lama sampai candle per jam pertama yang lengkap setelah 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 Berbasis Aktivitas
Bar tick, volume, dan dolar: tiga cara membiarkan partisipasi pasar — bukan jam — menentukan batas bar.
Alih-alih sampling pada interval waktu tetap, lakukan sampling setelah jumlah aktivitas pasar yang tetap. Ini menghasilkan bar dengan "konten informasi" yang kurang lebih sama terlepas dari waktu dalam sehari.
2. Bar Tick
Bar baru terbentuk setelah setiap N transaksi (tick). Selama aktivitas tinggi, bar terbentuk dengan cepat. Selama periode sepi, satu bar mungkin mencakup 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:
"""
Menghasilkan bar baru setiap `threshold` transaksi.
Setiap bar berisi jumlah "opini" pasar yang sama.
"""
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: Beradaptasi secara alami dengan aktivitas pasar. Distribusi imbal hasil dari bar tick cenderung lebih mendekati distribusi normal dibandingkan imbal hasil bar waktu — properti yang meningkatkan kinerja banyak model statistik.
Kekurangan: Membutuhkan stream transaksi mentah (tidak tersedia dari semua penyedia data untuk data historis). Waktu bar tidak dapat diprediksi — Anda tidak bisa mengatakan "bar berikutnya akan ditutup pada X."
3. Bar Volume
Bar baru terbentuk setelah N kontrak (atau koin, dalam kripto) diperdagangkan. Mirip dengan bar tick, namun diberi bobot berdasarkan ukuran transaksi — satu transaksi 100-BTC berkontribusi 100x lebih banyak daripada transaksi 1-BTC.
class VolumeBarGenerator:
"""
Menghasilkan bar baru setiap `threshold` unit volume.
Menormalisasi ukuran transaksi: satu order besar ≠ satu order kecil.
"""
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 setelah nilai nosional tetap (dalam USD/USDT) dipertukarkan. Yang paling robust di antara bar berbasis aktivitas karena menormalisasi baik jumlah transaksi maupun tingkat harga.
Pertimbangkan: jika ETH naik dari 4.000, menjual ETH senilai 4.000 tetapi 10 ETH pada harga $1.000. Bar volume akan memperlakukan ini secara berbeda; bar dolar memperlakukannya sama.
class DollarBarGenerator:
"""
Menghasilkan bar baru setiap `threshold` dolar (USDT) volume nosional.
Normalisasi paling robust: independen dari tingkat harga.
Lopez de Prado (2018) merekomendasikan bar dolar sebagai default
untuk sebagian besar aplikasi kuantitatif.
"""
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 Threshold
Threshold untuk bar berbasis aktivitas harus menghasilkan jumlah bar per hari yang kurang lebih sama dengan bar waktu yang Anda gantikan. Untuk BTCUSDT di Binance:
| Jenis Bar | Threshold Tipikal | ~Bar/Hari | TF Setara |
|---|---|---|---|
| Tick | 1.000 transaksi | ~1.400 | ~1m |
| Tick | 50.000 transaksi | ~28 | ~1h |
| Volume | 100 BTC | ~600 | ~2-3m |
| Volume | 2.400 BTC | ~25 | ~1h |
| Dolar | $1 juta | ~1.400 | ~1m |
| Dolar | $50 juta | ~28 | ~1h |
Angka-angka ini perkiraan dan berubah secara dramatis dengan rezim pasar. Selama rally atau crash, bar berbasis aktivitas akan menghasilkan 5-10x lebih banyak bar dari biasanya — yang memang merupakan tujuannya.
5–7. Bar Berbasis Harga
Bata Renko, bar range, dan bar volatilitas: sampling hanya ketika harga bergerak cukup jauh untuk bermakna.
Bar berbasis harga mengabaikan waktu maupun aktivitas. Bar baru terbentuk hanya ketika harga bergerak sebesar jumlah yang ditentukan. Ini secara alami menyaring noise sideways dan menyoroti tren.
5. Bar Renko
"Bata" Renko baru terbentuk ketika harga penutupan bergerak setidaknya N unit dari penutupan bata sebelumnya. Bata selalu berukuran sama, menciptakan representasi visual yang bersih dari arah tren.
class RenkoBarGenerator:
"""
Menghasilkan bata Renko berdasarkan pergerakan harga.
Properti utama: selama pergerakan sideways, tidak ada bata baru yang terbentuk.
Selama tren kuat, bata terbentuk dengan cepat.
"""
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 Dinamis menggunakan ATR (Average True Range) alih-alih ukuran bata tetap, beradaptasi dengan volatilitas secara otomatis.
6. Bar Range
Setiap bar memiliki rentang high-low tetap. Ketika rentang terlampaui, bar ditutup dan bar baru dimulai. Berbeda dengan Renko, bar range mencakup wick dan dapat menunjukkan volatilitas intra-bar.
class RangeBarGenerator:
"""
Menghasilkan bar dengan rentang high-low tetap.
Perbedaan dari Renko: bar range menampilkan OHLC lengkap dalam
rentang tersebut, bukan hanya arah bata. Lebih kaya informasi.
"""
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
Perbedaan utama antara Renko dan bar Range: Renko hanya melacak harga penutupan dan menunjukkan arah; bar range melacak rentang harga penuh dan menunjukkan struktur dalam bar. Bar range umumnya lebih berguna untuk trading algoritmik karena mempertahankan informasi high-low yang diperlukan untuk simulasi stop-loss dan take-profit.
7. Bar Volatilitas
Bar baru terbentuk ketika volatilitas intra-bar mencapai threshold dinamis — misalnya, kelipatan ATR terkini. Berbeda dengan bar range (threshold tetap), bar volatilitas beradaptasi dengan kondisi pasar.
class VolatilityBarGenerator:
"""
Menghasilkan bar ketika volatilitas intra-bar mencapai threshold.
Mirip dengan bar range, tetapi threshold beradaptasi dengan kondisi pasar
menggunakan ukuran ATR rolling. Di pasar tenang, bar memerlukan lebih sedikit
pergerakan absolut untuk ditutup; di pasar volatil, lebih banyak.
"""
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 Diperhalus)
Heikin-Ashi: rata-rata mengubah candle yang berisik menjadi sinyal tren yang halus — tetapi dengan biaya informasi harga yang tepat.
Heikin-Ashi (dalam bahasa Jepang berarti "bar rata-rata") bukan jenis bar — ini adalah transformasi yang dapat diterapkan di atas jenis bar dasar apa pun. Transformasi ini menghaluskan candle dengan merata-ratakan nilai bar saat ini dan sebelumnya:
- HA Close = (Open + High + Low + Close) / 4
- HA Open = (HA Open Sebelumnya + HA Close Sebelumnya) / 2
- HA High = max(High, HA Open, HA Close)
- HA Low = min(Low, HA Open, HA Close)
Tren tampak sebagai urutan candle warna yang sama tanpa wick bawah (uptrend) atau tanpa wick atas (downtrend).
class HeikinAshiTransformer:
"""
Mengubah candle OHLCV standar menjadi candle Heikin-Ashi.
Dapat diterapkan di atas JENIS bar APA PUN: bar waktu, bar volume,
bar rolling, dll. Ini adalah transformasi, bukan metode sampling.
PERINGATAN: Harga HA bersifat sintetis — tidak mewakili harga
yang benar-benar diperdagangkan. Jangan pernah gunakan HA close untuk
penempatan order atau perhitungan PnL. Gunakan HA hanya untuk
pembuatan sinyal, lalu eksekusi pada harga nyata.
"""
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]:
"""Transformasi seluruh seri. Reset state terlebih dahulu."""
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:
"""
Sinyal tren HA sederhana.
Mengembalikan:
+1: bullish (N candle HA hijau berturut-turut tanpa wick bawah)
-1: bearish (N candle HA merah berturut-turut tanpa wick atas)
0: tidak ada tren yang jelas
"""
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
Peringatan penting untuk backtesting: Harga Heikin-Ashi bersifat sintetis. Jika backtesting Anda menggunakan HA close sebagai harga entry, hasilnya akan salah. Selalu gunakan HA hanya untuk pembuatan sinyal dan eksekusi pada harga OHLC nyata.
Kapan HA berguna: Strategi mengikuti tren yang membutuhkan sinyal "tetap di dalam" yang bersih. Terapkan HA pada jenis bar dasar apa pun — bar waktu, bar volume, bar dolar — untuk menyaring crossover palsu.
Kapan HA merugikan: Strategi apa pun yang membutuhkan level harga yang tepat — support/resistance, analisis order book, PIQ (Position In Queue). Proses rata-rata menghancurkan informasi harga yang tepat.
9–11. Grafik Pembalikan Jepang
Kagi, Line Break, dan Point & Figure: metode charting bebas waktu yang berfokus murni pada struktur harga.
Ini adalah metode charting Jepang tradisional (bersama Renko) yang membuang waktu sepenuhnya dan berfokus pada struktur harga.
9. Grafik Kagi
Grafik Kagi terdiri dari garis vertikal yang mengubah arah ketika harga berbalik sebesar jumlah yang ditentukan. Garis berubah ketebalan ketika harga menembus high sebelumnya (tebal = "yang" = permintaan) atau low sebelumnya (tipis = "yin" = penawaran).
class KagiChartGenerator:
"""
Menghasilkan garis grafik Kagi berdasarkan pembalikan harga.
Berbeda dengan Renko (ukuran bata tetap), Kagi melacak besaran
aktual setiap pergerakan dan mengubah ketebalan garis pada titik breakout.
Berguna untuk mengidentifikasi breakout support/resistance dan
pergeseran penawaran/permintaan tanpa noise waktu.
"""
def __init__(self, reversal_amount: float = 10.0):
self.reversal_amount = reversal_amount
self.lines: list[dict] = []
self.current_direction: int = 0 # 1=naik, -1=turun
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' (tebal) atau 'yin' (tipis)
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. Grafik Line Break
Grafik line break menggambar garis baru (kotak) hanya ketika harga penutupan melampaui high atau low dari N garis sebelumnya (biasanya 3). Tidak ada garis baru yang digambar jika harga tetap dalam rentang tersebut.
class LineBreakGenerator:
"""
Menghasilkan bar Line Break (Three Line Break secara default).
Bar baru digambar hanya ketika close melampaui high atau low
dari N bar terakhir. Menyaring noise kecil dengan mensyaratkan harga
menembus rentang multi-bar.
Parameter 'N' (line_count) mengontrol sensitivitas:
- N=2: lebih sensitif, lebih banyak bar, lebih banyak noise
- N=3: standar (Three Line Break)
- N=4+: kurang sensitif, lebih sedikit bar, sinyal lebih kuat
"""
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. Grafik Point & Figure
Grafik Point & Figure (P&F) menggunakan kolom X (harga naik) dan O (harga turun). Pergantian kolom memerlukan pembalikan biasanya 3 ukuran kotak. Salah satu metode tertua untuk menyaring noise dan mengidentifikasi support/resistance.
class PointAndFigureGenerator:
"""
Menghasilkan data grafik Point & Figure.
Kolom X: harga naik dengan kelipatan box_size.
Kolom O: harga turun dengan kelipatan box_size.
Pergantian kolom: memerlukan pergerakan reversal_boxes * box_size
ke arah berlawanan.
Pengaturan klasik: box_size berdasarkan 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 trading algoritmik: Terutama digunakan untuk deteksi tren jangka panjang dan identifikasi support/resistance. Sebagai lapisan filter — "jangan ambil sinyal long ketika grafik Kagi dalam mode yin" — ketiganya menambah nilai dengan menyelaraskan transaksi dengan struktur makro.
12–14. Bar Berbasis Informasi
Bar imbalance, bar run, filter CUSUM, dan bar entropi: sampling ketika pasar memberi tahu kita bahwa sesuatu telah berubah.
Pendekatan paling canggih, dari buku Marcos Lopez de Prado Advances in Financial Machine Learning (2018). Wawasan intinya: lakukan sampling ketika informasi baru tiba ke pasar, bukan pada interval tetap.
12. Tick Imbalance Bars (TIB)
Jika pasar dalam keseimbangan, transaksi yang diprakarsai pembeli dan yang diprakarsai penjual seharusnya kurang lebih seimbang. Ketika ketidakseimbangan melampaui ekspektasi kita, sesuatu telah berubah. Sampel bar pada saat itu.
Setiap transaksi diklasifikasikan sebagai diprakarsai pembeli (+1) atau diprakarsai penjual (-1) menggunakan aturan tick. Kita melacak ketidakseimbangan kumulatif θ dan melakukan sampling ketika |θ| melampaui threshold dinamis.
class TickImbalanceBarGenerator:
"""
Menghasilkan bar ketika ketidakseimbangan tick kumulatif melampaui
tingkat yang diharapkan — yaitu, ketika "informasi baru" tiba.
Berdasarkan Lopez de Prado (2018), Bab 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:
"""Klasifikasikan transaksi sebagai beli (+1) atau jual (-1) menggunakan aturan tick."""
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)
Perluasan dari TIB: alih-alih menghitung setiap transaksi sebagai ±1, berikan bobot berdasarkan volume bertanda. Pembelian 100-BTC berkontribusi +100, penjualan 1-BTC berkontribusi -1. Menangkap order besar yang terinformasi yang mungkin dipecah menjadi banyak transaksi kecil.
class VolumeImbalanceBarGenerator:
"""
Seperti TIB, tetapi menggunakan volume bertanda alih-alih tick bertanda.
Menangkap wawasan bahwa sinyal beli 100-BTC 100x lebih
informatif daripada sinyal beli 1-BTC.
"""
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 Ledakan
Masalah yang diketahui dengan bar imbalance: threshold berbasis EWMA dapat masuk ke dalam loop umpan balik positif. Solusinya: klem dengan batas min_ticks dan max_ticks.
self.expected_ticks = max(
self.min_ticks, # Lantai: tidak pernah kurang dari 100 tick
min(
self.max_ticks, # Langit-langit: tidak pernah lebih dari 50000 tick
new_expected_ticks
)
)
14. Bar Run
Bar run melacak panjang run direktional saat ini — urutan berturut-turut terpanjang dari pembelian atau penjualan. Ketika trader besar yang terinformasi membagi ordernya menjadi banyak transaksi kecil, urutan menjadi sangat panjang. Bar run mendeteksi ini.
class TickRunBarGenerator:
"""
Menghasilkan bar ketika panjang run direktional melampaui ekspektasi.
Berdasarkan Lopez de Prado (2018), Bab 2.
Perbedaan dari bar imbalance:
- Bar imbalance melacak ketidakseimbangan NET (pembelian dikurangi penjualan)
- Bar run melacak panjang run MAKSIMUM (pembelian ATAU penjualan berturut-turut)
"""
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 run dapat diperluas menjadi volume run dan dollar run.
15. Bar Filter CUSUM
Filter CUSUM (Cumulative Sum) menentukan kapan melakukan sampling dengan melacak imbal hasil kumulatif. Berbeda dengan bar imbalance (yang bekerja pada transaksi mentah), CUSUM dapat diterapkan pada data OHLCV 1m yang sudah ada — tanpa memerlukan data tick.
class CUSUMFilterBarGenerator:
"""
Filter CUSUM simetris untuk sampling berbasis event.
Berdasarkan Lopez de Prado (2018), Bab 2.5.
Keunggulan utama dibanding Bollinger Bands: CUSUM memerlukan
run PENUH dengan magnitude threshold sebelum memicu. Bollinger Bands
memicu berulang kali ketika harga melayang di dekat band.
Dapat diterapkan pada data OHLCV 1m — tanpa data tick diperlukan.
"""
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 + Metode Triple Barrier: Dalam kerangka Lopez de Prado, event CUSUM digunakan sebagai titik entry untuk metode Triple Barrier — di mana setiap event memicu perdagangan dengan barrier stop-loss, take-profit, dan kedaluwarsa. Untuk validasi yang kuat dari strategi berbasis event seperti ini, lihat Walk-Forward Optimization dan Monte Carlo Bootstrap for Backtesting.
16. Bar Entropi
Pendekatan yang paling elegan secara teoritis: lakukan sampling ketika konten informasi (entropi Shannon) dari seri harga intra-bar melampaui threshold.
class EntropyBarGenerator:
"""
Menghasilkan bar ketika entropi imbal hasil intra-bar melampaui
threshold.
Berdasarkan teori informasi Shannon: bar disampling ketika
"informasi baru" tiba, diukur sebagai entropi distribusi
imbal hasil dalam bar saat ini.
Ini adalah bar berbasis informasi yang paling "murni" secara teoritis.
"""
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
Catatan praktis: Bar entropi secara komputasi mahal dan terutama merupakan kepentingan penelitian — tetapi untuk strategi berbasis ML, keduanya menghasilkan fitur dengan properti statistik yang lebih baik karena setiap bar berisi kurang lebih "informasi" yang sama.
17. Bar Delta (Order Flow)
Delta kumulatif: mengukur kekuatan net pembeli agresif vs penjual secara real time.
Bar delta melakukan sampling berdasarkan delta kumulatif — perbedaan berjalan antara volume beli dan volume jual. Berbeda dengan bar imbalance (yang menggunakan tanda tick ±1), bar delta menggunakan order flow berbobot volume aktual.
class DeltaBarGenerator:
"""
Menghasilkan bar berdasarkan delta order flow kumulatif.
Delta = Volume Beli - Volume Jual (diklasifikasikan berdasarkan sisi agressor).
Memerlukan data tingkat transaksi dengan klasifikasi sisi
(tersedia dari Binance aggTrades, Bybit trades, dll.)
"""
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
Divergensi delta: Salah satu sinyal paling kuat — harga naik sementara delta kumulatif negatif (penjual agresif tetapi harga tetap naik, menunjukkan absorpsi limit buy). Langsung relevan dengan pendekatan fingerprinting perilaku yang dijelaskan dalam artikel Digital Fingerprint: Trader Identification. Untuk market maker yang menggunakan model Avellaneda-Stoikov, bar delta memberikan pandangan real-time tentang risiko inventaris dan tekanan agressor.
Buffer sirkular dari bar dasar: data baru masuk, data lama keluar, dan candle yang diagregasi selalu valid.
Metode agregasi menentukan bagaimana bar dasar disusun menjadi candle timeframe lebih tinggi (HTF). Metode ini independen dari jenis bar — Anda dapat menerapkan metode agregasi apa pun pada jenis bar apa pun.
Metode A: Agregasi Selaras Kalender
Agregat semua bar dasar yang jatuh dalam batas kalender tetap. Candle "1 jam" mencakup semua bar dari 14:00:00 hingga 14:59:59.
Properti:
- Semua peserta pasar melihat batas yang sama — penting untuk analisis struktur pasar, support/resistance, pemicu PIQ
- Masalah cold start: candle parsial setelah restart
- Alami untuk bar waktu (inilah yang disediakan bursa secara native)
- Juga berfungsi untuk bar non-waktu: "semua bar volume yang ditutup antara 14:00 dan 15:00" = candle per jam selaras kalender dari bar volume
Metode B: Agregasi Rolling Window
Agregat N bar dasar terakhir yang ditutup, dihitung ulang setiap bar baru. Candle rolling "1 jam" = 60 bar waktu 1-menit terakhir yang ditutup, diperbarui setiap menit.
Unit atomik adalah bar dasar yang ditutup. Pilihan desain ini memberikan:
- Tidak ada cold start. Setelah N bar, candle valid. Tidak ada noise candle parsial.
- Paritas backtest. Jika live trading menggunakan unit atomik yang sama dengan mesin backtest, sinyal identik.
- Validasi sederhana. Satu aturan:
if buffer not full: skip.
import numpy as np
class RollingCandleAggregator:
"""
Menghasilkan candle timeframe lebih tinggi rolling dari bar dasar yang ditutup.
Bekerja dengan jenis bar APA PUN: bar waktu, bar tick, bar volume,
bar dolar, bar delta — apa pun yang menghasilkan output OHLCV.
Contoh: RollingCandleAggregator(window=60) dengan bar waktu 1m
menghasilkan candle "1h" yang diperbarui setiap menit.
Contoh: RollingCandleAggregator(window=24) dengan bar volume
menghasilkan candle yang mencakup 24 bar volume terakhir.
"""
def __init__(self, window: int):
self.window = window
self.buffer: deque[OHLCV] = deque(maxlen=window)
def push(self, bar: OHLCV) -> OHLCV | None:
"""
Tambahkan bar dasar yang ditutup. Mengembalikan candle yang diagregasi
hanya ketika buffer penuh (= candle 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
Trade-off pergeseran fase: Candle rolling ditutup pada :37 jika Anda mulai pada :37, bukan pada :00 seperti milik orang lain. Ini penting untuk strategi yang bergantung pada level yang terlihat oleh massa. Solusinya: gunakan keduanya — kalender untuk struktur pasar, rolling untuk sinyal.
Metode C: Agregasi Rolling Adaptif
Seperti rolling, tetapi ukuran window beradaptasi dengan volatilitas saat ini. Pasar tenang → window lebih lebar (lebih banyak smoothing). Pasar volatil → window lebih sempit (reaksi lebih cepat).
class AdaptiveRollingAggregator:
"""
Rolling window di mana ukuran window beradaptasi dengan volatilitas.
Bekerja dengan jenis bar dasar apa pun. Menggunakan ATR bar terkini
sebagai ukuran volatilitas.
Volatilitas rendah → window lebih lebar (lebih banyak smoothing, lebih sedikit sinyal)
Volatilitas tinggi → window lebih sempit (reaksi lebih cepat)
"""
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 dasar dapat dikombinasikan dengan setiap metode agregasi. Beberapa kombinasi bersifat standar (bar waktu selaras kalender = apa yang diberikan bursa), yang lain eksotis tetapi kuat.
Contoh Kombinasi
| Jenis Bar Dasar | Kalender | Rolling | Adaptif |
|---|---|---|---|
| Waktu | Candle bursa standar | HTF selalu valid, tanpa cold start | Timeframe adaptif volatilitas |
| Volume | "Semua bar volume jam ini" | 24 bar volume terakhir | Window lebih lebar di pasar tenang |
| Dolar | Agregat bar dolar per jam | N bar dolar terakhir | Window dolar adaptif |
| Tick Imbalance | Agregat imbalance per jam | N event imbalance terakhir | Reaksi cepat di rezim volatil |
| Delta | Order flow net per jam | Snapshot delta rolling | Window flow adaptif |
| Renko | "Bata jam ini" | N bata terakhir | Jumlah bata adaptif |
Mesin Hybrid: Kalender + Rolling
Dalam praktiknya, Anda menginginkan agregasi kalender dan rolling secara bersamaan. Overhead memori dapat diabaikan — dua buffer deque per timeframe per simbol.
class HybridCandleEngine:
"""
Mempertahankan candle selaras kalender dan rolling
untuk jenis bar dasar apa pun.
Candle kalender: untuk struktur pasar, support/resistance, PIQ.
Candle rolling: untuk indikator, pembuatan sinyal, entry/exit.
"""
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):
"""Proses jenis bar dasar apa pun — waktu, volume, tick, delta, dll."""
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] = []
Hybrid Waktu-Volume: Kalender dengan Pemisahan Volume
Varian agregasi khusus: candle selaras kalender yang menutup lebih awal paksa ketika volume melampaui threshold. Mempertahankan sinkronisasi waktu sambil beradaptasi dengan lonjakan aktivitas.
class TimeVolumeHybridGenerator:
"""
Candle selaras kalender yang memisah saat volume melonjak.
Aturan: tutup candle pada batas kalender ATAU ketika
volume yang terakumulasi melampaui vol_threshold, mana yang lebih dulu.
Bekerja dengan jenis bar dasar apa pun — pemicu volume menambahkan
dimensi pemisahan ekstra di atas penyelarasan kalender.
"""
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
Agregasi Praktis: Cascading Preload
Cascading preload: menyusun candle harian dari per jam, dan per jam dari per menit — melewati batas API.
Bursa membatasi berapa banyak data historis yang mereka layani. Binance memberikan ~1000 candle per permintaan REST, OKX dibatasi hingga 300. Jika Anda membutuhkan candle rolling 1D (1440 menit), Anda tidak selalu bisa mendapatkan cukup riwayat 1m. Untuk streaming real-time transaksi dan order book melalui WebSocket, lihat CCXT Pro WebSocket Methods.
Solusinya: agregasi bertingkat — bangun timeframe lebih tinggi dari resolusi tertinggi yang tersedia pada setiap kedalaman, lalu gabungkan.
Candle rolling 1W:
├── 6 candle 1D yang telah selesai ← ambil dari REST /klines?interval=1d
├── 1 hari parsial:
│ ├── 23 candle 1h yang telah selesai ← ambil dari REST /klines?interval=1h
│ └── 1 jam parsial:
│ └── N candle 1m yang telah selesai ← ambil dari REST /klines?interval=1m
└── Live: setiap candle 1m yang ditutup memperbarui seluruh rantai
Ini berfungsi karena agregasi OHLCV bersifat composable: high dari candle 1D adalah max dari 24 high 1h, yang merupakan max dari 1440 high 1m.
Batas Multi-Bursa
| Bursa | Maks Candle 1m | Maks Candle 1h | Interval Terkenal |
|---|---|---|---|
| Binance | 1.000 | 1.000 | 1m–1M, rentang penuh |
| Bybit | 1.000 | 1.000 | 1–720, D/W/M |
| OKX | 300 | 300 | 1m–1M (lebih restriktif) |
| Gate.io | 1.000 | 1.000 | 10d–30d |
Pemeriksaan Konsistensi Agregasi
Candle 1h dari REST API mungkin tidak cocok dengan yang akan Anda hitung dari 60 candle 1m. Selalu validasi:
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 validasi gagal secara konsisten, selalu agregat dari 1m sendiri — jangan pernah mempercayai candle HTF bursa untuk paritas backtesting.
Matriks Perbandingan
Sumbu 1: Jenis Bar Dasar
| # | Jenis Bar | Pemicu | Data Tick Diperlukan | Terbaik Untuk |
|---|---|---|---|---|
| 1 | Waktu | Interval tetap | Tidak | Struktur pasar, perilaku massa |
| 2 | Tick | N transaksi | Ya | Fitur ML, sampling opini setara |
| 3 | Volume | N unit diperdagangkan | Ya | Analisis aktivitas ternormalisasi |
| 4 | Dolar | $N nosional | Ya | Perbandingan lintas aset |
| 5 | Renko | Harga ± N unit | Tidak | Mengikuti tren, penyaringan noise |
| 6 | Range | High-Low ≥ N | Ya | Deteksi breakout |
| 7 | Volatilitas | Range adaptif | Ya | Analisis adaptif rezim |
| 8 | Heikin-Ashi | Transformasi | Tidak | Konfirmasi tren (harga sintetis!) |
| 9 | Kagi | Pembalikan harga | Tidak | Struktur penawaran/permintaan |
| 10 | Line Break | Breakout N-garis | Tidak | Filter tren makro |
| 11 | Point & Figure | Kotak + pembalikan | Tidak | Pemetaan support/resistance |
| 12 | TIB | Ketidakseimbangan tick | Ya | Deteksi aliran terinformasi |
| 13 | VIB | Ketidakseimbangan volume | Ya | Deteksi order besar |
| 14 | Run | Panjang run | Ya | Deteksi pemisahan order |
| 15 | CUSUM | Imbal hasil kumulatif | Tidak (penutupan 1m) | Event structural break |
| 16 | Entropi | Entropi Shannon | Ya | Riset ML, kemurnian fitur |
| 17 | Delta | Delta order flow | Ya (aggTrades) | Analisis aliran agressor |
Sumbu 2: Metode Agregasi
| Metode | Penyelarasan | Cold Start | Pergeseran Fase | Terbaik Untuk |
|---|---|---|---|---|
| Kalender | Jam dinding | Risiko bar parsial | Tidak ada (selaras massa) | Struktur pasar, PIQ, S/R |
| Rolling | N bar | Tidak ada (setelah pemanasan) | Ya (bergeser dari :00) | Indikator, sinyal |
| Adaptif | N berbasis volatilitas | Setelah kalibrasi ATR | Ya | Strategi adaptif volatilitas |
Rekomendasi Praktis
Arsitektur candle empat lapis: sinyal rolling, struktur kalender, aliran mikrostruktur, dan filter tren.
Jika mesin backtest Anda berjalan pada data OHLCV 1m:
- Bar waktu rolling — peningkatan paling sederhana. Tidak ada data tambahan. Menghilangkan cold start.
- Bar waktu hybrid (rolling + kalender) — kalender untuk struktur pasar, rolling untuk sinyal.
- Filter CUSUM — bekerja pada penutupan 1m, tanpa data tick. "Sesuatu bergerak cukup untuk menarik perhatian."
Jika Anda memiliki data tick/transaksi:
- Bar dolar + rolling — default yang direkomendasikan dari literatur keuangan kuantitatif.
- Bar volume imbalance + rolling — mendeteksi aliran terinformasi, lebih banyak sampling selama event penting.
- Bar delta + kalender — jika Anda memiliki klasifikasi sisi agressor, pandangan paling langsung tentang siapa yang mendorong pasar.
Sebagai filter (terapkan Heikin-Ashi atau Line Break di atas kombinasi dasar+agregasi apa pun):
- Heikin-Ashi atas bar volume rolling — sinyal tren bersih pada data ternormalisasi aktivitas.
- Line Break / Kagi atas bar kalender harian — filter tren makro.
Untuk Marketmaker.cc secara spesifik — pendekatan berlapis:
- Lapis 1 (sinyal): Agregasi rolling bar waktu untuk indikator dan sinyal entry/exit. Tanpa cold start, paritas backtest sempurna.
- Lapis 2 (struktur pasar): Bar waktu selaras kalender untuk support/resistance, analisis penutupan per jam, dan pemicu PIQ.
- Lapis 3 (mikrostruktur): Bar volume imbalance + bar delta dari stream transaksi mentah untuk mendeteksi aliran terinformasi, pemisahan order, dan mengantisipasi pergerakan besar. Lihat juga Digital Fingerprint: Trader Identification untuk pengenalan pola perilaku pada data order flow.
- Lapis 4 (filter tren): Transformasi Heikin-Ashi pada bar rolling, atau Line Break pada penutupan kalender 4h, untuk menjaga sinyal selaras dengan arah makro.
Kesimpulan
Konstruksi candle bukan pilihan tunggal — ini adalah dua keputusan independen:
-
Jenis bar apa? Waktu menangkap interval jam. Aktivitas (tick, volume, dolar) menangkap partisipasi pasar. Harga (Renko, range, volatilitas) menangkap pergerakan. Informasi (imbalance, run, CUSUM, entropi) menangkap kedatangan informasi baru. Order flow (delta) menangkap tekanan agresif.
-
Bagaimana mengagregasi ke timeframe lebih tinggi? Kalender selaras dengan massa. Rolling menghilangkan cold start. Adaptif bereaksi terhadap volatilitas.
"Candle 1 jam dari Binance" standar hanyalah satu sel dalam matriks 17×3. Lima puluh kombinasi lainnya tersedia bagi siapa saja yang mau mengimplementasikannya. Untuk sistem produksi, jawabannya adalah "pilih kombinasi yang tepat untuk setiap lapis mesin keputusan Anda."
Unit atomik — bar dasar yang ditutup — tetap menjadi fondasi. Segala sesuatu yang lain adalah agregasi.
Untuk informasi lebih lanjut tentang akurasi backtest dengan data granular, lihat Adaptive Drill-Down: Backtest with Variable Granularity. Untuk dampak prakomputasi indikator pada strategi multi-timeframe, lihat Aggregated Parquet Cache.
Tautan Berguna
- Lopez de Prado — Advances in Financial Machine Learning (2018)
- Easley, Lopez de Prado, O'Hara — The Volume Clock: Insights into the High Frequency Paradigm (2012)
- mlfinlab — Pustaka Python yang mengimplementasikan bar berbasis informasi
- Binance — Historical Market Data
- Apache Parquet — format penyimpanan kolumnar
Kutipan
@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 sumbu konstruksi candle: 17 jenis bar dasar × 3 metode agregasi = 51 kombinasi, dengan kode implementasi dan rekomendasi praktis untuk algotrading kripto.}
}
Penulis
Trading-systems engineer
Trading-systems engineer building bots since 2017: cross-exchange arbitrage (connected up to 30 venues), cointegration-based pairs arbitrage across spot and futures, scalping, news and sentiment-driven strategies, trend algorithms, and portfolio management and balancing algorithms. Also builds sub-millisecond order execution, big-data warehouses, backtesting engines, AI agents, and trading interfaces (incl. open-source profitmaker.cc). Stack: JS/TS, Python, Rust/Zig/Go, DevOps, backend, frontend, architecture.
Baca Selengkapnya
VectorBT: Framework Backtesting Tercepat untuk Python
OneTick: Platform Tempat Bursa Menangkap Spoofer dan Hedge Fund Memburu Alpha
