Creating Stablecoin Depeg Alert Bot: Price Deviation Notification System

# My terrible first attempt - don't do this
import time
import requests

while True:
    # Query 20 different APIs every 10 seconds
    # No error handling, no rate limiting, no persistence
    price = requests.get("https://api.coinbase.com/v2/exchange-rates").json()
    if price['data']['rates']['USDC'] != '1.00':
        print("ALERT!")  # Yes, I actually used print statements
    time.sleep(10)

import asyncio
import aiohttp
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class PriceData:
    symbol: str
    price: float
    exchange: str
    volume_24h: float
    timestamp: datetime
    
# I learned to use dataclasses after debugging JSON parsing errors for hours
# This saved me from so many "KeyError" exceptions

class StablecoinMonitor:
    def __init__(self):
        self.target_price = 1.00
        self.minor_threshold = 0.005  # 0.5% - learned this from monitoring real depegs
        self.major_threshold = 0.02   # 2% - point of no return in my experience
        self.price_history = {}
        self.exchange_weights = {
            'binance': 0.35,    # Highest volume, most reliable
            'coinbase': 0.25,   # Good US coverage
            'kraken': 0.20,     # European liquidity
            'uniswap': 0.20     # DeFi representation
        }
    
    async def collect_prices(self, symbols: List[str]) -> Dict[str, List[PriceData]]:
        """
        Collects prices from multiple exchanges simultaneously.
        I use asyncio here because fetching from 4 exchanges sequentially 
        was taking 3+ seconds, making alerts too slow.
        """
        tasks = []
        for exchange in self.exchange_weights.keys():
            for symbol in symbols:
                tasks.append(self.fetch_price(exchange, symbol))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return self.organize_results(results)
    
    async def fetch_price(self, exchange: str, symbol: str) -> Optional[PriceData]:
        """
        Fetches price from a specific exchange.
        Error handling here is crucial - I learned this after the bot crashed
        during a network outage and I missed a real depeg event.
        """
        try:
            async with aiohttp.ClientSession() as session:
                url = self.get_api_url(exchange, symbol)
                async with session.get(url, timeout=5) as response:
                    data = await response.json()
                    return self.parse_response(exchange, symbol, data)
        except Exception as e:
            logging.error(f"Failed to fetch {symbol} from {exchange}: {e}")
            return None
    
    def calculate_weighted_price(self, prices: List[PriceData]) -> float:
        """
        Calculates weighted average price across exchanges.
        I added volume weighting after noticing that low-volume exchanges
        showed false depegs during market stress.
        """
        if not prices:
            return None
            
        total_weight = 0
        weighted_sum = 0
        
        for price_data in prices:
            # Weight by exchange preference and volume
            exchange_weight = self.exchange_weights.get(price_data.exchange, 0.1)
            volume_weight = min(price_data.volume_24h / 1000000, 1.0)  # Cap at 1M
            total_weight_for_price = exchange_weight * (0.7 + 0.3 * volume_weight)
            
            weighted_sum += price_data.price * total_weight_for_price
            total_weight += total_weight_for_price
        
        return weighted_sum / total_weight if total_weight > 0 else None

class AlertManager:
    def __init__(self):
        self.cooldown_periods = {
            'minor': timedelta(minutes=5),    # Don't spam minor alerts
            'major': timedelta(minutes=1),    # But be aggressive on major ones
            'critical': timedelta(seconds=30) # Critical needs immediate attention
        }
        self.last_alert_times = {}
        
    def should_send_alert(self, symbol: str, deviation: float, severity: str) -> bool:
        """
        Smart alerting logic that prevents spam while ensuring critical alerts get through.
        This logic evolved from my painful experience of 200+ false alerts in the first week.
        """
        alert_key = f"{symbol}_{severity}"
        now = datetime.now()
        
        # Always send critical alerts (>3% deviation)
        if severity == 'critical':
            return True
            
        # Check cooldown for other severities
        if alert_key in self.last_alert_times:
            time_since_last = now - self.last_alert_times[alert_key]
            if time_since_last < self.cooldown_periods[severity]:
                return False
                
        # Additional context checks I added after too many false positives
        return self.validate_market_context(symbol, deviation)
    
    def validate_market_context(self, symbol: str, deviation: float) -> bool:
        """
        Checks if the deviation is likely real or just noise.
        I added this after getting alerts during normal market-making activities.
        """
        # Check if multiple stablecoins are depegging (systematic risk)
        other_stable_deviations = self.check_correlated_depegs()
        
        # Check recent volume spikes (could indicate real stress)
        volume_spike = self.detect_volume_anomaly(symbol)
        
        # For minor deviations, require additional confirmation
        if abs(deviation) < 0.01:  # Less than 1%
            return other_stable_deviations or volume_spike
            
        return True  # Major deviations always warrant alerts

