"""Influence network calculation for multi-persona interactions""" from typing import List, Dict, Tuple, Literal import math import random import networkx as nx from ..personas.models import Persona from .models import InfluenceWeight, NetworkMetrics NetworkType = Literal["fully_connected", "scale_free", "small_world"] class InfluenceNetwork: """ Calculates influence weights between personas based on their characteristics. Supports multiple network topologies: - fully_connected: Everyone influences everyone (weighted by characteristics) - scale_free: Power-law distribution (some hubs, realistic for social networks) - small_world: Clustered with shortcuts (Watts-Strogatz model) Influence is determined by: 1. Shared values and priorities 2. Expertise credibility (knowledge domains) 3. Political alignment 4. Professional relationships 5. Trust factors """ def __init__( self, personas: List[Persona], network_type: NetworkType = "scale_free", random_seed: int = None, homophily: float = 0.0, ): """ Initialize influence network. Args: personas: List of personas to include network_type: Network topology ("fully_connected", "scale_free", "small_world") random_seed: Random seed for reproducibility homophily: Homophily parameter (0-1). Higher = similar personas cluster together """ self.personas = {p.persona_id: p for p in personas} self.network_type = network_type self.homophily = homophily self.influence_matrix: Dict[Tuple[str, str], InfluenceWeight] = {} self.persona_assignment: Dict[int, str] = {} # node_id -> persona_id mapping if random_seed is not None: random.seed(random_seed) self._calculate_influence_network() def _calculate_influence_network(self): """Calculate pairwise influence weights based on network topology""" persona_ids = list(self.personas.keys()) if self.network_type == "fully_connected": # Everyone influences everyone connections = [ (i, j) for i in persona_ids for j in persona_ids if i != j ] elif self.network_type == "scale_free": # Barabási-Albert: preferential attachment creates hubs connections = self._create_scale_free_topology(persona_ids) elif self.network_type == "small_world": # Watts-Strogatz: clustered with random shortcuts connections = self._create_small_world_topology(persona_ids) else: raise ValueError(f"Unknown network type: {self.network_type}") # Calculate weights for established connections for influencer_id, influenced_id in connections: weight = self._calculate_influence_weight( influencer_id, influenced_id ) self.influence_matrix[(influencer_id, influenced_id)] = weight def _create_scale_free_topology( self, persona_ids: List[str] ) -> List[Tuple[str, str]]: """ Create scale-free network using Barabási-Albert model. Properties: - Power-law degree distribution (few hubs, many peripheral nodes) - Realistic for social networks - Some personas have disproportionate influence """ n = len(persona_ids) if n < 3: # Too small for BA model, use fully connected return [(i, j) for i in persona_ids for j in persona_ids if i != j] # Use networkx to generate scale-free network # m = edges to attach from new node (controls connectivity) m = max(2, min(3, n // 2)) G = nx.barabasi_albert_graph(n, m, seed=None) # Convert to directed graph with bidirectional edges connections = [] for i, j in G.edges(): # Add both directions connections.append((persona_ids[i], persona_ids[j])) connections.append((persona_ids[j], persona_ids[i])) return connections def _create_small_world_topology( self, persona_ids: List[str] ) -> List[Tuple[str, str]]: """ Create small-world network using Watts-Strogatz model. Properties: - High clustering (people in same group know each other) - Short path lengths (few degrees of separation) - Realistic for community-based networks """ n = len(persona_ids) if n < 4: # Too small, use fully connected return [(i, j) for i in persona_ids for j in persona_ids if i != j] # k = each node connected to k nearest neighbors (must be even) k = max(2, min(4, (n - 1) // 2)) if k % 2 != 0: k -= 1 # p = rewiring probability (0.1 = 10% of edges become shortcuts) p = 0.1 G = nx.watts_strogatz_graph(n, k, p, seed=None) # Convert to directed graph with bidirectional edges connections = [] for i, j in G.edges(): # Add both directions connections.append((persona_ids[i], persona_ids[j])) connections.append((persona_ids[j], persona_ids[i])) return connections def _calculate_influence_weight( self, influencer_id: str, influenced_id: str ) -> InfluenceWeight: """Calculate influence weight from one persona to another""" influencer = self.personas[influencer_id] influenced = self.personas[influenced_id] factors = {} # 1. Shared values (0-1) shared_values = self._calculate_shared_values(influencer, influenced) factors["shared_values"] = shared_values # 2. Expertise credibility (0-1) expertise = self._calculate_expertise_credibility(influencer) factors["expertise_credibility"] = expertise # 3. Political alignment (0-1) political = self._calculate_political_alignment(influencer, influenced) factors["political_alignment"] = political # 4. Trust based on openness and community engagement (0-1) trust = self._calculate_trust_factor(influenced) factors["trust_receptivity"] = trust # 5. Professional credibility based on role (0-1) professional = self._calculate_professional_credibility(influencer) factors["professional_credibility"] = professional # Weighted combination weights = { "shared_values": 0.25, "expertise_credibility": 0.20, "political_alignment": 0.20, "trust_receptivity": 0.20, "professional_credibility": 0.15, } total_weight = sum( factors[key] * weights[key] for key in weights.keys() ) return InfluenceWeight( influencer_id=influencer_id, influenced_id=influenced_id, weight=total_weight, factors=factors, ) def _calculate_shared_values( self, p1: Persona, p2: Persona ) -> float: """Calculate overlap in core values (0-1)""" values1 = set(p1.psychographics.core_values) values2 = set(p2.psychographics.core_values) if not values1 or not values2: return 0.5 # Neutral if no values specified intersection = len(values1 & values2) union = len(values1 | values2) return intersection / union if union > 0 else 0.5 def _calculate_expertise_credibility(self, persona: Persona) -> float: """Calculate expertise credibility (0-1)""" if not persona.knowledge_domains: return 0.5 # Neutral credibility # Average expertise level across domains avg_expertise = sum( kd.expertise_level for kd in persona.knowledge_domains ) / len(persona.knowledge_domains) # Normalize to 0-1 return avg_expertise / 10.0 def _calculate_political_alignment( self, p1: Persona, p2: Persona ) -> float: """Calculate political alignment (0-1)""" # Map political leanings to numeric scale scale = { "very_progressive": -2, "progressive": -1, "moderate": 0, "independent": 0, "conservative": 1, "very_conservative": 2, } pos1 = scale.get(p1.psychographics.political_leaning, 0) pos2 = scale.get(p2.psychographics.political_leaning, 0) # Distance on scale (0-4, normalize to 0-1) distance = abs(pos1 - pos2) alignment = 1 - (distance / 4.0) return alignment def _calculate_trust_factor(self, persona: Persona) -> float: """Calculate how receptive persona is to influence (0-1)""" # Based on openness to change and community engagement openness = persona.psychographics.openness_to_change / 10.0 engagement = persona.psychographics.community_engagement / 10.0 # Higher openness = more receptive to influence return (openness * 0.7 + engagement * 0.3) def _calculate_professional_credibility(self, persona: Persona) -> float: """Calculate professional credibility based on role (0-1)""" # Higher credibility for expert roles high_credibility_roles = [ "planner", "engineer", "architect", "economist", "researcher", ] role_lower = persona.role.lower() for keyword in high_credibility_roles: if keyword in role_lower: return 0.8 # Medium credibility for stakeholder roles medium_credibility_roles = [ "advocate", "organizer", "developer", "business", ] for keyword in medium_credibility_roles: if keyword in role_lower: return 0.6 # Default credibility return 0.5 def get_influence_weight( self, influencer_id: str, influenced_id: str ) -> float: """Get influence weight between two personas""" key = (influencer_id, influenced_id) return self.influence_matrix.get(key).weight if key in self.influence_matrix else 0.0 def get_influencers( self, persona_id: str, min_weight: float = 0.5 ) -> List[InfluenceWeight]: """Get personas who significantly influence this persona""" influencers = [] for (influencer_id, influenced_id), weight in self.influence_matrix.items(): if influenced_id == persona_id and weight.weight >= min_weight: influencers.append(weight) return sorted(influencers, key=lambda x: x.weight, reverse=True) def get_influenced_by( self, persona_id: str, min_weight: float = 0.5 ) -> List[InfluenceWeight]: """Get personas significantly influenced by this persona""" influenced = [] for (influencer_id, influenced_id), weight in self.influence_matrix.items(): if influencer_id == persona_id and weight.weight >= min_weight: influenced.append(weight) return sorted(influenced, key=lambda x: x.weight, reverse=True) def calculate_network_metrics(self) -> NetworkMetrics: """Calculate overall network metrics""" # Centrality: how influential each persona is overall centrality = {} for persona_id in self.personas.keys(): influenced_list = self.get_influenced_by(persona_id, min_weight=0.0) centrality[persona_id] = sum( w.weight for w in influenced_list ) / len(self.personas) if len(self.personas) > 1 else 0.0 # Average influence weight total_weight = sum(w.weight for w in self.influence_matrix.values()) avg_influence = ( total_weight / len(self.influence_matrix) if self.influence_matrix else 0.0 ) # Simple clustering coefficient (approximate) # Measure how connected the influence network is strong_connections = sum( 1 for w in self.influence_matrix.values() if w.weight > 0.6 ) clustering = ( strong_connections / len(self.influence_matrix) if self.influence_matrix else 0.0 ) return NetworkMetrics( centrality_scores=centrality, clustering_coefficient=clustering, average_influence=avg_influence, ) def get_network_edges( self, min_weight: float = 0.5 ) -> List[Dict[str, any]]: """Get network edges for visualization (above threshold)""" edges = [] for weight in self.influence_matrix.values(): if weight.weight >= min_weight: edges.append({ "source": weight.influencer_id, "target": weight.influenced_id, "weight": weight.weight, "factors": weight.factors, }) return edges def calculate_persona_similarity(self, p1: Persona, p2: Persona) -> float: """ Calculate overall similarity between two personas (0-1). Used for homophily-based network assignment. Higher values = more similar personas. """ # Reuse existing similarity calculations shared_values = self._calculate_shared_values(p1, p2) political = self._calculate_political_alignment(p1, p2) # Add demographic similarity age_diff = abs(p1.demographics.age - p2.demographics.age) / 100.0 age_similarity = 1.0 - min(age_diff, 1.0) # Education similarity (same level = 1.0, different = 0.5) edu_similarity = 1.0 if p1.demographics.education == p2.demographics.education else 0.5 # Weighted combination similarity = ( shared_values * 0.4 + political * 0.3 + age_similarity * 0.15 + edu_similarity * 0.15 ) return similarity @staticmethod def get_persona_base_type(persona: Persona) -> str: """ Extract base persona type from persona_id. For variants, returns the base persona name. E.g., "sarah_chen_v0" -> "sarah_chen" """ persona_id = persona.persona_id # Remove variant suffix if present if "_v" in persona_id: return persona_id.rsplit("_v", 1)[0] return persona_id