Behavioral Analysis

graph TB
    subgraph "Behavioral Intelligence Collection"
        A1["🗳️ Voting Behavior<br/>3.5M+ votes analyzed<br/>Deviation tracking"]
        A2["👥 Social Networks<br/>Collaboration patterns<br/>Influence metrics"]
        A3["📄 Productivity Signals<br/>Document authorship<br/>Committee activity"]
        A4["🎤 Communication Style<br/>Speech analysis<br/>Rhetoric patterns"]
        A5["⏱️ Temporal Patterns<br/>Attendance trends<br/>Engagement cycles"]
    end
    
    subgraph "Psychological Analysis"
        A1 --> B1[Decision-Making Analysis]
        A2 --> B2[Group Dynamics Assessment]
        A3 --> B3[Motivation Evaluation]
        A4 --> B4[Leadership Style Profiling]
        A5 --> B5[Behavioral Consistency Check]
    end
    
    subgraph "Cognitive Bias Detection"
        B1 --> C1{Confirmation Bias}
        B2 --> C2{Groupthink}
        B3 --> C3{Status Quo Bias}
        B4 --> C4{Authority Bias}
        B5 --> C5{Recency Bias}
    end
    
    subgraph "Intelligence Product"
        C1 & C2 & C3 & C4 & C5 --> D["🧠 Behavioral Profile"]
        D --> E[Predictive Insights]
        D --> F[Risk Indicators]
        D --> G[Leadership Assessment]
    end
    
    style A1 fill:#e1f5ff
    style A2 fill:#e1f5ff
    style A3 fill:#e1f5ff
    style A4 fill:#e1f5ff
    style A5 fill:#e1f5ff
    style D fill:#ffe6cc
    style E fill:#ccffcc
    style F fill:#ffcccc
    style G fill:#fff9cc

@Component
public class PartyConformityAnalyzer {
    
    /**
     * Analyzes voting deviation patterns to identify cognitive dissonance.
     * 
     * High deviation indicates:
     * - Internal conflict with party platform
     * - Constituency pressure overriding party discipline
     * - Personal ideology asserting independence
     * - Strategic positioning for leadership
     */
    @Transactional(readOnly = true)
    public PartyConformityProfile analyzeConformity(String politicianId, String partyId) {
        String sql = """
            SELECT 
                p.person_id,
                p.first_name || ' ' || p.last_name as name,
                p.party as current_party,
                vbs.total_votes,
                vbs.won_votes,
                vbs.lost_votes,
                vbs.rebel_votes,
                vbs.avg_vote_win_rate,
                vbs.vote_effectiveness_score,
                ROUND(100.0 * vbs.rebel_votes / NULLIF(vbs.total_votes, 0), 2) as deviation_rate,
                
                -- Behavioral indicators
                CASE 
                    WHEN vbs.rebel_votes::float / NULLIF(vbs.total_votes, 0) < 0.02 THEN 'CONFORMIST'
                    WHEN vbs.rebel_votes::float / NULLIF(vbs.total_votes, 0) < 0.05 THEN 'MODERATE'
                    WHEN vbs.rebel_votes::float / NULLIF(vbs.total_votes, 0) < 0.10 THEN 'INDEPENDENT'
                    ELSE 'MAVERICK'
                END as conformity_type,
                
                -- Cognitive dissonance indicators
                CASE 
                    WHEN vbs.rebel_votes > 50 AND vbs.rebel_votes::float / vbs.total_votes > 0.10 
                        THEN 'HIGH_DISSONANCE'
                    WHEN vbs.rebel_votes > 20 AND vbs.rebel_votes::float / vbs.total_votes > 0.05 
                        THEN 'MODERATE_DISSONANCE'
                    ELSE 'LOW_DISSONANCE'
                END as dissonance_level
                
            FROM view_riksdagen_politician p
            JOIN view_riksdagen_politician_ballot_summary vbs ON p.person_id = vbs.person_id
            WHERE p.person_id = :politicianId
                AND p.party = :partyId
            """;
        
        return jdbcTemplate.queryForObject(sql, PartyConformityProfile.class,
            Map.of("politicianId", politicianId, "partyId", partyId));
    }
}

