How I Built a Stablecoin Incident Analytics Framework After Losing $50K in UST

import asyncio
import websockets
from web3 import Web3
import pandas as pd
from dataclasses import dataclass
from typing import Dict, List
import logging

@dataclass
class StablecoinEvent:
    """Core event structure for incident tracking"""
    timestamp: int
    token_address: str
    event_type: str  # 'large_redemption', 'liquidity_drain', 'price_deviation'
    severity: str    # 'low', 'medium', 'high', 'critical'
    raw_data: Dict
    impact_score: float

class StablecoinMonitor:
    def __init__(self, rpc_urls: Dict[str, str], monitored_tokens: List[str]):
        self.rpc_urls = rpc_urls
        self.monitored_tokens = monitored_tokens
        self.w3_connections = {}
        self.event_buffer = []
        
        # Initialize connections to multiple chains
        for chain, url in rpc_urls.items():
            self.w3_connections[chain] = Web3(Web3.HTTPProvider(url))
    
    async def monitor_liquidity_events(self, token_address: str, chain: str):
        """
        Monitor DEX liquidity changes that precede depegging
        This caught the Curve 3pool drain 4 hours before UST hit $0.60
        """
        w3 = self.w3_connections[chain]
        
        # Monitor major DEX pools for this token
        pool_addresses = self.get_major_pools(token_address, chain)
        
        for pool in pool_addresses:
            try:
                # Get current reserves
                current_reserves = self.get_pool_reserves(pool, w3)
                
                # Check for unusual outflows (>10% in 1 hour)
                if self.detect_liquidity_drain(current_reserves, pool):
                    event = StablecoinEvent(
                        timestamp=int(time.time()),
                        token_address=token_address,
                        event_type='liquidity_drain',
                        severity=self.calculate_severity(current_reserves),
                        raw_data={'pool': pool, 'reserves': current_reserves},
                        impact_score=self.calculate_impact(current_reserves)
                    )
                    await self.process_event(event)
                    
            except Exception as e:
                logging.error(f"Error monitoring pool {pool}: {e}")
                continue
    
    def detect_liquidity_drain(self, reserves: Dict, pool_address: str) -> bool:
        """
        The key insight: massive redemptions happen before price crashes
        I track the rate of change, not just absolute levels
        """
        historical_reserves = self.get_historical_reserves(pool_address, hours=1)
        
        for token, current_amount in reserves.items():
            if token in historical_reserves:
                hour_ago = historical_reserves[token]
                drain_rate = (hour_ago - current_amount) / hour_ago
                
                # Flag if >15% drained in 1 hour (learned from UST analysis)
                if drain_rate > 0.15:
                    return True
        
        return False

import heapq
from enum import Enum
import asyncio
from datetime import datetime, timedelta

class EventPriority(Enum):
    LOW = 4
    MEDIUM = 3  
    HIGH = 2
    CRITICAL = 1

class EventProcessor:
    def __init__(self):
        self.event_queue = []
        self.processing = False
        self.alert_thresholds = {
            'liquidity_drain': 0.20,  # 20% drain triggers alert
            'price_deviation': 0.02,   # 2% depeg triggers alert  
            'volume_spike': 5.0,       # 5x normal volume
            'redemption_rush': 0.30    # 30% of supply redeemed
        }
    
    async def add_event(self, event: StablecoinEvent):
        """Add event to priority queue based on severity"""
        priority = self.get_event_priority(event)
        heapq.heappush(self.event_queue, (priority.value, event.timestamp, event))
        
        if not self.processing:
            await self.start_processing()
    
    def get_event_priority(self, event: StablecoinEvent) -> EventPriority:
        """
        Priority logic I learned the hard way during various depegs
        Multiple simultaneous events = crisis mode
        """
        if event.impact_score > 0.8:
            return EventPriority.CRITICAL
        elif event.event_type == 'liquidity_drain' and event.impact_score > 0.5:
            return EventPriority.HIGH
        elif self.detect_cascade_risk(event):
            return EventPriority.HIGH
        else:
            return EventPriority.MEDIUM
    
    def detect_cascade_risk(self, event: StablecoinEvent) -> bool:
        """
        Check if this event could trigger cascading failures
        Learned this during the March 2023 banking crisis
        """
        recent_events = [e for _, _, e in self.event_queue[-10:]]
        
        # Multiple events in short timeframe = cascade risk
        high_impact_events = [e for e in recent_events if e.impact_score > 0.6]
        
        return len(high_impact_events) >= 3
    
    async def process_event(self, event: StablecoinEvent):
        """Process individual events and trigger appropriate responses"""
        if event.severity == 'critical':
            await self.send_immediate_alert(event)
            await self.trigger_emergency_analysis(event)
        
        # Store for post-mortem analysis
        await self.store_event(event)
        
        # Update real-time dashboard
        await self.update_dashboard(event)

