Polymarket Trading Bot Code Review: warproxxx/poly-maker Analysis

• Market Scanner: Identifies markets with sufficient liquidity and favorable spreads (typically 2-5% bid-ask spread)
• Orderbook Analyzer: Calculates optimal bid/ask prices based on current orderbook depth and recent price movements
• Order Manager: Places, updates, and cancels limit orders using the py-clob-client library
• Risk Manager: Monitors positions, enforces maximum exposure limits, and implements stop-loss mechanisms
• Position Tracker: Tracks P&L, position sizes, and fill rates in real-time

1. Orderbook Spread Detection

def calculate_spread(orderbook):
    """Calculate bid-ask spread for market making."""
    best_bid = orderbook['bids'][0]['price']
    best_ask = orderbook['asks'][0]['price']
    spread = best_ask - best_bid
    spread_pct = (spread / best_ask) * 100
    return spread, spread_pct

# Target markets with 2-5% spread
if spread_pct >= 2.0 and spread_pct <= 5.0:
    place_market_making_orders(orderbook)

2. Limit Order Placement

from py_clob_client.client import ClobClient
from py_clob_client.order_builder import OrderBuilder

def place_limit_order(client, market_id, side, price, size):
    """Place a limit order with proper error handling."""
    try:
        order = OrderBuilder.create_limit_order(
            market_id=market_id,
            side=side,  # 'BUY' or 'SELL'
            price=price,
            size=size,
            order_type='GTC'  # Good-till-cancel
        )
        signed_order = client.sign_order(order)
        response = client.create_order(signed_order)
        return response
    except Exception as e:
        logger.error(f"Order placement failed: {e}")
        return None

3. Risk Management

class RiskManager:
    def __init__(self, max_position_size=1000, max_exposure=5000):
        self.max_position_size = max_position_size
        self.max_exposure = max_exposure
        self.current_positions = {}
    
    def can_place_order(self, market_id, size, price):
        """Check if order is within risk limits."""
        current_exposure = sum(
            pos['size'] * pos['price'] 
            for pos in self.current_positions.values()
        )
        new_exposure = size * price
        
        if current_exposure + new_exposure > self.max_exposure:
            return False
        if size > self.max_position_size:
            return False
        return True

Spike Detection Logic

import time
from collections import deque

class SpikeDetector:
    def __init__(self, spike_threshold=0.05, time_window=60):
        self.spike_threshold = spike_threshold  # 5% price change
        self.time_window = time_window  # 60 seconds
        self.price_history = deque(maxlen=100)
        self.volume_history = deque(maxlen=100)
    
    def detect_spike(self, current_price, current_volume):
        """Detect if current price movement is a spike."""
        self.price_history.append({
            'price': current_price,
            'volume': current_volume,
            'timestamp': time.time()
        })
        
        if len(self.price_history) < 2:
            return None
        
        # Calculate price change over time window
        recent_prices = [
            p for p in self.price_history 
            if time.time() - p['timestamp'] <= self.time_window
        ]
        
        if len(recent_prices) < 2:
            return None
        
        oldest_price = recent_prices[0]['price']
        price_change = (current_price - oldest_price) / oldest_price
        
        # Check if spike meets threshold and volume requirements
        if abs(price_change) >= self.spike_threshold:
            avg_volume = sum(p['volume'] for p in recent_prices) / len(recent_prices)
            if current_volume >= avg_volume * 1.5:  # 50% volume increase
                return {
                    'direction': 'UP' if price_change > 0 else 'DOWN',
                    'magnitude': abs(price_change),
                    'volume': current_volume
                }
        
        return None

Automated Trade Execution

def execute_spike_trade(client, market_id, spike_signal):
    """Execute trade when spike is detected."""
    direction = spike_signal['direction']
    magnitude = spike_signal['magnitude']
    
    # Position sizing based on spike magnitude
    # Larger spikes = larger positions (up to max)
    base_size = 100
    size_multiplier = min(magnitude / 0.05, 3.0)  # Cap at 3x
    position_size = int(base_size * size_multiplier)
    
    # Place market order
    order = OrderBuilder.create_market_order(
        market_id=market_id,
        side='BUY' if direction == 'UP' else 'SELL',
        size=position_size
    )
    
    signed_order = client.sign_order(order)
    response = client.create_order(signed_order)
    
    # Set stop-loss at 2% below entry
    stop_loss_price = response['price'] * 0.98 if direction == 'UP' else response['price'] * 1.02
    
    return {
        'order_id': response['id'],
        'entry_price': response['price'],
        'stop_loss': stop_loss_price,
        'take_profit': response['price'] * 1.05 if direction == 'UP' else response['price'] * 0.95
    }