import pandas as pd
import networkx as nx
from typing import Dict, List, Tuple

class EchoChamberDetector:
    """
    Detects echo chambers in parliamentary committees using network analysis.
    
    Indicators of echo chambers:
    - High internal connectivity, low external bridges
    - Ideological homogeneity exceeding party baseline
    - Resistance to cross-party collaboration
    - Information isolation from opposing viewpoints
    """
    
    def analyze_committee_network(self, committee_id: str) -> Dict:
        """
        Analyzes committee collaboration networks for echo chamber indicators.
        
        Returns metrics:
        - Internal density: Collaboration within ideological cluster
        - Bridge centrality: Cross-cluster information flow
        - Homophily index: Ideological similarity preference
        - Polarization score: Cluster separation intensity
        """
        
        query = """
        SELECT 
            c.org_code,
            c.committee_name,
            
            -- Network structure metrics
            COUNT(DISTINCT cm.person_id) as member_count,
            COUNT(DISTINCT cm.party) as party_diversity,
            
            -- Collaboration patterns (co-authorship, co-sponsorship)
            (SELECT COUNT(*) 
             FROM document_person dp1 
             JOIN document_person dp2 ON dp1.document_id = dp2.document_id 
             WHERE dp1.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
               AND dp2.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
               AND dp1.person_id < dp2.person_id
            ) as internal_collaboration,
            
            -- Cross-party bridge activity
            (SELECT COUNT(*) 
             FROM document_person dp1 
             JOIN document_person dp2 ON dp1.document_id = dp2.document_id 
             JOIN person p1 ON dp1.person_id = p1.person_id
             JOIN person p2 ON dp2.person_id = p2.person_id
             WHERE dp1.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
               AND dp2.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
               AND p1.party != p2.party
               AND dp1.person_id < dp2.person_id
            ) as cross_party_bridges,
            
            -- Ideological homogeneity (voting similarity)
            AVG(
                (SELECT AVG(
                    CASE WHEN v1.vote = v2.vote THEN 1.0 ELSE 0.0 END
                ) FROM vote v1, vote v2 
                WHERE v1.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
                  AND v2.person_id IN (SELECT person_id FROM committee_member WHERE org_code = c.org_code)
                  AND v1.ballot_id = v2.ballot_id
                  AND v1.person_id < v2.person_id
                )
            ) as internal_voting_similarity
            
        FROM committee c
        JOIN committee_member cm ON c.org_code = cm.org_code
        WHERE c.org_code = %s
        GROUP BY c.org_code, c.committee_name
        """
        
        df = pd.read_sql(query, self.connection, params=[committee_id])
        
        # Calculate echo chamber indicators
        internal_density = df['internal_collaboration'].iloc[0] / (df['member_count'].iloc[0] ** 2)
        bridge_ratio = df['cross_party_bridges'].iloc[0] / max(df['internal_collaboration'].iloc[0], 1)
        homophily_index = df['internal_voting_similarity'].iloc[0]
        
        # Echo chamber score (0-100, higher = stronger echo chamber)
        echo_chamber_score = (
            (internal_density * 30) +
            ((1 - bridge_ratio) * 30) +
            (homophily_index * 40)
        )
        
        return {
            'committee_id': committee_id,
            'echo_chamber_score': round(echo_chamber_score, 2),
            'internal_density': round(internal_density, 3),
            'bridge_ratio': round(bridge_ratio, 3),
            'homophily_index': round(homophily_index, 3),
            'classification': self._classify_echo_chamber(echo_chamber_score)
        }
    
    def _classify_echo_chamber(self, score: float) -> str:
        """Classify echo chamber severity."""
        if score >= 75:
            return "SEVERE_ECHO_CHAMBER"
        elif score >= 60:
            return "MODERATE_ECHO_CHAMBER"
        elif score >= 40:
            return "MILD_POLARIZATION"
        else:
            return "HEALTHY_DIVERSITY"

