Các Loại Nến và Phương Pháp Tổng Hợp Cho Giao Dịch Thuật Toán

Mọi biểu đồ nến bạn từng thấy trên Binance, TradingView, hay bất kỳ giao diện sàn giao dịch nào đều được xây dựng theo cùng một cách: tổng hợp các giao dịch trong một khoảng thời gian cố định — 1 phút, 5 phút, 1 giờ — và tạo ra một thanh OHLCV. Điều này phổ biến đến mức hầu hết các trader không bao giờ đặt câu hỏi về nó. Nhưng đối với giao dịch thuật toán, việc lựa chọn loại thanh và phương pháp tổng hợp là hai quyết định độc lập — và hầu hết các hệ thống đều gộp chúng lại với nhau.

Bài viết này tách biệt hai trục trong xây dựng nến: loại thanh bạn xây dựng (17 loại) và cách bạn tổng hợp chúng thành khung thời gian cao hơn (3 phương pháp). Sự kết hợp cho 51 cấu hình có thể, mỗi cấu hình có các thuộc tính khác nhau cho backtesting, giao dịch trực tiếp và tạo tín hiệu.

Để biết giới thiệu về cách các giao dịch thô trở thành nến tiêu chuẩn, xem Nến Giao Dịch Được Giải Thích.

TL;DR

• Xây dựng nến có hai trục độc lập: loại thanh và phương pháp tổng hợp
• 17 loại thanh cơ bản: thời gian, tick, khối lượng, đô la, Renko, phạm vi, biến động, Heikin-Ashi, Kagi, Line Break, P&F, tick imbalance (TIB), volume imbalance (VIB), run, CUSUM, entropy, delta
• 3 phương pháp tổng hợp: theo lịch, cửa sổ cuộn, cuộn thích nghi
• 17 × 3 = 51 tổ hợp có thể, mỗi tổ hợp có thuộc tính khác nhau
• Hầu hết các hệ thống chỉ sử dụng một tổ hợp: thanh thời gian theo lịch. 50 tổ hợp còn lại chưa được khai thác.
• Khuyến nghị thực tiễn: sử dụng nhiều tổ hợp theo lớp — thanh thời gian cuộn cho tín hiệu, thanh thời gian theo lịch cho cấu trúc thị trường, thanh hướng thông tin cho vi cấu trúc

Hai Trục Xây Dựng Nến

Quan điểm truyền thống đặt tất cả các loại thanh trong một danh sách phẳng: thanh thời gian, thanh tick, thanh khối lượng, Renko, v.v. Điều này gây hiểu lầm. Thực ra có hai lựa chọn trực giao:

Trục 1 — Loại Thanh Cơ Bản (17 loại): Bạn quyết định khi nào một thanh mới đóng như thế nào? Sau một khoảng thời gian cố định? Sau N giao dịch? Sau một chuyển động giá? Khi nội dung thông tin thay đổi? Điều này xác định "một thanh" có nghĩa là gì.

Trục 2 — Phương Pháp Tổng Hợp (3 phương pháp): Bạn tổng hợp các thanh cơ bản thành nến khung thời gian cao hơn như thế nào? Căn chỉnh theo ranh giới lịch (00:00, 01:00, ...)? Sử dụng cửa sổ cuộn của N thanh cuối cùng? Điều chỉnh kích thước cửa sổ theo biến động?

Hai trục này độc lập nhau. Bạn có thể có:

• Thanh tick theo lịch — tổng hợp các thanh tick đóng giữa 14:00 và 14:59 thành một nến giờ duy nhất
• Thanh khối lượng cuộn — lấy 24 thanh khối lượng cuối cùng bất kể khi nào chúng đóng
• Thanh delta thích nghi — sử dụng cửa sổ điều khiển biến động trên các thanh delta

Nến "1 giờ" tiêu chuẩn chỉ là một điểm trong ma trận 17×3 này: thanh thời gian + căn chỉnh lịch. Mọi tổ hợp khác đều là một lựa chọn thay thế đáng xem xét.

1. Thanh Thời Gian (Tiêu Chuẩn)

