Network Theory in Political Systems: Modeling Power Dynamics Through Graph Algorithms

import numpy as np
import networkx as nx
from scipy.sparse import csr_matrix

class PoliticalNetwork:
    def __init__(self, actors, relationships):
        self.G = nx.DiGraph()
        self.G.add_nodes_from(actors)
        self.G.add_weighted_edges_from(relationships)
        
    def adjacency_matrix(self):
        return nx.adjacency_matrix(self.G, weight='weight')
    
    def network_density(self):
        n = len(self.G.nodes())
        possible_edges = n * (n - 1)  # directed graph
        actual_edges = len(self.G.edges())
        return actual_edges / possible_edges
    
    def degree_distribution(self):
        degrees = dict(self.G.degree())
        return list(degrees.values())

# Example: Senate voting similarity network
senators = ['Warren', 'Sanders', 'Cruz', 'Paul', 'Schumer']
voting_similarity = [
    ('Warren', 'Sanders', 0.89),
    ('Warren', 'Schumer', 0.82),
    ('Cruz', 'Paul', 0.76),
    ('Sanders', 'Schumer', 0.71)
]

senate_net = PoliticalNetwork(senators, voting_similarity)
print(f"Network density: {senate_net.network_density():.3f}")

import networkx as nx
import numpy as np
from collections import defaultdict

def compute_all_centralities(G):
    """Compute multiple centrality measures for political analysis"""
    centralities = defaultdict(dict)
    
    # Degree centrality
    degree_cent = nx.degree_centrality(G)
    
    # Betweenness centrality
    between_cent = nx.betweenness_centrality(G, weight='weight')
    
    # Closeness centrality
    close_cent = nx.closeness_centrality(G, distance='weight')
    
    # Eigenvector centrality
    try:
        eigen_cent = nx.eigenvector_centrality(G, weight='weight')
    except:
        eigen_cent = {node: 0 for node in G.nodes()}
    
    # PageRank (modified eigenvector centrality)
    pagerank_cent = nx.pagerank(G, weight='weight')
    
    for node in G.nodes():
        centralities[node] = {
            'degree': degree_cent[node],
            'betweenness': between_cent[node],
            'closeness': close_cent[node],
            'eigenvector': eigen_cent[node],
            'pagerank': pagerank_cent[node]
        }
    
    return centralities

def rank_actors_by_influence(centralities, measure='pagerank'):
    """Rank political actors by specified centrality measure"""
    scores = [(actor, data[measure]) for actor, data in centralities.items()]
    return sorted(scores, key=lambda x: x[1], reverse=True)

# Example usage
centralities = compute_all_centralities(senate_net.G)
influence_ranking = rank_actors_by_influence(centralities)
print("Influence ranking (PageRank):")
for i, (actor, score) in enumerate(influence_ranking, 1):
    print(f"{i}. {actor}: {score:.4f}")

import networkx as nx
from networkx.algorithms import community
import matplotlib.pyplot as plt
from itertools import combinations

def detect_political_coalitions(G, algorithm='louvain'):
    """Detect coalition structure using community detection"""
    
    if algorithm == 'louvain':
        # Louvain algorithm for modularity optimization
        communities = community.greedy_modularity_communities(G, weight='weight')
    elif algorithm == 'girvan_newman':
        # Girvan-Newman algorithm
        communities = next(community.girvan_newman(G))
    elif algorithm == 'label_propagation':
        # Label propagation algorithm
        communities = community.label_propagation_communities(G)
    
    # Convert to dictionary format
    coalition_assignment = {}
    for i, coalition in enumerate(communities):
        for actor in coalition:
            coalition_assignment[actor] = i
    
    # Calculate modularity
    modularity = community.modularity(G, communities, weight='weight')
    
    return coalition_assignment, modularity, communities

def analyze_coalition_stability(G, coalitions):
    """Analyze internal vs external connection strength"""
    stability_metrics = {}
    
    for i, coalition in enumerate(coalitions):
        internal_weight = 0
        external_weight = 0
        
        # Internal connections
        for u, v in combinations(coalition, 2):
            if G.has_edge(u, v):
                internal_weight += G[u][v].get('weight', 1)
            if G.has_edge(v, u):
                internal_weight += G[v][u].get('weight', 1)
        
        # External connections
        other_actors = set(G.nodes()) - set(coalition)
        for actor in coalition:
            for other in other_actors:
                if G.has_edge(actor, other):
                    external_weight += G[actor][other].get('weight', 1)
        
        stability = internal_weight / (internal_weight + external_weight) if (internal_weight + external_weight) > 0 else 0
        stability_metrics[f'Coalition_{i}'] = {
            'actors': list(coalition),
            'internal_strength': internal_weight,
            'external_strength': external_weight,
            'stability_ratio': stability
        }
    
    return stability_metrics

