Types of arbitrage
1. Internal arbitrage (YES + NO)
Buy both YES and NO shares when their sum is less than 1 USDC:YES ask: 0.48 USDC
NO ask: 0.49 USDC
Total: 0.97 USDC
Profit: 1.00 - 0.97 = 0.03 USDC per share
2. Cross-market arbitrage
Find the same event on multiple platforms with different prices:Kuest: YES @ 0.65
Polymarket: YES @ 0.60
Strategy: Buy on Polymarket, sell on Kuest
Profit: 0.05 USDC per share
3. Kalshi arbitrage (coming soon)
Arbitrage between Kuest and Kalshi regulated markets:Kuest: Bitcoin > $100k = 72%
Kalshi: Bitcoin > $100k = 68%
Buy on Kalshi, sell on Kuest
Kalshi connector is in development. Track progress on the roadmap.
Detecting opportunities
Python example
from kuest import KuestClient
client = KuestClient(api_key=..., api_secret=..., passphrase=...)
def find_internal_arbitrage(threshold=0.02):
"""Find YES+NO arbitrage opportunities"""
markets = client.get_markets(status="active")
for market in markets:
orderbook = client.get_orderbook(market.id)
if not orderbook.asks:
continue
# Get best ask prices
yes_ask = min([o.price for o in orderbook.asks if o.outcome == "YES"])
no_ask = min([o.price for o in orderbook.asks if o.outcome == "NO"])
total = yes_ask + no_ask
if total < 1.0 - threshold:
profit = 1.0 - total
print(f"Arbitrage in {market.title}:")
print(f" YES: {yes_ask:.4f}, NO: {no_ask:.4f}")
print(f" Total: {total:.4f}")
print(f" Profit: {profit:.4f} USDC ({profit*100:.2f}%)")
yield {
"market_id": market.id,
"yes_ask": yes_ask,
"no_ask": no_ask,
"profit": profit,
}
# Find opportunities
for opportunity in find_internal_arbitrage(threshold=0.02):
print(f"Found arbitrage: {opportunity}")
Rust example
use kuest_sdk::{KuestClient, Outcome};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let client = KuestClient::new(/* ... */)?;
let markets = client.get_markets().await?;
for market in markets {
if let Some(opportunity) = check_arbitrage(&client, &market.id).await? {
println!(
"Arbitrage in {}: profit {:.4} USDC",
market.title, opportunity.profit
);
}
}
Ok(())
}
async fn check_arbitrage(
client: &KuestClient,
market_id: &str,
) -> Result<Option<ArbitrageOpportunity>, Box<dyn std::error::Error>> {
let orderbook = client.get_orderbook(market_id).await?;
let yes_ask = orderbook
.asks
.iter()
.filter(|o| o.outcome == Outcome::Yes)
.map(|o| o.price)
.min_by(|a, b| a.partial_cmp(b).unwrap())
.unwrap_or(1.0);
let no_ask = orderbook
.asks
.iter()
.filter(|o| o.outcome == Outcome::No)
.map(|o| o.price)
.min_by(|a, b| a.partial_cmp(b).unwrap())
.unwrap_or(1.0);
let total = yes_ask + no_ask;
if total < 0.98 {
Ok(Some(ArbitrageOpportunity {
market_id: market_id.to_string(),
yes_ask,
no_ask,
profit: 1.0 - total,
}))
} else {
Ok(None)
}
}
struct ArbitrageOpportunity {
market_id: String,
yes_ask: f64,
no_ask: f64,
profit: f64,
}
Executing arbitrage
Step 1: Calculate position size
def calculate_position_size(yes_ask, no_ask, max_capital=100, max_slippage=0.01):
"""Calculate optimal position size"""
cost_per_share = yes_ask + no_ask
max_shares = max_capital / cost_per_share
# Account for slippage
adjusted_size = max_shares * (1 - max_slippage)
return round(adjusted_size, 2)
Step 2: Place simultaneous orders
import asyncio
async def execute_arbitrage(market_id, yes_ask, no_ask, size):
"""Execute arbitrage trade"""
# Place both orders concurrently
yes_order, no_order = await asyncio.gather(
client.place_order_async(
market_id=market_id,
outcome="YES",
side="BUY",
price=yes_ask,
size=size,
),
client.place_order_async(
market_id=market_id,
outcome="NO",
side="BUY",
price=no_ask,
size=size,
),
)
print(f"YES order: {yes_order.id}")
print(f"NO order: {no_order.id}")
return yes_order, no_order
# Execute
asyncio.run(execute_arbitrage("0x123...", 0.48, 0.49, 10.0))
Step 3: Wait for settlement
After the market resolves, redeem your position:def redeem_arbitrage_position(market_id, size):
"""Redeem position after settlement"""
# Market resolves - you hold both YES and NO
# One will be worth 1 USDC, the other 0 USDC
# Net: 1.00 USDC per share
client.redeem_position(market_id, size)
profit = size * (1.0 - (yes_ask + no_ask))
print(f"Arbitrage profit: {profit:.2f} USDC")
Cross-platform arbitrage
Monitor multiple platforms
import requests
def get_polymarket_price(market_id):
"""Fetch price from Polymarket API"""
response = requests.get(
f"https://clob.polymarket.com/markets/{market_id}"
)
data = response.json()
return data["bestAsk"]
def find_cross_platform_arbitrage():
"""Compare Kuest vs Polymarket"""
# Get Kuest price
kuest_market = client.get_market("0x123...")
kuest_price = kuest_market.yes_price
# Get Polymarket price
polymarket_price = get_polymarket_price("polymarket-id")
spread = abs(kuest_price - polymarket_price)
if spread > 0.03: # 3% spread
print(f"Cross-platform arbitrage:")
print(f" Kuest: {kuest_price:.4f}")
print(f" Polymarket: {polymarket_price:.4f}")
print(f" Spread: {spread:.4f}")
if kuest_price > polymarket_price:
print(" Strategy: Buy Polymarket, Sell Kuest")
else:
print(" Strategy: Buy Kuest, Sell Polymarket")
Risk management
Partial fill risk
If only one order fills, you’re exposed to price risk.
import time
def execute_with_fill_check(market_id, yes_ask, no_ask, size):
"""Execute with partial fill protection"""
# Place orders
yes_order = client.place_order(market_id, "YES", "BUY", yes_ask, size)
no_order = client.place_order(market_id, "NO", "BUY", no_ask, size)
time.sleep(2) # Wait for fills
# Check fill status
yes_status = client.get_order(yes_order.id)
no_status = client.get_order(no_order.id)
if yes_status.filled != no_status.filled:
print("WARNING: Asymmetric fill detected!")
# Cancel unfilled order
if yes_status.filled < size:
client.cancel_order(yes_order.id)
if no_status.filled < size:
client.cancel_order(no_order.id)
# Close partial position
filled_size = min(yes_status.filled, no_status.filled)
# ... close logic
Gas cost considerations
def calculate_net_profit(yes_ask, no_ask, size, gas_cost_usdc=0.01):
"""Calculate profit after gas"""
gross_profit = size * (1.0 - (yes_ask + no_ask))
net_profit = gross_profit - gas_cost_usdc
if net_profit <= 0:
print("Unprofitable after gas")
return None
return net_profit
Slippage monitoring
def check_orderbook_depth(market_id, target_size):
"""Ensure sufficient liquidity"""
orderbook = client.get_orderbook(market_id)
yes_liquidity = sum(o.size for o in orderbook.asks if o.outcome == "YES")
no_liquidity = sum(o.size for o in orderbook.asks if o.outcome == "NO")
if yes_liquidity < target_size or no_liquidity < target_size:
print(f"Insufficient liquidity: YES={yes_liquidity}, NO={no_liquidity}")
return False
return True
Advanced strategies
Statistical arbitrage
Identify mispriced markets using historical data:import pandas as pd
import numpy as np
def find_statistical_arbitrage():
"""Find markets deviating from historical average"""
markets = client.get_markets()
for market in markets:
# Fetch price history
history = client.get_price_history(market.id, interval="1h", limit=168)
df = pd.DataFrame(history)
# Calculate statistics
mean_price = df['price'].mean()
std_price = df['price'].std()
current_price = df['price'].iloc[-1]
# Check for deviation
z_score = (current_price - mean_price) / std_price
if abs(z_score) > 2: # 2 standard deviations
print(f"Statistical arbitrage in {market.title}:")
print(f" Current: {current_price:.4f}")
print(f" Mean: {mean_price:.4f}")
print(f" Z-score: {z_score:.2f}")
Triangular arbitrage
Arbitrage across correlated markets:def triangular_arbitrage():
"""Find arbitrage across 3 related markets"""
# Example: Presidential election markets
# Market A: Democrat wins
# Market B: Republican wins
# Market C: Other wins
price_a = client.get_market("market-a").yes_price
price_b = client.get_market("market-b").yes_price
price_c = client.get_market("market-c").yes_price
total = price_a + price_b + price_c
if total < 0.97: # Should sum to 1.0
print(f"Triangular arbitrage: {total:.4f}")
# Buy all three outcomes
Monitoring and alerts
import smtplib
from email.mime.text import MIMEText
def send_alert(opportunity):
"""Send email alert for arbitrage opportunity"""
msg = MIMEText(f"""
Arbitrage opportunity detected:
Market: {opportunity['market_id']}
Profit: {opportunity['profit']:.4f} USDC
""")
msg['Subject'] = f"Arbitrage Alert: {opportunity['profit']*100:.2f}%"
msg['From'] = "[email protected]"
msg['To'] = "[email protected]"
# Send email
# smtp = smtplib.SMTP('localhost')
# smtp.send_message(msg)
# smtp.quit()
# Monitor continuously
while True:
for opportunity in find_internal_arbitrage(threshold=0.02):
send_alert(opportunity)
time.sleep(30)
Performance metrics
Track your arbitrage bot performance:class ArbitrageMetrics:
def __init__(self):
self.total_trades = 0
self.profitable_trades = 0
self.total_profit = 0.0
self.total_capital_deployed = 0.0
def record_trade(self, profit, capital):
self.total_trades += 1
if profit > 0:
self.profitable_trades += 1
self.total_profit += profit
self.total_capital_deployed += capital
def get_stats(self):
return {
"total_trades": self.total_trades,
"win_rate": self.profitable_trades / self.total_trades,
"total_profit": self.total_profit,
"roi": self.total_profit / self.total_capital_deployed,
"avg_profit_per_trade": self.total_profit / self.total_trades,
}
metrics = ArbitrageMetrics()
# After each trade
metrics.record_trade(profit=0.05, capital=10.0)
print(metrics.get_stats())
Next steps
Python SDK
Build bots with Python SDK
Rust SDK
High-performance Rust SDK
API reference
Complete API documentation
Trading API
Order placement endpoint