एल्गोरिदमिक ट्रेडिंग के लिए बार प्रकार और एग्रीगेशन विधियाँ

Binance, TradingView, या किसी भी एक्सचेंज UI पर आपने जो भी कैंडलस्टिक चार्ट देखा है, वह एक ही तरीके से बनाया गया है: एक निश्चित समय विंडो — 1 मिनट, 5 मिनट, 1 घंटा — के भीतर ट्रेड्स को एकत्रित करना और एक OHLCV बार बनाना। यह इतना सर्वव्यापी है कि अधिकांश ट्रेडर इस पर कभी सवाल नहीं उठाते। लेकिन एल्गोरिदमिक ट्रेडिंग के लिए, बार प्रकार और एग्रीगेशन विधि का चुनाव दो स्वतंत्र निर्णय हैं — और अधिकांश सिस्टम इन्हें मिला देते हैं।

यह लेख कैंडल निर्माण के दो अक्षों को अलग करता है: आप किस प्रकार का बार बनाते हैं (17 प्रकार) और आप उन्हें उच्च टाइमफ्रेम में कैसे एग्रीगेट करते हैं (3 विधियाँ)। यह संयोजन 51 संभावित कॉन्फ़िगरेशन देता है, प्रत्येक की बैकटेस्टिंग, लाइव ट्रेडिंग और सिग्नल जनरेशन के लिए अलग विशेषताएँ हैं।

कच्चे ट्रेड्स मानक कैंडल कैसे बनते हैं, इसका परिचय पाने के लिए देखें Trading Candles Demystified।

TL;DR

• कैंडल निर्माण में दो स्वतंत्र अक्ष हैं: बार प्रकार और एग्रीगेशन विधि
• 17 आधार बार प्रकार: समय, टिक, वॉल्यूम, डॉलर, रेंको, रेंज, वोलैटिलिटी, हेइकिन-आशी, कागी, लाइन ब्रेक, 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. रेंज बार

हर बार का एक निश्चित हाई-लो रेंज होता है। जब रेंज पार हो जाती है, तो बार बंद हो जाता है और एक नया शुरू होता है। रेंको के विपरीत, रेंज बार में विक्स शामिल होते हैं और इंट्रा-बार वोलैटिलिटी दिखा सकते हैं।

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. जापानी रिवर्सल चार्ट

कागी, लाइन ब्रेक और पॉइंट एंड फिगर: समय-मुक्त चार्टिंग विधियाँ जो पूरी तरह से मूल्य संरचना पर केंद्रित हैं।

ये पारंपरिक जापानी चार्टिंग विधियाँ हैं (रेंको के साथ-साथ) जो समय को पूरी तरह से त्याग देती हैं और मूल्य संरचना पर ध्यान केंद्रित करती हैं।

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. लाइन ब्रेक चार्ट

लाइन ब्रेक चार्ट एक नई लाइन (बॉक्स) तभी बनाते हैं जब क्लोजिंग प्राइस पिछले 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. पॉइंट एंड फिगर चार्ट

पॉइंट एंड फिगर (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

एल्गोरिदमिक ट्रेडिंग में कागी, लाइन ब्रेक और 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 को मौजूदा 1m 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 मिनट) चाहिए, तो आप हमेशा पर्याप्त 1m इतिहास प्राप्त नहीं कर सकते। 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 1h हाई का अधिकतम है, जो 1440 1m हाई का अधिकतम है।

मल्टी-एक्सचेंज सीमाएँ

एग्रीगेशन कंसिस्टेंसी चेक

REST API से 1h कैंडल शायद उससे मेल न खाए जो आप 60 1m कैंडल से गणना करते हैं। हमेशा सत्यापित करें:

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

यदि सत्यापन लगातार विफल होता है, तो हमेशा खुद 1m से एग्रीगेट करें — बैकटेस्टिंग पैरिटी के लिए कभी भी एक्सचेंज की HTF कैंडल पर भरोसा न करें।

