---
name: risk-assessment-frameworks
description: Political risk indicators, institutional risk, corruption risk, democratic backsliding, early warning systems for Swedish political intelligence
license: Apache-2.0
---
# 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:
```mermaid
graph TB
subgraph "Layer 1: Data Collection"
A1["🗳️ Behavioral Data
3.5M+ votes, attendance
Productivity metrics"]
A2["💰 Financial Data
World Bank, ESV
Economic indicators"]
A3["📊 Democracy Indices
V-Dem, Freedom House
International benchmarks"]
A4["📰 Media Coverage
Sentiment analysis
Scandal tracking"]
end
subgraph "Layer 2: Risk Rules Engine (Drools)"
A1 --> B1[Behavioral Risk Rules
24 politician rules
12 party rules]
A2 --> B2[Financial Risk Rules
8 corruption indicators]
A3 --> B3[Democratic Health Rules
6 institutional rules]
A4 --> B4[Reputational Risk Rules
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
```python
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.
```java
@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.
```sql
-- 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.
```python
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',
'severity': 'MODERATE',
'value': int(features['months_in_office']),
'impact': 'Long tenure increases internal tensions and public fatigue'
})
if features['recent_scandals'] > 2:
risks.append({
'factor': 'Scandal Exposure',
'severity': 'HIGH',
'value': int(features['recent_scandals']),
'impact': 'Multiple scandals eroding public trust and coalition unity'
})
return sorted(risks, key=lambda x:
{'CRITICAL': 3, 'HIGH': 2, 'MODERATE': 1}.get(x['severity'], 0),
reverse=True)
def _generate_scenarios(
self,
features: pd.Series,
base_probability: float
) -> List[Dict]:
"""Generate potential collapse scenarios with probabilities."""
scenarios = []
# Scenario 1: Policy Conflict Rupture
if features['critical_policy_conflicts'] > 1:
scenarios.append({
'scenario': 'Policy Conflict Rupture',
'trigger': 'Irreconcilable disagreement on major legislation',
'probability': round(
base_probability * (1 + features['critical_policy_conflicts'] * 0.1),
3
),
'timeline': '3-6 months',
'warning_signs': [
'Increased voting deviations on key issues',
'Public disagreements between coalition leaders',
'Failure to pass priority legislation'
]
})
# Scenario 2: Leadership Crisis
if features['min_leadership_approval'] < 35:
scenarios.append({
'scenario': 'Leadership Crisis',
'trigger': 'Prime Minister or key party leader resignation',
'probability': round(
base_probability * (1 + (35 - features['min_leadership_approval']) / 100),
3
),
'timeline': '1-3 months',
'warning_signs': [
'Plummeting approval ratings',
'Calls for leadership change within party',
'Loss of confidence votes discussed'
]
})
# Scenario 3: Electoral Pressure
if features['months_in_office'] > 30:
scenarios.append({
'scenario': 'Pre-Election Defection',
'trigger': 'Party leaves coalition to improve electoral positioning',
'probability': round(
base_probability * (1 + features['months_in_office'] / 100),
3
),
'timeline': '6-12 months',
'warning_signs': [
'Party distancing from coalition decisions',
'Increased rebel voting to differentiate',
'Campaign-style criticism of coalition partners'
]
})
# Scenario 4: Scandal Cascade
if features['recent_scandals'] > 1:
scenarios.append({
'scenario': 'Scandal Cascade Collapse',
'trigger': 'Multiple scandals forcing coalition crisis',
'probability': round(
base_probability * (1 + features['recent_scandals'] * 0.15),
3
),
'timeline': '1-2 months',
'warning_signs': [
'Media feeding frenzy',
'Opposition calls for no-confidence vote',
'Coalition partners demanding action/resignations'
]
})
return sorted(scenarios, key=lambda x: x['probability'], reverse=True)
def _classify_stability(self, probability: float) -> str:
"""Classify coalition stability."""
if probability >= 0.80:
return "HIGHLY_STABLE"
elif probability >= 0.65:
return "MODERATELY_STABLE"
elif probability >= 0.45:
return "UNSTABLE"
else:
return "CRITICAL_INSTABILITY"
def _recommend_monitoring(self, features: pd.Series) -> List[str]:
"""Recommend monitoring priorities."""
priorities = []
if features['min_cross_party_alignment'] < 75:
priorities.append("PRIORITY 1: Daily monitoring of cross-party voting alignment")
if features['min_leadership_approval'] < 40:
priorities.append("PRIORITY 1: Weekly leadership approval tracking")
if features['critical_policy_conflicts'] > 0:
priorities.append("PRIORITY 2: Monitor voting on critical policy areas")
if features['recent_scandals'] > 0:
priorities.append("PRIORITY 2: Media sentiment analysis for scandal escalation")
if features['months_in_office'] > 30:
priorities.append("PRIORITY 3: Electoral positioning indicators")
return priorities if priorities else [
"STANDARD: Routine coalition monitoring (monthly deviation analysis)"
]
```
## 5. Political Violence Risk Indicators
### Early Warning System for Political Violence
Political violence threatens democratic stability. The CIA platform monitors behavioral and contextual indicators.
```sql
-- Political Violence Risk Assessment
WITH violence_risk_indicators AS (
SELECT
-- Rhetorical escalation
COUNT(CASE WHEN dc.contains_violent_rhetoric = TRUE THEN 1 END) as violent_rhetoric_count,
COUNT(CASE WHEN dc.contains_dehumanizing_language = TRUE THEN 1 END) as dehumanization_count,
COUNT(CASE WHEN dc.contains_threat = TRUE THEN 1 END) as threat_count,
-- Polarization indicators
AVG(pp.polarization_index) as avg_polarization,
MAX(pp.polarization_index) as max_polarization,
-- Protest activity
COUNT(DISTINCT pe.protest_event_id) as protest_count,
AVG(pe.violence_level) as avg_protest_violence,
COUNT(CASE WHEN pe.violence_level >= 3 THEN 1 END) as violent_protests,
-- Hate crime correlation
(SELECT COUNT(*) FROM hate_crime_incident
WHERE incident_date >= NOW() - INTERVAL '6 months'
AND political_motivation = TRUE
) as political_hate_crimes,
-- Online extremism
COUNT(DISTINCT oec.extremist_content_id) as extremist_content_items,
-- Media incitement
COUNT(CASE WHEN ma.incitement_score > 0.7 THEN 1 END) as high_incitement_articles
FROM document_content dc
JOIN party_polarization pp ON 1=1
LEFT JOIN protest_event pe ON pe.event_date >= NOW() - INTERVAL '6 months'
LEFT JOIN online_extremist_content oec ON oec.detected_date >= NOW() - INTERVAL '6 months'
LEFT JOIN media_article ma ON ma.published_date >= NOW() - INTERVAL '6 months'
WHERE dc.created_date >= NOW() - INTERVAL '6 months'
)
SELECT
vri.*,
-- Violence Risk Score (0-100, higher = higher risk)
(
LEAST(vri.violent_rhetoric_count / 10.0, 1.0) * 15 +
LEAST(vri.dehumanization_count / 15.0, 1.0) * 15 +
LEAST(vri.threat_count / 5.0, 1.0) * 20 +
vri.avg_polarization * 15 +
LEAST(vri.violent_protests / 5.0, 1.0) * 15 +
LEAST(vri.political_hate_crimes / 20.0, 1.0) * 10 +
LEAST(vri.extremist_content_items / 100.0, 1.0) * 10
) * 100 as violence_risk_score,
-- Risk Classification
CASE
WHEN (
LEAST(vri.violent_rhetoric_count / 10.0, 1.0) * 15 +
LEAST(vri.dehumanization_count / 15.0, 1.0) * 15 +
LEAST(vri.threat_count / 5.0, 1.0) * 20 +
vri.avg_polarization * 15 +
LEAST(vri.violent_protests / 5.0, 1.0) * 15 +
LEAST(vri.political_hate_crimes / 20.0, 1.0) * 10 +
LEAST(vri.extremist_content_items / 100.0, 1.0) * 10
) * 100 >= 70 THEN 'CRITICAL_VIOLENCE_RISK'
WHEN (
LEAST(vri.violent_rhetoric_count / 10.0, 1.0) * 15 +
LEAST(vri.dehumanization_count / 15.0, 1.0) * 15 +
LEAST(vri.threat_count / 5.0, 1.0) * 20 +
vri.avg_polarization * 15 +
LEAST(vri.violent_protests / 5.0, 1.0) * 15 +
LEAST(vri.political_hate_crimes / 20.0, 1.0) * 10 +
LEAST(vri.extremist_content_items / 100.0, 1.0) * 10
) * 100 >= 50 THEN 'ELEVATED_VIOLENCE_RISK'
WHEN (
LEAST(vri.violent_rhetoric_count / 10.0, 1.0) * 15 +
LEAST(vri.dehumanization_count / 15.0, 1.0) * 15 +
LEAST(vri.threat_count / 5.0, 1.0) * 20 +
vri.avg_polarization * 15 +
LEAST(vri.violent_protests / 5.0, 1.0) * 15 +
LEAST(vri.political_hate_crimes / 20.0, 1.0) * 10 +
LEAST(vri.extremist_content_items / 100.0, 1.0) * 10
) * 100 >= 30 THEN 'MODERATE_VIOLENCE_RISK'
ELSE 'LOW_VIOLENCE_RISK'
END as risk_classification,
-- Immediate action required?
CASE
WHEN vri.threat_count > 0 OR vri.violent_protests > 2
THEN TRUE
ELSE FALSE
END as immediate_action_required
FROM violence_risk_indicators vri;
```
## ISMS Compliance Mapping
### ISO 27001:2022 Controls
| Control | Risk Assessment Application |
|---------|---------------------------|
| **A.5.7 - Threat intelligence** | Systematic threat intelligence from risk frameworks |
| **A.5.10 - Acceptable use of information and other associated assets** | Ethical use of political risk intelligence |
| **A.8.16 - Monitoring activities** | Continuous risk monitoring and early warning |
### NIST Cybersecurity Framework 2.0
| Function | Risk Assessment Integration |
|----------|---------------------------|
| **IDENTIFY (ID.RA)** | Comprehensive political risk identification |
| **DETECT (DE.CM)** | Early warning detection systems |
| **RESPOND (RS.AN)** | Risk-based response prioritization |
### CIS Controls v8
| Control | Application |
|---------|-------------|
| **CIS Control 4 - Secure Configuration** | Secure risk assessment system configuration |
| **CIS Control 8 - Audit Log Management** | Risk assessment audit trail |
## Hack23 ISMS Policy References
This skill implements requirements from:
- **[Secure Development Policy](https://github.com/Hack23/ISMS-PUBLIC/blob/main/Secure_Development_Policy.md)** - Risk assessment methodology standards
- **[Information Security Policy](https://github.com/Hack23/ISMS-PUBLIC/blob/main/Information_Security_Policy.md)** - Ethical risk assessment practices
- **[Risk Assessment Methodology](https://github.com/Hack23/ISMS-PUBLIC/blob/main/Risk_Assessment_Methodology.md)** - Risk calculation methods
- **[Threat Modeling](https://github.com/Hack23/ISMS-PUBLIC/blob/main/Threat_Modeling.md)** - Threat-based risk assessment
## References
### Risk Assessment Literature
1. **Coppedge, M., et al. (2021)**. *V-Dem Codebook v11*. Varieties of Democracy (V-Dem) Project.
2. **Transparency International (2023)**. *Corruption Perceptions Index: Methodology*.
3. **Lührmann, A., & Lindberg, S. I. (2019)**. "A Third Wave of Autocratization is Here: What is New About It?" *Democratization*, 26(7), 1095-1113.
4. **Schedler, A. (2013)**. *The Politics of Uncertainty: Sustaining and Subverting Electoral Authoritarianism*. Oxford University Press.
### Database Intelligence Sources
- **[DATABASE_VIEW_INTELLIGENCE_CATALOG.md](../../DATABASE_VIEW_INTELLIGENCE_CATALOG.md)** - Complete view documentation
- **[RISK_RULES_INTOP_OSINT.md](../../RISK_RULES_INTOP_OSINT.md)** - 50+ risk rule specifications
- **[DATA_ANALYSIS_INTOP_OSINT.md](../../DATA_ANALYSIS_INTOP_OSINT.md)** - Analysis frameworks
- **[INTELLIGENCE_DATA_FLOW.md](../../INTELLIGENCE_DATA_FLOW.md)** - Risk data flow mapping