Mật độ thông tin không đồng đều: ranh giới thời gian cứng nhắc đối xử với 200 giao dịch trong giờ yên tĩnh giống như 50.000 giao dịch trong giờ thông báo.

Mặc định. Một thanh mới hình thành sau một khoảng thời gian cố định: 1 phút, 5 phút, 1 giờ. Mọi sàn giao dịch đều cung cấp những điều này natively.

Thuộc tính:

• Trong phiên Châu Á (00:00–08:00 UTC), một nến 1 giờ có thể chứa 200 giao dịch. Trong thông báo niêm yết Binance, cùng một khoảng thời gian có thể chứa 50.000 giao dịch. Thanh thời gian đối xử cả hai như nhau. Phát hiện các đột biến hoạt động như vậy rất quan trọng cho bảo vệ bot — xem Phát Hiện Bất Thường Cho Bot Giao Dịch.
• Tất cả các thành viên thị trường thấy cùng ranh giới nến — một điểm Schelling. Điều này làm cho thanh thời gian cần thiết để phân tích hành vi đám đông.
• Các chỉ số được tính toán trên nến một phần (sau khi khởi động lại) tạo ra các giá trị rác.

from datetime import datetime

def time_until_valid_hourly_candle():
    """Bao lâu cho đến nến giờ hoàn chỉnh đầu tiên sau khi khởi động lại."""
    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. Thanh Dựa Trên Hoạt Động

Thanh tick, khối lượng và đô la: ba cách để để sự tham gia thị trường — không phải đồng hồ — xác định ranh giới thanh.

Thay vì lấy mẫu theo khoảng thời gian cố định, lấy mẫu sau một lượng hoạt động thị trường cố định. Điều này tạo ra các thanh có "nội dung thông tin" gần như bằng nhau bất kể giờ trong ngày.

2. Thanh Tick

Một thanh mới hình thành sau mỗi N giao dịch (tick). Trong hoạt động cao, các thanh hình thành nhanh chóng. Trong các giai đoạn yên tĩnh, một thanh duy nhất có thể kéo dài hàng giờ.

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

Ưu điểm: Tự nhiên thích nghi với hoạt động thị trường. Lợi nhuận từ thanh tick có xu hướng gần với phân phối chuẩn hơn so với lợi nhuận thanh thời gian — một thuộc tính cải thiện hiệu suất của nhiều mô hình thống kê.

Nhược điểm: Yêu cầu luồng giao dịch thô (không có sẵn từ tất cả các nhà cung cấp dữ liệu cho dữ liệu lịch sử). Thời điểm thanh là không thể đoán trước — bạn không thể nói "thanh tiếp theo sẽ đóng lúc X."

3. Thanh Khối Lượng

Một thanh mới hình thành sau khi N hợp đồng (hoặc đồng tiền, trong crypto) đã được giao dịch. Tương tự thanh tick, nhưng có trọng số theo kích thước giao dịch — một giao dịch 100 BTC đóng góp nhiều hơn 100 lần so với giao dịch 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. Thanh Đô La

Một thanh mới hình thành sau khi một giá trị danh nghĩa cố định (bằng USD/USDT) đã được trao đổi. Mạnh mẽ nhất trong các thanh dựa trên hoạt động vì nó chuẩn hóa cả số lượng giao dịch và mức giá.

Hãy xem xét: nếu ETH tăng từ $1.000 lên$4.000, bán $10.000 ETH cần 2,5 ETH ở$4.000 nhưng 10 ETH ở $1.000. Thanh khối lượng sẽ đối xử khác nhau; thanh đô la đối xử chúng như nhau.

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

Chọn Ngưỡng

Ngưỡng cho các thanh dựa trên hoạt động nên tạo ra khoảng cùng số lượng thanh mỗi ngày như các thanh thời gian bạn đang thay thế. Đối với BTCUSDT trên Binance:

Những con số này là xấp xỉ và thay đổi đáng kể theo chế độ thị trường. Trong một đợt tăng giá hoặc sụp đổ, các thanh dựa trên hoạt động sẽ tạo ra nhiều hơn 5-10 lần thanh so với bình thường — đó chính là điểm mấu chốt.

