Алгоритмдік трейдинг үшін бар түрлері мен агрегациялау әдістері

Binance, TradingView немесе кез келген биржа интерфейсінде сіз көрген кез келген шамдар графигі бір ғана тәсілмен құрылады: белгіленген уақыт терезесінде — 1 минут, 5 минут, 1 сағат — сауда-саттықтарды агрегациялап, OHLCV барын шығару. Бұл соншалықты кең таралған, көптеген трейдерлер оны ешқашан сұрақ астына алмайды. Бірақ алгоритмдік трейдинг үшін бар түрін және агрегациялау әдісін таңдау — екі тәуелсіз шешім, ал көптеген жүйелер оларды араластырады.

Бұл мақала шам құрылымының екі осін бөліп қарастырады: қандай түрдегі бар құрасыз (17 түр) және оларды жоғарырақ уақыт кезеңдеріне қалай агрегациялайсыз (3 әдіс). Комбинация 51 мүмкін конфигурацияны береді, олардың әрқайсысының бэктестинг, тірі сауда және сигнал генерациялау үшін әртүрлі қасиеттері бар.

Шикі сауда-саттықтардың стандартты шамдарға қалай айналатыны туралы кіріспе үшін қараңыз Trading Candles Demystified.

TL;DR

• Шам құрылымында екі тәуелсіз ось бар: бар түрі мен агрегациялау әдісі
• 17 негізгі бар түрі: уақыт, тик, көлем, доллар, Ренко, диапазон, волатильділік, Хейкин-Аши, Каги, Line Break, P&F, тик теңсіздігі (TIB), көлем теңсіздігі (VIB), серия, CUSUM, энтропия, дельта
• 3 агрегациялау әдісі: күнтізбеге тураланған, жылжымалы терезе, бейімделгіш жылжымалы
• 17 × 3 = 51 мүмкін комбинация, әрқайсысының қасиеттері әртүрлі
• Көптеген жүйелер тек бір комбинацияны қолданады: күнтізбеге тураланған уақыт барлары. Қалған 50-і пайдаланылмаған.
• Практикалық ұсыныс: бірнеше комбинацияны қабаттар түрінде қолданыңыз — сигналдар үшін жылжымалы уақыт барлары, нарық құрылымы үшін күнтізбелік уақыт барлары, микроқұрылым үшін ақпаратқа негізделген барлар

Шам құрылымының екі осі

Дәстүрлі көзқарас барлық бар түрлерін тегіс тізімге қояды: уақыт барлары, тик барлары, көлем барлары, Ренко және т.б. Бұл жаңылыстырады. Шын мәнінде екі ортогональ таңдау бар:

Ось 1 — негізгі бар түрі (17 түр): Жаңа бар қашан жабылатынын қалай шешесіз? Белгіленген уақыт аралығынан кейін ме? N сауда-саттықтан кейін ме? Баға қозғалысынан кейін ме? Ақпараттық мазмұн өзгергенде ме? Бұл "бір бар" деген не екенін анықтайды.

Ось 2 — агрегациялау әдісі (3 әдіс): Негізгі барларды жоғары уақыт кезеңіндегі шамдарға қалай құрастырасыз? Күнтізбелік шекараларға (00:00, 01:00, ...) тураластырасыз ба? Соңғы N бардың жылжымалы терезесін қолданасыз ба? Терезе өлшемін волатильділікке бейімдейсіз бе?

Бұл екі ось тәуелсіз. Сізде мыналар болуы мүмкін:

• Күнтізбеге тураланған тик барлары — 14:00 мен 14:59 арасында жабылған тик барларын бір сағаттық шамға агрегациялау
• Жылжымалы көлем барлары — олар қашан жабылғанына қарамастан соңғы 24 көлем барын алу
• Бейімделгіш дельта барлары — дельта барларына волатильділікке негізделген терезені қолдану

Стандартты "1 сағаттық шам" бұл 17×3 матрицадағы бір ғана нүкте: уақыт барлары + күнтізбелік тураланым. Басқа әрбір комбинация қарастыруға тұрарлық балама.

1. Уақыт барлары (стандартты)

