How I Uncovered Hidden Voting Patterns in Stablecoin Governance (And You Can Too)

# governance_collector.py - My battle-tested data collection system
import asyncio
from web3 import Web3
from dataclasses import dataclass
from typing import Dict, List, Optional
import pandas as pd
from datetime import datetime, timedelta

@dataclass
class GovernanceVote:
    proposal_id: str
    voter_address: str
    vote_choice: int  # 0=against, 1=for, 2=abstain
    voting_power: float
    block_number: int
    timestamp: datetime
    gas_used: int

class StablecoinGovernanceCollector:
    def __init__(self, rpc_url: str, governance_contract: str):
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        self.governance_address = governance_contract
        # This ABI includes the critical VoteCast event
        self.governance_abi = self._load_governance_abi()
        self.contract = self.w3.eth.contract(
            address=governance_contract, 
            abi=self.governance_abi
        )
    
    async def collect_voting_history(self, start_block: int, end_block: int) -> List[GovernanceVote]:
        """
        The key insight: collect ALL vote events, not just successful proposals
        I spent 2 days debugging why my analysis was wrong before realizing
        I was only looking at passed proposals
        """
        votes = []
        
        # Get VoteCast events in chunks to avoid RPC limits
        chunk_size = 10000  # Learned this from hitting rate limits repeatedly
        
        for block_start in range(start_block, end_block, chunk_size):
            block_end = min(block_start + chunk_size, end_block)
            
            vote_events = self.contract.events.VoteCast.get_logs(
                fromBlock=block_start,
                toBlock=block_end
            )
            
            for event in vote_events:
                # This is where I extract the crucial voting behavior data
                vote = GovernanceVote(
                    proposal_id=str(event.args.proposalId),
                    voter_address=event.args.voter.lower(),
                    vote_choice=event.args.support,
                    voting_power=float(event.args.weight) / 1e18,  # Convert from wei
                    block_number=event.blockNumber,
                    timestamp=self._get_block_timestamp(event.blockNumber),
                    gas_used=self._get_transaction_gas(event.transactionHash)
                )
                votes.append(vote)
            
            # Rate limiting - learned this after getting banned from Infura
            await asyncio.sleep(0.1)
        
        return votes
    
    def _get_block_timestamp(self, block_number: int) -> datetime:
        """Convert block number to timestamp - essential for temporal analysis"""
        block = self.w3.eth.get_block(block_number)
        return datetime.fromtimestamp(block.timestamp)

# governance_enricher.py - Where the magic happens
class VotingPatternEnricher:
    def __init__(self, votes_df: pd.DataFrame):
        self.votes_df = votes_df
        self.enriched_df = None
    
    def enrich_voting_data(self) -> pd.DataFrame:
        """
        This enrichment process took me weeks to perfect
        Each step reveals different aspects of voting behavior
        """
        df = self.votes_df.copy()
        
        # 1. Temporal features - timing is everything in governance
        df['hour_of_day'] = df['timestamp'].dt.hour
        df['day_of_week'] = df['timestamp'].dt.dayofweek
        df['days_since_proposal'] = self._calculate_proposal_age(df)
        
        # 2. Voter behavior patterns - this changed everything for me
        df['voter_total_votes'] = df.groupby('voter_address')['proposal_id'].transform('nunique')
        df['voter_avg_power'] = df.groupby('voter_address')['voting_power'].transform('mean')
        df['voter_consistency'] = self._calculate_voting_consistency(df)
        
        # 3. Proposal context - success patterns emerge here
        df['proposal_total_power'] = df.groupby('proposal_id')['voting_power'].transform('sum')
        df['proposal_participation'] = df.groupby('proposal_id')['voter_address'].transform('nunique')
        df['voter_power_rank'] = df.groupby('proposal_id')['voting_power'].rank(method='dense', ascending=False)
        
        # 4. Network effects - the breakthrough insight
        df['coordinated_voting_score'] = self._calculate_coordination_score(df)
        df['influence_network_position'] = self._calculate_network_centrality(df)
        
        self.enriched_df = df
        return df
    
    def _calculate_coordination_score(self, df: pd.DataFrame) -> pd.Series:
        """
        This metric identifies coordinated voting behavior
        High scores indicate addresses that consistently vote together
        """
        coordination_scores = []
        
        for _, vote in df.iterrows():
            voter = vote['voter_address']
            proposal = vote['proposal_id']
            vote_choice = vote['vote_choice']
            
            # Find other voters on this proposal with same choice
            same_choice_voters = df[
                (df['proposal_id'] == proposal) & 
                (df['vote_choice'] == vote_choice) &
                (df['voter_address'] != voter)
            ]['voter_address'].tolist()
            
            # Calculate how often this voter agrees with others
            if len(same_choice_voters) > 0:
                historical_agreement = self._calculate_historical_agreement(
                    voter, same_choice_voters, df
                )
                coordination_scores.append(historical_agreement)
            else:
                coordination_scores.append(0.0)
        
        return pd.Series(coordination_scores, index=df.index)