@Service
public class LeadershipStyleAnalyzer {
    
    /**
     * Analyzes leadership effectiveness through behavioral indicators.
     * 
     * Based on transformational leadership theory (Bass & Riggio, 2006)
     * and political leadership research (Burns, 1978).
     */
    public LeadershipProfile analyzeLeadership(String politicianId) {
        String sql = """
            WITH leadership_metrics AS (
                SELECT 
                    p.person_id,
                    p.first_name || ' ' || p.last_name as name,
                    p.party,
                    
                    -- Dimension 1: Collaborative vs. Authoritarian
                    vim.collaboration_score,
                    vim.network_centrality,
                    
                    -- Dimension 2: Ideological vs. Pragmatic
                    vbs.rebel_votes::float / NULLIF(vbs.total_votes, 0) as ideological_purity,
                    vbs.vote_effectiveness_score as pragmatic_success,
                    
                    -- Dimension 3: Proactive vs. Reactive
                    COUNT(DISTINCT d.document_id) as initiated_documents,
                    vbs.total_votes as participation_votes,
                    
                    -- Dimension 4: Consensus-builder vs. Confrontational
                    vim.cross_party_collaboration_score,
                    vbs.rebel_votes as confrontational_votes,
                    
                    -- Dimension 5: Visible vs. Behind-scenes
                    COUNT(DISTINCT CASE WHEN d.document_type = 'motion' THEN d.document_id END) as public_initiatives,
                    COUNT(DISTINCT CASE WHEN d.document_type = 'interpellation' THEN d.document_id END) as oversight_activity
                    
                FROM view_riksdagen_politician p
                LEFT JOIN view_riksdagen_politician_ballot_summary vbs ON p.person_id = vbs.person_id
                LEFT JOIN view_riksdagen_politician_influence_metrics vim ON p.person_id = vim.person_id
                LEFT JOIN view_riksdagen_politician_document d ON p.person_id = d.person_id
                WHERE p.person_id = :politicianId
                GROUP BY p.person_id, p.first_name, p.last_name, p.party,
                         vim.collaboration_score, vim.network_centrality,
                         vbs.rebel_votes, vbs.total_votes, vbs.vote_effectiveness_score
            )
            SELECT 
                *,
                -- Leadership style classification
                CASE 
                    WHEN collaboration_score > 0.7 AND cross_party_collaboration_score > 0.6 
                        THEN 'TRANSFORMATIONAL'
                    WHEN ideological_purity > 0.15 AND confrontational_votes > 100 
                        THEN 'IDEOLOGICAL_PURIST'
                    WHEN pragmatic_success > 0.75 AND cross_party_collaboration_score > 0.5 
                        THEN 'PRAGMATIC_DEALMAKER'
                    WHEN initiated_documents > 50 AND public_initiatives > 30 
                        THEN 'POLICY_ENTREPRENEUR'
                    WHEN network_centrality > 0.8 AND collaboration_score < 0.4 
                        THEN 'AUTHORITARIAN_BROKER'
                    ELSE 'BACKBENCHER'
                END as leadership_style
            FROM leadership_metrics
            """;
        
        return jdbcTemplate.queryForObject(sql, LeadershipProfile.class,
            Map.of("politicianId", politicianId));
    }
}

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta

@dataclass
class CognitiveBiasIndicators:
    """Indicators of cognitive biases in political decision-making."""
    politician_id: str
    confirmation_bias_score: float
    status_quo_bias_score: float
    authority_bias_score: float
    recency_bias_score: float
    availability_bias_score: float
    