Біркелкі емес ақпараттық тығыздық: қатаң уақыт шекаралары 200 сауда-саттықты тыныш сағаттарды 50,000 сауда-саттықты хабарландыру сағаттарымен бірдей етіп қарайды.

Әдепкі. Белгіленген уақыт аралығынан кейін жаңа бар пайда болады: 1 минут, 5 минут, 1 сағат. Әрбір биржа бұларды өздігінен ұсынады.

Қасиеттері:

• Азиялық сессия (00:00–08:00 UTC) кезінде 1 сағаттық шамда 200 сауда-саттық болуы мүмкін. Binance листинг хабарландыруы кезінде дәл сол терезеде 50,000 сауда-саттық болуы мүмкін. Уақыт барлары екеуін бірдей санайды. Мұндай белсенділік секірулерін анықтау боттарды қорғау үшін маңызды — қараңыз Anomaly Detection for Trading Bots.
• Барлық нарық қатысушылары бір шам шекараларын көреді — Шеллинг нүктесі. Бұл уақыт барларын көпшіліктің мінез-құлқын талдау үшін маңызды етеді.
• Ішінара шамдарда (қайта іске қосудан кейін) есептелген индикаторлар қате мәндер шығарады.

from datetime import datetime

def time_until_valid_hourly_candle():
    """How long until the first complete hourly candle after restart."""
    now = datetime.utcnow()
    minutes_into_hour = now.minute
    seconds_into_minute = now.second

    wait_seconds = (60 - minutes_into_hour) * 60 - seconds_into_minute
    wait_seconds += 3600

    return wait_seconds

2–4. Белсенділікке негізделген барлар

Тик, көлем және доллар барлары: бар шекараларын анықтауда сағаттың орнына нарықтық қатысуды пайдаланудың үш жолы.

Белгіленген уақыт аралықтарында іріктеудің орнына, белгіленген мөлшердегі нарықтық белсенділіктен кейін іріктеу жасаңыз. Бұл күннің қай уақыты болса да шамамен бірдей "ақпараттық мазмұны" бар барларды береді.

2. Тик барлары

Әрбір N сауда-саттықтан (тиктерден) кейін жаңа бар қалыптасады. Жоғары белсенділік кезінде барлар тез қалыптасады. Тыныш кезеңдерде бір бар сағаттарға созылуы мүмкін.

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

Артықшылықтары: Нарықтық белсенділікке табиғи түрде бейімделеді. Тик барларынан алынған кірістер уақыт барлары кірістеріне қарағанда қалыпты үлестіруге жақынырақ болады — бұл көптеген статистикалық модельдердің өнімділігін жақсартатын қасиет.

Кемшіліктері: Шикі сауда-саттық ағынын талап етеді (тарихи деректер үшін барлық деректер провайдерлерінде қолжетімді емес). Бар уақыты болжанбайды — келесі бар нақты X уақытында жабылады деп айта алмайсыз.

3. Көлем барлары

N келісім-шарт (немесе криптода тиын) сауда-саттыққа түскеннен кейін жаңа бар пайда болады. Тик барларына ұқсас, бірақ сауда көлеміне қарай өлшенген — бір 100-BTC мәмілесі 1-BTC мәмілесінен 100 есе көп үлес қосады.

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. Доллар барлары

Белгіленген номиналды құн (USD/USDT-де) алмасқаннан кейін жаңа бар пайда болады. Белсенділікке негізделген барлардың ішінде ең берік, себебі ол сауда санын да, баға деңгейін де қалыпқа келтіреді.

Ойлап көріңіз: егер ETH $1,000-нан$4,000-ге дейін көтерілсе, $10,000 тұратын ETH сату үшін$4,000 бағамен 2.5 ETH керек, ал $1,000 бағамен 10 ETH керек. Көлем барлары бұларды әртүрлі қарайды; доллар барлары бұларды бірдей қарайды.

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

Шектеуді таңдау

Белсенділікке негізделген барлар үшін шектеу сіз ауыстырып жатқан уақыт барларымен шамамен бірдей күндік бар санын беруі керек. Binance-тегі BTCUSDT үшін:

