Building Stablecoin Security Score Calculator: Risk Assessment Framework I Built for $2B TVL

// Security scoring engine that evaluates technical architecture
interface TechnicalSecurityMetrics {
    smartContractSecurity: number;
    accessControlRobustness: number;
    upgradeabilityRisks: number;
    oracleReliability: number;
    emergencyControls: number;
}

class TechnicalSecurityScorer {
    async evaluateSmartContractSecurity(contractAddress: string): Promise<number> {
        let score = 0;
        const maxScore = 100;
        
        // Check for common vulnerabilities I've seen cause failures
        const vulnerabilities = await this.scanForVulnerabilities(contractAddress);
        
        // Reentrancy protection (learned from multiple exploits)
        if (await this.hasReentrancyGuards(contractAddress)) {
            score += 20;
        }
        
        // Access control patterns
        const accessControlScore = await this.evaluateAccessControl(contractAddress);
        score += accessControlScore * 0.25; // 25 points max
        
        // Emergency pause mechanisms (critical for incident response)
        if (await this.hasEmergencyPause(contractAddress)) {
            score += 15;
        }
        
        // Timelock protection for critical functions
        if (await this.hasTimelockProtection(contractAddress)) {
            score += 10;
        }
        
        // Multi-signature requirements
        const multisigScore = await this.evaluateMultisigSecurity(contractAddress);
        score += multisigScore * 0.30; // 30 points max
        
        return Math.min(score, maxScore);
    }

    async scanForVulnerabilities(contractAddress: string): Promise<string[]> {
        // Integration with multiple security scanners
        const vulnerabilities = [];
        
        try {
            // Slither static analysis
            const slitherResults = await this.runSlitherAnalysis(contractAddress);
            vulnerabilities.push(...slitherResults.high_risk);
            
            // Mythril symbolic execution
            const mythrilResults = await this.runMythrilAnalysis(contractAddress);
            vulnerabilities.push(...mythrilResults.critical);
            
            // Custom vulnerability patterns I've identified
            const customVulns = await this.checkCustomPatterns(contractAddress);
            vulnerabilities.push(...customVulns);
            
        } catch (error) {
            console.error(`Vulnerability scan failed for ${contractAddress}:`, error);
        }
        
        return vulnerabilities;
    }

    async evaluateAccessControl(contractAddress: string): Promise<number> {
        let score = 0;
        const contract = await this.getContractInstance(contractAddress);
        
        // Check for role-based access control
        if (await this.implementsRBAC(contract)) {
            score += 40;
        }
        
        // Verify principle of least privilege
        const privilegeScore = await this.checkPrivilegeDistribution(contract);
        score += privilegeScore * 0.6; // 60 points max
        
        return score;
    }
}

# Economic stability analyzer that predicted Terra's collapse
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Tuple

@dataclass
class EconomicMetrics:
    collateral_ratio: float
    peg_stability: float
    liquidity_depth: float
    redemption_capacity: float
    yield_sustainability: float

