Stablecoin Depeg Detection with Ollama: Monitor USDT, USDC, and DAI Price Stability

# Download and install Ollama
curl -fsSL https://ollama.com/install.sh | sh

# Verify installation
ollama --version

# Pull the required model for Data Analysis
ollama pull llama3.1:8b

pip install requests pandas numpy matplotlib seaborn python-telegram-bot schedule

# config.py
import os

# CoinGecko API (free tier: 50 calls/minute)
COINGECKO_API_URL = "https://api.coingecko.com/api/v3"

# Alternative APIs for redundancy
BINANCE_API_URL = "https://api.binance.com/api/v3"
COINBASE_API_URL = "https://api.coinbase.com/v2"

# Alert thresholds
DEPEG_THRESHOLDS = {
    'minor': {'lower': 0.995, 'upper': 1.005},
    'moderate': {'lower': 0.99, 'upper': 1.01},
    'critical': {'lower': 0.95, 'upper': 1.05}
}

# Monitoring intervals
CHECK_INTERVAL_SECONDS = 60  # Check every minute
ALERT_COOLDOWN_MINUTES = 15  # Prevent spam alerts

# price_collector.py
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
from typing import Dict, List, Optional

class StablecoinPriceCollector:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'StablecoinMonitor/1.0'
        })
        
    def get_current_prices(self) -> Dict[str, float]:
        """Fetch current prices for USDT, USDC, and DAI"""
        
        # Primary source: CoinGecko
        try:
            response = self.session.get(
                f"{COINGECKO_API_URL}/simple/price",
                params={
                    'ids': 'tether,usd-coin,dai',
                    'vs_currencies': 'usd'
                },
                timeout=10
            )
            
            if response.status_code == 200:
                data = response.json()
                return {
                    'USDT': data['tether']['usd'],
                    'USDC': data['usd-coin']['usd'], 
                    'DAI': data['dai']['usd']
                }
        except Exception as e:
            print(f"CoinGecko API error: {e}")
            
        # Fallback to Binance API
        return self._get_binance_prices()
    
    def _get_binance_prices(self) -> Dict[str, float]:
        """Fallback price source"""
        pairs = ['USDTUSD', 'USDCUSD', 'DAIUSD']
        prices = {}
        
        for pair in pairs:
            try:
                response = self.session.get(
                    f"{BINANCE_API_URL}/ticker/price",
                    params={'symbol': pair}
                )
                if response.status_code == 200:
                    data = response.json()
                    token = pair[:4] if pair.startswith('USDT') else pair[:3]
                    prices[token] = float(data['price'])
            except:
                continue
                
        return prices
    
    def get_price_history(self, days: int = 7) -> pd.DataFrame:
        """Fetch historical price data"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        # Simulate historical data collection
        # In production, store in database
        dates = pd.date_range(start_date, end_date, freq='H')
        
        return pd.DataFrame({
            'timestamp': dates,
            'USDT': 1.0 + (pd.Series(range(len(dates))) * 0.001).cumsum(),
            'USDC': 1.0 + (pd.Series(range(len(dates))) * 0.0005).cumsum(),
            'DAI': 1.0 + (pd.Series(range(len(dates))) * 0.002).cumsum()
        })

# depeg_detector.py
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime

@dataclass
class DepegAlert:
    token: str
    current_price: float
    deviation: float
    severity: str
    timestamp: datetime
    duration_minutes: int

class DepegDetector:
    def __init__(self, thresholds: Dict):
        self.thresholds = thresholds
        self.price_history = []
        self.active_depegs = {}
        
    def analyze_price_deviation(self, prices: Dict[str, float]) -> List[DepegAlert]:
        """Detect depegging events across all monitored stablecoins"""
        alerts = []
        current_time = datetime.now()
        
        for token, price in prices.items():
            deviation = abs(price - 1.0)
            severity = self._classify_severity(deviation)
            
            if severity != 'normal':
                # Track depeg duration
                if token in self.active_depegs:
                    duration = (current_time - self.active_depegs[token]).total_seconds() / 60
                else:
                    self.active_depegs[token] = current_time
                    duration = 0
                
                alerts.append(DepegAlert(
                    token=token,
                    current_price=price,
                    deviation=deviation,
                    severity=severity,
                    timestamp=current_time,
                    duration_minutes=int(duration)
                ))
            else:
                # Clear resolved depeg
                self.active_depegs.pop(token, None)
                
        return alerts
    
    def _classify_severity(self, deviation: float) -> str:
        """Classify depeg severity based on price deviation"""
        if deviation >= (1 - self.thresholds['critical']['lower']):
            return 'critical'
        elif deviation >= (1 - self.thresholds['moderate']['lower']):
            return 'moderate'  
        elif deviation >= (1 - self.thresholds['minor']['lower']):
            return 'minor'
        else:
            return 'normal'
    
    def calculate_volatility_metrics(self, token: str, prices: List[float]) -> Dict:
        """Calculate advanced volatility metrics"""
        if len(prices) < 10:
            return {'std_dev': 0, 'max_deviation': 0, 'trend': 'stable'}
            
        prices_array = np.array(prices)
        
        return {
            'std_dev': np.std(prices_array),
            'max_deviation': max(abs(prices_array - 1.0)),
            'trend': 'increasing' if prices_array[-1] > prices_array[0] else 'decreasing',
            'volatility_score': np.std(prices_array) * 100
        }

# token_monitors.py
from abc import ABC, abstractmethod

class TokenMonitor(ABC):
    @abstractmethod
    def get_risk_factors(self) -> Dict[str, float]:
        pass
    
    @abstractmethod
    def custom_alert_logic(self, price: float, history: List[float]) -> bool:
        pass

class USDTMonitor(TokenMonitor):
    """Tether-specific monitoring logic"""
    
    def get_risk_factors(self) -> Dict[str, float]:
        return {
            'regulatory_risk': 0.7,  # High due to opacity
            'liquidity_risk': 0.3,   # Generally high liquidity
            'counterparty_risk': 0.8 # Centralized backing
        }
    
    def custom_alert_logic(self, price: float, history: List[float]) -> bool:
        # USDT specific: Alert on sustained deviation
        if len(history) >= 5:
            recent_avg = sum(history[-5:]) / 5
            return abs(recent_avg - 1.0) > 0.005
        return False

class USDCMonitor(TokenMonitor):
    """USD Coin specific monitoring"""
    
    def get_risk_factors(self) -> Dict[str, float]:
        return {
            'regulatory_risk': 0.2,  # Well regulated
            'liquidity_risk': 0.2,   # High liquidity
            'counterparty_risk': 0.4 # Coinbase backing
        }
    
    def custom_alert_logic(self, price: float, history: List[float]) -> bool:
        # USDC: More sensitive to banking sector news
        return abs(price - 1.0) > 0.003

class DAIMonitor(TokenMonitor):
    """DAI algorithmic stablecoin monitoring"""
    
    def get_risk_factors(self) -> Dict[str, float]:
        return {
            'regulatory_risk': 0.4,  # DeFi regulatory uncertainty
            'liquidity_risk': 0.5,   # Lower than centralized coins
            'counterparty_risk': 0.3 # Decentralized collateral
        }
    
    def custom_alert_logic(self, price: float, history: List[float]) -> bool:
        # DAI: Monitor for algorithmic instability
        if len(history) >= 3:
            volatility = np.std(history[-10:]) if len(history) >= 10 else 0
            return volatility > 0.01 or abs(price - 1.0) > 0.008
        return False

# main_monitor.py
import asyncio
import logging
from datetime import datetime
import json

class StablecoinMonitoringSystem:
    def __init__(self):
        self.collector = StablecoinPriceCollector()
        self.detector = DepegDetector(DEPEG_THRESHOLDS)
        self.monitors = {
            'USDT': USDTMonitor(),
            'USDC': USDCMonitor(), 
            'DAI': DAIMonitor()
        }
        self.alert_manager = AlertManager()
        
    async def run_monitoring_loop(self):
        """Main monitoring loop"""
        while True:
            try:
                # Collect current prices
                current_prices = self.collector.get_current_prices()
                
                if not current_prices:
                    logging.warning("Failed to fetch prices, retrying...")
                    await asyncio.sleep(30)
                    continue
                
                # Detect depegging events
                alerts = self.detector.analyze_price_deviation(current_prices)
                
                # Apply token-specific logic
                for token, price in current_prices.items():
                    if token in self.monitors:
                        monitor = self.monitors[token]
                        if monitor.custom_alert_logic(price, []):  # Add history
                            # Create custom alert
                            pass
                
                # Process alerts
                if alerts:
                    await self.alert_manager.process_alerts(alerts)
                
                # Log status
                self._log_status(current_prices, alerts)
                
                # Wait for next check
                await asyncio.sleep(CHECK_INTERVAL_SECONDS)
                
            except Exception as e:
                logging.error(f"Monitoring loop error: {e}")
                await asyncio.sleep(60)
    
    def _log_status(self, prices: Dict, alerts: List):
        """Log current system status"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        status = {
            'timestamp': timestamp,
            'prices': prices,
            'active_alerts': len(alerts),
            'system_status': 'healthy' if not alerts else 'alerts_active'
        }
        
        logging.info(f"Status: {json.dumps(status, indent=2)}")