def identify_whale_dominators(enriched_df: pd.DataFrame) -> pd.DataFrame:
    """
    Whale dominators: addresses whose individual vote can swing outcomes
    I define this as having >5% of total voting power in >50% of their votes
    """
    voter_stats = enriched_df.groupby('voter_address').agg({
        'voting_power': ['mean', 'max', 'std'],
        'proposal_total_power': 'mean',
        'proposal_id': 'nunique'
    }).round(4)
    
    # Flatten column names
    voter_stats.columns = ['avg_power', 'max_power', 'power_std', 'avg_total_power', 'proposals_voted']
    
    # Calculate dominance metrics
    voter_stats['power_percentage'] = (voter_stats['avg_power'] / voter_stats['avg_total_power']) * 100
    voter_stats['dominance_score'] = voter_stats['power_percentage'] * voter_stats['proposals_voted']
    
    # My criteria: >5% average power AND >10 votes
    whale_dominators = voter_stats[
        (voter_stats['power_percentage'] > 5.0) & 
        (voter_stats['proposals_voted'] > 10)
    ].sort_values('dominance_score', ascending=False)
    
    return whale_dominators

def detect_coordination_clusters(enriched_df: pd.DataFrame, min_agreement_rate: float = 0.8) -> Dict:
    """
    Detect groups of addresses that vote together suspiciously often
    This analysis revealed manipulation attempts I never would have found manually
    """
    from sklearn.cluster import DBSCAN
    from itertools import combinations
    
    # Calculate pairwise agreement rates
    voters = enriched_df['voter_address'].unique()
    agreement_matrix = np.zeros((len(voters), len(voters)))
    
    for i, voter1 in enumerate(voters):
        for j, voter2 in enumerate(voters):
            if i != j:
                agreement_rate = calculate_pairwise_agreement(voter1, voter2, enriched_df)
                agreement_matrix[i][j] = agreement_rate
    
    # Use DBSCAN to identify clusters of highly coordinated voters
    # eps=0.2 means addresses must agree >80% of the time to cluster
    clustering = DBSCAN(eps=1-min_agreement_rate, min_samples=3)
    cluster_labels = clustering.fit_predict(agreement_matrix)
    
    # Package results
    clusters = {}
    for cluster_id in set(cluster_labels):
        if cluster_id != -1:  # -1 is noise in DBSCAN
            cluster_addresses = [voters[i] for i, label in enumerate(cluster_labels) if label == cluster_id]
            clusters[f"cluster_{cluster_id}"] = {
                'addresses': cluster_addresses,
                'size': len(cluster_addresses),
                'avg_agreement': np.mean([agreement_matrix[i][j] for i in range(len(voters)) 
                                        for j in range(len(voters)) 
                                        if cluster_labels[i] == cluster_labels[j] == cluster_id and i != j])
            }
    
    return clusters