Бұл сандар шамамен алынған және нарық режиміне қарай күрт өзгереді. Ралли немесе құлдырау кезінде белсенділікке негізделген барлар әдеттегіден 5-10 есе көп бар шығарады — бұл дәл керек нәрсе.

5–7. Бағаға негізделген барлар

Ренко кірпіштері, диапазон барлары және волатильділік барлары: баға маңызды дәрежеде қозғалғанда ғана іріктеу.

Бағаға негізделген барлар уақытты да, белсенділікті де ескермейді. Жаңа бар тек баға көрсетілген мөлшерде қозғалғанда ғана пайда болады. Бұл табиғи түрде бүйірлік шуды сүзеді және трендтерді ерекшелейді.

5. Ренко барлары

Жаңа Ренко "кірпіші" жабылу бағасы алдыңғы кірпіштің жабылуынан кемінде N бірлікке қозғалғанда пайда болады. Кірпіштер әрқашан бірдей өлшемде болады, бұл тренд бағытының таза визуалды көрінісін жасайды.

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

Динамикалық Ренко белгіленген кірпіш өлшемінің орнына ATR (орташа шынайы диапазон) қолданады, волатильділікке автоматты түрде бейімделеді.

6. Диапазон барлары

Әрбір бардың белгіленген жоғары-төмен диапазоны бар. Диапазон асып кеткенде бар жабылады, жаңасы басталады. Ренкодан айырмашылығы, диапазон барларында білектер (wicks) болады және бар ішіндегі волатильділікті көрсете алады.

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

Ренко мен диапазон барларының негізгі айырмашылығы: Ренко тек жабылу бағаларын қадағалайды және бағытты көрсетеді; диапазон барлары толық баға ауқымын қадағалайды және бар ішіндегі құрылымды көрсетеді. Диапазон барлары алгоритмдік трейдинг үшін жалпы алғанда пайдалырақ, себебі олар стоп-лосс пен тейк-профит симуляциясы үшін қажетті жоғары-төмен ақпаратты сақтайды.

7. Волатильділік барлары

Жаңа бар бар ішіндегі волатильділік динамикалық шектіге жеткенде пайда болады — мысалы, соңғы ATR-дің еселігіне. Диапазон барларынан (белгіленген шекті) айырмашылығы, волатильділік барлары нарық жағдайларына бейімделеді.

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. Хейкин-Аши (тегістелген түрлендіру)

Хейкин-Аши: орташалау шулы шамдарды тегіс тренд сигналдарына айналдырады — бірақ дәл баға ақпаратының есебінен.

Хейкин-Аши (жапон тілінде "орташа бар") бар түрі емес — бұл кез келген негізгі бар түрінің үстіне қолданылатын түрлендіру. Ол ағымдағы және алдыңғы бар мәндерін орташалау арқылы шамдарды тегістейді:

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

Трендтер төменгі білексіз (өсу трендінде) немесе жоғарғы білексіз (құлдырау трендінде) бірдей түсті шамдардың тізбегі ретінде көрінеді.

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

Бэктестинг үшін маңызды ескерту: Хейкин-Аши бағалары синтетикалық. Егер бэктестіңіз кіру бағасы ретінде HA close қолданса, нәтижелер қате болады. HA-ны әрқашан тек сигнал генерациялау үшін қолданыңыз, ал орындауды нақты OHLC бағаларымен жасаңыз.

HA қашан пайдалы: Таза "ұстап тұру" сигналдарын қажет ететін трендке ілесу стратегиялары. HA-ны кез келген негізгі бар түрінің — уақыт, көлем, доллар барларының — үстіне қолданыңыз, жалған кесіп өтулерді сүзу үшін.

HA қашан зиянды: Нақты баға деңгейлерін қажет ететін кез келген стратегия — қолдау/қарсылық, тапсырыстар кітабын талдау, PIQ (Position In Queue). Орташалау дәл баға ақпаратын жояды.

9–11. Жапон реверсия графиктері

Каги, Line Break және Point & Figure: тек баға құрылымына назар аударатын уақытсыз графика әдістері.

Бұл (Ренкомен қатар) уақытты толығымен алып тастап, баға құрылымына назар аударатын дәстүрлі жапон графика әдістері.