# Example: Analyze legislative coalition structure
coalitions, modularity, communities = detect_political_coalitions(senate_net.G)
stability = analyze_coalition_stability(senate_net.G, communities)

print(f"Network modularity: {modularity:.3f}")
print("\nDetected coalitions:")
for name, metrics in stability.items():
    print(f"{name}: {metrics['actors']} (stability: {metrics['stability_ratio']:.3f})")

import networkx as nx
import numpy as np
from collections import defaultdict
import pandas as pd

class TemporalPoliticalNetwork:
    def __init__(self):
        self.snapshots = {}  # time -> networkx.Graph
        self.actor_history = defaultdict(list)
    
    def add_snapshot(self, time_period, actors, relationships):
        """Add a network snapshot for a specific time period"""
        G = nx.DiGraph()
        G.add_nodes_from(actors)
        G.add_weighted_edges_from(relationships)
        self.snapshots[time_period] = G
    
    def track_centrality_evolution(self, actor, centrality_type='pagerank'):
        """Track how an actor's centrality changes over time"""
        evolution = []
        for time, G in sorted(self.snapshots.items()):
            if actor in G.nodes():
                if centrality_type == 'pagerank':
                    centrality = nx.pagerank(G, weight='weight')[actor]
                elif centrality_type == 'betweenness':
                    centrality = nx.betweenness_centrality(G, weight='weight')[actor]
                elif centrality_type == 'degree':
                    centrality = nx.degree_centrality(G)[actor]
                else:
                    centrality = 0
            else:
                centrality = 0
            evolution.append((time, centrality))
        return evolution
    
    def detect_structural_changes(self):
        """Identify significant changes in network structure over time"""
        changes = []
        time_points = sorted(self.snapshots.keys())
        
        for i in range(1, len(time_points)):
            prev_time, curr_time = time_points[i-1], time_points[i]
            prev_G, curr_G = self.snapshots[prev_time], self.snapshots[curr_time]
            
            # Calculate change metrics
            prev_edges = set(prev_G.edges())
            curr_edges = set(curr_G.edges())
            
            added_edges = curr_edges - prev_edges
            removed_edges = prev_edges - curr_edges
            
            # Modularity change
            prev_communities = community.greedy_modularity_communities(prev_G)
            curr_communities = community.greedy_modularity_communities(curr_G)
            prev_mod = community.modularity(prev_G, prev_communities)
            curr_mod = community.modularity(curr_G, curr_communities)
            
            changes.append({
                'time_period': f"{prev_time} -> {curr_time}",
                'edges_added': len(added_edges),
                'edges_removed': len(removed_edges),
                'modularity_change': curr_mod - prev_mod,
                'new_relationships': list(added_edges)[:5],  # Show first 5
                'broken_relationships': list(removed_edges)[:5]
            })
        
        return changes
    
    def predict_future_connections(self, prediction_time):
        """Simple link prediction based on network evolution patterns"""
        # This is a simplified example - real prediction would use more sophisticated methods
        if len(self.snapshots) < 2:
            return []
        
        # Get the two most recent snapshots
        times = sorted(self.snapshots.keys())[-2:]
        G1, G2 = self.snapshots[times[0]], self.snapshots[times[1]]
        
        # Find nodes that gained connections
        growth_nodes = []
        for node in G2.nodes():
            if node in G1.nodes():
                if G2.degree(node) > G1.degree(node):
                    growth_nodes.append(node)
        
        # Predict new connections for growing nodes
        predictions = []
        for node in growth_nodes[:3]:  # Top 3 growing nodes
            # Simple preferential attachment prediction
            candidates = [n for n in G2.nodes() if n != node and not G2.has_edge(node, n)]
            if candidates:
                # Prefer connecting to high-degree nodes
                best_candidate = max(candidates, key=lambda x: G2.degree(x))
                predictions.append((node, best_candidate))
        
        return predictions

# Example usage with legislative session data
temporal_net = TemporalPoliticalNetwork()

# Add snapshots for different congressional sessions
temporal_net.add_snapshot('2019-2020', 
    ['Warren', 'Sanders', 'Cruz', 'Paul', 'Schumer'],
    [('Warren', 'Sanders', 0.85), ('Cruz', 'Paul', 0.72)]
)