class EconomicStabilityScorer:
    def __init__(self):
        # Thresholds learned from analyzing protocol failures
        self.critical_thresholds = {
            'min_collateral_ratio': 1.5,  # 150% minimum
            'max_peg_deviation': 0.02,     # 2% maximum deviation
            'min_liquidity_depth': 0.1,   # 10% of supply in DEX
            'max_concentration_risk': 0.4, # 40% max single collateral
            'sustainable_yield_rate': 0.15 # 15% APY maximum sustainable
        }
    
    async def calculate_economic_score(self, protocol_data: Dict) -> float:
        """Calculate economic stability score based on real metrics"""
        
        # 1. Collateral Analysis (most critical factor)
        collateral_score = await self.evaluate_collateral_health(protocol_data)
        
        # 2. Peg Stability Analysis
        peg_score = await self.analyze_peg_stability(protocol_data)
        
        # 3. Liquidity Depth Analysis
        liquidity_score = await self.assess_liquidity_resilience(protocol_data)
        
        # 4. Economic Model Sustainability
        sustainability_score = await self.evaluate_economic_model(protocol_data)
        
        # Weighted average with penalties for critical failures
        base_score = (
            collateral_score * 0.4 +    # 40% weight - most important
            peg_score * 0.25 +          # 25% weight
            liquidity_score * 0.20 +    # 20% weight  
            sustainability_score * 0.15 # 15% weight
        )
        
        # Apply penalties for critical risks
        penalty_factor = await self.calculate_penalty_factor(protocol_data)
        
        return max(0, base_score * penalty_factor)
    
    async def evaluate_collateral_health(self, data: Dict) -> float:
        """Analyze collateral composition and health"""
        score = 0
        
        # Get collateral composition
        collateral_types = data.get('collateral_breakdown', {})
        total_collateral = sum(collateral_types.values())
        
        if total_collateral == 0:
            return 0  # No collateral = immediate failure
        
        # Check diversification (learned from UST single-point failure)
        concentration_risk = max(collateral_types.values()) / total_collateral
        if concentration_risk < self.critical_thresholds['max_concentration_risk']:
            score += 30
        else:
            score += max(0, 30 - (concentration_risk - 0.4) * 100)
        
        # Evaluate collateral quality
        high_quality_ratio = await self.calculate_high_quality_collateral_ratio(collateral_types)
        score += high_quality_ratio * 40
        
        # Check collateral ratio
        current_ratio = data.get('collateral_ratio', 0)
        if current_ratio >= self.critical_thresholds['min_collateral_ratio']:
            score += 30
        else:
            # Severe penalty for undercollateralization
            score += max(0, (current_ratio - 1.0) * 60)
        
        return min(100, score)
    
    async def analyze_peg_stability(self, data: Dict) -> float:
        """Analyze historical peg stability"""
        price_history = data.get('price_history', [])
        if not price_history:
            return 50  # Neutral score for new protocols
        
        # Calculate price deviations over different time periods
        deviations_24h = self.calculate_price_deviations(price_history[-24:])
        deviations_7d = self.calculate_price_deviations(price_history[-168:])
        deviations_30d = self.calculate_price_deviations(price_history[-720:])
        
        # Score based on stability
        score = 100
        
        # Penalize recent large deviations more heavily
        if deviations_24h['max_deviation'] > 0.05:  # 5%
            score -= 40
        elif deviations_24h['max_deviation'] > 0.02:  # 2%
            score -= 20
        
        # Consider medium-term stability
        if deviations_7d['avg_deviation'] > 0.01:  # 1%
            score -= 15
        
        # Long-term stability matters for institutional confidence
        if deviations_30d['avg_deviation'] > 0.005:  # 0.5%
            score -= 10
        
        return max(0, score)
    
    def calculate_price_deviations(self, prices: List[float]) -> Dict[str, float]:
        """Calculate price deviation statistics"""
        if not prices:
            return {'max_deviation': 0, 'avg_deviation': 0}
        
        target_price = 1.0
        deviations = [abs(price - target_price) for price in prices]
        
        return {
            'max_deviation': max(deviations),
            'avg_deviation': np.mean(deviations),
            'std_deviation': np.std(deviations)
        }
    
    async def assess_liquidity_resilience(self, data: Dict) -> float:
        """Evaluate liquidity depth and resilience"""
        score = 0
        
        # Get DEX liquidity data
        dex_liquidity = data.get('dex_liquidity', {})
        total_supply = data.get('total_supply', 0)
        
        if total_supply == 0:
            return 0
        
        # Calculate liquidity ratio
        total_liquidity = sum(dex_liquidity.values())
        liquidity_ratio = total_liquidity / total_supply
        
        if liquidity_ratio >= self.critical_thresholds['min_liquidity_depth']:
            score += 40
        else:
            score += liquidity_ratio / self.critical_thresholds['min_liquidity_depth'] * 40
        
        # Check liquidity distribution across DEXes
        if len(dex_liquidity) >= 3:  # Multiple DEX presence
            score += 20
        elif len(dex_liquidity) == 2:
            score += 10
        
        # Analyze liquidity depth (can it handle large redemptions?)
        large_trade_impact = await self.calculate_trade_impact(dex_liquidity, total_supply * 0.05)
        if large_trade_impact < 0.02:  # 2% impact for 5% supply trade
            score += 40
        else:
            score += max(0, 40 - large_trade_impact * 1000)
        
        return min(100, score)