import numpy as np
from scipy.stats import pearsonr
from collections import defaultdict
import asyncio

class CrossChainAnalyzer:
    def __init__(self, chains: List[str]):
        self.chains = chains
        self.correlation_window = 3600  # 1 hour correlation window
        self.event_history = defaultdict(list)
        
    async def analyze_cross_chain_correlation(self, token_address: str) -> Dict:
        """
        Analyze how events correlate across different chains
        Saved me during the USDC crisis by showing Polygon stress first
        """
        correlations = {}
        chain_events = {}
        
        # Get recent events for each chain
        for chain in self.chains:
            events = await self.get_recent_events(token_address, chain)
            if events:
                chain_events[chain] = self.aggregate_impact_scores(events)
        
        # Calculate pairwise correlations
        for chain1 in chain_events:
            for chain2 in chain_events:
                if chain1 != chain2:
                    correlation = self.calculate_correlation(
                        chain_events[chain1], 
                        chain_events[chain2]
                    )
                    correlations[f"{chain1}_{chain2}"] = correlation
        
        return {
            'correlations': correlations,
            'risk_spreading': self.detect_risk_spreading(correlations),
            'lead_chain': self.identify_lead_indicator(chain_events)
        }
    
    def detect_risk_spreading(self, correlations: Dict) -> bool:
        """
        Detect if stress is spreading across chains
        High correlation during crisis = contagion risk
        """
        high_correlations = [v for v in correlations.values() if v > 0.7]
        return len(high_correlations) > len(self.chains) * 0.5
    
    def identify_lead_indicator(self, chain_events: Dict) -> str:
        """
        Find which chain typically shows stress signals first
        Usually it's the chain with highest DEX volume
        """
        lead_scores = {}
        
        for chain in chain_events:
            # Calculate how often this chain shows stress first
            events = chain_events[chain]
            early_signals = sum(1 for event in events if event['is_early_signal'])
            lead_scores[chain] = early_signals / len(events) if events else 0
        
        return max(lead_scores, key=lead_scores.get) if lead_scores else None

from dataclasses import dataclass
from typing import List, Optional
import matplotlib.pyplot as plt
import pandas as pd

@dataclass
class IncidentTimeline:
    incident_id: str
    start_time: int
    end_time: Optional[int]
    events: List[StablecoinEvent]
    severity_peak: float
    recovery_time: Optional[int]