temporal_net.add_snapshot('2021-2022',
    ['Warren', 'Sanders', 'Cruz', 'Paul', 'Schumer', 'AOC'],
    [('Warren', 'Sanders', 0.89), ('Warren', 'AOC', 0.78), ('Cruz', 'Paul', 0.74)]
)

# Track Warren's influence over time
warren_evolution = temporal_net.track_centrality_evolution('Warren', 'pagerank')
print("Warren's PageRank evolution:", warren_evolution)

# Detect structural changes
changes = temporal_net.detect_structural_changes()
for change in changes:
    print(f"\nPeriod: {change['time_period']}")
    print(f"New relationships: {change['new_relationships']}")
    print(f"Modularity change: {change['modularity_change']:.3f}")

import itertools
from fractions import Fraction
import numpy as np
from collections import defaultdict

class VotingPowerAnalyzer:
    def __init__(self, players, weights, quota):
        self.players = players  # List of player names
        self.weights = weights  # List of voting weights
        self.quota = quota      # Minimum votes needed to win
        self.n = len(players)
    
    def is_winning_coalition(self, coalition_indices):
        """Check if a coalition has enough votes to win"""
        total_weight = sum(self.weights[i] for i in coalition_indices)
        return total_weight >= self.quota
    
    def compute_shapley_values(self):
        """Compute Shapley values for all players"""
        shapley_values = [0.0] * self.n
        
        # Iterate over all possible coalitions
        for r in range(self.n + 1):
            for coalition in itertools.combinations(range(self.n), r):
                coalition_set = set(coalition)
                
                for i in range(self.n):
                    if i not in coalition_set:
                        # Calculate marginal contribution
                        without_i = list(coalition)
                        with_i = list(coalition) + [i]
                        
                        value_without = 1 if self.is_winning_coalition(without_i) else 0
                        value_with = 1 if self.is_winning_coalition(with_i) else 0
                        
                        marginal = value_with - value_without
                        
                        # Weight by coalition size probabilities
                        coalition_size = len(coalition)
                        weight = (1.0 * coalition_size * 
                                (self.n - coalition_size - 1)) / self.n
                        weight *= marginal
                        
                        shapley_values[i] += weight / (2**(self.n - 1))
        
        return {self.players[i]: shapley_values[i] for i in range(self.n)}
    
    def compute_banzhaf_indices(self):
        """Compute Banzhaf power indices for all players"""
        swing_counts = [0] * self.n
        total_coalitions = 0
        
        # Check all possible coalitions
        for r in range(self.n + 1):
            for coalition in itertools.combinations(range(self.n), r):
                coalition_set = set(coalition)
                
                for i in range(self.n):
                    if i not in coalition_set:
                        # Check if player i is pivotal
                        without_i = list(coalition)
                        with_i = list(coalition) + [i]
                        
                        wins_without = self.is_winning_coalition(without_i)
                        wins_with = self.is_winning_coalition(with_i)
                        
                        if not wins_without and wins_with:
                            swing_counts[i] += 1
                
                total_coalitions += 1
        
        # Normalize to get Banzhaf indices
        total_swings = sum(swing_counts)
        if total_swings == 0:
            banzhaf_indices = {self.players[i]: 0 for i in range(self.n)}
        else:
            banzhaf_indices = {self.players[i]: swing_counts[i] / total_swings 
                             for i in range(self.n)}
        
        return banzhaf_indices
    
    def analyze_voting_power(self):
        """Complete voting power analysis"""
        shapley = self.compute_shapley_values()
        banzhaf = self.compute_banzhaf_indices()
        
        # Calculate raw voting weight proportions
        total_weight = sum(self.weights)
        weight_props = {self.players[i]: self.weights[i] / total_weight 
                       for i in range(self.n)}
        
        analysis = {
            'players': self.players,
            'voting_weights': {self.players[i]: self.weights[i] for i in range(self.n)},
            'weight_proportions': weight_props,
            'shapley_values': shapley,
            'banzhaf_indices': banzhaf,
            'quota': self.quota,
            'total_weight': total_weight
        }
        
        return analysis

# Example: UN Security Council voting power
# 5 permanent members (veto power) + 10 non-permanent members
# Simplified model: permanent members have weight 7, non-permanent weight 1
# Need 9 votes including no vetoes from permanent members

council_members = ['US', 'Russia', 'China', 'UK', 'France'] + \
                 [f'NonPerm_{i}' for i in range(1, 11)]
council_weights = [7] * 5 + [1] * 10  # Permanent members have higher weight
council_quota = 9

