Алгоритмдик трейдинг үчүн бар түрлөрү жана агрегациялоо ыкмалары

Binance, TradingView же башка каалаган биржа UI'да сиз көргөн ар кандай шам графиги бир эле жол менен түзүлөт: белгиленген убакыт терезесинде — 1 мүнөт, 5 мүнөт, 1 саат — соодаларды агрегациялап, OHLCV барын чыгаруу. Бул ушунчалык кеңири таралган, көпчүлүк трейдерлер аны эч качан суроого алышпайт. Бирок алгоритмдик трейдинг үчүн бар түрүн жана агрегациялоо ыкмасын тандоо — эки көз каранды эмес чечим, ал эми көпчүлүк системалар аларды аралаштырышат.

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

Чийки соодалардын стандарттуу шамдарга кантип айланарын түшүнүү үчүн караңыз Trading Candles Demystified.

TL;DR

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

Шам түзүлүшүнүн эки огу

Салттуу көз караш бардык бар түрлөрүн тегиз тизмеге коёт: убакыт барлары, тик барлары, көлөм барлары, Ренко ж.б. Бул чаташтырат. Чындыгында эки ортогоналдуу тандоо бар:

Ok 1 — негизги бар түрү (17 түр): Жаңы бар качан жабылаарын кантип чечесиз? Белгиленген убакыт аралыгынан кийинби? N сооданан кийинби? Баа кыймылынан кийинби? Маалымат мазмуну өзгөргөндөбү? Бул "бир бар" деген эмнени билдирерин аныктайт.

Ok 2 — агрегациялоо ыкмасы (3 ыкма): Негизги барларды жогорку убакыт алкагындагы шамдарга кантип түзөсүз? Календардык чекараларга (00:00, 01:00, ...) шайкештиресизби? Акыркы N бардын жылжымалуу терезесин колдоносузбу? Терезе өлчөмүн волатилдүүлүккө ыңгайлаштыраcызбы?

Бул эки ok көз каранды эмес. Сизде булар болушу мүмкүн:

• Календарга шайкештирилген тик барлары — 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

Артыкчылыктары: Рыноктук активдүүлүккө табигый түрдө ыңгайлашат. Тик барлардан алынган кирешелер убакыт-бар кирешелерине караганда нормалдуу бөлүштүрүүгө жакыныраак болот — бул көптөгөн статистикалык моделдердин натыйжалуулугун жакшыртат.

Кемчиликтери: Чийки сооda агымын талап кылат (тарыхый маалыматтар үчүн бардык маалымат берүүчүлөрдө жеткиликтүү эмес). Бар мезгили алдын ала белгисиз — сиз "кийинки бар так X мезгилинде жабылат" деп айта албайсыз.

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

N келишим (же криптодо тыйын) соодаланганда жаңы бар пайда болот. Тик барларына окшош, бирок сооda көлөмүнө жараша салмактанган — бир 100-BTC сооda, 1-BTC сооdaдан 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'де) алмашылгандан кийин жаңы бар пайда болот. Активдүүлүккө негизделген барлардын арасында эң бекем, анткени ал сооda санын да, баа деңгээлин да нормалдаштырат.

Ойлонуп көрүңүз: эгер 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)

Эгер рынок тең салмактуулукта болсо, сатып алуучу баштаган жана сатуучу баштаган соодалар болжол менен теңдешиши керек. Теңсиздик биздин күткөндөн ашканда, бир нерсе өзгөрдү. Ошол учурда бар тандаңыз.

Ар бир сооda тик эрежесин колдонуп сатып алуучу баштаган (+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 шамына ишенбеңиз.

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

Ok 1: негизги бар түрлөрү

Ok 2: агрегациялоо ыкмалары

Практикалык сунуштар

Төрт катмарлуу шам архитектурасы: жылжымалуу сигналдар, календардык структура, микроструктуралык агым жана тенденция фильтрлери.

Эгер сиздин бэктест кыймылдаткычыңыз 1мин OHLCV маалыматтарында иштесе:

1. Жылжымалуу убакыт барлары — эң жөнөкөй жаңыртуу. Кошумча маалымат жок. Суук стартты жоёт.
2. Гибрид (жылжымалуу + календарь) убакыт барлары — рынок структурасы үчүн календарь, сигналдар үчүн жылжымалуу.
3. CUSUM фильтри — 1мин жабууларда иштейт, тик маалыматтары керек эмес. "Бир нерсе кызыктуу болушу үчүн жетиштүү жылды."

Эгер сизде тик/сооda маалыматтары болсо:

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