Cumulative delta indicator development

Cumulative Volume Delta (CVD) Indicator Development

Cumulative Volume Delta (CVD — Cumulative Volume Delta) is the accumulated sum of the difference between aggressive buy and sell volumes over a selected period. Unlike delta of individual candle, CVD shows the accumulated balance of market forces, making hidden divergences with price visible.

How CVD Works

Candle 1: buy_vol=100, sell_vol=60 → delta=+40, CVD=+40
Candle 2: buy_vol=80,  sell_vol=90 → delta=-10, CVD=+30
Candle 3: buy_vol=50,  sell_vol=120 → delta=-70, CVD=-40
Candle 4: buy_vol=200, sell_vol=80 → delta=+120, CVD=+80

CVD grows when buyers are more aggressive. With rising price and falling CVD — bullish narrative is not confirmed by volume (bearish divergence).

CVD Calculation

import pandas as pd
from decimal import Decimal
import asyncio

class CVDCalculator:
    def __init__(self, symbol: str, reset_period: str = 'session'):
        """
        reset_period: 'session' (daily), 'week', 'never'
        """
        self.symbol = symbol
        self.reset_period = reset_period

    def calculate_from_trades(self, trades: list[dict]) -> pd.DataFrame:
        """
        Calculate CVD from raw aggTrades data
        trades: list {price, quantity, time, is_buyer_maker}
        """
        df = pd.DataFrame(trades)

        # Determine trade side
        df['buy_vol'] = df.apply(
            lambda row: row['quantity'] if not row['is_buyer_maker'] else 0,
            axis=1
        )
        df['sell_vol'] = df.apply(
            lambda row: row['quantity'] if row['is_buyer_maker'] else 0,
            axis=1
        )
        df['delta'] = df['buy_vol'] - df['sell_vol']

        # Group by candles (e.g., 1-minute)
        df['time'] = pd.to_datetime(df['time'], unit='ms')
        df = df.set_index('time')

        candle_delta = df['delta'].resample('1min').sum()
        candle_buy = df['buy_vol'].resample('1min').sum()
        candle_sell = df['sell_vol'].resample('1min').sum()

        result = pd.DataFrame({
            'delta': candle_delta,
            'buy_vol': candle_buy,
            'sell_vol': candle_sell,
        })

        # CVD with session reset
        if self.reset_period == 'session':
            result['session'] = result.index.date
            result['cvd'] = result.groupby('session')['delta'].cumsum()
        else:
            result['cvd'] = result['delta'].cumsum()

        return result

Detecting Divergences

class CVDDivergenceDetector:
    def find_divergences(
        self,
        price_series: pd.Series,
        cvd_series: pd.Series,
        lookback: int = 20
    ) -> pd.DataFrame:
        divergences = []

        for i in range(lookback, len(price_series)):
            window_price = price_series.iloc[i-lookback:i+1]
            window_cvd = cvd_series.iloc[i-lookback:i+1]

            current_price = window_price.iloc[-1]
            current_cvd = window_cvd.iloc[-1]

            # Bearish divergence: price higher, CVD below previous high
            prev_price_high = window_price.iloc[:-1].max()
            prev_cvd_at_high = window_cvd.iloc[window_price.iloc[:-1].argmax()]

            if current_price > prev_price_high and current_cvd < prev_cvd_at_high:
                divergences.append({
                    'time': price_series.index[i],
                    'type': 'bearish',
                    'price': current_price,
                    'cvd': current_cvd,
                    'strength': (prev_cvd_at_high - current_cvd) / abs(prev_cvd_at_high) * 100
                })

            # Bullish divergence: price lower, CVD above previous low
            prev_price_low = window_price.iloc[:-1].min()
            prev_cvd_at_low = window_cvd.iloc[window_price.iloc[:-1].argmin()]

            if current_price < prev_price_low and current_cvd > prev_cvd_at_low:
                divergences.append({
                    'time': price_series.index[i],
                    'type': 'bullish',
                    'price': current_price,
                    'cvd': current_cvd,
                    'strength': (current_cvd - prev_cvd_at_low) / abs(prev_cvd_at_low) * 100
                })

        return pd.DataFrame(divergences)

Pine Script Implementation

//@version=5
indicator("Cumulative Volume Delta", shorttitle="CVD", overlay=false)

reset_on_session = input.bool(true, "Reset daily")

// Approximate delta from OHLCV
candle_up = close >= open
delta_approx = candle_up ?
    volume * ((close - open) / (high - low + 0.001)) :
    -volume * ((open - close) / (high - low + 0.001))

// CVD with session reset
var float cvd = 0.0
new_session = ta.change(time("D")) != 0

if reset_on_session and new_session
    cvd := delta_approx
else
    cvd := cvd + delta_approx

// Color by direction
cvd_color = cvd >= cvd[1] ? color.new(color.green, 40) : color.new(color.red, 40)

hline(0, color=color.gray, linestyle=hline.style_dotted)
plot(cvd, "CVD", cvd_color, linewidth=2)

// Divergence markers (simplified)
price_up = close > close[20]
cvd_down = cvd < cvd[20]
bearish_div = price_up and cvd_down
plotshape(bearish_div, "Bear Div", shape.circle, location.top,
          color.new(color.red, 0), size=size.tiny)

Real-time WebSocket Update

class CVDWebSocketStreamer:
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.cvd = 0.0
        self.session_date = None

    async def stream(self):
        url = f"wss://stream.binance.com:9443/ws/{self.symbol.lower()}@aggTrade"

        async with websockets.connect(url) as ws:
            async for message in ws:
                trade = json.loads(message)
                await self.process_trade(trade)

    async def process_trade(self, trade: dict):
        import datetime
        today = datetime.date.today()

        # Reset CVD at session start
        if self.session_date != today:
            self.cvd = 0.0
            self.session_date = today

        qty = float(trade['q'])
        is_buyer_maker = trade['m']

        delta = -qty if is_buyer_maker else qty
        self.cvd += delta

        # Publish update to subscribers
        await self.broadcast({
            'type': 'cvd_update',
            'symbol': self.symbol,
            'cvd': self.cvd,
            'delta': delta,
            'timestamp': trade['T']
        })

CVD is a more advanced tool than regular bar delta. Accumulated divergence is visible immediately: when price rose for 3 days but CVD fell all that time — serious warning signal about trend weakness.