9. Каги графиктері

Каги графиктері баға белгіленген мөлшерге кері бұрылғанда бағытын өзгертетін тік сызықтардан тұрады. Баға алдыңғы жоғары деңгейді (қалың = "ян" = сұраныс) немесе алдыңғы төмен деңгейді (жіңішке = "инь" = ұсыныс) бұзғанда сызықтар қалыңдығын өзгертеді.

class KagiChartGenerator:
    """
    Generates Kagi chart lines based on price reversals.

    Unlike Renko (fixed brick size), Kagi tracks the actual magnitude
    of each move and changes line thickness at breakout points.

    Useful for identifying support/resistance breaks and
    supply/demand shifts without time noise.
    """

    def __init__(self, reversal_amount: float = 10.0):
        self.reversal_amount = reversal_amount
        self.lines: list[dict] = []
        self.current_direction: int = 0  # 1=up, -1=down
        self.current_price: float | None = None
        self.extreme_price: float | None = None
        self.prev_high: float | None = None
        self.prev_low: float | None = None
        self.line_type: str = 'yang'  # 'yang' (thick) or 'yin' (thin)

    def on_price(self, timestamp: int, price: float):
        if self.current_price is None:
            self.current_price = price
            self.extreme_price = price
            return None

        if self.current_direction == 0:
            if price - self.current_price >= self.reversal_amount:
                self.current_direction = 1
                self.extreme_price = price
            elif self.current_price - price >= self.reversal_amount:
                self.current_direction = -1
                self.extreme_price = price
            return None

        if self.current_direction == 1:
            if price > self.extreme_price:
                self.extreme_price = price
                if self.prev_high is not None and price > self.prev_high:
                    self.line_type = 'yang'
            elif self.extreme_price - price >= self.reversal_amount:
                line = {
                    'timestamp': timestamp,
                    'start': self.current_price,
                    'end': self.extreme_price,
                    'direction': 'up',
                    'type': self.line_type,
                }
                self.lines.append(line)
                self.prev_high = self.extreme_price
                self.current_price = self.extreme_price
                self.extreme_price = price
                self.current_direction = -1
                if self.prev_low is not None and price < self.prev_low:
                    self.line_type = 'yin'
                return line
        else:
            if price < self.extreme_price:
                self.extreme_price = price
                if self.prev_low is not None and price < self.prev_low:
                    self.line_type = 'yin'
            elif price - self.extreme_price >= self.reversal_amount:
                line = {
                    'timestamp': timestamp,
                    'start': self.current_price,
                    'end': self.extreme_price,
                    'direction': 'down',
                    'type': self.line_type,
                }
                self.lines.append(line)
                self.prev_low = self.extreme_price
                self.current_price = self.extreme_price
                self.extreme_price = price
                self.current_direction = 1
                if self.prev_high is not None and price > self.prev_high:
                    self.line_type = 'yang'
                return line

        return None

10. Line Break графиктері

Line Break графиктері жаңа сызықты (қорапты) тек жабылу бағасы алдыңғы N сызықтың (әдетте 3) жоғары немесе төмен деңгейінен асып кеткенде ғана салады. Баға ауқымда қалса, жаңа сызық салынбайды.

class LineBreakGenerator:
    """
    Generates Line Break bars (Three Line Break by default).

    A new bar is drawn only when the close exceeds the high or low
    of the last N bars. Filters out minor noise by requiring price
    to break through a multi-bar range.

    The 'N' parameter (line_count) controls sensitivity:
    - N=2: more sensitive, more bars, more noise
    - N=3: standard (Three Line Break)
    - N=4+: less sensitive, fewer bars, stronger signals
    """

    def __init__(self, line_count: int = 3):
        self.line_count = line_count
        self.lines: list[dict] = []

    def on_close(self, timestamp: int, close: float) -> dict | None:
        if not self.lines:
            self.lines.append({
                'timestamp': timestamp,
                'open': close,
                'close': close,
                'high': close,
                'low': close,
                'direction': 0,
            })
            return None

        lookback = self.lines[-self.line_count:] if len(self.lines) >= self.line_count else self.lines

        highest = max(l['high'] for l in lookback)
        lowest = min(l['low'] for l in lookback)
        last = self.lines[-1]

        new_line = None

        if close > highest:
            new_line = {
                'timestamp': timestamp,
                'open': last['close'],
                'close': close,
                'high': close,
                'low': last['close'],
                'direction': 1,
            }
        elif close < lowest:
            new_line = {
                'timestamp': timestamp,
                'open': last['close'],
                'close': close,
                'high': last['close'],
                'low': close,
                'direction': -1,
            }

        if new_line:
            self.lines.append(new_line)
            return new_line

        return None

