Building a Cryptocurrency Exchange Rate Monitor
Crypto markets run 24/7 and prices can swing 10% in minutes. In this tutorial, we will build a Python-based cryptocurrency exchange rate monitor that tracks prices across exchanges, detects arbitrage opportunities, and sends alerts.
Setup
pip install requests pandas websocket-client
Fetching Prices from CoinGecko
CoinGecko offers a generous free API with no key required:
import requests
import pandas as pd
from datetime import datetime
def get_crypto_prices(coins, vs_currency="usd"):
url = "https://api.coingecko.com/api/v3/simple/price"
params = {
"ids": ",".join(coins),
"vs_currencies": vs_currency,
"include_24hr_change": "true",
"include_market_cap": "true",
"include_24hr_vol": "true"
}
response = requests.get(url, params=params)
data = response.json()
results = []
for coin, info in data.items():
results.append({
"coin": coin,
"price": info.get(f"{vs_currency}", 0),
"change_24h": info.get(f"{vs_currency}_24h_change", 0),
"market_cap": info.get(f"{vs_currency}_market_cap", 0),
"volume_24h": info.get(f"{vs_currency}_24h_vol", 0),
"timestamp": datetime.now().isoformat()
})
return pd.DataFrame(results)
coins = ["bitcoin", "ethereum", "solana", "cardano", "polkadot"]
df = get_crypto_prices(coins)
print(df[["coin", "price", "change_24h"]].to_string(index=False))
Multi-Exchange Price Comparison
Different exchanges have different prices. This creates arbitrage opportunities:
def get_binance_price(symbol):
url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}USDT"
response = requests.get(url)
if response.status_code == 200:
return float(response.json()["price"])
return None
def get_kraken_price(pair):
url = f"https://api.kraken.com/0/public/Ticker?pair={pair}"
response = requests.get(url)
if response.status_code == 200:
data = response.json()
result = data.get("result", {})
for key, value in result.items():
return float(value["c"][0])
return None
def compare_exchanges(symbol, kraken_pair):
binance = get_binance_price(symbol)
kraken = get_kraken_price(kraken_pair)
if binance and kraken:
spread = abs(binance - kraken)
spread_pct = (spread / min(binance, kraken)) * 100
return {
"symbol": symbol,
"binance": binance,
"kraken": kraken,
"spread": spread,
"spread_pct": round(spread_pct, 4),
"arbitrage": "BUY Kraken" if kraken < binance else "BUY Binance"
}
return None
for symbol, pair in [("BTC", "XBTUSD"), ("ETH", "ETHUSD")]:
result = compare_exchanges(symbol, pair)
if result:
print(f"{result['symbol']}: Binance ${result['binance']:.2f} | Kraken ${result['kraken']:.2f} | Spread {result['spread_pct']}%")
Real-Time WebSocket Monitoring
import websocket
import json
def monitor_binance_ws(symbols, callback):
streams = "/".join([f"{s.lower()}usdt@ticker" for s in symbols])
ws_url = f"wss://stream.binance.com:9443/ws/{streams}"
def on_message(ws, message):
data = json.loads(message)
ticker = {
"symbol": data["s"],
"price": float(data["c"]),
"change_pct": float(data["P"]),
"volume": float(data["v"]),
"timestamp": datetime.now().isoformat()
}
callback(ticker)
ws = websocket.WebSocketApp(ws_url, on_message=on_message)
ws.run_forever()
def print_ticker(ticker):
direction = "+" if ticker["change_pct"] > 0 else ""
print(f"{ticker['symbol']}: ${ticker['price']:.2f} ({direction}{ticker['change_pct']:.2f}%)")
Price Alert System
class PriceAlertMonitor:
def __init__(self):
self.alerts = []
self.triggered = set()
def add_alert(self, coin, condition, threshold):
self.alerts.append({
"coin": coin,
"condition": condition,
"threshold": threshold
})
def check_alerts(self, prices_df):
triggered = []
for alert in self.alerts:
row = prices_df[prices_df["coin"] == alert["coin"]]
if row.empty:
continue
price = row.iloc[0]["price"]
alert_key = f"{alert['coin']}_{alert['condition']}_{alert['threshold']}"
if alert["condition"] == "above" and price > alert["threshold"]:
if alert_key not in self.triggered:
triggered.append(f"{alert['coin']} is ABOVE ${alert['threshold']:,.2f} (current: ${price:,.2f})")
self.triggered.add(alert_key)
elif alert["condition"] == "below" and price < alert["threshold"]:
if alert_key not in self.triggered:
triggered.append(f"{alert['coin']} is BELOW ${alert['threshold']:,.2f} (current: ${price:,.2f})")
self.triggered.add(alert_key)
return triggered
monitor = PriceAlertMonitor()
monitor.add_alert("bitcoin", "below", 50000)
monitor.add_alert("ethereum", "above", 5000)
Historical Data Collection
def get_price_history(coin, days=30):
url = f"https://api.coingecko.com/api/v3/coins/{coin}/market_chart"
params = {"vs_currency": "usd", "days": days}
response = requests.get(url, params=params)
data = response.json()
prices = pd.DataFrame(data["prices"], columns=["timestamp", "price"])
prices["date"] = pd.to_datetime(prices["timestamp"], unit="ms")
prices = prices.drop("timestamp", axis=1)
return prices
btc_history = get_price_history("bitcoin", days=90)
print(f"BTC 90-day range: ${btc_history['price'].min():,.0f} - ${btc_history['price'].max():,.0f}")
Scaling and Reliability
For monitoring many coins across many exchanges, use ScraperAPI when scraping exchange websites that block bots. ThorData proxies help access geo-restricted exchanges. ScrapeOps monitors your data pipeline uptime.
Tips
- Use WebSockets for real-time data instead of polling REST APIs
- Cache historical data locally to reduce API calls
- Account for fees when calculating arbitrage opportunities
- Use UTC timestamps consistently across all data sources
- Rate limit your requests — CoinGecko allows 10-30 calls/minute on free tier
Follow for more Python cryptocurrency and finance tutorials!
Top comments (0)