// Governance security analyzer that evaluates attack resistance
pragma solidity ^0.8.19;

interface IGovernanceSecurityAnalyzer {
    struct GovernanceMetrics {
        uint256 proposalThreshold;
        uint256 quorumRequirement;
        uint256 timelockDelay;
        uint256 votingPower;
        bool hasEmergencyPause;
        bool hasMultisigOverride;
    }
    
    function analyzeGovernanceRisks(address governanceContract) 
        external view returns (uint256 securityScore);
}

contract GovernanceSecurityAnalyzer is IGovernanceSecurityAnalyzer {
    // Minimum security thresholds learned from governance attack analysis
    uint256 constant MIN_PROPOSAL_THRESHOLD = 100000e18; // 100k tokens
    uint256 constant MIN_QUORUM_PERCENT = 4; // 4% of total supply
    uint256 constant MIN_TIMELOCK_DELAY = 2 days;
    uint256 constant MAX_SINGLE_HOLDER_PERCENT = 20; // 20% max concentration
    
    mapping(address => GovernanceMetrics) public governanceData;
    
    function analyzeGovernanceRisks(address governanceContract) 
        external view override returns (uint256 securityScore) {
        
        GovernanceMetrics memory metrics = governanceData[governanceContract];
        uint256 score = 0;
        
        // 1. Proposal Threshold Analysis (prevents spam attacks)
        if (metrics.proposalThreshold >= MIN_PROPOSAL_THRESHOLD) {
            score += 20;
        } else {
            score += (metrics.proposalThreshold * 20) / MIN_PROPOSAL_THRESHOLD;
        }
        
        // 2. Quorum Requirements (prevents low-participation attacks)
        uint256 quorumPercent = calculateQuorumPercent(governanceContract);
        if (quorumPercent >= MIN_QUORUM_PERCENT) {
            score += 25;
        } else {
            score += (quorumPercent * 25) / MIN_QUORUM_PERCENT;
        }
        
        // 3. Timelock Protection (critical for security)
        if (metrics.timelockDelay >= MIN_TIMELOCK_DELAY) {
            score += 20;
        } else {
            score += (metrics.timelockDelay * 20) / MIN_TIMELOCK_DELAY;
        }
        
        // 4. Token Distribution Analysis
        uint256 distributionScore = analyzeTokenDistribution(governanceContract);
        score += distributionScore * 20 / 100;
        
        // 5. Emergency Controls
        if (metrics.hasEmergencyPause) {
            score += 10;
        }
        
        if (metrics.hasMultisigOverride) {
            score += 5;
        }
        
        return score;
    }
    
    function analyzeTokenDistribution(address governanceContract) 
        internal view returns (uint256 distributionScore) {
        
        // Get top holder concentrations
        uint256[] memory topHolderPercentages = getTopHolderPercentages(governanceContract, 10);
        
        uint256 score = 100;
        
        // Penalize high concentration (learned from multiple governance attacks)
        for (uint i = 0; i < topHolderPercentages.length; i++) {
            if (topHolderPercentages[i] > MAX_SINGLE_HOLDER_PERCENT) {
                score -= (topHolderPercentages[i] - MAX_SINGLE_HOLDER_PERCENT) * 2;
            }
        }
        
        // Check if top 5 holders control majority
        uint256 top5Total = 0;
        for (uint i = 0; i < 5 && i < topHolderPercentages.length; i++) {
            top5Total += topHolderPercentages[i];
        }
        
        if (top5Total > 50) {
            score -= 30; // Major penalty for majority control
        }
        
        return score > 0 ? score : 0;
    }
    
    function calculateQuorumPercent(address governanceContract) 
        internal view returns (uint256) {
        // Implementation depends on specific governance contract
        // This is a simplified version
        return 5; // Placeholder - implement based on actual contract
    }
    
    function getTopHolderPercentages(address tokenContract, uint256 count) 
        internal view returns (uint256[] memory) {
        // Implementation to get top holder percentages
        // This would integrate with token contract to analyze distribution
        uint256[] memory percentages = new uint256[](count);
        // Placeholder implementation
        return percentages;
    }
}