• Market Matching: Identifies equivalent markets (same event, same resolution date)
• Price Comparison: Calculates price difference between platforms
• Arbitrage Detection: Finds opportunities where combined cost < $1 (guaranteed profit)
• Simultaneous Execution: Places orders on both platforms simultaneously
• Profit Locking: One side always wins, ensuring profit regardless of outcome

Arbitrage Opportunity Detection

def find_arbitrage_opportunity(polymarket_price, kalshi_price):
    """
    Find arbitrage opportunity between Polymarket and Kalshi.
    
    Strategy: Buy both YES on Polymarket and Kalshi
    If combined cost < $1, profit is guaranteed.
    """
    # Normalize prices (Polymarket uses 0-1, Kalshi uses 0-100)
    pm_price = polymarket_price  # Already 0-1
    kalshi_price_normalized = kalshi_price / 100  # Convert to 0-1
    
    # Calculate combined cost
    combined_cost = pm_price + kalshi_price_normalized
    
    # Arbitrage exists if combined cost < 1.0
    if combined_cost < 0.98:  # 2% minimum profit margin
        profit_margin = 1.0 - combined_cost
        return {
            'opportunity': True,
            'polymarket_price': pm_price,
            'kalshi_price': kalshi_price_normalized,
            'combined_cost': combined_cost,
            'profit_margin': profit_margin,
            'recommended_size': calculate_position_size(profit_margin)
        }
    
    return {'opportunity': False}

def calculate_position_size(profit_margin):
    """Calculate optimal position size based on profit margin."""
    # Larger profit margins = larger positions
    base_size = 1000  # $1,000 base
    size_multiplier = profit_margin / 0.02  # Scale with margin
    return int(base_size * min(size_multiplier, 5.0))  # Cap at 5x

Simultaneous Order Execution

import asyncio
from py_clob_client.client import ClobClient
from kalshi_client import KalshiClient

async def execute_arbitrage(arb_opportunity):
    """Execute arbitrage trade on both platforms simultaneously."""
    pm_client = ClobClient()
    kalshi_client = KalshiClient()
    
    size = arb_opportunity['recommended_size']
    
    # Execute on both platforms concurrently
    tasks = [
        place_polymarket_order(pm_client, size),
        place_kalshi_order(kalshi_client, size)
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Verify both orders filled
    if any(isinstance(r, Exception) for r in results):
        # Cancel successful order if one failed
        cancel_filled_orders(results)
        return {'success': False, 'error': 'Partial fill'}
    
    return {
        'success': True,
        'polymarket_order': results[0],
        'kalshi_order': results[1],
        'locked_profit': size * arb_opportunity['profit_margin']
    }

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def place_order_with_retry(client, order):
    """Place order with exponential backoff retry."""
    try:
        return client.create_order(order)
    except APIError as e:
        if e.status_code == 429:  # Rate limit
            raise  # Retry with backoff
        elif e.status_code >= 500:  # Server error
            raise  # Retry
        else:
            raise  # Don't retry client errors

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    POLYMARKET_API_KEY = os.getenv('POLYMARKET_API_KEY')
    POLYMARKET_API_SECRET = os.getenv('POLYMARKET_API_SECRET')
    POLYMARKET_API_URL = os.getenv('POLYMARKET_API_URL', 'https://clob.polymarket.com')
    
    MAX_POSITION_SIZE = int(os.getenv('MAX_POSITION_SIZE', '1000'))
    MAX_EXPOSURE = int(os.getenv('MAX_EXPOSURE', '5000'))
    SPREAD_THRESHOLD = float(os.getenv('SPREAD_THRESHOLD', '0.02'))  # 2%
    
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('bot.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def log_trade(trade_data):
    """Log trade execution with full context."""
    logger.info(f"Trade executed: {trade_data}")
    logger.info(f"Market: {trade_data['market_id']}")
    logger.info(f"Side: {trade_data['side']}")
    logger.info(f"Size: {trade_data['size']}")
    logger.info(f"Price: {trade_data['price']}")
    logger.info(f"P&L: {trade_data.get('pnl', 'N/A')}")

• Market Making (poly-maker): Moderate latency acceptable (~500ms). Limit orders can be placed ahead of time.
• Spike Detection (spike-bot): Fast execution required (~500ms). Need to capture spikes before they reverse.
• Arbitrage (arbitrage-bot): Fast execution critical (~500ms). Opportunities disappear quickly.

Choose poly-maker if:

• You have significant capital ($5,000+)
• You understand orderbook dynamics
• You want steady, lower-risk returns
• You can monitor markets continuously

Choose spike-bot if:

• You want to capitalize on news events
• You can execute trades quickly
• You're comfortable with higher risk/reward
• You want automated momentum trading

Choose arbitrage-bot if:

• You have accounts on multiple platforms
• You want risk-free profit opportunities
• You can execute trades simultaneously
• You're comfortable with lower frequency, higher certainty trades