# alert_manager.py
import asyncio
from telegram import Bot
import smtplib
from email.mime.text import MIMEText
import json
import requests

class AlertManager:
    def __init__(self):
        self.telegram_bot = Bot(token=os.getenv('TELEGRAM_BOT_TOKEN'))
        self.alert_history = []
        self.cooldown_tracker = {}
        
    async def process_alerts(self, alerts: List[DepegAlert]):
        """Process and route alerts to appropriate channels"""
        for alert in alerts:
            if self._should_send_alert(alert):
                await asyncio.gather(
                    self._send_telegram_alert(alert),
                    self._send_email_alert(alert),
                    self._log_alert(alert)
                )
                
                self.cooldown_tracker[alert.token] = datetime.now()
    
    def _should_send_alert(self, alert: DepegAlert) -> bool:
        """Prevent alert spam with cooldown logic"""
        if alert.token not in self.cooldown_tracker:
            return True
            
        last_alert = self.cooldown_tracker[alert.token]
        cooldown_period = timedelta(minutes=ALERT_COOLDOWN_MINUTES)
        
        # Send if cooldown expired or severity increased
        return (datetime.now() - last_alert) > cooldown_period
    
    async def _send_telegram_alert(self, alert: DepegAlert):
        """Send alert via Telegram"""
        try:
            severity_emoji = {
                'minor': '🟡',
                'moderate': '🟠', 
                'critical': '🔴'
            }
            
            message = f"""
{severity_emoji.get(alert.severity, '⚪')} **{alert.severity.upper()} DEPEG ALERT**

🪙 **Token**: {alert.token}
💰 **Price**: ${alert.current_price:.6f}
📊 **Deviation**: {alert.deviation:.4f} ({alert.deviation*100:.2f}%)
⏱️ **Duration**: {alert.duration_minutes} minutes
🕒 **Time**: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')}

Monitor your positions closely!
            """
            
            chat_id = os.getenv('TELEGRAM_CHAT_ID')
            await self.telegram_bot.send_message(
                chat_id=chat_id,
                text=message,
                parse_mode='Markdown'
            )
            
        except Exception as e:
            logging.error(f"Telegram alert failed: {e}")
    
    async def _send_email_alert(self, alert: DepegAlert):
        """Send email alert for critical events"""
        if alert.severity != 'critical':
            return
            
        try:
            subject = f"CRITICAL: {alert.token} Depeg Detected - ${alert.current_price:.4f}"
            
            body = f"""
CRITICAL STABLECOIN DEPEGGING ALERT

Token: {alert.token}
Current Price: ${alert.current_price:.6f}
Deviation: {alert.deviation:.4f} ({alert.deviation*100:.2f}%)
Duration: {alert.duration_minutes} minutes
Timestamp: {alert.timestamp}

Immediate action may be required to protect your positions.

This is an automated alert from your stablecoin monitoring system.
            """
            
            msg = MIMEText(body)
            msg['Subject'] = subject
            msg['From'] = os.getenv('SMTP_FROM_EMAIL')
            msg['To'] = os.getenv('ALERT_EMAIL')
            
            # Send via SMTP (configure your email provider)
            # Implementation depends on your email service
            
        except Exception as e:
            logging.error(f"Email alert failed: {e}")
    
    def _log_alert(self, alert: DepegAlert):
        """Log alert to file and database"""
        alert_data = {
            'timestamp': alert.timestamp.isoformat(),
            'token': alert.token,
            'price': alert.current_price,
            'deviation': alert.deviation,
            'severity': alert.severity,
            'duration_minutes': alert.duration_minutes
        }
        
        self.alert_history.append(alert_data)
        
        # Write to log file
        with open('alerts.log', 'a') as f:
            f.write(f"{json.dumps(alert_data)}\n")