5–7. Thanh Dựa Trên Giá

Gạch Renko, thanh phạm vi và thanh biến động: chỉ lấy mẫu khi giá di chuyển đủ để quan trọng.

Các thanh dựa trên giá bỏ qua cả thời gian và hoạt động. Một thanh mới chỉ hình thành khi giá di chuyển một lượng cụ thể. Điều này tự nhiên lọc nhiễu ngang và làm nổi bật xu hướng.

5. Thanh Renko

Một "gạch" Renko mới hình thành khi giá đóng cửa di chuyển ít nhất N đơn vị so với giá đóng cửa của gạch trước. Các gạch luôn có cùng kích thước, tạo ra biểu diễn trực quan rõ ràng về hướng xu hướng.

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 Động sử dụng ATR (Average True Range) thay vì kích thước gạch cố định, tự động thích nghi với biến động.

6. Thanh Phạm Vi

Mỗi thanh có phạm vi cao-thấp cố định. Khi phạm vi bị vượt quá, thanh đóng và một thanh mới bắt đầu. Không giống Renko, thanh phạm vi bao gồm bóng và có thể hiển thị biến động trong thanh.

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

Sự khác biệt chính giữa Renko và thanh Phạm Vi: Renko chỉ theo dõi giá đóng cửa và hiển thị hướng; thanh phạm vi theo dõi toàn bộ phạm vi giá và hiển thị cấu trúc trong thanh. Thanh phạm vi thường hữu ích hơn cho giao dịch thuật toán vì chúng bảo tồn thông tin cao-thấp cần thiết để mô phỏng dừng lỗ và chốt lời.

7. Thanh Biến Động

Một thanh mới hình thành khi biến động trong thanh đạt đến ngưỡng động — ví dụ, bội số của ATR gần đây. Không giống thanh phạm vi (ngưỡng cố định), thanh biến động thích nghi với điều kiện thị trường.

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 (Biến Đổi Làm Mịn)

Heikin-Ashi: biến đổi trung bình chuyển đổi nến nhiễu thành tín hiệu xu hướng mượt mà — nhưng với cái giá là thông tin giá chính xác.

Heikin-Ashi (tiếng Nhật: "thanh trung bình") không phải là loại thanh — đó là một biến đổi có thể được áp dụng trên bất kỳ loại thanh cơ bản nào. Nó làm mịn nến bằng cách lấy trung bình các giá trị thanh hiện tại và trước:

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

Xu hướng xuất hiện dưới dạng chuỗi nến cùng màu không có bóng dưới (xu hướng tăng) hoặc không có bóng trên (xu hướng giảm).

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

Cảnh báo quan trọng cho backtesting: Giá Heikin-Ashi là tổng hợp. Nếu backtest của bạn sử dụng HA close làm giá vào lệnh, kết quả sẽ sai. Luôn sử dụng HA chỉ để tạo tín hiệu và thực hiện theo giá OHLC thực.

Khi nào HA hữu ích: Các chiến lược theo xu hướng cần tín hiệu "duy trì vào" sạch. Áp dụng HA trên bất kỳ loại thanh cơ bản nào — thanh thời gian, thanh khối lượng, thanh đô la — để lọc các giao cắt giả.

Khi nào HA có hại: Bất kỳ chiến lược nào cần mức giá chính xác — hỗ trợ/kháng cự, phân tích sổ lệnh, PIQ (Position In Queue). Việc lấy trung bình phá hủy thông tin giá chính xác.

9–11. Biểu Đồ Đảo Chiều Nhật Bản

Kagi, Line Break và Point & Figure: các phương pháp lập biểu đồ không có thời gian tập trung thuần túy vào cấu trúc giá.

Đây là các phương pháp lập biểu đồ Nhật Bản truyền thống (cùng với Renko) bỏ qua hoàn toàn thời gian và tập trung vào cấu trúc giá.

9. Biểu Đồ Kagi

Biểu đồ Kagi bao gồm các đường thẳng đứng thay đổi hướng khi giá đảo chiều một lượng cụ thể. Các đường thay đổi độ dày khi giá phá vỡ mức cao trước (dày = "yang" = cầu) hoặc mức thấp trước (mỏng = "yin" = cung).

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. Biểu Đồ Line Break