11. Point & Figure графиктері

Point & Figure (P&F) графиктерінде X (өсіп жатқан баға) және O (төмендеп жатқан баға) бағаналары қолданылады. Баған ауысуы әдетте 3 қорап өлшемінің реверсиясын талап етеді. Шуды сүзу мен қолдау/қарсылықты анықтаудың ең көне әдістерінің бірі.

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

Алгоритмдік трейдингте Каги, Line Break және P&F: Негізінен ұзақ мерзімді тренд анықтау және қолдау/қарсылық анықтау үшін қолданылады. Сүзгі қабаты ретінде — "Каги графигі инь режимінде болғанда лонг сигналдарын алма" — олар мәмілелерді макро құрылыммен сәйкестендіру арқылы құн қосады.

12–14. Ақпаратқа негізделген барлар

Теңсіздік барлары, серия барлары, CUSUM сүзгілері мен энтропия барлары: нарық бізге бірдеңе өзгергенін айтқанда іріктеу.

Marcos Lopez de Prado-ның Advances in Financial Machine Learning (2018) кітабынан ең күрделі тәсіл. Негізгі түсінік: белгіленген аралықтарда емес, нарыққа жаңа ақпарат келгенде іріктеу.

12. Тик теңсіздігі барлары (TIB)

Егер нарық тепе-теңдікте болса, сатып алушы бастаған және сатушы бастаған мәмілелер шамамен теңдесуі керек. Теңсіздік біздің күткенімізден асып кеткенде, бірдеңе өзгерді. Сол сәтте бар іріктеу.

Әрбір мәміле тик ережесін қолдана отырып сатып алушы бастаған (+1) немесе сатушы бастаған (-1) деп жіктеледі. Біз жинақталған теңсіздік θ-ны қадағалаймыз және |θ| динамикалық шектіден асып кеткенде іріктейміз.

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. Көлем теңсіздігі барлары (VIB)

TIB-тің кеңеюі: әрбір мәмілені ±1 деп санаудың орнына, таңбаланған көлемге қарай өлшеу. 100-BTC сатып алу +100 үлес қосады, ал 1-BTC сату -1 үлес қосады. Көптеген кіші мәмілелерге бөлінуі мүмкін үлкен ақпараттандырылған тапсырыстарды анықтайды.

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

Жарылу мәселесі

Теңсіздік барларымен байланысты белгілі мәселе: EWMA негізіндегі шекті оң кері байланыс циклына түсуі мүмкін. Шешім: min_ticks және 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. Серия барлары

Серия барлары ағымдағы бағыттық серияның ұзындығын қадағалайды — сатып алулардың немесе сатулардың ең ұзын тізбегін. Үлкен ақпараттандырылған трейдер тапсырысты көптеген кіші мәмілелерге бөлгенде, тізбек әдеттен тыс ұзарады. Серия барлары мұны анықтайды.

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

Серия барларын көлем серияларына және доллар серияларына дейін кеңейтуге болады.

15. CUSUM сүзгі барлары

CUSUM (жинақталған сома) сүзгісі жинақталған кірістерді қадағалау арқылы қашан іріктеу жасау керектігін анықтайды. Теңсіздік барларынан (шикі мәмілелермен жұмыс істейтін) айырмашылығы, CUSUM бар 1 минуттық OHLCV деректеріне қолданылуы мүмкін — тик деректері қажет емес.

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 + үштік бөгет әдісі: Lopez de Prado әдісінде CUSUM оқиғалары үштік бөгет әдісі үшін кіру нүктелері ретінде қолданылады — мұнда әрбір оқиға стоп-лосс, тейк-профит және мерзімі бітетін бөгеттермен мәмілені іске қосады. Мұндай оқиғаға негізделген стратегияларды берік тексеру үшін қараңыз Walk-Forward Optimization және Monte Carlo Bootstrap for Backtesting.