class CognitiveBiasDetector:
    """
    Identifies cognitive biases through voting pattern analysis.
    
    Based on Kahneman & Tversky's cognitive bias research
    applied to political decision-making contexts.
    """
    
    def detect_confirmation_bias(self, politician_id: str) -> float:
        """
        Detects confirmation bias: Tendency to vote with pre-existing beliefs.
        
        Measured by:
        - Consistency with historical positions
        - Resistance to policy evolution despite new evidence
        - Selective attention to information supporting prior stance
        """
        
        query = """
        WITH politician_voting AS (
            SELECT 
                v.person_id,
                v.vote,
                b.issue_category,
                b.vote_date,
                LAG(v.vote) OVER (
                    PARTITION BY v.person_id, b.issue_category 
                    ORDER BY b.vote_date
                ) as previous_vote,
                LAG(b.vote_date) OVER (
                    PARTITION BY v.person_id, b.issue_category 
                    ORDER BY b.vote_date
                ) as previous_date
            FROM vote v
            JOIN ballot b ON v.ballot_id = b.ballot_id
            WHERE v.person_id = %s
                AND b.vote_date >= NOW() - INTERVAL '4 years'
        )
        SELECT 
            person_id,
            -- Consistency score: How often votes align with historical position
            AVG(CASE WHEN vote = previous_vote THEN 1.0 ELSE 0.0 END) as consistency_rate,
            
            -- Rigidity score: Resistance to policy evolution over time
            COUNT(CASE WHEN vote != previous_vote 
                       AND previous_date < vote_date - INTERVAL '1 year' 
                       THEN 1 END)::float / COUNT(*) as evolution_resistance,
            
            COUNT(*) as total_comparable_votes
            
        FROM politician_voting
        WHERE previous_vote IS NOT NULL
        GROUP BY person_id
        """
        
        result = pd.read_sql(query, self.connection, params=[politician_id])
        
        if result.empty or result['total_comparable_votes'].iloc[0] < 10:
            return 0.0
        
        # Confirmation bias score: High consistency + high resistance = stronger bias
        consistency_rate = result['consistency_rate'].iloc[0]
        evolution_resistance = result['evolution_resistance'].iloc[0]
        
        bias_score = (consistency_rate * 0.6) + (evolution_resistance * 0.4)
        return round(bias_score * 100, 2)
    
    def detect_status_quo_bias(self, politician_id: str) -> float:
        """
        Detects status quo bias: Preference for maintaining current state.
        
        Measured by:
        - Voting against reform proposals
        - Supporting incumbent policies
        - Resisting change initiatives
        """
        
        query = """
        SELECT 
            v.person_id,
            COUNT(CASE WHEN b.is_reform_proposal = TRUE AND v.vote = 'Nej' THEN 1 END)::float /
            NULLIF(COUNT(CASE WHEN b.is_reform_proposal = TRUE THEN 1 END), 0) as reform_opposition_rate,
            
            COUNT(CASE WHEN b.is_status_quo_motion = TRUE AND v.vote = 'Ja' THEN 1 END)::float /
            NULLIF(COUNT(CASE WHEN b.is_status_quo_motion = TRUE THEN 1 END), 0) as status_quo_support_rate,
            
            COUNT(*) as total_policy_votes
            
        FROM vote v
        JOIN ballot b ON v.ballot_id = b.ballot_id
        WHERE v.person_id = %s
            AND (b.is_reform_proposal = TRUE OR b.is_status_quo_motion = TRUE)
            AND b.vote_date >= NOW() - INTERVAL '2 years'
        GROUP BY v.person_id
        """
        
        result = pd.read_sql(query, self.connection, params=[politician_id])
        
        if result.empty or result['total_policy_votes'].iloc[0] < 5:
            return 0.0
        
        reform_opposition = result['reform_opposition_rate'].iloc[0] or 0.0
        status_quo_support = result['status_quo_support_rate'].iloc[0] or 0.0
        
        bias_score = (reform_opposition * 0.5) + (status_quo_support * 0.5)
        return round(bias_score * 100, 2)
    
    def detect_authority_bias(self, politician_id: str) -> float:
        """
        Detects authority bias: Over-reliance on party leadership guidance.
        
        Measured by:
        - Voting alignment with party leadership
        - Lack of independent positions
        - Deference to authority figures
        """
        
        query = """
        WITH party_leader_votes AS (
            SELECT 
                v.ballot_id,
                v.vote as leader_vote
            FROM vote v
            JOIN person p ON v.person_id = p.person_id
            WHERE p.is_party_leader = TRUE
                AND p.party = (SELECT party FROM person WHERE person_id = %s)
        )
        SELECT 
            v.person_id,
            COUNT(CASE WHEN v.vote = plv.leader_vote THEN 1 END)::float /
            NULLIF(COUNT(*), 0) as leadership_alignment_rate,
            
            COUNT(*) as total_votes_with_leader
            
        FROM vote v
        JOIN party_leader_votes plv ON v.ballot_id = plv.ballot_id
        WHERE v.person_id = %s
        GROUP BY v.person_id
        """
        
        result = pd.read_sql(query, self.connection, params=[politician_id, politician_id])
        
        if result.empty or result['total_votes_with_leader'].iloc[0] < 20:
            return 0.0
        
        alignment_rate = result['leadership_alignment_rate'].iloc[0]
        
        # Authority bias score: Very high alignment suggests deference
        if alignment_rate > 0.95:
            return 100.0
        elif alignment_rate > 0.90:
            return 75.0
        elif alignment_rate > 0.85:
            return 50.0
        else:
            return round((alignment_rate - 0.70) * 200, 2)  # Scale 70-85% to 0-30
    
    def detect_recency_bias(self, politician_id: str) -> float:
        """
        Detects recency bias: Disproportionate weight on recent information.
        
        Measured by:
        - Vote position changes after recent media coverage
        - Inconsistency with long-term stance based on recent events
        - Rapid policy shifts following public attention
        """
        
        query = """
        WITH recent_votes AS (
            SELECT 
                v.person_id,
                v.vote,
                b.issue_category,
                b.vote_date,
                CASE WHEN b.vote_date >= NOW() - INTERVAL '90 days' THEN 'recent'
                     WHEN b.vote_date >= NOW() - INTERVAL '1 year' THEN 'medium_term'
                     ELSE 'historical' END as time_period
            FROM vote v
            JOIN ballot b ON v.ballot_id = b.ballot_id
            WHERE v.person_id = %s
                AND b.vote_date >= NOW() - INTERVAL '3 years'
        ),
        consistency_analysis AS (
            SELECT 
                person_id,
                issue_category,
                AVG(CASE WHEN time_period = 'recent' AND vote = 'Ja' THEN 1.0 ELSE 0.0 END) as recent_support,
                AVG(CASE WHEN time_period = 'historical' AND vote = 'Ja' THEN 1.0 ELSE 0.0 END) as historical_support
            FROM recent_votes
            GROUP BY person_id, issue_category
            HAVING COUNT(CASE WHEN time_period = 'recent' THEN 1 END) >= 3
               AND COUNT(CASE WHEN time_period = 'historical' THEN 1 END) >= 5
        )
        SELECT 
            person_id,
            AVG(ABS(recent_support - historical_support)) as avg_shift_magnitude,
            COUNT(*) as analyzed_categories
        FROM consistency_analysis
        WHERE ABS(recent_support - historical_support) > 0.20  -- Significant shift threshold
        GROUP BY person_id
        """
        
        result = pd.read_sql(query, self.connection, params=[politician_id])
        
        if result.empty or result['analyzed_categories'].iloc[0] < 3:
            return 0.0
        
        shift_magnitude = result['avg_shift_magnitude'].iloc[0]
        
        # Recency bias score: Larger shifts = stronger bias
        bias_score = min(shift_magnitude * 150, 100)  # Cap at 100
        return round(bias_score, 2)