This bot uses CVD (Cumulative Volume Delta) divergence signals to trade BTC 5-minute Polymarket markets. It analyzes real-time tick data from the Moon Dev API to detect when price movement disagrees with actual buying/selling pressure.
CVD measures the net buying vs selling pressure over time. It works using the "tick rule":
The CVD is the running total of all these +1/-1 values. A rising CVD means buyers are aggressive. A falling CVD means sellers are aggressive.
Divergence happens when price and CVD disagree:
Divergence is a powerful signal because it reveals what's happening under the surface - not just what the price shows.
The bot checks CVD on several timeframes: 1m, 3m, 5m, 10m, and 15m. A signal is stronger when multiple timeframes agree.
| Signal | Price | CVD | Meaning | Bet Direction |
|---|---|---|---|---|
| BEARISH DIV | Going UP | Going DOWN | Weak rally, sellers aggressive | DOWN |
| BULLISH DIV | Going DOWN | Going UP | Fake drop, buyers aggressive | UP |
| STRONG BULL | Going UP | Going UP | Strong momentum, buyers in control | UP |
| STRONG BEAR | Going DOWN | Going DOWN | Strong momentum, sellers in control | DOWN |
def compute_tick_cvd(ticks):
# Extract prices from tick data
prices = [t.get('p', t.get('price', 0)) for t in ticks]
cvd = 0
deltas = []
for i in range(1, len(prices)):
diff = prices[i] - prices[i-1]
if diff > 0: # Price went up = aggressive buy
delta = 1
elif diff < 0: # Price went down = aggressive sell
delta = -1
else: # Same price = inherit last direction
delta = last_direction
cvd += delta
deltas.append(delta)
return cvd, price_change, deltas, prices
What it does: Takes raw price tick data and computes the CVD value. Each price change is classified as buyer-driven (+1) or seller-driven (-1). The running sum gives you the CVD - a positive value means net buying pressure, negative means net selling pressure.
def detect_divergence(price_change, cvd_value):
# BEARISH: price up but sellers aggressive
if price_change > THRESH and cvd_value < -THRESH:
return "BEARISH_DIV", "DOWN"
# BULLISH: price down but buyers aggressive
if price_change < -THRESH and cvd_value > THRESH:
return "BULLISH_DIV", "UP"
# STRONG BULL: both positive
if price_change > THRESH and cvd_value > THRESH:
return "STRONG_BULL", "UP"
# STRONG BEAR: both negative
if price_change < -THRESH and cvd_value < -THRESH:
return "STRONG_BEAR", "DOWN"
What it does: Compares price movement with CVD. When they disagree (divergence), it generates a contrarian signal. When they agree, it generates a momentum signal. The function also calculates a strength score (0-100) based on how extreme the divergence is.
Like Tutorial 03, the bot uses the stink bid + hedge approach:
pip install py-clob-client hyperliquid-python-sdk web3 termcolor pandas eth-account python-dotenv
| Term | Meaning |
|---|---|
| CVD | Cumulative Volume Delta - running total of buy vs sell pressure |
| Tick Data | Individual price changes as they happen in real-time |
| Tick Rule | Method to classify each trade as buyer or seller initiated |
| Divergence | When price and volume indicators disagree |
| Order Flow | The actual buying and selling activity behind price moves |
| Timeframe | The period of time used for analysis (1m, 5m, 15m, etc.) |
| Contrarian | Trading against the current trend, expecting a reversal |
# --- PYTHON ---
def compute_tick_cvd(ticks):
prices = [t.get('p', t.get('price', 0)) for t in ticks]
cvd = 0; deltas = []; last_direction = 0
for i in range(1, len(prices)):
diff = prices[i] - prices[i-1]
if diff > 0:
delta = 1
elif diff < 0:
delta = -1
else:
delta = last_direction
cvd += delta
deltas.append(delta)
price_change = ((prices[-1] - prices[0]) / prices[0] * 100)
return cvd, price_change, deltas, prices
# --- PSEUDO-CODE ---
FUNCTION compute_tick_cvd(list of price ticks):
EXTRACT just the prices from the tick data
START with CVD = 0 (neutral)
REMEMBER the last direction (for when price doesn't change)
FOR each pair of consecutive prices:
CALCULATE the difference between current and previous price
IF price went UP: delta = +1 (buyer was aggressive)
IF price went DOWN: delta = -1 (seller was aggressive)
IF price stayed SAME: delta = same as last time
ADD delta to running CVD total
CALCULATE overall price change as a percentage
RETURN:
- The CVD value (positive = net buying, negative = net selling)
- The price change percentage
- The list of individual deltas
- The list of prices
# --- PYTHON ---
def detect_divergence(price_change, cvd_value):
if abs(price_change) < 0.01 and abs(cvd_value) < 5:
return "NEUTRAL", "NEUTRAL", "FLAT", 0
if price_change > THRESH and cvd_value < -THRESH:
return "BEARISH_DIV", "BEARISH DIV", "DOWN", strength
if price_change < -THRESH and cvd_value > THRESH:
return "BULLISH_DIV", "BULLISH DIV", "UP", strength
if price_change > THRESH and cvd_value > THRESH:
return "STRONG_BULL", "STRONG BULL", "UP", strength
if price_change < -THRESH and cvd_value < -THRESH:
return "STRONG_BEAR", "STRONG BEAR", "DOWN", strength
return "NEUTRAL", "NEUTRAL", "FLAT", 0
# --- PSEUDO-CODE ---
FUNCTION detect_divergence(price change %, CVD value):
IF both price and CVD are barely moving:
RETURN "NEUTRAL" (no signal)
IF price went UP but CVD is NEGATIVE (sellers aggressive):
RETURN "BEARISH DIVERGENCE" -> bet DOWN
CALCULATE strength = based on how extreme the divergence is
IF price went DOWN but CVD is POSITIVE (buyers aggressive):
RETURN "BULLISH DIVERGENCE" -> bet UP
CALCULATE strength = based on how extreme the divergence is
IF price went UP and CVD is also POSITIVE (both agree = strong):
RETURN "STRONG BULL" -> bet UP
IF price went DOWN and CVD is also NEGATIVE (both agree = strong):
RETURN "STRONG BEAR" -> bet DOWN
OTHERWISE: RETURN "NEUTRAL" (no clear signal)
# --- PYTHON ---
def check_cvd_signal():
signals = []
tick_response = api.get_ticks("BTC", "1h", limit=10000)
all_ticks = tick_response.get('ticks', [])
for tf in CVD_SIGNAL_TIMEFRAMES:
tick_data[tf] = slice_ticks_by_time(all_ticks, TF_SECONDS[tf])
for tf in CVD_SIGNAL_TIMEFRAMES:
cvd_val, price_chg, _, _ = compute_tick_cvd(ticks)
signal_type, _, direction, strength = detect_divergence(price_chg, cvd_val)
if signal_type is a real signal:
signals.append((direction, signal_type, detail, strength, tf))
signals.sort(key=lambda x: x[3], reverse=True)
best = signals[0]
return best direction, signal_type, detail
# --- PSEUDO-CODE ---
FUNCTION check_cvd_signal():
FETCH the last 1 hour of BTC tick data from Moon Dev API
FOR EACH timeframe (1min, 3min, 5min, 10min, 15min):
SLICE the tick data to only include ticks from that timeframe window
FOR EACH timeframe:
COMPUTE CVD and price change for that window
DETECT if there's a divergence signal
IF a real signal exists (not neutral):
RECORD it with direction, strength, and which timeframe
SORT all found signals by strength (strongest first)
IF we found any signals:
RETURN the strongest one (direction, type, and details)
ELSE:
RETURN no signal
# --- PYTHON ---
class CVDStinkBot:
def run_market_cycle(self, market_ts):
self.market_info = get_market_info(market_ts)
while True:
time_remaining = get_time_remaining(market_ts)
if time_remaining <= 0: self.cancel_orders(); break
if self.poly_filled: time.sleep(2); continue
if self.order_placed and not self.poly_filled:
if self.check_for_fill():
fill_hyperliquid_hedge(self.target_outcome)
time.sleep(2); continue
if not self.signal_fired:
direction, signal_type, detail = check_cvd_signal()
if direction:
self.signal_fired = True
self.target_outcome = direction
self.place_stink_bid()
time.sleep(BOT_POLL_INTERVAL)
# --- PSEUDO-CODE ---
CLASS CVDStinkBot:
FUNCTION run_market_cycle(market timestamp):
FIND the current Polymarket 5-minute market
LOOP until market expires:
CHECK time remaining
IF time is up: CANCEL orders and EXIT
IF Polymarket order already filled:
Just wait (position will resolve at market end)
IF we placed an order, check if it got filled:
IF filled: PLACE the Hyperliquid hedge
WAIT 2 seconds and check again
IF no signal fired yet:
CHECK CVD across all timeframes for a divergence signal
IF a signal is found:
RECORD the signal direction (UP or DOWN)
PLACE a stink bid at 30% below current price
WAIT 10 seconds before next check