voting_analyzer = VotingPowerAnalyzer(council_members, council_weights, council_quota)
power_analysis = voting_analyzer.analyze_voting_power()

print("UN Security Council Voting Power Analysis:")
print("\nShapley Values (Decision-making power):")
for member, power in sorted(power_analysis['shapley_values'].items(), 
                           key=lambda x: x[1], reverse=True)[:8]:
    print(f"{member}: {power:.4f}")

print("\nBanzhaf Indices (Swing power):")
for member, power in sorted(power_analysis['banzhaf_indices'].items(),
                           key=lambda x: x[1], reverse=True)[:8]:
    print(f"{member}: {power:.4f}")

import networkx as nx
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
import matplotlib.pyplot as plt
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import AgglomerativeClustering

class PoliticalNetworkAnalysisSystem:
    def __init__(self):
        self.networks = {}  # time_period -> network
        self.metadata = {}  # Additional actor information
        self.analysis_cache = {}  # Cached analysis results
    
    def load_voting_data(self, csv_file, time_column='date', 
                        actor_columns=['senator_1', 'senator_2'],
                        weight_column='voting_similarity'):
        """Load voting similarity data from CSV"""
        df = pd.read_csv(csv_file)
        
        # Group by time periods (e.g., by year or session)
        df['time_period'] = pd.to_datetime(df[time_column]).dt.year
        
        for period, group in df.groupby('time_period'):
            G = nx.Graph()
            
            for _, row in group.iterrows():
                actor1, actor2 = row[actor_columns[0]], row[actor_columns[1]]
                weight = row[weight_column]
                
                G.add_edge(actor1, actor2, weight=weight,
                          votes_together=row.get('votes_together', 0),
                          total_votes=row.get('total_votes', 1))
            
            self.networks[period] = G
    
    def load_financial_data(self, csv_file, time_period):
        """Load campaign contribution networks"""
        df = pd.read_csv(csv_file)
        
        G = nx.DiGraph()  # Directed for donations
        
        for _, row in df.iterrows():
            donor = row['donor']
            recipient = row['recipient']
            amount = row['amount']
            
            if G.has_edge(donor, recipient):
                G[donor][recipient]['weight'] += amount
            else:
                G.add_edge(donor, recipient, weight=amount)
        
        self.networks[f"finance_{time_period}"] = G
    
    def comprehensive_analysis(self, time_period):
        """Perform complete network analysis for a time period"""
        if time_period not in self.networks:
            raise ValueError(f"No network data for period {time_period}")
        
        G = self.networks[time_period]
        analysis = {}
        
        # Basic network properties
        analysis['basic_properties'] = {
            'nodes': G.number_of_nodes(),
            'edges': G.number_of_edges(),
            'density': nx.density(G),
            'is_connected': nx.is_connected(G) if not G.is_directed() else nx.is_weakly_connected(G)
        }
        
        # Centrality measures
        analysis['centrality'] = {
            'degree': nx.degree_centrality(G),
            'betweenness': nx.betweenness_centrality(G, weight='weight'),
            'closeness': nx.closeness_centrality(G, distance='weight'),
            'eigenvector': nx.eigenvector_centrality(G, weight='weight', max_iter=1000),
            'pagerank': nx.pagerank(G, weight='weight')
        }
        
        # Community detection
        if not G.is_directed():
            communities = community.greedy_modularity_communities(G, weight='weight')
            analysis['communities'] = {
                'communities': [list(c) for c in communities],
                'modularity': community.modularity(G, communities, weight='weight'),
                'num_communities': len(communities)
            }
        
        # Network efficiency and robustness
        analysis['robustness'] = self._analyze_robustness(G)
        
        # Top actors by different measures
        analysis['rankings'] = {}
        for measure, scores in analysis['centrality'].items():
            ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
            analysis['rankings'][measure] = ranked[:10]
        
        self.analysis_cache[time_period] = analysis
        return analysis
    
    def _analyze_robustness(self, G):
        """Analyze network robustness to node removal"""
        original_efficiency = nx.global_efficiency(G)
        
        # Test robustness to random node removal
        robustness_scores = []
        nodes = list(G.nodes())
        
        for i in range(min(10, len(nodes))):
            G_copy = G.copy()
            if nodes:
                node_to_remove = np.random.choice(nodes)
                G_copy.remove_node(node_to_remove)
                
                if G_copy.number_of_nodes() > 0:
                    efficiency = nx.global_efficiency(G_copy)
                    robustness_scores.append(efficiency / original_efficiency)
        
        # Test robustness to targeted attacks (remove highest degree nodes)
        degree_centrality = nx.degree_centrality(G)
        high_degree_nodes = sorted(degree_centrality.items(), 
                                 key=lambda x: x[1], reverse=True)
        
        targeted_robustness = []
        for i in range(min(5, len(high_degree_nodes))):
            G_copy = G.copy()
            node_to_remove = high_degree_nodes[i][0]
            G_copy.remove_node(node_to_remove)
            
            if G_copy.number_of_nodes() > 0:
                efficiency = nx.global_efficiency(G_copy)
                targeted_robustness.append(efficiency / original_efficiency)
        
        return {
            'random_failure_robustness': np.mean(robustness_scores) if robustness_scores else 0,
            'targeted_attack_robustness': np.mean(targeted_robustness) if targeted_robustness else 0,
            'original_efficiency': original_efficiency
        }
    
    def temporal_analysis(self):
        """Analyze how the network evolves over time"""
        if len(self.networks) < 2:
            return "Need at least 2 time periods for temporal analysis"
        
        temporal_results = {}
        time_periods = sorted([t for t in self.networks.keys() 
                             if not str(t).startswith('finance')])
        
        for i, period in enumerate(time_periods):
            if period not in self.analysis_cache:
                self.comprehensive_analysis(period)
            
            analysis = self.analysis_cache[period]
            
            temporal_results[period] = {
                'basic_properties': analysis['basic_properties'],
                'top_actors': {
                    measure: ranking[:5] for measure, ranking in analysis['rankings'].items()
                },
                'community_structure': analysis.get('communities', {})
            }
        
        # Calculate change metrics
        changes = []
        for i in range(1, len(time_periods)):
            prev_period, curr_period = time_periods[i-1], time_periods[i]
            
            density_change = (temporal_results[curr_period]['basic_properties']['density'] - 
                            temporal_results[prev_period]['basic_properties']['density'])
            
            changes.append({
                'from_period': prev_period,
                'to_period': curr_period,
                'density_change': density_change,
                'nodes_change': (temporal_results[curr_period]['basic_properties']['nodes'] - 
                               temporal_results[prev_period]['basic_properties']['nodes'])
            })
        
        temporal_results['changes'] = changes
        return temporal_results
    
    def generate_report(self, time_period=None):
        """Generate a comprehensive analysis report"""
        if time_period:
            analysis = self.comprehensive_analysis(time_period)
            
            report = f"""
    POLITICAL NETWORK ANALYSIS REPORT
    Time Period: {time_period}
    =====================================
    
    NETWORK OVERVIEW:
    - Nodes (Actors): {analysis['basic_properties']['nodes']}
    - Edges (Relationships): {analysis['basic_properties']['edges']}
    - Network Density: {analysis['basic_properties']['density']:.3f}
    - Connected: {analysis['basic_properties']['is_connected']}
    
    TOP ACTORS BY INFLUENCE (PageRank):
    """
            
            for i, (actor, score) in enumerate(analysis['rankings']['pagerank'][:5], 1):
                report += f"    {i}. {actor}: {score:.4f}\n"
            
            if 'communities' in analysis:
                report += f"""
    
    COALITION STRUCTURE:
    - Number of Communities: {analysis['communities']['num_communities']}
    - Modularity Score: {analysis['communities']['modularity']:.3f}
    
    NETWORK ROBUSTNESS:
    - Random Failure Tolerance: {analysis['robustness']['random_failure_robustness']:.3f}
    - Targeted Attack Tolerance: {analysis['robustness']['targeted_attack_robustness']:.3f}
    """
            
            return report
        else:
            # Multi-period temporal report
            temporal = self.temporal_analysis()
            return json.dumps(temporal, indent=2, default=str)

# Example usage
analysis_system = PoliticalNetworkAnalysisSystem()

# Simulate loading data (in practice, load from real CSV files)
sample_data = {
    2020: [('Warren', 'Sanders', 0.89), ('Cruz', 'Paul', 0.76)],
    2021: [('Warren', 'Sanders', 0.91), ('Cruz', 'Paul', 0.74), ('Warren', 'AOC', 0.82)],
    2022: [('Warren', 'Sanders', 0.88), ('Cruz', 'Paul', 0.78), ('Warren', 'AOC', 0.85)]
}

for year, relationships in sample_data.items():
    G = nx.Graph()
    G.add_weighted_edges_from(relationships)
    analysis_system.networks[year] = G

# Generate comprehensive report
report = analysis_system.generate_report(2022)
print(report)