class PostMortemAnalyzer:
    def __init__(self):
        self.incident_patterns = {}
        self.recovery_strategies = {}
    
    async def generate_incident_report(self, incident_id: str) -> Dict:
        """
        Generate comprehensive post-mortem analysis
        This analysis helped me understand why I missed the UST signals
        """
        timeline = await self.reconstruct_timeline(incident_id)
        
        analysis = {
            'timeline': timeline,
            'root_cause_analysis': await self.identify_root_cause(timeline),
            'early_warning_missed': await self.analyze_missed_signals(timeline),
            'similar_incidents': await self.find_similar_patterns(timeline),
            'recovery_analysis': await self.analyze_recovery_pattern(timeline),
            'lessons_learned': await self.extract_lessons(timeline)
        }
        
        # Generate visual timeline
        await self.create_timeline_visualization(timeline, incident_id)
        
        return analysis
    
    async def identify_root_cause(self, timeline: IncidentTimeline) -> Dict:
        """
        Identify the fundamental cause of the incident
        For UST: algorithmic design flaw + reflexivity death spiral
        """
        events = timeline.events
        
        # Categorize events by type
        liquidity_events = [e for e in events if e.event_type == 'liquidity_drain']
        price_events = [e for e in events if e.event_type == 'price_deviation'] 
        volume_events = [e for e in events if e.event_type == 'volume_spike']
        
        # Analyze sequence to find trigger
        if liquidity_events and liquidity_events[0].timestamp < price_events[0].timestamp:
            root_cause = "liquidity_crisis"
            confidence = 0.8
        elif len(volume_events) > 10 and volume_events[0].impact_score > 0.9:
            root_cause = "coordinated_attack"  
            confidence = 0.7
        else:
            root_cause = "market_stress"
            confidence = 0.6
        
        return {
            'primary_cause': root_cause,
            'confidence': confidence,
            'contributing_factors': self.identify_contributing_factors(events),
            'trigger_event': events[0] if events else None
        }
    
    async def analyze_missed_signals(self, timeline: IncidentTimeline) -> List[Dict]:
        """
        Identify early warning signals that were missed
        This is the most painful but valuable part of the analysis
        """
        missed_signals = []
        
        # Check for gradual liquidity decline before crisis
        pre_crisis_period = timeline.start_time - 86400  # 24 hours before
        pre_crisis_events = await self.get_events_in_period(
            timeline.incident_id, 
            pre_crisis_period, 
            timeline.start_time
        )
        
        # Analyze what we should have caught
        for event in pre_crisis_events:
            if event.impact_score > 0.3 and event.severity == 'low':
                missed_signals.append({
                    'event': event,
                    'should_have_alerted': True,
                    'reason': 'Cumulative impact underestimated',
                    'recommended_threshold': event.impact_score * 0.7
                })
        
        return missed_signals

import sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
import joblib

class IncidentPredictor:
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.feature_columns = [
            'liquidity_drain_rate',
            'price_deviation_magnitude', 
            'volume_anomaly_score',
            'cross_chain_correlation',
            'redemption_rate',
            'collateral_ratio_change'
        ]
    
    def prepare_training_data(self, historical_incidents: List[IncidentTimeline]):
        """
        Prepare features from historical incidents for ML training
        Using 50+ historical depegging events as training data
        """
        features = []
        labels = []
        
        for incident in historical_incidents:
            # Extract features from the 6 hours before incident peak
            pre_peak_events = self.get_pre_peak_events(incident, hours=6)
            feature_vector = self.extract_features(pre_peak_events)
            
            features.append(feature_vector)
            # Label: 1 for major incident (>5% depeg), 0 for minor
            labels.append(1 if incident.severity_peak > 0.05 else 0)
        
        return np.array(features), np.array(labels)
    
    def extract_features(self, events: List[StablecoinEvent]) -> List[float]:
        """Extract numerical features for ML model"""
        if not events:
            return [0.0] * len(self.feature_columns)
        
        # Aggregate events into feature vector
        liquidity_events = [e for e in events if e.event_type == 'liquidity_drain']
        price_events = [e for e in events if e.event_type == 'price_deviation']
        volume_events = [e for e in events if e.event_type == 'volume_spike']
        
        features = [
            np.mean([e.impact_score for e in liquidity_events]) if liquidity_events else 0,
            np.max([e.impact_score for e in price_events]) if price_events else 0,
            len(volume_events) / 6.0,  # Normalized by 6-hour window
            self.calculate_correlation_score(events),
            self.calculate_redemption_rate(events),
            self.calculate_collateral_change(events)
        ]
        
        return features
    
    async def predict_incident_probability(self, recent_events: List[StablecoinEvent]) -> Dict:
        """
        Predict probability of major incident in next 24 hours
        This is what would have saved me during UST
        """
        if not self.model:
            return {'error': 'Model not trained'}
        
        features = self.extract_features(recent_events)
        features_scaled = self.scaler.transform([features])
        
        probability = self.model.predict_proba(features_scaled)[0][1]
        prediction = self.model.predict(features_scaled)[0]
        
        # Get feature importance for explanation
        feature_importance = dict(zip(
            self.feature_columns, 
            self.model.feature_importances_
        ))
        
        return {
            'incident_probability': probability,
            'binary_prediction': bool(prediction),
            'confidence': max(self.model.predict_proba(features_scaled)[0]),
            'key_factors': sorted(feature_importance.items(), key=lambda x: x[1], reverse=True)[:3],
            'recommendation': self.get_recommendation(probability)
        }
    
    def get_recommendation(self, probability: float) -> str:
        """Provide actionable recommendations based on prediction"""
        if probability > 0.8:
            return "CRITICAL: Consider immediate position reduction"
        elif probability > 0.6:
            return "HIGH RISK: Increase monitoring, prepare exit strategy"
        elif probability > 0.4:
            return "ELEVATED RISK: Monitor closely"
        else:
            return "LOW RISK: Normal monitoring sufficient"