def calculate_pairwise_agreement(voter1: str, voter2: str, df: pd.DataFrame) -> float:
    """Calculate how often two voters choose the same option"""
    voter1_votes = df[df['voter_address'] == voter1][['proposal_id', 'vote_choice']]
    voter2_votes = df[df['voter_address'] == voter2][['proposal_id', 'vote_choice']]
    
    # Find proposals both voted on
    common_proposals = pd.merge(voter1_votes, voter2_votes, on='proposal_id', suffixes=('_1', '_2'))
    
    if len(common_proposals) == 0:
        return 0.0
    
    agreements = (common_proposals['vote_choice_1'] == common_proposals['vote_choice_2']).sum()
    return agreements / len(common_proposals)

def analyze_strategic_timing(enriched_df: pd.DataFrame) -> pd.DataFrame:
    """
    Identify voters who strategically time their votes for maximum impact
    This pattern often indicates sophisticated governance strategies
    """
    # Calculate proposal duration for each vote
    proposal_durations = enriched_df.groupby('proposal_id').agg({
        'timestamp': ['min', 'max'],
        'days_since_proposal': 'max'
    })
    proposal_durations.columns = ['start_time', 'end_time', 'total_duration']
    
    # Merge back to get relative timing for each vote
    timing_analysis = enriched_df.merge(
        proposal_durations, 
        left_on='proposal_id', 
        right_index=True
    )
    
    # Calculate relative timing (0 = first vote, 1 = last vote)
    timing_analysis['relative_timing'] = (
        timing_analysis['days_since_proposal'] / timing_analysis['total_duration']
    )
    
    # Identify strategic timers - those who consistently vote late
    voter_timing_stats = timing_analysis.groupby('voter_address').agg({
        'relative_timing': ['mean', 'std', 'count']
    })
    voter_timing_stats.columns = ['avg_timing', 'timing_consistency', 'vote_count']
    
    # Strategic timers: vote in final 20% of proposal period, >70% of the time
    strategic_timers = voter_timing_stats[
        (voter_timing_stats['avg_timing'] > 0.8) & 
        (voter_timing_stats['vote_count'] > 5)
    ].sort_values('avg_timing', ascending=False)
    
    return strategic_timers

# governance_dashboard.py - Real-time governance monitoring
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
import networkx as nx