# auto_response.py
class AutomatedResponseSystem:
    def __init__(self):
        self.response_config = {
            'critical': {
                'enable_trading_halt': True,
                'liquidate_percentage': 0.5,  # Liquidate 50% on critical depeg
                'notify_all_channels': True
            },
            'moderate': {
                'enable_trading_halt': False,
                'liquidate_percentage': 0.2,  # Liquidate 20% on moderate depeg
                'notify_all_channels': True
            }
        }
    
    async def execute_response(self, alert: DepegAlert):
        """Execute automated response based on severity"""
        config = self.response_config.get(alert.severity)
        
        if not config:
            return
            
        if config['enable_trading_halt']:
            await self._halt_trading(alert.token)
            
        if config['liquidate_percentage'] > 0:
            await self._partial_liquidation(alert.token, config['liquidate_percentage'])
    
    async def _halt_trading(self, token: str):
        """Halt automated trading for the affected token"""
        # Implement your trading bot integration
        logging.warning(f"Trading halted for {token}")
    
    async def _partial_liquidation(self, token: str, percentage: float):
        """Execute partial liquidation"""
        # Implement your exchange integration
        logging.warning(f"Liquidating {percentage*100}% of {token} positions")

# system_health.py
class SystemHealthMonitor:
    def __init__(self):
        self.health_metrics = {
            'api_response_times': [],
            'failed_requests': 0,
            'last_successful_check': None,
            'system_uptime': datetime.now()
        }
    
    def check_system_health(self) -> Dict[str, any]:
        """Comprehensive system health check"""
        health_status = {
            'overall_status': 'healthy',
            'api_connectivity': self._test_api_connectivity(),
            'alert_system': self._test_alert_system(),
            'data_freshness': self._check_data_freshness(),
            'memory_usage': self._get_memory_usage()
        }
        
        # Determine overall health
        if not all([
            health_status['api_connectivity'],
            health_status['alert_system'],
            health_status['data_freshness']
        ]):
            health_status['overall_status'] = 'degraded'
            
        return health_status
    
    def _test_api_connectivity(self) -> bool:
        """Test connectivity to price data APIs"""
        try:
            response = requests.get(f"{COINGECKO_API_URL}/ping", timeout=5)
            return response.status_code == 200
        except:
            return False
    
    def _test_alert_system(self) -> bool:
        """Test alert system functionality"""
        # Test telegram bot connection
        try:
            bot = Bot(token=os.getenv('TELEGRAM_BOT_TOKEN'))
            bot.get_me()  # Simple API call to test connection
            return True
        except:
            return False
    
    def _check_data_freshness(self) -> bool:
        """Ensure data is recent"""
        if self.health_metrics['last_successful_check']:
            time_diff = datetime.now() - self.health_metrics['last_successful_check']
            return time_diff.total_seconds() < 300  # 5 minutes
        return False
    
    def _get_memory_usage(self) -> float:
        """Get current memory usage percentage"""
        import psutil
        return psutil.virtual_memory().percent