// Operational security assessment based on team practices
class OperationalSecurityScorer {
    constructor() {
        this.securityPractices = {
            multiSigUsage: { weight: 25, description: "Multi-signature wallet usage" },
            auditFrequency: { weight: 20, description: "Regular security audits" },
            bugBountyProgram: { weight: 15, description: "Active bug bounty program" },
            incidentResponse: { weight: 20, description: "Incident response procedures" },
            teamTransparency: { weight: 20, description: "Team transparency and KYC" }
        };
    }

    async evaluateOperationalSecurity(protocolData) {
        let totalScore = 0;
        const maxScore = 100;

        // Multi-signature wallet usage
        const multisigScore = await this.assessMultiSigSecurity(protocolData);
        totalScore += multisigScore * (this.securityPractices.multiSigUsage.weight / 100);

        // Security audit practices
        const auditScore = await this.evaluateAuditPractices(protocolData);
        totalScore += auditScore * (this.securityPractices.auditFrequency.weight / 100);

        // Bug bounty program effectiveness
        const bountyScore = await this.assessBugBountyProgram(protocolData);
        totalScore += bountyScore * (this.securityPractices.bugBountyProgram.weight / 100);

        // Incident response readiness
        const responseScore = await this.evaluateIncidentResponse(protocolData);
        totalScore += responseScore * (this.securityPractices.incidentResponse.weight / 100);

        // Team transparency and accountability
        const transparencyScore = await this.assessTeamTransparency(protocolData);
        totalScore += transparencyScore * (this.securityPractices.teamTransparency.weight / 100);

        return Math.min(totalScore, maxScore);
    }

    async assessMultiSigSecurity(protocolData) {
        let score = 0;
        const treasuryMultisig = protocolData.treasuryMultisig;
        const adminMultisig = protocolData.adminMultisig;

        // Check treasury multi-sig configuration
        if (treasuryMultisig && treasuryMultisig.signers >= 3) {
            score += 50;
            
            // Bonus for higher security
            if (treasuryMultisig.threshold >= Math.ceil(treasuryMultisig.signers * 0.6)) {
                score += 20;
            }
        }

        // Check admin multi-sig for critical functions
        if (adminMultisig && adminMultisig.signers >= 2) {
            score += 30;
        }

        return Math.min(score, 100);
    }

    async evaluateAuditPractices(protocolData) {
        let score = 0;
        const audits = protocolData.securityAudits || [];

        // Recent audits (within 6 months)
        const recentAudits = audits.filter(audit => {
            const auditDate = new Date(audit.date);
            const sixMonthsAgo = new Date();
            sixMonthsAgo.setMonth(sixMonthsAgo.getMonth() - 6);
            return auditDate > sixMonthsAgo;
        });

        if (recentAudits.length >= 2) {
            score += 60;
        } else if (recentAudits.length === 1) {
            score += 30;
        }

        // Quality of audit firms
        const topTierFirms = ['Trail of Bits', 'OpenZeppelin', 'ConsenSys Diligence', 'Quantstamp'];
        const hasTopTierAudit = audits.some(audit => 
            topTierFirms.includes(audit.auditor)
        );

        if (hasTopTierAudit) {
            score += 40;
        }

        return Math.min(score, 100);
    }

    async assessBugBountyProgram(protocolData) {
        const bountyProgram = protocolData.bugBountyProgram;
        
        if (!bountyProgram || !bountyProgram.active) {
            return 0;
        }

        let score = 40; // Base score for having a program

        // Bounty amounts indicate seriousness
        if (bountyProgram.maxPayout >= 1000000) { // $1M+
            score += 30;
        } else if (bountyProgram.maxPayout >= 100000) { // $100k+
            score += 20;
        } else if (bountyProgram.maxPayout >= 10000) { // $10k+
            score += 10;
        }

        // Platform quality
        if (bountyProgram.platform === 'Immunefi' || bountyProgram.platform === 'HackerOne') {
            score += 20;
        }

        // Response track record
        if (bountyProgram.averageResponseTime <= 7) { // 7 days
            score += 10;
        }

        return Math.min(score, 100);
    }
}

# Real-time risk monitoring system
import asyncio
import websockets
import json
from datetime import datetime, timedelta