16. Энтропия барлары

Теориялық жағынан ең сымбатты тәсіл: бар ішіндегі баға қатарының ақпараттық мазмұны (Шеннон энтропиясы) шектіден асып кеткенде іріктеу.

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

Практикалық ескерту: Энтропия барлары есептеу жағынан қымбат және негізінен зерттеу қызығушылығын білдіреді — бірақ ML негізіндегі стратегиялар үшін олар жақсырақ статистикалық қасиеттері бар белгілерді шығарады, себебі әрбір барда шамамен бірдей "ақпарат" болады.

17. Дельта барлары (тапсырыс ағыны)

Жинақталған дельта: нақты уақытта агрессивті сатып алушылар мен сатушылардың таза күшін өлшеу.

Дельта барлары жинақталған дельта негізінде іріктеледі — сатып алу көлемі мен сату көлемі арасындағы жүріп жатқан айырмашылық. Теңсіздік барларынан (тик белгілерін ±1 қолданатын) айырмашылығы, дельта барлары нақты көлем бойынша өлшенген тапсырыс ағынын қолданады.

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

Дельта дивергенциясы: Ең күшті сигналдардың бірі — баға өсіп жатқанда жинақталған дельта теріс болғанда (сатушылар агрессивті, бірақ баға бәрібір көтеріледі, бұл лимиттік сатып алудың сіңірілуін білдіреді). Digital Fingerprint: Trader Identification мақаласында сипатталған мінез-құлықтық саусақ ізі тәсіліне тікелей қатысты. Avellaneda-Stoikov моделін қолданатын маркет-мейкерлер үшін дельта барлары тауарлық қор тәуекелі мен агрессор қысымының нақты уақыттағы көрінісін береді.

Негізгі барлардың дөңгелек буфері: жаңа деректер кіреді, ескі деректер шығады, ал агрегацияланған шам әрқашан жарамды.

Агрегациялау әдістері негізгі барлардың жоғары уақыт кезеңіндегі (HTF) шамдарға қалай құрастырылатынын анықтайды. Олар бар түрінен тәуелсіз — кез келген агрегациялау әдісін кез келген негізгі бар түріне қолдана аласыз.

Әдіс A: күнтізбеге тураланған агрегациялау

Белгіленген күнтізбелік шекара ішінде түсетін барлық негізгі барларды агрегациялау. "1 сағаттық" шам 14:00:00-ден 14:59:59-ге дейінгі барлық барларды қамтиды.

Қасиеттері:

• Барлық нарық қатысушылары бір шекараларды көреді — нарық құрылымын талдау, қолдау/қарсылық, PIQ триггерлер үшін маңызды
• Суық старт мәселесі: қайта іске қосудан кейінгі ішінара шам
• Уақыт барлары үшін табиғи (бұл биржалар өздігінен ұсынатын нәрсе)
• Уақытсыз барлар үшін де жұмыс істейді: "14:00 мен 15:00 арасында жабылған барлық көлем барлары" = көлем барларынан алынған күнтізбеге тураланған сағаттық шам

Әдіс B: жылжымалы терезе агрегациялауы

Соңғы N жабылған негізгі барды агрегациялау, әрбір жаңа барда қайта есептеу. "1 сағаттық" жылжымалы шам = соңғы 60 жабылған 1 минуттық уақыт бары, әр минут сайын жаңартылады.

Атомдық бірлік — жабылған негізгі бар. Бұл дизайн шешімі мыналарды береді:

1. Суық старт жоқ. N бардан кейін шам жарамды. Ешбір ішінара шам шуы жоқ.
2. Бэктест паритеті. Егер тірі сауда бэктест қозғалтқышымен бірдей атомдық бірлікті қолданса, сигналдар бірдей болады.
3. Қарапайым тексеру. Бір ереже: 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