Biểu đồ Line Break vẽ một đường mới (hộp) chỉ khi giá đóng cửa vượt quá mức cao hoặc thấp của N đường trước (thường là 3). Không có đường mới nào được vẽ nếu giá ở trong phạm vi.

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. Biểu Đồ Point & Figure

Biểu đồ Point & Figure (P&F) sử dụng các cột X (giá tăng) và O (giá giảm). Chuyển đổi cột yêu cầu đảo chiều thường là 3 kích thước hộp. Một trong những phương pháp lâu đời nhất để lọc nhiễu và xác định hỗ trợ/kháng cự.

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 và P&F trong giao dịch thuật toán: Chủ yếu được sử dụng để phát hiện xu hướng dài hạn và xác định hỗ trợ/kháng cự. Như một lớp lọc — "đừng lấy tín hiệu mua khi biểu đồ Kagi đang ở chế độ yin" — chúng thêm giá trị bằng cách căn chỉnh giao dịch với cấu trúc vĩ mô.

12–14. Thanh Hướng Thông Tin

Thanh imbalance, thanh run, bộ lọc CUSUM và thanh entropy: lấy mẫu khi thị trường cho chúng ta biết điều gì đó đã thay đổi.

Cách tiếp cận tinh vi nhất, từ Advances in Financial Machine Learning (2018) của Marcos Lopez de Prado. Ý tưởng cốt lõi: lấy mẫu khi thông tin mới đến thị trường, không phải theo khoảng cố định.

12. Thanh Tick Imbalance (TIB)

Nếu thị trường ở trạng thái cân bằng, các giao dịch do người mua và người bán khởi xướng nên xấp xỉ cân bằng. Khi sự mất cân bằng vượt quá kỳ vọng của chúng ta, điều gì đó đã thay đổi. Lấy mẫu một thanh tại thời điểm đó.

Mỗi giao dịch được phân loại là do người mua khởi xướng (+1) hoặc người bán khởi xướng (-1) bằng quy tắc tick. Chúng ta theo dõi sự mất cân bằng tích lũy θ và lấy mẫu khi |θ| vượt quá ngưỡng động.

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. Thanh Volume Imbalance (VIB)

Mở rộng của TIB: thay vì đếm mỗi giao dịch là ±1, có trọng số theo khối lượng có dấu. Một lệnh mua 100 BTC đóng góp +100, một lệnh bán 1 BTC đóng góp -1. Nắm bắt các lệnh lớn có thông tin có thể được chia thành nhiều lệnh nhỏ.

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

Vấn Đề Bùng Nổ

Một vấn đề đã biết với các thanh imbalance: ngưỡng dựa trên EWMA có thể vào vòng phản hồi dương. Giải pháp: kẹp với các giới hạn min_ticks và 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. Thanh Run

Thanh run theo dõi độ dài của chuỗi định hướng hiện tại — chuỗi liên tiếp dài nhất của lệnh mua hoặc bán. Khi một trader lớn có thông tin chia một lệnh thành nhiều giao dịch nhỏ, chuỗi trở nên dài bất thường. Thanh run phát hiện điều này.

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

Thanh run có thể được mở rộng thành volume runs và dollar runs.

15. Thanh Bộ Lọc CUSUM

Bộ lọc CUSUM (Cumulative Sum) xác định khi nào lấy mẫu bằng cách theo dõi lợi nhuận tích lũy. Không giống các thanh imbalance (hoạt động trên giao dịch thô), CUSUM có thể được áp dụng cho dữ liệu OHLCV 1m hiện có — không cần dữ liệu tick.

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 + Phương Pháp Triple Barrier: Trong khung của Lopez de Prado, các sự kiện CUSUM được sử dụng như các điểm vào lệnh cho phương pháp Triple Barrier — nơi mỗi sự kiện kích hoạt một giao dịch với rào cản dừng lỗ, chốt lời và hết hạn. Để xác thực mạnh mẽ các chiến lược dựa trên sự kiện như vậy, xem Walk-Forward Optimization và Monte Carlo Bootstrap cho Backtesting.