async def send_alert(self, alert_data: dict):
    """
    Sends alerts through appropriate channels based on severity.
    I learned to vary the urgency after my girlfriend got angry about
    3 AM Discord pings for 0.2% USDT deviations.
    """
    severity = alert_data['severity']
    
    if severity == 'critical':
        # All channels immediately
        await asyncio.gather(
            self.send_discord_alert(alert_data, urgent=True),
            self.send_email_alert(alert_data),
            self.send_sms_alert(alert_data)  # Only for critical events
        )
    elif severity == 'major':
        # Discord and email, but not SMS at night
        tasks = [self.send_discord_alert(alert_data)]
        if self.is_daytime():
            tasks.append(self.send_email_alert(alert_data))
        await asyncio.gather(*tasks)
    else:
        # Just Discord during business hours
        if self.is_business_hours():
            await self.send_discord_alert(alert_data, urgent=False)

def format_alert_message(self, symbol: str, current_price: float, deviation: float, context: dict) -> str:
    """
    Creates informative alert messages with context.
    The format evolved from my early cryptic alerts like "USDC ALERT!"
    which told me nothing useful.
    """
    severity_emoji = {
        'minor': '⚠️',
        'major': '🚨', 
        'critical': '💥'
    }
    
    message = f"{severity_emoji.get(context['severity'], '❓')} **{symbol} Depeg Alert**\n\n"
    message += f"**Current Price**: ${current_price:.4f}\n"
    message += f"**Deviation**: {deviation*100:+.2f}%\n"
    message += f"**Severity**: {context['severity'].upper()}\n\n"
    
    # Add context that helps with decision making
    if context.get('volume_spike'):
        message += "📈 **High volume detected** - potential stress event\n"
    if context.get('correlated_depegs'):
        message += "🔗 **Multiple stablecoins affected** - systematic risk\n"
    if context.get('exchange_discrepancy'):
        message += "🏦 **Large exchange price differences** - arbitrage opportunity\n"
        
    message += f"\n🕐 Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}"
    
    return message

-- Price history table for trend analysis
CREATE TABLE price_history (
    id SERIAL PRIMARY KEY,
    symbol VARCHAR(10) NOT NULL,
    exchange VARCHAR(20) NOT NULL,
    price DECIMAL(10,6) NOT NULL,
    volume_24h BIGINT,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_symbol_timestamp (symbol, timestamp)
);

