Tutorial 03: Polymarket Liquidation Statistical Arbitrage Advanced

1. What Does This Bot Do?

This is a statistical arbitrage bot that uses BTC liquidation data to trade 5-minute binary markets on Polymarket, with a hedge on Hyperliquid. It's called a "stink bid" bot because it places lowball orders and waits for a fill.

2. Key Concepts

Polymarket 5-Minute Markets

Stink Bid

A limit order placed well below the current market price. The idea is: "I'm only buying if the price drops to my cheap level." It's like offering $200 for a $300 item and hoping the seller gets desperate.

Hyperliquid Hedge

A position on Hyperliquid that moves in the opposite direction of your Polymarket bet. If Polymarket pays you, the hedge loses (but less). If Polymarket loses, the hedge gains. This reduces overall risk.

Liquidation Signal

When many BTC traders get force-closed (liquidated), it often indicates a momentum wave. Large long liquidations suggest bearish pressure (price might go down). Large short liquidations suggest bullish pressure (price might go up).

3. The Strategy Explained

4. How the Bot Works

1 Moon Dev API - Getting Liquidation Data

def moondev_api_get(endpoint):
    url = f"https://api.moondev.com{endpoint}"
    headers = {'X-API-Key': MOONDEV_API_KEY}
    response = requests.get(url, headers=headers, timeout=30)
    return response.json()

def get_all_liquidations(timeframe="10m"):
    # Gets BTC liquidations from ALL exchanges combined
    return moondev_api_get(f"/api/all_liquidations/{timeframe}.json")

What it does: Calls the Moon Dev API to get BTC liquidation data from all major exchanges (Binance, Bybit, OKX, Hyperliquid) combined. The timeframe="10m" looks at the last 10 minutes of data.

2 LiqStinkBot Class - The Bot's Brain

class LiqStinkBot:
    def __init__(self):
        self.reset()  # Start fresh

    def reset(self):
        # Clear all state for the next 5-minute market
        self.signal_fired = False
        self.order_placed = False
        self.poly_filled = False
        self.hedge_filled = False
        # ... and more state variables

What it does: The LiqStinkBot class manages the entire lifecycle of a trade. It tracks whether a signal was detected, if an order was placed, if it was filled, and if the hedge was placed. At the start of each new 5-minute market, everything resets.

3 place_stink_bid() - Placing the Cheap Order

def place_stink_bid(self):
    # Get current market price
    book = get_order_book(self.target_token_id)
    self.signal_price = book['best_ask']

    # Calculate stink bid price = 30% below current price
    self.stink_bid_price = round(
        self.signal_price * (1 - PULLBACK_PCT), 4
    )

    # Place the limit order
    response = place_limit_order(
        token_id=self.target_token_id,
        side="BUY",
        price=self.stink_bid_price,
        size=self.stink_bid_shares,
    )

What it does: Calculates a price 30% below the current market price (PULLBACK_PCT = 0.30) and places a buy limit order at that price. The bot is saying: "I'll only buy if the price drops to my cheap level."

4 fill_hyperliquid_hedge() - Insurance Position

def fill_hyperliquid_hedge(poly_outcome):
    # If Polymarket bet UP -> hedge by going SHORT on Hyperliquid
    if poly_outcome == "UP":
        is_buy = False   # SHORT
    else:
        is_buy = True    # LONG

    # Loop until filled, bumping price each attempt
    for attempt in range(HEDGE_MAX_ATTEMPTS):
        # ... place order, check if filled ...

What it does: After the Polymarket order fills, places an inverse position on Hyperliquid. If you bet UP on Polymarket, you go SHORT on Hyperliquid (and vice versa). This is your insurance. It uses a loop that bumps the price closer to market on each attempt until it fills.

5 run_market_cycle() - The Main Loop

def run_market_cycle(self, market_ts):
    # 1. Find the current 5-minute market
    self.market_info = get_market_info(market_ts)

    # 2. Loop until market expires
    while True:
        # Check if time's up
        if time_remaining <= 0: break

        # Cancel if too little time left
        if time_remaining < MIN_TIME_LEFT: break

        # Check for liquidation signal
        signal = check_liquidation_signal()
        if signal:
            self.place_stink_bid()

        # Check if stink bid was filled
        if self.order_placed:
            if self.check_for_fill():
                fill_hyperliquid_hedge(...)

        time.sleep(BOT_POLL_INTERVAL)