class GovernanceDashboard:
    def __init__(self, enriched_df: pd.DataFrame):
        self.df = enriched_df
        self.figures = {}
    
    def create_power_concentration_chart(self) -> go.Figure:
        """
        Visualize voting power concentration - this chart always shocks clients
        """
        # Calculate cumulative power distribution
        power_by_voter = self.df.groupby('voter_address')['voting_power'].mean().sort_values(ascending=False)
        cumulative_power = power_by_voter.cumsum() / power_by_voter.sum() * 100
        
        fig = go.Figure()
        
        # Add the Lorenz curve
        fig.add_trace(go.Scatter(
            x=list(range(1, len(cumulative_power) + 1)),
            y=cumulative_power.values,
            mode='lines',
            name='Actual Distribution',
            line=dict(color='red', width=3)
        ))
        
        # Add perfect equality line
        fig.add_trace(go.Scatter(
            x=[1, len(cumulative_power)],
            y=[0, 100],
            mode='lines',
            name='Perfect Equality',
            line=dict(color='blue', dash='dash')
        ))
        
        fig.update_layout(
            title='Voting Power Concentration (Lorenz Curve)',
            xaxis_title='Voter Rank',
            yaxis_title='Cumulative Power %',
            showlegend=True
        )
        
        return fig
    
    def create_coordination_network(self, min_agreement: float = 0.7) -> go.Figure:
        """
        Network visualization of coordinated voting behavior
        This visualization makes manipulation immediately obvious
        """
        # Build network from coordination data
        G = nx.Graph()
        
        # Add nodes (voters)
        voters = self.df['voter_address'].unique()
        for voter in voters:
            avg_power = self.df[self.df['voter_address'] == voter]['voting_power'].mean()
            G.add_node(voter, power=avg_power)
        
        # Add edges (coordinated relationships)
        for voter1, voter2 in combinations(voters, 2):
            agreement = calculate_pairwise_agreement(voter1, voter2, self.df)
            if agreement >= min_agreement:
                G.add_edge(voter1, voter2, weight=agreement)
        
        # Create layout
        pos = nx.spring_layout(G, k=1, iterations=50)
        
        # Extract coordinates
        edge_x, edge_y = [], []
        for edge in G.edges():
            x0, y0 = pos[edge[0]]
            x1, y1 = pos[edge[1]]
            edge_x.extend([x0, x1, None])
            edge_y.extend([y0, y1, None])
        
        node_x = [pos[node][0] for node in G.nodes()]
        node_y = [pos[node][1] for node in G.nodes()]
        node_size = [G.nodes[node]['power'] * 100 for node in G.nodes()]  # Scale for visibility
        
        fig = go.Figure()
        
        # Add edges
        fig.add_trace(go.Scatter(
            x=edge_x, y=edge_y,
            line=dict(width=1, color='lightgray'),
            hoverinfo='none',
            mode='lines'
        ))
        
        # Add nodes
        fig.add_trace(go.Scatter(
            x=node_x, y=node_y,
            mode='markers',
            hoverinfo='text',
            text=[f"Address: {node}<br>Avg Power: {G.nodes[node]['power']:.2f}" for node in G.nodes()],
            marker=dict(
                size=node_size,
                color='red',
                line=dict(width=2, color='darkred')
            )
        ))
        
        fig.update_layout(
            title=f'Coordination Network (Agreement ≥ {min_agreement*100}%)',
            showlegend=False,
            xaxis=dict(showgrid=False, zeroline=False, showticklabels=False),
            yaxis=dict(showgrid=False, zeroline=False, showticklabels=False)
        )
        
        return fig

# main_analysis.py - Complete governance analysis pipeline
async def analyze_protocol_governance(
    rpc_url: str, 
    governance_contract: str, 
    start_block: int,
    protocol_name: str
) -> Dict:
    """
    Complete governance analysis pipeline
    This is the exact process I use for client assessments
    """
    
    print(f"Starting governance analysis for {protocol_name}...")
    
    # Step 1: Data Collection
    collector = StablecoinGovernanceCollector(rpc_url, governance_contract)
    raw_votes = await collector.collect_voting_history(start_block, current_block())
    print(f"Collected {len(raw_votes)} votes")
    
    # Step 2: Data Enrichment
    votes_df = pd.DataFrame([vote.__dict__ for vote in raw_votes])
    enricher = VotingPatternEnricher(votes_df)
    enriched_df = enricher.enrich_voting_data()
    print("Data enrichment complete")
    
    # Step 3: Pattern Analysis
    results = {
        'protocol': protocol_name,
        'analysis_date': datetime.now().isoformat(),
        'total_votes': len(enriched_df),
        'unique_voters': enriched_df['voter_address'].nunique(),
        'unique_proposals': enriched_df['proposal_id'].nunique(),
    }
    
    # Whale analysis
    whales = identify_whale_dominators(enriched_df)
    results['whale_dominators'] = {
        'count': len(whales),
        'top_whales': whales.head().to_dict('index'),
        'power_concentration': whales['power_percentage'].sum()
    }
    
    # Coordination analysis
    clusters = detect_coordination_clusters(enriched_df)
    results['coordination_clusters'] = {
        'cluster_count': len(clusters),
        'largest_cluster_size': max([cluster['size'] for cluster in clusters.values()]) if clusters else 0,
        'total_coordinated_addresses': sum([cluster['size'] for cluster in clusters.values()])
    }
    
    # Timing analysis
    strategic_timers = analyze_strategic_timing(enriched_df)
    results['strategic_timing'] = {
        'strategic_timer_count': len(strategic_timers),
        'avg_late_voting_rate': strategic_timers['avg_timing'].mean() if len(strategic_timers) > 0 else 0
    }
    
    # Step 4: Generate Dashboard
    dashboard = GovernanceDashboard(enriched_df)
    results['visualizations'] = {
        'power_concentration': dashboard.create_power_concentration_chart(),
        'coordination_network': dashboard.create_coordination_network(),
        'temporal_patterns': dashboard.create_temporal_analysis()
    }
    
    # Step 5: Risk Assessment
    results['risk_assessment'] = calculate_governance_risk_score(results)
    
    print(f"Analysis complete. Risk score: {results['risk_assessment']['overall_score']}")
    return results