16. Thanh Entropy

Cách tiếp cận thanh lịch nhất về mặt lý thuyết: lấy mẫu khi nội dung thông tin (entropy Shannon) của chuỗi giá trong thanh vượt quá ngưỡng.

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

Lưu ý thực tiễn: Thanh entropy tốn kém về mặt tính toán và chủ yếu là mối quan tâm nghiên cứu — nhưng đối với các chiến lược dựa trên ML, chúng tạo ra các đặc trưng có thuộc tính thống kê tốt hơn vì mỗi thanh chứa xấp xỉ "thông tin" bằng nhau.

17. Thanh Delta (Dòng Lệnh)

Delta tích lũy: đo lực thuần của người mua tích cực so với người bán theo thời gian thực.

Thanh delta lấy mẫu dựa trên delta tích lũy — sự khác biệt liên tục giữa khối lượng mua và khối lượng bán. Không giống các thanh imbalance (sử dụng dấu tick ±1), thanh delta sử dụng dòng lệnh theo trọng số khối lượng thực tế.

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

Phân kỳ delta: Một trong những tín hiệu mạnh mẽ nhất — giá tăng trong khi delta tích lũy âm (người bán tích cực nhưng giá vẫn tăng, cho thấy sự hấp thụ lệnh mua giới hạn). Trực tiếp liên quan đến cách tiếp cận nhận dạng hành vi được mô tả trong bài viết Dấu Vân Tay Số: Nhận Dạng Trader. Đối với các nhà tạo lập thị trường sử dụng mô hình Avellaneda-Stoikov, thanh delta cung cấp cái nhìn theo thời gian thực về rủi ro tồn kho và áp lực từ người tấn công.

Một vùng đệm vòng tròn của các thanh cơ bản: dữ liệu mới vào, dữ liệu cũ ra và nến tổng hợp luôn hợp lệ.

Các phương pháp tổng hợp xác định cách các thanh cơ bản được tổng hợp thành nến khung thời gian cao hơn (HTF). Chúng độc lập với loại thanh — bạn có thể áp dụng bất kỳ phương pháp tổng hợp nào cho bất kỳ loại thanh cơ bản nào.

Phương Pháp A: Tổng Hợp Theo Lịch

Tổng hợp tất cả các thanh cơ bản nằm trong ranh giới lịch cố định. Nến "1 giờ" bao gồm tất cả các thanh từ 14:00:00 đến 14:59:59.

Thuộc tính:

• Tất cả các thành viên thị trường thấy cùng ranh giới — cần thiết để phân tích cấu trúc thị trường, hỗ trợ/kháng cự, PIQ triggers
• Vấn đề khởi động lạnh: nến một phần sau khi khởi động lại
• Tự nhiên cho thanh thời gian (đây là những gì sàn giao dịch cung cấp natively)
• Cũng hoạt động cho các thanh không phải thời gian: "tất cả thanh khối lượng đóng giữa 14:00 và 15:00" = nến giờ theo lịch từ thanh khối lượng

Phương Pháp B: Tổng Hợp Cửa Sổ Cuộn

Tổng hợp N thanh cơ bản đã đóng cuối cùng, được tính lại trên mỗi thanh mới. Nến cuộn "1 giờ" = 60 thanh thời gian 1 phút đã đóng cuối cùng, được cập nhật mỗi phút.

Đơn vị nguyên tử là thanh cơ bản đã đóng. Lựa chọn thiết kế này cho:

1. Không có khởi động lạnh. Sau N thanh, nến hợp lệ. Không có nhiễu nến một phần.
2. Tương đương backtest. Nếu giao dịch trực tiếp sử dụng cùng đơn vị nguyên tử như engine backtest, các tín hiệu giống nhau.
3. Xác thực đơn giản. Một quy tắc: 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

Đánh đổi lệch pha: Nến cuộn đóng lúc :37 nếu bạn bắt đầu lúc :37, không phải lúc :00 như mọi người khác. Điều này quan trọng đối với các chiến lược phụ thuộc vào các mức có thể thấy bởi đám đông. Giải pháp: sử dụng cả hai — lịch cho cấu trúc thị trường, cuộn cho tín hiệu.