# troubleshooting.py
class TroubleshootingSystem:
    def __init__(self):
        self.error_patterns = {
            'rate_limit': ['429', 'rate limit', 'too many requests'],
            'network_error': ['connection', 'timeout', 'unreachable'],
            'api_error': ['400', '401', '403', '500', '502', '503']
        }
    
    def diagnose_error(self, error_message: str) -> Dict[str, str]:
        """Diagnose common errors and suggest solutions"""
        error_message_lower = error_message.lower()
        
        for error_type, patterns in self.error_patterns.items():
            if any(pattern in error_message_lower for pattern in patterns):
                return {
                    'error_type': error_type,
                    'suggestion': self._get_solution(error_type)
                }
        
        return {'error_type': 'unknown', 'suggestion': 'Check logs for details'}
    
    def _get_solution(self, error_type: str) -> str:
        solutions = {
            'rate_limit': 'Implement exponential backoff, reduce request frequency',
            'network_error': 'Check internet connection, use VPN if needed',
            'api_error': 'Verify API keys, check service status'
        }
        return solutions.get(error_type, 'Contact support')

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Ollama
RUN curl -fsSL https://ollama.com/install.sh | sh

# Copy requirements and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user
RUN useradd -m -u 1000 monitor
USER monitor

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python health_check.py