Фазалық ауысу трейд-офы: Жылжымалы шамдар егер :37-де бастасаңыз :37-де жабылады, барлықтардай :00-де емес. Бұл көпшілікке көрінетін деңгейлерге тәуелді стратегиялар үшін маңызды. Шешім: екеуін де қолдану — нарық құрылымы үшін күнтізбе, сигналдар үшін жылжымалы.

Әдіс C: бейімделгіш жылжымалы агрегациялау

Жылжымалыға ұқсас, бірақ терезе өлшемі ағымдағы волатильділікке бейімделеді. Тыныш нарықтар → кеңірек терезе (көбірек тегістеу). Волатильді нарықтар → тарырақ терезе (жылдамырақ реакция).

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

Әрбір негізгі бар түрін әрбір агрегациялау әдісімен біріктіруге болады. Кейбір комбинациялар стандартты (күнтізбелік уақыт барлары = биржалар сізге беретін нәрсе), басқалары экзотикалық, бірақ күшті.

Комбинация мысалдары

Гибридті қозғалтқыш: күнтізбе + жылжымалы

Іс жүзінде сізге күнтізбелік және жылжымалы агрегациялау бір мезгілде керек. Жад шығыны шамалы — символ басына, уақыт кезеңі басына екі deque буфері.

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

Уақыт-көлем гибриді: көлем бөліністерімен күнтізбе

Ерекше агрегациялау варианты: көлем шектіден асып кеткенде ерте мәжбүрлі жабылатын күнтізбеге тураланған шамдар. Уақыт синхронизациясын сақтай отырып, белсенділік секірулеріне бейімделеді.

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

Практикалық агрегациялау: каскадты алдын ала жүктеу

Каскадты алдын ала жүктеу: сағаттық шамдардан күндік, минуттық шамдардан сағаттық шамдарды құрастыру — API шектеулерін айналып өту.

Биржалар қанша тарихи дерек беретінін шектейді. Binance бір REST сұрауына ~1000 шам береді, OKX 300-де шектелген. Егер сізге жылжымалы 1D шам (1440 минут) керек болса, әрдайым жеткілікті 1 минуттық тарихты ала алмайсыз. WebSocket арқылы мәмілелер мен тапсырыстар кітабының нақты уақыттағы ағынын алу үшін қараңыз CCXT Pro WebSocket Methods.

Шешім: каскадты агрегациялау — әрбір тереңдікте қолжетімді ең жоғары рұқсаттан жоғары уақыт кезеңдерін құрастыру, содан кейін оларды бір-біріне тіркеу.

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

Бұл жұмыс істейді, себебі OHLCV агрегациясы құрастырылатын: 1D шамның жоғары деңгейі 24 сағаттық жоғары деңгейлердің максимумы, ол 1440 минуттық жоғары деңгейлердің максимумы.

Көп биржалық шектеулер

Агрегациялау сәйкестігін тексеру

REST API-ден алынған 1 сағаттық шам 60 минуттық шамнан есептелетінмен сәйкес келмеуі мүмкін. Әрдайым тексеріңіз:

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

Егер тексеру тұрақты түрде сәтсіз болса, әрдайым 1 минуттан өзіңіз агрегациялаңыз — бэктест паритеті үшін ешқашан биржаның HTF шамына сенбеңіз.

Салыстыру матрицасы

Ось 1: негізгі бар түрлері

Ось 2: агрегациялау әдістері

Практикалық ұсыныстар

Төрт қабатты шам архитектурасы: жылжымалы сигналдар, күнтізбелік құрылым, микроқұрылымдық ағын және тренд сүзгілері.

Егер бэктест қозғалтқышыңыз 1 минуттық OHLCV деректерінде жұмыс істесе:

1. Жылжымалы уақыт барлары — ең қарапайым жаңарту. Қосымша дерек жоқ. Суық стартты жояды.
2. Гибрид (жылжымалы + күнтізбе) уақыт барлары — нарық құрылымы үшін күнтізбе, сигналдар үшін жылжымалы.
3. CUSUM сүзгісі — 1 минуттық жабылуларда жұмыс істейді, тик деректері қажет емес. "Бірдеңе қызықты болу үшін жеткілікті қозғалды."

