Hack23 Risk Assessment Frameworks Skill
Purpose
This skill provides comprehensive risk assessment methodologies for evaluating political, institutional, and democratic risks within the Swedish political system. It integrates international frameworks (V-Dem, Transparency International, Freedom House) with CIA platform's proprietary 50+ Drools risk rules to create systematic early warning capabilities for democratic backsliding, corruption, institutional erosion, political violence, and coalition instability.
When to Use This Skill
Apply this skill when:
- ✅ Conducting democratic health assessments of Swedish institutions
- ✅ Identifying early warning signs of institutional erosion
- ✅ Assessing corruption risk at politician or party level
- ✅ Evaluating coalition stability and government sustainability
- ✅ Detecting democratic backsliding indicators
- ✅ Measuring institutional accountability effectiveness
- ✅ Analyzing political violence risk factors
- ✅ Creating risk-based intelligence priorities
- ✅ Benchmarking Sweden against international democracy standards
- ✅ Generating risk reports for stakeholders and media
Do NOT use for:
- ❌ Political persecution or targeting of legitimate opposition
- ❌ Fabricating risks to manipulate public opinion
- ❌ Undermining democratic institutions through false alarms
- ❌ Violating privacy or conducting surveillance without legal basis
Risk Assessment Framework Architecture
Integrated Risk Intelligence System
The CIA platform integrates four layers of risk intelligence to create comprehensive political risk profiles:
graph TB
subgraph "Layer 1: Data Collection"
A1["🗳️ Behavioral Data<br/>3.5M+ votes, attendance<br/>Productivity metrics"]
A2["💰 Financial Data<br/>World Bank, ESV<br/>Economic indicators"]
A3["📊 Democracy Indices<br/>V-Dem, Freedom House<br/>International benchmarks"]
A4["📰 Media Coverage<br/>Sentiment analysis<br/>Scandal tracking"]
end
subgraph "Layer 2: Risk Rules Engine (Drools)"
A1 --> B1[Behavioral Risk Rules<br/>24 politician rules<br/>12 party rules]
A2 --> B2[Financial Risk Rules<br/>8 corruption indicators]
A3 --> B3[Democratic Health Rules<br/>6 institutional rules]
A4 --> B4[Reputational Risk Rules<br/>4 scandal detection rules]
end
subgraph "Layer 3: Risk Aggregation"
B1 --> C1[Individual Risk Profiles]
B2 --> C2[Institutional Risk Profiles]
B3 --> C3[Systemic Risk Profiles]
B4 --> C4[Reputational Risk Profiles]
end
subgraph "Layer 4: Risk Intelligence"
C1 & C2 & C3 & C4 --> D["🎯 Composite Risk Score"]
D --> E[Early Warning Alerts]
D --> F[Risk Mitigation Strategies]
D --> G[Intelligence Priorities]
end
style A1 fill:#e1f5ff
style A2 fill:#e1f5ff
style A3 fill:#e1f5ff
style A4 fill:#e1f5ff
style B1 fill:#fff9cc
style B2 fill:#fff9cc
style B3 fill:#fff9cc
style B4 fill:#fff9cc
style D fill:#ffe6cc
style E fill:#ffcccc
style F fill:#ccffcc
style G fill:#e6ccff
1. Democratic Backsliding Detection
V-Dem Integration Framework
The Varieties of Democracy (V-Dem) project provides the world's most comprehensive democracy measurement. The CIA platform integrates V-Dem indicators with behavioral data.
V-Dem Core Indicators Tracked:
- Liberal Democracy Index - Rule of law, checks on government
- Electoral Democracy Index - Free and fair elections
- Participatory Democracy Index - Citizen participation
- Deliberative Democracy Index - Quality of public discourse
- Egalitarian Democracy Index - Equal access to power
from typing import Dict, List, Tuple
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class DemocraticBackslidingDetector:
"""
Detects democratic backsliding through trend analysis and threshold monitoring.
Based on V-Dem Early Warning of Democratic Decline (Edda) methodology
and combines international indices with CIA platform behavioral data.
"""
# V-Dem backsliding thresholds (0-1 scale)
CRITICAL_THRESHOLDS = {
'liberal_democracy_index': 0.50, # Below = autocratization
'electoral_democracy_index': 0.60, # Below = electoral manipulation
'participatory_democracy_index': 0.45, # Below = citizen disengagement
'deliberative_democracy_index': 0.50, # Below = discourse degradation
'egalitarian_democracy_index': 0.55 # Below = inequality deepening
}
def assess_democratic_health(self, country_code: str = 'SWE') -> Dict:
"""
Comprehensive democratic health assessment for Sweden.
Combines:
1. V-Dem historical trends (5-year analysis)
2. CIA behavioral indicators (parliamentary effectiveness)
3. International comparison (Nordic benchmarking)
4. Early warning signals (acceleration detection)
"""
# Fetch V-Dem data
vdem_query = """
SELECT
year,
v2x_libdem as liberal_democracy_index,
v2x_polyarchy as electoral_democracy_index,
v2x_partipdem as participatory_democracy_index,
v2x_delibdem as deliberative_democracy_index,
v2x_egaldem as egalitarian_democracy_index,
-- Component indicators
v2x_judicind as judicial_independence,
v2x_frassoc_thick as freedom_association,
v2x_freexp_altinf as freedom_expression,
v2x_elecoff as elected_officials_index,
v2xlg_legcon as legislative_constraints,
v2x_corr as political_corruption_index,
-- Backsliding indicators
v2x_regime as regime_type
FROM vdem_data
WHERE country_code = %s
AND year >= EXTRACT(YEAR FROM NOW()) - 10
ORDER BY year DESC
"""
vdem_df = pd.read_sql(vdem_query, self.connection, params=[country_code])
# Calculate trends (5-year linear regression slopes)
trends = {}
for column in vdem_df.columns:
if column not in ['year', 'country_code', 'regime_type']:
X = vdem_df['year'].values.reshape(-1, 1)
y = vdem_df[column].values
# Simple linear regression
slope = np.polyfit(X.flatten(), y, 1)[0]
trends[column] = round(slope, 4)
# Fetch CIA behavioral indicators
behavioral_query = """
SELECT
-- Parliamentary effectiveness
AVG(ce.overall_effectiveness_score) as avg_committee_effectiveness,
-- Party discipline (inverse of deviation)
AVG(100 - pd.avg_deviation_rate) as avg_party_discipline,
-- Oversight activity
COUNT(DISTINCT oa.document_id) as oversight_action_count,
AVG(oa.oversight_effectiveness_score) as avg_oversight_effectiveness,
-- Cross-party collaboration
AVG(cpc.collaboration_intensity) as avg_cross_party_collaboration,
-- Voting participation
AVG(100 - vbs.avg_absent_percentage) as avg_participation_rate
FROM committee_effectiveness ce,
party_deviation pd,
oversight_activity oa,
cross_party_collaboration cpc,
vote_ballot_summary vbs
WHERE pd.analysis_date >= NOW() - INTERVAL '2 years'
AND oa.created_date >= NOW() - INTERVAL '2 years'
"""
behavioral_data = pd.read_sql(behavioral_query, self.connection).iloc[0]
# Current V-Dem scores
current_vdem = vdem_df.iloc[0]
# Identify risks
risks = self._identify_risks(current_vdem, trends, behavioral_data)
# Calculate composite democratic health score (0-100)
health_score = self._calculate_health_score(current_vdem, behavioral_data)
# Early warning assessment
early_warnings = self._detect_early_warnings(trends, current_vdem)
return {
'country': country_code,
'assessment_date': datetime.now().isoformat(),
'current_scores': {
'liberal_democracy': round(current_vdem['liberal_democracy_index'], 3),
'electoral_democracy': round(current_vdem['electoral_democracy_index'], 3),
'participatory_democracy': round(current_vdem['participatory_democracy_index'], 3),
'deliberative_democracy': round(current_vdem['deliberative_democracy_index'], 3),
'egalitarian_democracy': round(current_vdem['egalitarian_democracy_index'], 3)
},
'5_year_trends': trends,
'behavioral_indicators': {
'committee_effectiveness': round(behavioral_data['avg_committee_effectiveness'], 2),
'party_discipline': round(behavioral_data['avg_party_discipline'], 2),
'oversight_effectiveness': round(behavioral_data['avg_oversight_effectiveness'], 2),
'cross_party_collaboration': round(behavioral_data['avg_cross_party_collaboration'], 3),
'participation_rate': round(behavioral_data['avg_participation_rate'], 2)
},
'composite_health_score': round(health_score, 2),
'health_classification': self._classify_health(health_score),
'identified_risks': risks,
'early_warnings': early_warnings,
'international_ranking': self._get_nordic_comparison(current_vdem)
}
def _identify_risks(
self,
current: pd.Series,
trends: Dict,
behavioral: pd.Series
) -> List[str]:
"""Identify specific democratic risks."""
risks = []
# Check V-Dem thresholds
for indicator, threshold in self.CRITICAL_THRESHOLDS.items():
if current.get(indicator, 1.0) < threshold:
risks.append(
f"CRITICAL: {indicator} below threshold "
f"({current[indicator]:.3f} < {threshold})"
)
# Check negative trends
for indicator, slope in trends.items():
if slope < -0.01: # Declining more than 0.01/year
risks.append(
f"WARNING: Declining {indicator} (trend: {slope:.4f}/year)"
)
# Check behavioral indicators
if behavioral['avg_committee_effectiveness'] < 50:
risks.append("Institutional dysfunction: Low committee effectiveness")
if behavioral['avg_oversight_effectiveness'] < 60:
risks.append("Accountability deficit: Weak oversight mechanisms")
if behavioral['avg_participation_rate'] < 85:
risks.append("Disengagement: Low parliamentary participation")
return risks if risks else ["No critical risks detected"]
def _calculate_health_score(
self,
vdem: pd.Series,
behavioral: pd.Series
) -> float:
"""Calculate composite democratic health score (0-100)."""
# V-Dem component (70% weight)
vdem_score = (
vdem['liberal_democracy_index'] * 20 +
vdem['electoral_democracy_index'] * 20 +
vdem['participatory_democracy_index'] * 10 +
vdem['deliberative_democracy_index'] * 10 +
vdem['egalitarian_democracy_index'] * 10
)
# Behavioral component (30% weight)
behavioral_score = (
(behavioral['avg_committee_effectiveness'] / 100) * 10 +
(behavioral['avg_oversight_effectiveness'] / 100) * 10 +
(behavioral['avg_participation_rate'] / 100) * 10
)
return vdem_score * 100 + behavioral_score
def _classify_health(self, score: float) -> str:
"""Classify democratic health."""
if score >= 85:
return "ROBUST_DEMOCRACY"
elif score >= 70:
return "HEALTHY_DEMOCRACY"
elif score >= 55:
return "FLAWED_DEMOCRACY"
elif score >= 40:
return "HYBRID_REGIME"
else:
return "AUTOCRATIC_REGIME"
def _detect_early_warnings(
self,
trends: Dict,
current: pd.Series
) -> List[str]:
"""Detect early warning signals of democratic decline."""
warnings = []
# Accelerating decline (second derivative)
declining_indicators = [k for k, v in trends.items() if v < -0.005]
if len(declining_indicators) >= 3:
warnings.append(
"EARLY WARNING: Multiple indicators declining simultaneously"
)
# Judicial independence warning
if (current.get('judicial_independence', 1.0) < 0.70 or
trends.get('judicial_independence', 0) < -0.01):
warnings.append(
"CRITICAL: Judicial independence erosion detected"
)
# Freedom of expression warning
if (current.get('freedom_expression', 1.0) < 0.75 or
trends.get('freedom_expression', 0) < -0.01):
warnings.append(
"WARNING: Press freedom and expression declining"
)
# Legislative constraints weakening
if (current.get('legislative_constraints', 1.0) < 0.70 or
trends.get('legislative_constraints', 0) < -0.01):
warnings.append(
"WARNING: Legislative oversight weakening"
)
# Corruption increasing
if trends.get('political_corruption_index', 0) > 0.01:
warnings.append(
"WARNING: Political corruption index increasing"
)
return warnings if warnings else ["No early warnings detected"]
def _get_nordic_comparison(self, current: pd.Series) -> Dict:
"""Compare Sweden to other Nordic countries."""
query = """
SELECT
country_name,
v2x_libdem as liberal_democracy_index
FROM vdem_data
WHERE country_code IN ('SWE', 'NOR', 'DNK', 'FIN', 'ISL')
AND year = (SELECT MAX(year) FROM vdem_data)
ORDER BY v2x_libdem DESC
"""
nordic_df = pd.read_sql(query, self.connection)
sweden_rank = nordic_df[
nordic_df['country_name'] == 'Sweden'
].index[0] + 1 if 'Sweden' in nordic_df['country_name'].values else None
return {
'nordic_ranking': f"{sweden_rank}/5" if sweden_rank else "N/A",
'regional_comparison': nordic_df.to_dict('records')
}
2. Corruption Risk Assessment
Transparency International Integration
The CIA platform integrates Transparency International's Corruption Perceptions Index (CPI) methodology with behavioral indicators to assess corruption risk.
@Service
public class CorruptionRiskAnalyzer {
/**
* Multi-dimensional corruption risk assessment.
*
* Risk dimensions:
* 1. Financial irregularities (unexplained wealth, conflict of interest)
* 2. Behavioral anomalies (voting patterns inconsistent with stated positions)
* 3. Network corruption (connections to sanctioned entities)
* 4. Transparency violations (disclosure failures, opacity)
* 5. Accountability evasion (oversight avoidance, question dodging)
*/
public CorruptionRiskProfile assessCorruptionRisk(String politicianId) {
String sql = """
WITH financial_risk AS (
SELECT
p.person_id,
-- Financial disclosure completeness
fd.disclosure_completeness_score,
fd.wealth_change_unexplained_ratio,
fd.conflict_of_interest_declarations,
-- Red flags
CASE WHEN fd.wealth_change_unexplained_ratio > 0.30 THEN 1 ELSE 0 END as wealth_anomaly_flag,
CASE WHEN fd.disclosure_completeness_score < 0.70 THEN 1 ELSE 0 END as disclosure_failure_flag,
CASE WHEN fd.conflict_of_interest_declarations = 0 AND fd.business_holdings > 0
THEN 1 ELSE 0 END as coi_omission_flag
FROM person p
LEFT JOIN financial_disclosure fd ON p.person_id = fd.person_id
WHERE p.person_id = :politicianId
),
behavioral_risk AS (
SELECT
p.person_id,
-- Rhetoric-action gaps (potential deception)
raa.credibility_score,
raa.contradiction_count,
-- Voting patterns (influence indicators)
vbs.rebel_votes,
vbs.total_votes,
-- Policy area concentration (capture risk)
(SELECT COUNT(DISTINCT issue_category)
FROM document WHERE person_id = p.person_id) as policy_focus_diversity,
-- Red flags
CASE WHEN raa.credibility_score < 50 THEN 1 ELSE 0 END as credibility_flag,
CASE WHEN raa.contradiction_count > 20 THEN 1 ELSE 0 END as contradiction_flag
FROM person p
LEFT JOIN rhetoric_action_alignment raa ON p.person_id = raa.person_id
LEFT JOIN vote_ballot_summary vbs ON p.person_id = vbs.person_id
WHERE p.person_id = :politicianId
),
network_risk AS (
SELECT
p.person_id,
-- Network connections to high-risk entities
COUNT(DISTINCT CASE WHEN ne.entity_risk_level = 'HIGH'
THEN ne.entity_id END) as high_risk_connections,
COUNT(DISTINCT CASE WHEN ne.entity_type = 'SANCTIONED_ENTITY'
THEN ne.entity_id END) as sanctioned_connections,
COUNT(DISTINCT CASE WHEN ne.entity_type = 'CONVICTED_CRIMINAL'
THEN ne.entity_id END) as criminal_connections,
-- Red flags
CASE WHEN COUNT(DISTINCT CASE WHEN ne.entity_risk_level = 'HIGH'
THEN ne.entity_id END) > 0
THEN 1 ELSE 0 END as network_risk_flag
FROM person p
LEFT JOIN network_entity ne ON p.person_id = ne.person_id
WHERE p.person_id = :politicianId
GROUP BY p.person_id
),
transparency_risk AS (
SELECT
p.person_id,
-- Response to oversight
oa.response_rate,
oa.substantive_response_rate,
oa.avg_response_time,
-- Media transparency
COUNT(DISTINCT mi.interview_id) as media_engagement_count,
-- Red flags
CASE WHEN oa.response_rate < 70 THEN 1 ELSE 0 END as evasion_flag,
CASE WHEN oa.substantive_response_rate < 50 THEN 1 ELSE 0 END as opacity_flag
FROM person p
LEFT JOIN oversight_activity oa ON p.person_id = oa.person_id
LEFT JOIN media_interview mi ON p.person_id = mi.person_id
WHERE p.person_id = :politicianId
GROUP BY p.person_id, oa.response_rate, oa.substantive_response_rate,
oa.avg_response_time
)
SELECT
p.person_id,
p.first_name || ' ' || p.last_name as name,
p.party,
-- Financial risk indicators
fr.wealth_anomaly_flag,
fr.disclosure_failure_flag,
fr.coi_omission_flag,
fr.wealth_change_unexplained_ratio,
-- Behavioral risk indicators
br.credibility_flag,
br.contradiction_flag,
br.credibility_score,
-- Network risk indicators
nr.network_risk_flag,
nr.high_risk_connections,
nr.sanctioned_connections,
-- Transparency risk indicators
tr.evasion_flag,
tr.opacity_flag,
tr.response_rate,
-- Total red flags
(fr.wealth_anomaly_flag + fr.disclosure_failure_flag + fr.coi_omission_flag +
br.credibility_flag + br.contradiction_flag +
nr.network_risk_flag +
tr.evasion_flag + tr.opacity_flag) as total_red_flags,
-- Corruption risk score (0-100, higher = higher risk)
(
(fr.wealth_anomaly_flag + fr.disclosure_failure_flag + fr.coi_omission_flag) * 8 +
(br.credibility_flag + br.contradiction_flag) * 6 +
nr.network_risk_flag * 10 +
(tr.evasion_flag + tr.opacity_flag) * 6 +
(fr.wealth_change_unexplained_ratio * 20) +
((100 - br.credibility_score) / 100 * 15) +
(nr.high_risk_connections * 3) +
((100 - tr.response_rate) / 100 * 10)
) as corruption_risk_score,
-- Risk classification
CASE
WHEN (
(fr.wealth_anomaly_flag + fr.disclosure_failure_flag + fr.coi_omission_flag) * 8 +
(br.credibility_flag + br.contradiction_flag) * 6 +
nr.network_risk_flag * 10 +
(tr.evasion_flag + tr.opacity_flag) * 6 +
(fr.wealth_change_unexplained_ratio * 20) +
((100 - br.credibility_score) / 100 * 15) +
(nr.high_risk_connections * 3) +
((100 - tr.response_rate) / 100 * 10)
) >= 70 THEN 'CRITICAL_CORRUPTION_RISK'
WHEN (
(fr.wealth_anomaly_flag + fr.disclosure_failure_flag + fr.coi_omission_flag) * 8 +
(br.credibility_flag + br.contradiction_flag) * 6 +
nr.network_risk_flag * 10 +
(tr.evasion_flag + tr.opacity_flag) * 6 +
(fr.wealth_change_unexplained_ratio * 20) +
((100 - br.credibility_score) / 100 * 15) +
(nr.high_risk_connections * 3) +
((100 - tr.response_rate) / 100 * 10)
) >= 50 THEN 'HIGH_CORRUPTION_RISK'
WHEN (
(fr.wealth_anomaly_flag + fr.disclosure_failure_flag + fr.coi_omission_flag) * 8 +
(br.credibility_flag + br.contradiction_flag) * 6 +
nr.network_risk_flag * 10 +
(tr.evasion_flag + tr.opacity_flag) * 6 +
(fr.wealth_change_unexplained_ratio * 20) +
((100 - br.credibility_score) / 100 * 15) +
(nr.high_risk_connections * 3) +
((100 - tr.response_rate) / 100 * 10)
) >= 30 THEN 'MODERATE_CORRUPTION_RISK'
ELSE 'LOW_CORRUPTION_RISK'
END as risk_classification
FROM person p
LEFT JOIN financial_risk fr ON p.person_id = fr.person_id
LEFT JOIN behavioral_risk br ON p.person_id = br.person_id
LEFT JOIN network_risk nr ON p.person_id = nr.person_id
LEFT JOIN transparency_risk tr ON p.person_id = tr.person_id
WHERE p.person_id = :politicianId
""";
return jdbcTemplate.queryForObject(sql, CorruptionRiskProfile.class,
Map.of("politicianId", politicianId));
}
}
3. Institutional Erosion Metrics
Measuring Parliamentary Effectiveness Decline
Institutional health requires effective parliamentary procedures, accountability mechanisms, and checks on executive power.
-- Institutional Erosion Index
WITH institutional_metrics AS (
SELECT
-- Executive-Legislative Balance
(SELECT AVG(oversight_effectiveness_score)
FROM oversight_activity
WHERE created_date >= NOW() - INTERVAL '2 years'
) as oversight_effectiveness,
-- Legislative Productivity
(SELECT COUNT(*)
FROM document
WHERE document_type = 'adopted_law'
AND created_date >= NOW() - INTERVAL '2 years'
)::float /
(SELECT COUNT(*)
FROM document
WHERE document_type = 'adopted_law'
AND created_date >= NOW() - INTERVAL '4 years'
AND created_date < NOW() - INTERVAL '2 years'
) as legislative_productivity_trend,
-- Committee Functionality
(SELECT AVG(overall_effectiveness_score)
FROM committee_effectiveness
WHERE analysis_date >= NOW() - INTERVAL '2 years'
) as avg_committee_effectiveness,
-- Parliamentary Participation
(SELECT AVG(100 - avg_absent_percentage)
FROM vote_ballot_summary
WHERE analysis_date >= NOW() - INTERVAL '2 years'
) as avg_participation_rate,
-- Opposition Effectiveness
(SELECT AVG(oversight_effectiveness_score)
FROM oversight_activity oa
JOIN person p ON oa.person_id = p.person_id
WHERE p.party NOT IN (SELECT party FROM government_coalition)
AND oa.created_date >= NOW() - INTERVAL '2 years'
) as opposition_effectiveness,
-- Procedural Fairness
(SELECT AVG(debate_time_allocated::float / debate_time_requested)
FROM parliamentary_debate
WHERE debate_date >= NOW() - INTERVAL '2 years'
) as debate_time_fairness,
-- Cross-Party Collaboration
(SELECT AVG(collaboration_intensity)
FROM cross_party_collaboration
WHERE analysis_date >= NOW() - INTERVAL '2 years'
) as cross_party_collaboration
),
historical_comparison AS (
-- Compare current metrics to 5-year historical baseline
SELECT
'oversight_effectiveness' as metric,
im.oversight_effectiveness as current_value,
(SELECT AVG(oversight_effectiveness_score)
FROM oversight_activity
WHERE created_date >= NOW() - INTERVAL '7 years'
AND created_date < NOW() - INTERVAL '2 years'
) as historical_baseline,
im.oversight_effectiveness -
(SELECT AVG(oversight_effectiveness_score)
FROM oversight_activity
WHERE created_date >= NOW() - INTERVAL '7 years'
AND created_date < NOW() - INTERVAL '2 years'
) as change_from_baseline
FROM institutional_metrics im
UNION ALL
SELECT
'committee_effectiveness' as metric,
im.avg_committee_effectiveness as current_value,
(SELECT AVG(overall_effectiveness_score)
FROM committee_effectiveness
WHERE analysis_date >= NOW() - INTERVAL '7 years'
AND analysis_date < NOW() - INTERVAL '2 years'
) as historical_baseline,
im.avg_committee_effectiveness -
(SELECT AVG(overall_effectiveness_score)
FROM committee_effectiveness
WHERE analysis_date >= NOW() - INTERVAL '7 years'
AND analysis_date < NOW() - INTERVAL '2 years'
) as change_from_baseline
FROM institutional_metrics im
UNION ALL
SELECT
'participation_rate' as metric,
im.avg_participation_rate as current_value,
(SELECT AVG(100 - avg_absent_percentage)
FROM vote_ballot_summary
WHERE analysis_date >= NOW() - INTERVAL '7 years'
AND analysis_date < NOW() - INTERVAL '2 years'
) as historical_baseline,
im.avg_participation_rate -
(SELECT AVG(100 - avg_absent_percentage)
FROM vote_ballot_summary
WHERE analysis_date >= NOW() - INTERVAL '7 years'
AND analysis_date < NOW() - INTERVAL '2 years'
) as change_from_baseline
FROM institutional_metrics im
)
SELECT
im.*,
-- Institutional Erosion Index (0-100, higher = more erosion)
(
CASE WHEN im.oversight_effectiveness < 60 THEN (60 - im.oversight_effectiveness) / 60 * 20 ELSE 0 END +
CASE WHEN im.legislative_productivity_trend < 0.9 THEN (0.9 - im.legislative_productivity_trend) * 15 ELSE 0 END +
CASE WHEN im.avg_committee_effectiveness < 65 THEN (65 - im.avg_committee_effectiveness) / 65 * 20 ELSE 0 END +
CASE WHEN im.avg_participation_rate < 85 THEN (85 - im.avg_participation_rate) / 85 * 15 ELSE 0 END +
CASE WHEN im.opposition_effectiveness < 55 THEN (55 - im.opposition_effectiveness) / 55 * 15 ELSE 0 END +
CASE WHEN im.debate_time_fairness < 0.70 THEN (0.70 - im.debate_time_fairness) / 0.70 * 10 ELSE 0 END +
CASE WHEN im.cross_party_collaboration < 2.0 THEN (2.0 - im.cross_party_collaboration) / 2.0 * 5 ELSE 0 END
) as institutional_erosion_index,
-- Erosion classification
CASE
WHEN (
CASE WHEN im.oversight_effectiveness < 60 THEN (60 - im.oversight_effectiveness) / 60 * 20 ELSE 0 END +
CASE WHEN im.legislative_productivity_trend < 0.9 THEN (0.9 - im.legislative_productivity_trend) * 15 ELSE 0 END +
CASE WHEN im.avg_committee_effectiveness < 65 THEN (65 - im.avg_committee_effectiveness) / 65 * 20 ELSE 0 END +
CASE WHEN im.avg_participation_rate < 85 THEN (85 - im.avg_participation_rate) / 85 * 15 ELSE 0 END +
CASE WHEN im.opposition_effectiveness < 55 THEN (55 - im.opposition_effectiveness) / 55 * 15 ELSE 0 END +
CASE WHEN im.debate_time_fairness < 0.70 THEN (0.70 - im.debate_time_fairness) / 0.70 * 10 ELSE 0 END +
CASE WHEN im.cross_party_collaboration < 2.0 THEN (2.0 - im.cross_party_collaboration) / 2.0 * 5 ELSE 0 END
) >= 50 THEN 'CRITICAL_EROSION'
WHEN (
CASE WHEN im.oversight_effectiveness < 60 THEN (60 - im.oversight_effectiveness) / 60 * 20 ELSE 0 END +
CASE WHEN im.legislative_productivity_trend < 0.9 THEN (0.9 - im.legislative_productivity_trend) * 15 ELSE 0 END +
CASE WHEN im.avg_committee_effectiveness < 65 THEN (65 - im.avg_committee_effectiveness) / 65 * 20 ELSE 0 END +
CASE WHEN im.avg_participation_rate < 85 THEN (85 - im.avg_participation_rate) / 85 * 15 ELSE 0 END +
CASE WHEN im.opposition_effectiveness < 55 THEN (55 - im.opposition_effectiveness) / 55 * 15 ELSE 0 END +
CASE WHEN im.debate_time_fairness < 0.70 THEN (0.70 - im.debate_time_fairness) / 0.70 * 10 ELSE 0 END +
CASE WHEN im.cross_party_collaboration < 2.0 THEN (2.0 - im.cross_party_collaboration) / 2.0 * 5 ELSE 0 END
) >= 30 THEN 'MODERATE_EROSION'
WHEN (
CASE WHEN im.oversight_effectiveness < 60 THEN (60 - im.oversight_effectiveness) / 60 * 20 ELSE 0 END +
CASE WHEN im.legislative_productivity_trend < 0.9 THEN (0.9 - im.legislative_productivity_trend) * 15 ELSE 0 END +
CASE WHEN im.avg_committee_effectiveness < 65 THEN (65 - im.avg_committee_effectiveness) / 65 * 20 ELSE 0 END +
CASE WHEN im.avg_participation_rate < 85 THEN (85 - im.avg_participation_rate) / 85 * 15 ELSE 0 END +
CASE WHEN im.opposition_effectiveness < 55 THEN (55 - im.opposition_effectiveness) / 55 * 15 ELSE 0 END +
CASE WHEN im.debate_time_fairness < 0.70 THEN (0.70 - im.debate_time_fairness) / 0.70 * 10 ELSE 0 END +
CASE WHEN im.cross_party_collaboration < 2.0 THEN (2.0 - im.cross_party_collaboration) / 2.0 * 5 ELSE 0 END
) >= 15 THEN 'MINOR_EROSION'
ELSE 'HEALTHY_INSTITUTION'
END as erosion_classification,
-- Historical trend assessment
(SELECT
CASE
WHEN COUNT(CASE WHEN change_from_baseline < -5 THEN 1 END) >= 2
THEN 'ACCELERATING_DECLINE'
WHEN COUNT(CASE WHEN change_from_baseline < 0 THEN 1 END) >= 2
THEN 'GRADUAL_DECLINE'
WHEN COUNT(CASE WHEN change_from_baseline > 5 THEN 1 END) >= 2
THEN 'IMPROVEMENT_TREND'
ELSE 'STABLE'
END
FROM historical_comparison
) as historical_trend
FROM institutional_metrics im;
4. Coalition Instability Prediction
Government Sustainability Forecasting
Coalition governments in parliamentary systems are vulnerable to collapse. The CIA platform predicts coalition stability.
from sklearn.ensemble import GradientBoostingClassifier
from typing import Dict, List
import pandas as pd
class CoalitionStabilityPredictor:
"""
Predicts coalition stability and government sustainability.
Features:
- Intra-party discipline (deviation rates)
- Inter-party alignment (voting agreement)
- Policy conflict indicators (deviation on key issues)
- Leadership approval ratings
- Economic conditions
- Scandal/crisis events
- Time in office (fatigue factor)
"""
def __init__(self):
self.model = GradientBoostingClassifier(n_estimators=200, max_depth=5)
self.trained = False
def predict_stability(
self,
coalition_parties: List[str],
prediction_horizon_months: int = 12
) -> Dict:
"""
Predicts coalition stability over specified time horizon.
Returns:
- Survival probability (0-1)
- Key risk factors
- Collapse scenarios
- Recommended monitoring priorities
"""
# Extract coalition features
query = """
WITH coalition_features AS (
SELECT
-- Party discipline
AVG(pd.avg_deviation_rate) as avg_intra_party_deviation,
MAX(pd.max_deviation_rate) as max_intra_party_deviation,
STDDEV(pd.avg_deviation_rate) as deviation_heterogeneity,
-- Cross-party alignment
AVG(cpa.alignment_rate) as avg_cross_party_alignment,
MIN(cpa.alignment_rate) as min_cross_party_alignment,
-- Policy conflict indicators
COUNT(DISTINCT CASE
WHEN pd.issue_category IN ('economic_policy', 'foreign_policy', 'justice')
AND pd.avg_deviation_rate > 15
THEN pd.issue_category
END) as critical_policy_conflicts,
-- Leadership factors
AVG(lp.approval_rating) as avg_leadership_approval,
MIN(lp.approval_rating) as min_leadership_approval,
-- Time factors
EXTRACT(MONTH FROM NOW() - MIN(gc.formation_date)) as months_in_office,
-- Crisis events
COUNT(DISTINCT ce.crisis_id) as recent_crises,
-- Scandal exposure
COUNT(DISTINCT se.scandal_id) as recent_scandals
FROM party_deviation pd
JOIN cross_party_alignment cpa ON pd.party IN (cpa.party_a, cpa.party_b)
JOIN leadership_profile lp ON pd.party = lp.party
JOIN government_coalition gc ON pd.party = gc.party
LEFT JOIN crisis_event ce ON ce.event_date >= NOW() - INTERVAL '6 months'
LEFT JOIN scandal_event se ON se.event_date >= NOW() - INTERVAL '6 months'
AND se.party IN (SELECT unnest(%s))
WHERE pd.party = ANY(%s)
AND pd.analysis_date >= NOW() - INTERVAL '6 months'
GROUP BY 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11
)
SELECT * FROM coalition_features
"""
features = pd.read_sql(
query,
self.connection,
params=[coalition_parties, coalition_parties]
).iloc[0]
if not self.trained:
self.train() # Train model if not already trained
# Prepare feature vector
X = self._prepare_features(features)
# Predict survival probability
survival_probability = self.model.predict_proba(X)[0][1]
# Identify risk factors
risk_factors = self._identify_risk_factors(features)
# Generate collapse scenarios
scenarios = self._generate_scenarios(features, survival_probability)
return {
'coalition_parties': coalition_parties,
'prediction_horizon_months': prediction_horizon_months,
'survival_probability': round(survival_probability, 3),
'stability_classification': self._classify_stability(survival_probability),
'confidence': 'HIGH' if abs(survival_probability - 0.5) > 0.3 else 'MODERATE',
'key_risk_factors': risk_factors,
'collapse_scenarios': scenarios,
'monitoring_priorities': self._recommend_monitoring(features)
}
def _identify_risk_factors(self, features: pd.Series) -> List[Dict]:
"""Identify and rank risk factors threatening coalition stability."""
risks = []
if features['avg_intra_party_deviation'] > 10:
risks.append({
'factor': 'High Intra-Party Deviation',
'severity': 'HIGH',
'value': round(features['avg_intra_party_deviation'], 2),
'impact': 'Party discipline breakdown threatens coalition cohesion'
})
if features['min_cross_party_alignment'] < 70:
risks.append({
'factor': 'Low Cross-Party Alignment',
'severity': 'CRITICAL',
'value': round(features['min_cross_party_alignment'], 2),
'impact': 'Coalition partners voting against each other'
})
if features['critical_policy_conflicts'] > 2:
risks.append({
'factor': 'Critical Policy Conflicts',
'severity': 'HIGH',
'value': int(features['critical_policy_conflicts']),
'impact': 'Fundamental disagreements on core policy areas'
})
if features['min_leadership_approval'] < 30:
risks.append({
'factor': 'Leadership Crisis',
'severity': 'CRITICAL',
'value': round(features['min_leadership_approval'], 2),
'impact': 'Public disapproval undermining government legitimacy'
})
if features['months_in_office'] > 36:
risks.append({
'factor': 'Coalition Fatigue',
'severit
…(truncated)