# Start the monitoring system
CMD ["python", "main_monitor.py"]

# docker-compose.yml
version: '3.8'

services:
  stablecoin-monitor:
    build: .
    restart: unless-stopped
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
      - ALERT_EMAIL=${ALERT_EMAIL}
      - COINGECKO_API_KEY=${COINGECKO_API_KEY}
    volumes:
      - ./logs:/app/logs
      - ./data:/app/data
    networks:
      - monitoring

  redis:
    image: redis:7-alpine
    restart: unless-stopped
    networks:
      - monitoring

  prometheus:
    image: prom/prometheus
    restart: unless-stopped
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    networks:
      - monitoring

networks:
  monitoring:
    driver: bridge

# ml_detector.py
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class MLDepegDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.scaler = StandardScaler()
        self.is_trained = False
        
    def train_model(self, historical_data: pd.DataFrame):
        """Train anomaly detection model"""
        features = self._extract_features(historical_data)
        
        # Scale features
        features_scaled = self.scaler.fit_transform(features)
        
        # Train isolation forest
        self.model.fit(features_scaled)
        self.is_trained = True
        
    def predict_anomaly(self, current_data: Dict) -> float:
        """Predict anomaly score for current data"""
        if not self.is_trained:
            return 0.0
            
        features = self._extract_current_features(current_data)
        features_scaled = self.scaler.transform([features])
        
        # Get anomaly score (-1 for anomaly, 1 for normal)
        anomaly_score = self.model.decision_function(features_scaled)[0]
        
        # Convert to probability (0-1 scale)
        return max(0, 1 - (anomaly_score + 0.5))
    
    def _extract_features(self, data: pd.DataFrame) -> np.ndarray:
        """Extract features for ML model"""
        features = []
        
        for _, row in data.iterrows():
            feature_vector = [
                row['USDT'], row['USDC'], row['DAI'],  # Current prices
                abs(row['USDT'] - 1.0),  # USDT deviation
                abs(row['USDC'] - 1.0),  # USDC deviation
                abs(row['DAI'] - 1.0),   # DAI deviation
            ]
            features.append(feature_vector)
            
        return np.array(features)