Егер сізде тик/мәміле деректері болса:

4. Доллар барлары + жылжымалы — квант қаржы әдебиетінен ұсынылатын әдепкі.
5. Көлем теңсіздігі барлары + жылжымалы — ақпараттандырылған ағынды анықтайды, маңызды оқиғалар кезінде көбірек іріктейді.
6. Дельта барлары + күнтізбе — егер сізде агрессор-жақ жіктеуі болса, нарықты кім итеріп жатқанын көрудің ең тікелей жолы.

Сүзгі ретінде (кез келген негіз+агрегация комбинациясының үстіне Хейкин-Аши немесе Line Break қолдану):

7. Жылжымалы көлем барларына Хейкин-Аши — белсенділікке қалыпқа келтірілген деректерде таза тренд сигналдары.
8. Күндік күнтізбелік барларға Line Break / Каги — макро тренд сүзгісі.

Marketmaker.cc үшін нақты — қабатты тәсіл:

• 1-қабат (сигналдар): Индикаторлар мен кіру/шығу сигналдары үшін уақыт барларының жылжымалы агрегациясы. Суық старт жоқ, тамаша бэктест паритеті.
• 2-қабат (нарық құрылымы): Қолдау/қарсылық, сағаттық жабылу талдауы және PIQ триггерлер үшін күнтізбеге тураланған уақыт барлары.
• 3-қабат (микроқұрылым): Ақпараттандырылған ағынды анықтау, тапсырыс бөлінуін анықтау және үлкен қозғалыстарды алдын ала болжау үшін шикі мәміле ағынынан алынған көлем теңсіздігі барлары + дельта барлары. Тапсырыс ағыны деректеріндегі мінез-құлықтық үлгіні тану үшін Digital Fingerprint: Trader Identification мақаласын да қараңыз.
• 4-қабат (тренд сүзгісі): Жылжымалы барларға Хейкин-Аши түрлендіруі немесе 4 сағаттық күнтізбелік жабылуларға Line Break, сигналдарды макро бағытпен сәйкестендіріп ұстау үшін.

Қорытынды

Шам құрылымы бір ғана таңдау емес — бұл екі тәуелсіз шешім:

1. Қандай түрдегі бар? Уақыт сағат аралықтарын қамтиды. Белсенділік (тик, көлем, доллар) нарықтық қатысуды қамтиды. Баға (Ренко, диапазон, волатильділік) қозғалыстарды қамтиды. Ақпарат (теңсіздік, серия, CUSUM, энтропия) жаңа ақпараттың келуін қамтиды. Тапсырыс ағыны (дельта) агрессивті қысымды қамтиды.

2. Жоғары уақыт кезеңдеріне қалай агрегациялау керек? Күнтізбе көпшілікпен сәйкестендіреді. Жылжымалы суық стартты жояды. Бейімделгіш волатильділікке жауап береді.

Binance-тен алынған стандартты "1 сағаттық шам" 17×3 матрицасындағы бір ғана ұяшық. Қалған 50 комбинация оларды іске асыруға дайын кез келген адамға қолжетімді. Өндірістік жүйе үшін жауап — "шешім қабылдау қозғалтқышыңыздың әрбір қабаты үшін дұрыс комбинацияны таңдау."

Атомдық бірлік — жабылған негізгі бар — негіз болып қала береді. Қалғанының бәрі агрегациялау.

Ұсақталған деректермен бэктест дәлдігі туралы көбірек білу үшін қараңыз Adaptive Drill-Down: Backtest with Variable Granularity. Мультиуақытты-кезеңдік стратегиялардағы индикаторды алдын ала есептеудің әсері үшін қараңыз Aggregated Parquet Cache.

Пайдалы сілтемелер

1. Lopez de Prado — Advances in Financial Machine Learning (2018)
2. Easley, Lopez de Prado, O'Hara — The Volume Clock: Insights into the High Frequency Paradigm (2012)
3. mlfinlab — Python library implementing information-driven bars
4. Binance — Historical Market Data
5. Apache Parquet — columnar storage format

Дәйексөз

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