Phương Pháp C: Tổng Hợp Cuộn Thích Nghi

Giống như cuộn, nhưng kích thước cửa sổ thích nghi với biến động hiện tại. Thị trường bình tĩnh → cửa sổ rộng hơn (làm mịn nhiều hơn). Thị trường biến động → cửa sổ hẹp hơn (phản ứng nhanh hơn).

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

Mỗi loại thanh cơ bản có thể được kết hợp với mọi phương pháp tổng hợp. Một số tổ hợp là tiêu chuẩn (thanh thời gian theo lịch = những gì sàn giao dịch cung cấp), những tổ hợp khác là kỳ lạ nhưng mạnh mẽ.

Ví Dụ Tổ Hợp

Engine Lai: Lịch + Cuộn

Trong thực tế, bạn muốn cả tổng hợp lịch và cuộn đồng thời. Chi phí bộ nhớ là không đáng kể — hai bộ đệm deque mỗi khung thời gian mỗi ký hiệu.

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

Lai Thời Gian-Khối Lượng: Lịch với Chia Nhỏ Khối Lượng

Một biến thể tổng hợp đặc biệt: nến theo lịch bắt buộc đóng sớm khi khối lượng vượt quá ngưỡng. Duy trì đồng bộ thời gian trong khi thích nghi với các đột biến hoạt động.

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

Tổng Hợp Thực Tiễn: Tải Trước Xếp Tầng

Tải trước xếp tầng: tổng hợp nến ngày từ nến giờ, và nến giờ từ nến phút — bỏ qua giới hạn API.

Các sàn giao dịch giới hạn lượng dữ liệu lịch sử họ phục vụ. Binance cung cấp ~1000 nến mỗi yêu cầu REST, OKX giới hạn ở 300. Nếu bạn cần nến cuộn 1D (1440 phút), bạn không thể luôn luôn có đủ lịch sử 1m. Để stream giao dịch và sổ lệnh theo thời gian thực qua WebSocket, xem Phương Pháp WebSocket CCXT Pro.

Giải pháp: tổng hợp xếp tầng — xây dựng các khung thời gian cao hơn từ độ phân giải cao nhất có sẵn ở mỗi độ sâu, sau đó ghép chúng lại.

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

Điều này hoạt động vì tổng hợp OHLCV có thể tổng hợp được: mức cao của nến 1D là max của 24 mức cao 1h, đó là max của 1440 mức cao 1m.

Giới Hạn Đa Sàn

Kiểm Tra Tính Nhất Quán Tổng Hợp

Nến 1h từ REST API có thể không khớp với những gì bạn tính toán từ 60 nến 1m. Luôn xác thực:

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

Nếu xác thực liên tục thất bại, luôn tự tổng hợp từ 1m — không bao giờ tin tưởng nến HTF của sàn giao dịch cho tương đương backtest.

Ma Trận So Sánh

Trục 1: Các Loại Thanh Cơ Bản

Trục 2: Phương Pháp Tổng Hợp

Khuyến Nghị Thực Tiễn

Kiến trúc nến bốn lớp: tín hiệu cuộn, cấu trúc lịch, dòng lệnh vi cấu trúc và bộ lọc xu hướng.

Nếu engine backtest của bạn chạy trên dữ liệu OHLCV 1m:

1. Thanh thời gian cuộn — nâng cấp đơn giản nhất. Không cần thêm dữ liệu. Loại bỏ khởi động lạnh.
2. Thanh thời gian lai (cuộn + lịch) — lịch cho cấu trúc thị trường, cuộn cho tín hiệu.
3. Bộ lọc CUSUM — hoạt động trên đóng cửa 1m, không cần dữ liệu tick. "Điều gì đó đã di chuyển đủ để thú vị."

Nếu bạn có dữ liệu tick/giao dịch:

4. Thanh đô la + cuộn — mặc định được khuyến nghị từ tài liệu tài chính định lượng.
5. Thanh volume imbalance + cuộn — phát hiện dòng lệnh có thông tin, lấy mẫu nhiều hơn trong các sự kiện quan trọng.
6. Thanh delta + lịch — nếu bạn có phân loại phía tấn công, cái nhìn trực tiếp nhất về ai đang đẩy thị trường.

Như bộ lọc (áp dụng Heikin-Ashi hoặc Line Break trên bất kỳ tổ hợp base+aggregation nào):

7. Heikin-Ashi trên thanh khối lượng cuộn — tín hiệu xu hướng sạch trên dữ liệu chuẩn hóa hoạt động.
8. Line Break / Kagi trên nến ngày theo lịch — bộ lọc xu hướng vĩ mô.

Đối với Marketmaker.cc cụ thể — cách tiếp cận theo lớp:

• Lớp 1 (tín hiệu): Tổng hợp cuộn của thanh thời gian cho chỉ số và tín hiệu vào/ra. Không khởi động lạnh, tương đương backtest hoàn hảo.
• Lớp 2 (cấu trúc thị trường): Thanh thời gian theo lịch cho hỗ trợ/kháng cự, phân tích đóng cửa giờ và PIQ triggers.
• Lớp 3 (vi cấu trúc): Thanh volume imbalance + thanh delta từ luồng giao dịch thô để phát hiện dòng lệnh có thông tin, chia nhỏ lệnh và dự đoán các động thái lớn. Xem thêm Dấu Vân Tay Số: Nhận Dạng Trader để nhận dạng mẫu hành vi trên dữ liệu dòng lệnh.
• Lớp 4 (bộ lọc xu hướng): Biến đổi Heikin-Ashi trên thanh cuộn, hoặc Line Break trên đóng cửa 4h theo lịch, để giữ tín hiệu căn chỉnh với hướng vĩ mô.

Kết Luận

Xây dựng nến không phải là một lựa chọn duy nhất — đó là hai quyết định độc lập:

1. Loại thanh nào? Thời gian nắm bắt khoảng thời gian đồng hồ. Hoạt động (tick, khối lượng, đô la) nắm bắt sự tham gia thị trường. Giá (Renko, phạm vi, biến động) nắm bắt các chuyển động. Thông tin (imbalance, run, CUSUM, entropy) nắm bắt sự xuất hiện của thông tin mới. Dòng lệnh (delta) nắm bắt áp lực tích cực.

2. Cách tổng hợp thành khung thời gian cao hơn? Lịch căn chỉnh với đám đông. Cuộn loại bỏ khởi động lạnh. Thích nghi phản ứng với biến động.

Nến "1 giờ tiêu chuẩn từ Binance" chỉ là một ô trong ma trận 17×3. 50 tổ hợp khác có sẵn cho bất kỳ ai sẵn sàng triển khai chúng. Đối với một hệ thống sản xuất, câu trả lời là "chọn tổ hợp phù hợp cho mỗi lớp của engine quyết định của bạn."

Đơn vị nguyên tử — thanh cơ bản đã đóng — vẫn là nền tảng. Mọi thứ khác là tổng hợp.

Để biết thêm về độ chính xác backtest với dữ liệu chi tiết, xem Adaptive Drill-Down: Backtest với Độ Phân Giải Biến Đổi. Để biết ảnh hưởng của tính toán trước chỉ số trên các chiến lược đa khung thời gian, xem Bộ Đệm Parquet Tổng Hợp.

Liên Kết Hữu Ích

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 — Thư viện Python triển khai thanh hướng thông tin
4. Binance — Dữ Liệu Thị Trường Lịch Sử
5. Apache Parquet — định dạng lưu trữ theo cột

Trích Dẫn

@article{soloviov2026bartypes,
  author = {Soloviov, Eugen},
  title = {Bar Types and Aggregation Methods for Algorithmic Trading},
  year = {2026},
  url = {https://marketmaker.cc/en/blog/post/beyond-time-bars-candle-construction},
  description = {Two-axis classification of candle construction: 17 base bar types × 3 aggregation methods = 51 combinations, with implementation code and practical recommendations for crypto algotrading.}
}