तुलना मैट्रिक्स

अक्ष 1: आधार बार प्रकार

अक्ष 2: एग्रीगेशन विधियाँ

व्यावहारिक सिफारिशें

चार-परत कैंडल आर्किटेक्चर: रोलिंग सिग्नल, कैलेंडर संरचना, माइक्रोस्ट्रक्चर फ्लो, और ट्रेंड फिल्टर।

यदि आपका बैकटेस्ट इंजन 1m OHLCV डेटा पर चलता है:

1. रोलिंग टाइम बार — सबसे सरल अपग्रेड। कोई अतिरिक्त डेटा नहीं। कोल्ड स्टार्ट को समाप्त करता है।
2. हाइब्रिड (रोलिंग + कैलेंडर) टाइम बार — बाज़ार संरचना के लिए कैलेंडर, सिग्नल के लिए रोलिंग।
3. CUSUM फ़िल्टर — 1m क्लोज़ पर काम करता है, टिक डेटा नहीं चाहिए। "कुछ इतना आगे बढ़ा कि दिलचस्प हो गया।"

यदि आपके पास टिक/ट्रेड डेटा है:

4. डॉलर बार + रोलिंग — क्वांट फाइनेंस साहित्य से अनुशंसित डिफ़ॉल्ट।
5. वॉल्यूम इम्बैलेंस बार + रोलिंग — इंफॉर्म्ड फ्लो का पता लगाता है, महत्वपूर्ण घटनाओं के दौरान अधिक सैंपल करता है।
6. डेल्टा बार + कैलेंडर — यदि आपके पास आक्रामक-पक्ष वर्गीकरण है, तो यह कि कौन बाज़ार को धकेल रहा है इसका सबसे सीधा दृश्य।

फ़िल्टर के रूप में (किसी भी बेस+एग्रीगेशन संयोजन पर हेइकिन-आशी या लाइन ब्रेक लागू करें):

7. रोलिंग वॉल्यूम बार पर हेइकिन-आशी — एक्टिविटी-नॉर्मलाइज़्ड डेटा पर साफ ट्रेंड सिग्नल।
8. दैनिक कैलेंडर बार पर लाइन ब्रेक / कागी — मैक्रो ट्रेंड फ़िल्टर।

Marketmaker.cc के लिए विशेष रूप से — एक परतदार दृष्टिकोण:

• परत 1 (सिग्नल): इंडिकेटर और एंट्री/एग्ज़िट सिग्नल के लिए टाइम बार का रोलिंग एग्रीगेशन। कोई कोल्ड स्टार्ट नहीं, बेहतरीन बैकटेस्ट पैरिटी।
• परत 2 (बाज़ार संरचना): सपोर्ट/रेजिस्टेंस, घंटे के क्लोज़ विश्लेषण, और PIQ ट्रिगर के लिए कैलेंडर-अलाइंड टाइम बार।
• परत 3 (माइक्रोस्ट्रक्चर): इंफॉर्म्ड फ्लो, ऑर्डर स्प्लिटिंग का पता लगाने और बड़ी चालों की भविष्यवाणी करने के लिए कच्चे ट्रेड स्ट्रीम से वॉल्यूम इम्बैलेंस बार + डेल्टा बार। ऑर्डर फ्लो डेटा पर व्यवहार पैटर्न पहचान के लिए Digital Fingerprint: Trader Identification भी देखें।
• परत 4 (ट्रेंड फ़िल्टर): रोलिंग बार पर हेइकिन-आशी ट्रांसफॉर्मेशन, या 4h कैलेंडर क्लोज़ पर लाइन ब्रेक, सिग्नल को मैक्रो दिशा के साथ संरेखित रखने के लिए।

निष्कर्ष

कैंडल निर्माण एक अकेला विकल्प नहीं है — यह दो स्वतंत्र निर्णय हैं:

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