class RealTimeRiskMonitor:
    def __init__(self):
        self.risk_factors = {
            'price_volatility': {'weight': 0.3, 'threshold': 0.02},
            'liquidity_changes': {'weight': 0.25, 'threshold': 0.15},
            'governance_activity': {'weight': 0.2, 'threshold': 5},
            'technical_incidents': {'weight': 0.15, 'threshold': 1},
            'market_sentiment': {'weight': 0.1, 'threshold': -0.3}
        }
        
        self.monitoring_active = False
        self.current_scores = {}

    async def start_monitoring(self, protocol_addresses):
        """Start real-time monitoring for given protocols"""
        self.monitoring_active = True
        
        tasks = [
            self.monitor_price_stability(protocol_addresses),
            self.monitor_liquidity_changes(protocol_addresses),
            self.monitor_governance_events(protocol_addresses),
            self.monitor_technical_health(protocol_addresses)
        ]
        
        await asyncio.gather(*tasks)

    async def monitor_price_stability(self, protocols):
        """Monitor price stability and update scores"""
        while self.monitoring_active:
            try:
                for protocol in protocols:
                    current_price = await self.get_current_price(protocol)
                    price_history = await self.get_price_history(protocol, hours=24)
                    
                    volatility = self.calculate_volatility(price_history)
                    
                    # Update score based on volatility
                    if volatility > self.risk_factors['price_volatility']['threshold']:
                        penalty = min(20, volatility * 100)
                        await self.apply_score_penalty(protocol, 'price_volatility', penalty)
                    
                await asyncio.sleep(300)  # Check every 5 minutes
                
            except Exception as e:
                print(f"Price monitoring error: {e}")
                await asyncio.sleep(60)

    async def monitor_liquidity_changes(self, protocols):
        """Monitor liquidity pool changes"""
        while self.monitoring_active:
            try:
                for protocol in protocols:
                    current_liquidity = await self.get_current_liquidity(protocol)
                    baseline_liquidity = await self.get_baseline_liquidity(protocol)
                    
                    if baseline_liquidity > 0:
                        change_ratio = (baseline_liquidity - current_liquidity) / baseline_liquidity
                        
                        if change_ratio > self.risk_factors['liquidity_changes']['threshold']:
                            penalty = min(30, change_ratio * 100)
                            await self.apply_score_penalty(protocol, 'liquidity_drop', penalty)
                
                await asyncio.sleep(600)  # Check every 10 minutes
                
            except Exception as e:
                print(f"Liquidity monitoring error: {e}")
                await asyncio.sleep(120)

    def calculate_composite_score(self, protocol_scores):
        """Calculate final composite security score"""
        weights = {
            'technical_security': 0.25,
            'economic_stability': 0.30,
            'governance_security': 0.20,
            'operational_security': 0.15,
            'real_time_risk': 0.10
        }
        
        composite_score = 0
        for category, score in protocol_scores.items():
            if category in weights:
                composite_score += score * weights[category]
        
        # Apply any active penalties from real-time monitoring
        penalties = self.get_active_penalties(protocol_scores.get('protocol_id'))
        total_penalty = sum(penalties.values())
        
        final_score = max(0, composite_score - total_penalty)
        
        return {
            'composite_score': final_score,
            'grade': self.score_to_grade(final_score),
            'risk_level': self.score_to_risk_level(final_score),
            'breakdown': protocol_scores,
            'active_penalties': penalties,
            'last_updated': datetime.now().isoformat()
        }

    def score_to_grade(self, score):
        """Convert numeric score to letter grade"""
        if score >= 90:
            return 'A+'
        elif score >= 85:
            return 'A'
        elif score >= 80:
            return 'B+'
        elif score >= 75:
            return 'B'
        elif score >= 70:
            return 'C+'
        elif score >= 60:
            return 'C'
        elif score >= 50:
            return 'D'
        else:
            return 'F'

    def score_to_risk_level(self, score):
        """Convert score to risk level for institutional use"""
        if score >= 80:
            return 'LOW_RISK'
        elif score >= 60:
            return 'MEDIUM_RISK'
        elif score >= 40:
            return 'HIGH_RISK'
        else:
            return 'CRITICAL_RISK'