import asyncio
import aioredis
from kubernetes import client, config
import logging
from dataclasses import asdict
import json

class ProductionDeployment:
    def __init__(self):
        self.redis_pool = None
        self.k8s_client = None
        self.health_check_interval = 30
        self.failover_timeout = 60
        
    async def setup_infrastructure(self):
        """
        Setup production infrastructure with redundancy
        Learned importance after missing USDC depeg due to API outage
        """
        # Redis cluster for event storage and caching
        self.redis_pool = aioredis.ConnectionPool.from_url(
            "redis://redis-cluster:6379", 
            max_connections=20,
            retry_on_timeout=True
        )
        
        # Kubernetes client for auto-scaling
        config.load_incluster_config()
        self.k8s_client = client.AppsV1Api()
        
        # Start health monitoring
        asyncio.create_task(self.health_monitor())
    
    async def deploy_monitoring_pods(self, token_configs: List[Dict]):
        """
        Deploy separate monitoring pods for each major stablecoin
        Isolated failures prevent cascade outages
        """
        for token_config in token_configs:
            deployment_name = f"monitor-{token_config['symbol'].lower()}"
            
            deployment = {
                'apiVersion': 'apps/v1',
                'kind': 'Deployment',
                'metadata': {'name': deployment_name},
                'spec': {
                    'replicas': 2,  # Always run 2 instances minimum
                    'selector': {'matchLabels': {'app': deployment_name}},
                    'template': {
                        'metadata': {'labels': {'app': deployment_name}},
                        'spec': {
                            'containers': [{
                                'name': 'monitor',
                                'image': 'stablecoin-monitor:latest',
                                'env': [
                                    {'name': 'TOKEN_ADDRESS', 'value': token_config['address']},
                                    {'name': 'CHAINS', 'value': ','.join(token_config['chains'])},
                                    {'name': 'ALERT_WEBHOOK', 'value': token_config['webhook']}
                                ],
                                'resources': {
                                    'requests': {'memory': '512Mi', 'cpu': '0.5'},
                                    'limits': {'memory': '1Gi', 'cpu': '1'}
                                }
                            }]
                        }
                    }
                }
            }
            
            try:
                self.k8s_client.create_namespaced_deployment(
                    namespace='monitoring',
                    body=deployment
                )
                logging.info(f"Deployed monitoring for {token_config['symbol']}")
            except Exception as e:
                logging.error(f"Failed to deploy {deployment_name}: {e}")
    
    async def handle_alert_fatigue(self, events: List[StablecoinEvent]):
        """
        Prevent alert fatigue during volatile periods
        During May 2022, I got 847 alerts in one day - learned to batch them
        """
        redis = aioredis.Redis(connection_pool=self.redis_pool)
        
        # Group events by severity and time window
        alert_groups = {}
        current_time = int(time.time())
        
        for event in events:
            # Group alerts in 15-minute windows
            time_bucket = current_time // 900 * 900
            key = f"{event.token_address}:{event.severity}:{time_bucket}"
            
            if key not in alert_groups:
                alert_groups[key] = []
            alert_groups[key].append(event)
        
        # Send consolidated alerts
        for group_key, group_events in alert_groups.items():
            if len(group_events) > 1:
                await self.send_consolidated_alert(group_events)
            else:
                await self.send_individual_alert(group_events[0])
    
    async def send_consolidated_alert(self, events: List[StablecoinEvent]):
        """Send batched alert to prevent spam"""
        token = events[0].token_address
        severity = events[0].severity
        count = len(events)
        
        message = f"🚨 {severity.upper()}: {count} events detected for {token}\n"
        message += f"Time range: {min(e.timestamp for e in events)} - {max(e.timestamp for e in events)}\n"
        message += f"Max impact: {max(e.impact_score for e in events):.2f}\n"
        message += "Check dashboard for details."
        
        await self.send_webhook_alert(message, severity)