def calculate_governance_risk_score(analysis_results: Dict) -> Dict:
    """
    Calculate overall governance health score
    Based on patterns I've observed across dozens of protocols
    """
    risk_factors = {
        'power_concentration': min(analysis_results['whale_dominators']['power_concentration'] / 50, 1.0),
        'coordination_risk': min(analysis_results['coordination_clusters']['total_coordinated_addresses'] / 100, 1.0),
        'timing_manipulation': min(len(analysis_results['strategic_timing']) / 20, 1.0)
    }
    
    # Weighted risk score (0-100, higher is riskier)
    overall_score = (
        risk_factors['power_concentration'] * 0.4 +
        risk_factors['coordination_risk'] * 0.4 +
        risk_factors['timing_manipulation'] * 0.2
    ) * 100
    
    return {
        'overall_score': overall_score,
        'risk_factors': risk_factors,
        'risk_level': 'HIGH' if overall_score > 70 else 'MEDIUM' if overall_score > 40 else 'LOW'
    }

# advanced_patterns.py - Next-level governance analysis
def detect_sandwich_voting(enriched_df: pd.DataFrame) -> pd.DataFrame:
    """
    Detect "sandwich" voting patterns where coordinated groups
    vote early and late to influence middle voters
    """
    sandwich_patterns = []
    
    for proposal_id in enriched_df['proposal_id'].unique():
        proposal_votes = enriched_df[enriched_df['proposal_id'] == proposal_id].sort_values('timestamp')
        
        if len(proposal_votes) < 10:  # Need sufficient votes to detect pattern
            continue
        
        # Check for coordinated early/late voting
        early_votes = proposal_votes.head(int(len(proposal_votes) * 0.2))  # First 20%
        late_votes = proposal_votes.tail(int(len(proposal_votes) * 0.2))   # Last 20%
        
        # Calculate coordination between early and late voters
        early_coordination = detect_coordination_in_timeframe(early_votes)
        late_coordination = detect_coordination_in_timeframe(late_votes)
        
        if early_coordination > 0.8 and late_coordination > 0.8:
            sandwich_patterns.append({
                'proposal_id': proposal_id,
                'early_coordination': early_coordination,
                'late_coordination': late_coordination,
                'total_votes': len(proposal_votes)
            })
    
    return pd.DataFrame(sandwich_patterns)

def analyze_proposal_timing_attacks(enriched_df: pd.DataFrame) -> Dict:
    """
    Detect if proposals are being timed to coincide with low participation periods
    (weekends, holidays, etc.)
    """
    proposal_starts = enriched_df.groupby('proposal_id')['timestamp'].min()
    
    timing_analysis = {
        'weekend_proposals': sum(proposal_starts.dt.dayofweek >= 5),
        'holiday_proposals': count_holiday_proposals(proposal_starts),
        'night_proposals': sum(proposal_starts.dt.hour < 6),  # UTC midnight-6am
        'total_proposals': len(proposal_starts)
    }
    
    # Calculate manipulation probability
    weekend_rate = timing_analysis['weekend_proposals'] / timing_analysis['total_proposals']
    expected_weekend_rate = 2/7  # Random timing would be ~28.6%
    
    timing_analysis['manipulation_probability'] = max(0, weekend_rate - expected_weekend_rate)
    
    return timing_analysis