What it does: This is the heart of the bot. For each 5-minute market window, it: (1) finds the market, (2) watches for liquidation signals, (3) places stink bids when triggered, (4) monitors for fills, (5) places hedges, and (6) logs everything.

5. Configuration

6. Dependencies

pip install requests pandas python-dotenv termcolor eth-account web3
pip install py-clob-client hyperliquid-python-sdk

Required in .env:

PRIVATE_KEY=your_polymarket_private_key
PUBLIC_KEY=your_polymarket_public_address
HYPER_LIQUID_KEY=your_hyperliquid_private_key
MOONDEV_API_KEY=your_moondev_api_key

7. Glossary

8. Full Code: Python to Pseudo-Code Translation

Configuration & Account Setup

# --- PYTHON ---
BOT_POLL_INTERVAL = 10
LIQUIDATION_THRESHOLD_MIN = 25_000
LIQUIDATION_THRESHOLD_MAX = 100_000
PULLBACK_PCT = 0.30
MIN_TIME_LEFT = 60
POLY_USD_PER_POSITION = 15.0
HEDGE_USD = 10.0
HEDGE_LEVERAGE = 3
MARKET_DURATION = 300
hl_account = eth_account.Account.from_key(HYPER_LIQUID_KEY)

# --- PSEUDO-CODE ---
SET check interval to every 10 seconds
SET minimum liquidation signal to $25,000
SET maximum liquidation signal to $100,000 (above this = missed the wave)
SET stink bid pullback to 30% below current price
SET minimum time remaining to 60 seconds (cancel orders if less)
SET Polymarket position size to $15 per bet
SET Hyperliquid hedge size to $10
SET Hyperliquid leverage to 3x
SET market duration to 300 seconds (5 minutes)
CREATE a Hyperliquid account object from the private key in .env

check_liquidation_signal() - Detect Liquidation Events

# --- PYTHON ---
def check_liquidation_signal():
    long_liq, short_liq = get_btc_liquidations()
    if long_liq >= LIQUIDATION_THRESHOLD_MIN and long_liq <= LIQUIDATION_THRESHOLD_MAX and long_liq > short_liq:
        return 'LONG_LIQ', long_liq
    elif short_liq >= LIQUIDATION_THRESHOLD_MIN and short_liq <= LIQUIDATION_THRESHOLD_MAX and short_liq > long_liq:
        return 'SHORT_LIQ', short_liq
    elif long_liq > LIQUIDATION_THRESHOLD_MAX or short_liq > LIQUIDATION_THRESHOLD_MAX:
        print("Missed the wave, waiting for next")
    return None, 0

# --- PSEUDO-CODE ---
FUNCTION check_liquidation_signal():
    CALL the Moon Dev API to get BTC liquidation totals from all exchanges
    SEPARATE into long liquidations (force-sold) and short liquidations (force-bought)

    IF long liquidations are between $25K-$100K AND more than short liquidations:
        RETURN "LONG_LIQ" signal with the dollar amount
        (This means: many longs were just force-sold = bearish = price may drop)

    IF short liquidations are between $25K-$100K AND more than long liquidations:
        RETURN "SHORT_LIQ" signal with the dollar amount
        (This means: many shorts were just force-bought = bullish = price may rise)

    IF either side is over $100K:
        PRINT "Missed the wave" (the move already happened)

    RETURN no signal

place_stink_bid() - Place the Discounted Order

# --- PYTHON ---
def place_stink_bid(self):
    book = get_order_book(self.target_token_id)
    self.signal_price = book['best_ask']
    self.stink_bid_price = round(self.signal_price * (1 - PULLBACK_PCT), 4)
    if self.stink_bid_price < 0.01:
        self.stink_bid_price = 0.01
    self.stink_bid_shares = calculate_shares(POLY_USD_PER_POSITION, self.stink_bid_price)
    cancel_token_orders(token_id)
    time.sleep(0.5)
    response = place_limit_order(token_id, "BUY", self.stink_bid_price, self.stink_bid_shares)

# --- PSEUDO-CODE ---
FUNCTION place_stink_bid():
    GET the current order book for the target token (UP or DOWN)
    RECORD the current asking price as our "signal price"

    CALCULATE our stink bid price = signal price minus 30%
    IF the stink bid would be less than $0.01: set it to $0.01 (minimum)

    CALCULATE how many shares we can buy with $15 at the stink bid price
    CANCEL any existing orders for this token
    WAIT half a second
    PLACE a BUY limit order at the stink bid price
    IF the order was accepted: mark order_placed as TRUE