-- Alert history to prevent duplicate notifications
CREATE TABLE alert_history (
    id SERIAL PRIMARY KEY,
    symbol VARCHAR(10) NOT NULL,
    severity VARCHAR(10) NOT NULL,
    deviation DECIMAL(6,4) NOT NULL,
    message TEXT,
    channels_sent JSON,  -- Track which channels received the alert
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- System health monitoring
CREATE TABLE monitor_health (
    id SERIAL PRIMARY KEY,
    component VARCHAR(50) NOT NULL,
    status VARCHAR(20) NOT NULL,
    last_successful_run TIMESTAMP,
    error_count INT DEFAULT 0,
    last_error TEXT,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

# config.py - Hard-learned production settings
import os

class Config:
    # API endpoints and keys
    EXCHANGE_APIS = {
        'binance': {
            'base_url': 'https://api.binance.com/api/v3/',
            'rate_limit': 1200,  # Requests per minute
            'timeout': 5
        },
        'coinbase': {
            'base_url': 'https://api.coinbase.com/v2/',
            'rate_limit': 10000,  # More generous limit
            'timeout': 5
        }
        # ... other exchanges
    }
    
    # Monitoring intervals I learned through trial and error
    PRICE_CHECK_INTERVAL = 30  # seconds - balance between responsiveness and rate limits
    HEALTH_CHECK_INTERVAL = 300  # 5 minutes - catch system issues quickly
    DATABASE_CLEANUP_INTERVAL = 86400  # Daily cleanup of old data
    
    # Alert thresholds based on 1+ years of observation
    THRESHOLDS = {
        'minor': 0.005,    # 0.5% - early warning
        'major': 0.015,    # 1.5% - requires attention
        'critical': 0.03   # 3% - emergency action needed
    }
    
    # Notification settings
    DISCORD_WEBHOOK = os.getenv('DISCORD_WEBHOOK_URL')
    EMAIL_SETTINGS = {
        'smtp_server': 'smtp.gmail.com',
        'smtp_port': 587,
        'username': os.getenv('EMAIL_USERNAME'),
        'password': os.getenv('EMAIL_APP_PASSWORD')  # Use app passwords!
    }
    
    # Database connection
    DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql://localhost/stablecoin_monitor')
    
    # Logging configuration 
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
    LOG_FILE = '/var/log/stablecoin_monitor.log'

# main.py
import asyncio
import logging
from monitor import StablecoinMonitor
from alerts import AlertManager
from database import DatabaseManager

async def main():
    """
    Main monitoring loop that I've refined through months of operation.
    Key insight: separate concerns and handle failures gracefully.
    """
    # Initialize components
    monitor = StablecoinMonitor()
    alert_manager = AlertManager()
    db_manager = DatabaseManager()
    
    # Symbols to monitor - I started with just USDC, now monitor all major stables
    symbols = ['USDC', 'USDT', 'DAI', 'FRAX', 'BUSD']
    
    logging.info("🚀 Starting stablecoin depeg monitor...")
    
    while True:
        try:
            # Collect prices from all exchanges
            start_time = time.time()
            price_data = await monitor.collect_prices(symbols)
            
            # Process each stablecoin
            for symbol in symbols:
                if symbol not in price_data or not price_data[symbol]:
                    logging.warning(f"No price data for {symbol}")
                    continue
                
                # Calculate weighted average price
                avg_price = monitor.calculate_weighted_price(price_data[symbol])
                if avg_price is None:
                    continue
                
                # Calculate deviation from $1.00 peg
                deviation = (avg_price - 1.0) / 1.0
                
                # Store in database for historical analysis
                await db_manager.store_price_data(symbol, avg_price, price_data[symbol])
                
                # Check if alert is needed
                severity = monitor.determine_severity(abs(deviation))
                if severity and alert_manager.should_send_alert(symbol, deviation, severity):
                    
                    alert_data = {
                        'symbol': symbol,
                        'price': avg_price,
                        'deviation': deviation,
                        'severity': severity,
                        'context': monitor.get_market_context(symbol),
                        'prices_by_exchange': price_data[symbol]
                    }
                    
                    await alert_manager.send_alert(alert_data)
                    await db_manager.log_alert(alert_data)
                    
                    logging.info(f"Alert sent for {symbol}: {deviation*100:.2f}% deviation")
            
            # Performance monitoring - I track this to optimize the system
            cycle_time = time.time() - start_time
            if cycle_time > 10:  # Warn if cycle takes too long
                logging.warning(f"Slow monitoring cycle: {cycle_time:.2f}s")
            
            # Wait for next cycle
            await asyncio.sleep(Config.PRICE_CHECK_INTERVAL)
            
        except Exception as e:
            logging.error(f"Error in main loop: {e}")
            # Continue running despite errors - resilience is key
            await asyncio.sleep(60)  # Wait a bit before retrying

if __name__ == "__main__":
    # Set up logging
    logging.basicConfig(
        level=Config.LOG_LEVEL,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(Config.LOG_FILE),
            logging.StreamHandler()
        ]
    )
    
    # Run the monitoring system
    asyncio.run(main())

# /etc/systemd/system/stablecoin-monitor.service
[Unit]
Description=Stablecoin Depeg Monitor
After=network.target postgresql.service

[Service]
Type=simple
User=monitor
WorkingDirectory=/opt/stablecoin-monitor
Environment=PYTHONPATH=/opt/stablecoin-monitor
ExecStart=/opt/stablecoin-monitor/venv/bin/python main.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

import asyncio
from collections import defaultdict
import time

class RateLimiter:
    def __init__(self):
        self.request_counts = defaultdict(list)
        self.limits = {
            'binance': (1200, 60),    # 1200 requests per 60 seconds
            'coinbase': (10000, 60),  # More generous
        }
    
    async def acquire(self, exchange: str):
        """
        Acquire permission to make a request.
        This saved me from getting banned again.
        """
        max_requests, window = self.limits[exchange]
        now = time.time()
        
        # Clean old requests outside the window
        self.request_counts[exchange] = [
            req_time for req_time in self.request_counts[exchange] 
            if now - req_time < window
        ]
        
        # Check if we're at the limit
        if len(self.request_counts[exchange]) >= max_requests:
            sleep_time = window - (now - self.request_counts[exchange][0])
            await asyncio.sleep(sleep_time)
        
        # Record this request
        self.request_counts[exchange].append(now)

async def fetch_with_retry(self, url: str, max_retries: int = 3) -> dict:
    """
    Robust fetching with exponential backoff.
    I learned this pattern after too many timeout failures.
    """
    for attempt in range(max_retries):
        try:
            timeout = 5 * (2 ** attempt)  # 5s, 10s, 20s
            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session:
                async with session.get(url) as response:
                    return await response.json()
        except asyncio.TimeoutError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

def calculate_dynamic_threshold(self, symbol: str, base_threshold: float) -> float:
    """
    Adjusts alert thresholds based on recent market volatility.
    During high volatility periods, use higher thresholds to reduce noise.
    """
    # Calculate recent volatility (standard deviation of prices)
    recent_prices = self.get_recent_prices(symbol, hours=24)
    if len(recent_prices) < 10:
        return base_threshold
    
    volatility = statistics.stdev(recent_prices)
    
    # Scale threshold based on volatility
    volatility_multiplier = min(1 + volatility * 10, 3.0)  # Cap at 3x
    adjusted_threshold = base_threshold * volatility_multiplier
    
    logging.debug(f"Adjusted threshold for {symbol}: {adjusted_threshold:.4f} (volatility: {volatility:.4f})")
    return adjusted_threshold

def detect_arbitrage_opportunities(self, prices: List[PriceData]) -> Optional[dict]:
    """
    Identifies arbitrage opportunities between exchanges.
    I added this after manually noticing profitable spreads during depegs.
    """
    if len(prices) < 2:
        return None
    
    prices_by_exchange = {p.exchange: p.price for p in prices}
    min_price = min(prices_by_exchange.values())
    max_price = max(prices_by_exchange.values())
    
    spread_percentage = (max_price - min_price) / min_price
    
    if spread_percentage > 0.005:  # 0.5% minimum for profitable arbitrage
        return {
            'buy_exchange': min(prices_by_exchange, key=prices_by_exchange.get),
            'sell_exchange': max(prices_by_exchange, key=prices_by_exchange.get),
            'buy_price': min_price,
            'sell_price': max_price,
            'spread_percentage': spread_percentage,
            'potential_profit': spread_percentage - 0.002  # Account for fees
        }
    
    return None

def analyze_depeg_patterns(self, symbol: str) -> dict:
    """
    Looks for patterns that historically preceded major depegs.
    This is based on analysis of past events like USDC and UST collapses.
    """
    recent_data = self.get_recent_prices(symbol, hours=6)
    
    # Look for concerning patterns
    patterns = {
        'increasing_volatility': self.detect_volatility_trend(recent_data),
        'volume_surge': self.detect_volume_anomaly(symbol),
        'cross_exchange_divergence': self.measure_exchange_divergence(symbol),
        'social_sentiment': self.check_social_mentions(symbol)  # Twitter/Reddit API
    }
    
    # Score the risk level
    risk_score = sum([
        patterns['increasing_volatility'] * 0.3,
        patterns['volume_surge'] * 0.4,
        patterns['cross_exchange_divergence'] * 0.2,
        patterns['social_sentiment'] * 0.1
    ])
    
    return {
        'risk_score': risk_score,
        'patterns': patterns,
        'recommendation': self.get_risk_recommendation(risk_score)
    }