fill_hyperliquid_hedge() - Place Insurance Position

# --- PYTHON ---
def fill_hyperliquid_hedge(poly_outcome):
    if poly_outcome.upper() == "UP":
        is_buy = False; hedge_side = "SHORT"
    else:
        is_buy = True; hedge_side = "LONG"
    for attempt in range(1, HEDGE_MAX_ATTEMPTS + 1):
        ask, bid, _ = hl_ask_bid(HEDGE_SYMBOL)
        mid_price = (ask + bid) / 2
        btc_size = (HEDGE_USD * HEDGE_LEVERAGE) / mid_price
        result = hl_limit_order(HEDGE_SYMBOL, is_buy, btc_size, limit_px, False, hl_account)
        if result filled:
            return True, hedge_side, btc_size, limit_px
        time.sleep(HEDGE_FILL_WAIT)
    return False, hedge_side, 0, 0

# --- PSEUDO-CODE ---
FUNCTION fill_hyperliquid_hedge(Polymarket bet outcome):
    IF we bet UP on Polymarket:
        We need to go SHORT on Hyperliquid (inverse hedge)
    IF we bet DOWN on Polymarket:
        We need to go LONG on Hyperliquid (inverse hedge)

    SET leverage to 3x on Hyperliquid

    TRY up to 10 times to fill the hedge:
        GET current BTC price (bid and ask) from Hyperliquid
        CALCULATE the middle price
        CALCULATE how much BTC to buy/sell = ($10 x 3x leverage) / BTC price
        ROUND to the correct number of decimal places

        TRY to place a limit order
        IF the order was filled immediately: RETURN success!

        On each failed attempt, move the price slightly closer to market price
        WAIT a couple seconds between attempts

    IF all attempts failed: RETURN failure

run_market_cycle() - The Core Loop

# --- PYTHON ---
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 time_remaining < MIN_TIME_LEFT and self.order_placed and not self.poly_filled:
            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 and time_remaining > MIN_TIME_LEFT:
            signal_type, signal_amount = check_liquidation_signal()
            if signal_type:
                self.signal_fired = True
                self.place_stink_bid()
        time.sleep(BOT_POLL_INTERVAL)

# --- PSEUDO-CODE ---
FUNCTION run_market_cycle(market timestamp):
    FIND the current 5-minute Polymarket market

    LOOP until the market expires:
        CALCULATE how many seconds are left in this 5-minute window

        IF time is up:
            CANCEL any open orders and EXIT the loop

        IF less than 60 seconds left AND we have an unfilled order:
            CANCEL the order (too late) and EXIT the loop

        IF our Polymarket order already filled:
            Just wait (nothing to do until market resolves)

        IF we placed an order but haven't checked if it filled:
            CHECK if it filled
            IF it filled: PLACE the Hyperliquid hedge
            WAIT 2 seconds, then check again

        IF we haven't fired a signal yet AND there's enough time left:
            CHECK for liquidation signals
            IF a signal is detected:
                MARK signal as fired
                DETERMINE direction (UP or DOWN)
                PLACE the stink bid

        WAIT 10 seconds before the next check

log_trade() - Record Everything

# --- PYTHON ---
def log_trade(signal_type, signal_amount, outcome, poly_price, stink_price,
              poly_shares, hedge_side, hedge_size, hedge_price, result, reason=""):
    new_row = pd.DataFrame([{ 'timestamp': datetime.now().isoformat(),
        'signal_type': signal_type, 'signal_amount_usd': round(signal_amount, 2),
        'outcome': outcome, 'stink_bid_price': round(stink_price, 4), ... }])
    if os.path.exists(TRADE_LOG_FILE):
        existing = pd.read_csv(TRADE_LOG_FILE)
        df = pd.concat([existing, new_row])
    else:
        df = new_row
    df.to_csv(TRADE_LOG_FILE, index=False)

# --- PSEUDO-CODE ---
FUNCTION log_trade(all the trade details):
    CREATE a new data row with:
        - Current timestamp
        - What type of liquidation signal triggered the trade
        - How big the liquidation was in USD
        - What outcome we bet on (UP or DOWN)
        - The original market price when we detected the signal
        - The stink bid price we placed
        - How many shares we bought
        - The hedge details (if placed)
        - The final result (BOTH_FILLED, POLY_ONLY, NO_FILL, etc.)

    IF a trade log CSV file already exists:
        READ the existing file
        APPEND the new row to the end
    ELSE:
        CREATE a new file with this row

    SAVE the CSV file