Geopolitical forecasting applies structured analytical techniques to predict political events, conflicts, policy changes, and their cascading effects on economies and societies. Drawing from intelligence analysis, political science, and superforecasting, this discipline combines qualitative judgment with quantitative methods to navigate the inherently uncertain world of international relations, elections, and governance.

## Key Points

1. ACH (Analysis of Competing Hypotheses) forces consideration of alternative explanations, reducing confirmation bias
2. Structural risk factors (regime type, economic level, recent conflict history) provide a quantitative baseline for conflict prediction
3. Election forecasting combines polls (informative close to election) with fundamentals (informative far out) using time-varying weights
4. Geopolitical scenarios should cover the full range from cooperative to conflictual, with assigned probabilities that sum to 1
5. Policy impacts cascade through domains; second-order effects often matter more than first-order and are harder to predict
6. The DIME framework ensures analysis covers all instruments of national power
7. Key Assumptions Check identifies which assumptions, if wrong, would invalidate the entire assessment
8. Anocracies (mixed regimes) face the highest conflict risk; stable autocracies and stable democracies are both more peaceful

Geopolitical Forecasting

Overview

Geopolitical forecasting applies structured analytical techniques to predict political events, conflicts, policy changes, and their cascading effects on economies and societies. Drawing from intelligence analysis, political science, and superforecasting, this discipline combines qualitative judgment with quantitative methods to navigate the inherently uncertain world of international relations, elections, and governance.

Frameworks for Geopolitical Analysis

DIME Framework (Diplomatic, Information, Military, Economic)

class DIMEAnalysis:
    """
    Analyze geopolitical situations through the DIME framework.
    Each instrument of national power affects outcomes differently.
    """

    def __init__(self, actor: str, situation: str):
        self.actor = actor
        self.situation = situation
        self.instruments = {
            'diplomatic': {
                'description': 'Alliances, treaties, negotiations, UN votes',
                'indicators': [],
                'current_posture': '',
                'constraints': []
            },
            'information': {
                'description': 'Media narratives, propaganda, cyber operations, intelligence',
                'indicators': [],
                'current_posture': '',
                'constraints': []
            },
            'military': {
                'description': 'Force posture, deployments, exercises, capability',
                'indicators': [],
                'current_posture': '',
                'constraints': []
            },
            'economic': {
                'description': 'Sanctions, trade policy, aid, financial leverage',
                'indicators': [],
                'current_posture': '',
                'constraints': []
            }
        }

    def assess_instrument(self, instrument: str, posture: str,
                          indicators: list, constraints: list):
        """Assess one instrument of power."""
        self.instruments[instrument]['current_posture'] = posture
        self.instruments[instrument]['indicators'] = indicators
        self.instruments[instrument]['constraints'] = constraints

    def generate_options(self) -> list:
        """Generate possible courses of action across instruments."""
        options = []

        # Escalatory options
        for inst in self.instruments:
            options.append({
                'action': f'Escalate {inst}',
                'instrument': inst,
                'direction': 'escalatory',
                'feasibility': self._assess_feasibility(inst, 'escalate'),
                'risks': self._assess_risks(inst, 'escalate')
            })

        # De-escalatory options
        for inst in self.instruments:
            options.append({
                'action': f'De-escalate {inst}',
                'instrument': inst,
                'direction': 'de-escalatory',
                'feasibility': self._assess_feasibility(inst, 'deescalate'),
                'risks': self._assess_risks(inst, 'deescalate')
            })

        return options

    def _assess_feasibility(self, instrument: str, direction: str) -> float:
        constraints = len(self.instruments[instrument]['constraints'])
        base = 0.7 if direction == 'deescalate' else 0.5
        return max(0.1, base - constraints * 0.1)

    def _assess_risks(self, instrument: str, direction: str) -> list:
        if direction == 'escalate':
            return ['Counter-escalation', 'Alliance strain', 'Domestic backlash']
        return ['Perception of weakness', 'Precedent setting', 'Ally abandonment']

Analysis of Competing Hypotheses (ACH)

class ACH:
    """
    Analysis of Competing Hypotheses (Heuer, 1999).
    Structured technique to evaluate multiple hypotheses
    against available evidence, avoiding confirmation bias.
    """

    def __init__(self, question: str):
        self.question = question
        self.hypotheses = []
        self.evidence = []
        self.matrix = None  # hypotheses x evidence

    def add_hypothesis(self, description: str, prior: float = None):
        self.hypotheses.append({
            'id': len(self.hypotheses),
            'description': description,
            'prior': prior or 1.0 / (len(self.hypotheses) + 1)
        })

    def add_evidence(self, description: str, source: str,
                     reliability: float = 0.8, relevance: float = 0.8):
        self.evidence.append({
            'id': len(self.evidence),
            'description': description,
            'source': source,
            'reliability': reliability,
            'relevance': relevance
        })

    def assess_consistency(self, hypothesis_id: int, evidence_id: int,
                           rating: str):
        """
        Rate how consistent evidence is with a hypothesis.
        Ratings: 'very_consistent', 'consistent', 'neutral',
                 'inconsistent', 'very_inconsistent'
        """
        if self.matrix is None:
            n_h = len(self.hypotheses)
            n_e = len(self.evidence)
            self.matrix = [[None] * n_e for _ in range(n_h)]

        rating_scores = {
            'very_consistent': 2,
            'consistent': 1,
            'neutral': 0,
            'inconsistent': -1,
            'very_inconsistent': -2
        }
        self.matrix[hypothesis_id][evidence_id] = rating_scores.get(rating, 0)

    def evaluate(self) -> dict:
        """
        Evaluate hypotheses using the ACH methodology.
        Key insight: focus on DISCONFIRMING evidence, not confirming.
        """
        if self.matrix is None:
            return {'error': 'No assessments made'}

        results = []
        for h_id, hypothesis in enumerate(self.hypotheses):
            weighted_score = 0
            inconsistency_count = 0
            total_weight = 0

            for e_id, evidence in enumerate(self.evidence):
                if self.matrix[h_id][e_id] is not None:
                    weight = evidence['reliability'] * evidence['relevance']
                    score = self.matrix[h_id][e_id]
                    weighted_score += score * weight
                    total_weight += weight

                    if score < 0:
                        inconsistency_count += 1

            normalized_score = weighted_score / total_weight if total_weight > 0 else 0

            results.append({
                'hypothesis': hypothesis['description'],
                'weighted_score': normalized_score,
                'inconsistencies': inconsistency_count,
                'prior': hypothesis['prior']
            })

        # Rank by fewest inconsistencies (ACH focuses on disconfirmation)
        results.sort(key=lambda x: x['inconsistencies'])

        # Normalize to probabilities
        total_score = sum(max(0.01, r['weighted_score'] + 3) for r in results)
        for r in results:
            r['probability'] = max(0.01, r['weighted_score'] + 3) / total_score

        return {
            'rankings': results,
            'most_likely': results[0],
            'diagnosticity': self._compute_diagnosticity()
        }

    def _compute_diagnosticity(self) -> list:
        """
        Identify most diagnostic evidence.
        Diagnostic evidence strongly favors one hypothesis over others.
        """
        diagnostic_scores = []

        for e_id, evidence in enumerate(self.evidence):
            ratings = [
                self.matrix[h_id][e_id]
                for h_id in range(len(self.hypotheses))
                if self.matrix[h_id][e_id] is not None
            ]

            if len(ratings) < 2:
                continue

            # High variance = high diagnosticity
            variance = np.var(ratings)
            diagnostic_scores.append({
                'evidence': evidence['description'],
                'diagnosticity': variance,
                'ratings_spread': max(ratings) - min(ratings)
            })

        diagnostic_scores.sort(key=lambda x: -x['diagnosticity'])
        return diagnostic_scores

Conflict Prediction

Quantitative Conflict Risk Assessment

class ConflictPredictor:
    """
    Predict conflict probability using structural indicators.
    Based on research from Uppsala Conflict Data Program and
    Political Instability Task Force.
    """

    def __init__(self):
        self.risk_factors = {}

    def assess_structural_risk(self, country_data: dict) -> dict:
        """
        Evaluate structural risk factors for conflict onset.
        Each factor contributes to overall conflict probability.
        """
        factors = {
            'regime_type': self._regime_risk(country_data.get('polity_score', 0)),
            'gdp_per_capita': self._economic_risk(country_data.get('gdp_pc', 5000)),
            'ethnic_fractionalization': self._ethnic_risk(country_data.get('ethnic_frac', 0.3)),
            'youth_bulge': self._youth_risk(country_data.get('youth_fraction', 0.2)),
            'recent_conflict': self._history_risk(country_data.get('years_since_conflict', 20)),
            'neighborhood_instability': self._neighborhood_risk(country_data.get('neighbor_conflicts', 0)),
            'state_capacity': self._capacity_risk(country_data.get('state_capacity', 0.5)),
            'natural_resources': self._resource_risk(country_data.get('resource_rents_pct', 5))
        }

        # Weighted combination
        weights = {
            'regime_type': 0.20,
            'gdp_per_capita': 0.15,
            'ethnic_fractionalization': 0.10,
            'youth_bulge': 0.10,
            'recent_conflict': 0.20,
            'neighborhood_instability': 0.10,
            'state_capacity': 0.10,
            'natural_resources': 0.05
        }

        overall_risk = sum(
            factors[f] * weights[f] for f in factors
        )

        return {
            'overall_risk': overall_risk,
            'risk_level': self._categorize(overall_risk),
            'factor_scores': factors,
            'top_risks': sorted(
                factors.items(), key=lambda x: -x[1]
            )[:3],
            'conflict_probability_5yr': self._risk_to_probability(overall_risk)
        }

    def _regime_risk(self, polity: int) -> float:
        """Anocracies (mixed regimes) have highest conflict risk."""
        # Polity ranges from -10 (autocracy) to 10 (democracy)
        # Anocracies (-5 to 5) are most at risk
        if -5 <= polity <= 5:
            return 0.8 - abs(polity) * 0.06
        return 0.3

    def _economic_risk(self, gdp_pc: float) -> float:
        """Lower GDP per capita = higher risk."""
        if gdp_pc < 1000:
            return 0.9
        elif gdp_pc < 3000:
            return 0.6
        elif gdp_pc < 10000:
            return 0.3
        return 0.1

    def _ethnic_risk(self, frac: float) -> float:
        """Ethnic fractionalization increases conflict risk, especially with exclusion."""
        return min(frac * 1.2, 1.0)

    def _youth_risk(self, fraction: float) -> float:
        """Large youth cohorts (youth bulge) correlate with instability."""
        if fraction > 0.35:
            return 0.8
        elif fraction > 0.25:
            return 0.5
        return 0.2

    def _history_risk(self, years: int) -> float:
        """Recent conflict strongly predicts future conflict (conflict trap)."""
        if years < 5:
            return 0.9
        elif years < 10:
            return 0.6
        elif years < 20:
            return 0.3
        return 0.1

    def _neighborhood_risk(self, n_conflicts: int) -> float:
        """Conflicts in neighboring countries increase risk (contagion)."""
        return min(n_conflicts * 0.2, 0.9)

    def _capacity_risk(self, capacity: float) -> float:
        """Weak states are more vulnerable to conflict."""
        return max(0, 1 - capacity)

    def _resource_risk(self, rents_pct: float) -> float:
        """Resource-dependent economies face 'resource curse' risks."""
        if rents_pct > 20:
            return 0.7
        elif rents_pct > 10:
            return 0.4
        return 0.2

    def _categorize(self, risk: float) -> str:
        if risk > 0.7: return 'critical'
        if risk > 0.5: return 'high'
        if risk > 0.3: return 'moderate'
        if risk > 0.15: return 'low'
        return 'minimal'

    def _risk_to_probability(self, risk: float) -> float:
        """Convert risk score to 5-year conflict onset probability."""
        # Calibrated against historical data
        return min(risk * 0.4, 0.95)

Election Forecasting

Multi-Factor Election Model

class ElectionForecaster:
    """
    Combine multiple signals for election forecasting:
    polls, fundamentals, expert judgment.
    """

    def __init__(self, election_name: str):
        self.election = election_name
        self.polls = []
        self.fundamentals = {}
        self.expert_forecasts = []

    def add_poll(self, pollster: str, date: str, candidate_a: float,
                 sample_size: int, methodology: str, pollster_rating: float = 0.5):
        self.polls.append({
            'pollster': pollster,
            'date': date,
            'candidate_a': candidate_a,
            'sample_size': sample_size,
            'methodology': methodology,
            'rating': pollster_rating,
            'margin_of_error': 1.96 * np.sqrt(candidate_a * (1-candidate_a) / sample_size)
        })

    def set_fundamentals(self, gdp_growth: float, incumbent_approval: float,
                         incumbent_running: bool, polarization: float,
                         time_in_power_years: int):
        """Economic and structural fundamentals."""
        self.fundamentals = {
            'gdp_growth': gdp_growth,
            'incumbent_approval': incumbent_approval,
            'incumbent_running': incumbent_running,
            'polarization': polarization,
            'time_in_power': time_in_power_years
        }

    def poll_average(self, recent_n: int = 10, decay_days: float = 14) -> dict:
        """Compute weighted poll average."""
        if not self.polls:
            return {'error': 'No polls'}

        sorted_polls = sorted(self.polls, key=lambda p: p['date'], reverse=True)
        recent = sorted_polls[:recent_n]

        total_weight = 0
        weighted_sum = 0

        for poll in recent:
            # Weight by sample size, recency, and pollster rating
            size_weight = np.sqrt(poll['sample_size'] / 1000)
            rating_weight = poll['rating']

            weight = size_weight * rating_weight
            weighted_sum += poll['candidate_a'] * weight
            total_weight += weight

        avg = weighted_sum / total_weight if total_weight > 0 else 0.5

        return {
            'weighted_average': avg,
            'n_polls': len(recent),
            'spread': max(p['candidate_a'] for p in recent) - min(p['candidate_a'] for p in recent),
            'average_moe': np.mean([p['margin_of_error'] for p in recent])
        }

    def fundamentals_prediction(self) -> dict:
        """Predict based on economic/structural fundamentals alone."""
        if not self.fundamentals:
            return {'error': 'No fundamentals set'}

        f = self.fundamentals

        # Simplified "Time for Change" model (Abramowitz-style)
        # Incumbent advantage with good economy, penalized by fatigue
        incumbent_advantage = 0.5

        # Economy effect
        if f['gdp_growth'] > 3:
            incumbent_advantage += 0.05
        elif f['gdp_growth'] > 1:
            incumbent_advantage += 0.02
        elif f['gdp_growth'] < 0:
            incumbent_advantage -= 0.05

        # Approval effect
        incumbent_advantage += (f['incumbent_approval'] - 0.45) * 0.3

        # Fatigue effect
        if f['time_in_power'] > 8:
            incumbent_advantage -= 0.03 * (f['time_in_power'] - 8) / 4

        # Running incumbent bonus
        if f['incumbent_running']:
            incumbent_advantage += 0.03

        return {
            'incumbent_vote_share': np.clip(incumbent_advantage, 0.3, 0.7),
            'model': 'fundamentals_only'
        }

    def combined_forecast(self, days_until_election: int) -> dict:
        """
        Combine polls and fundamentals.
        Weight shifts from fundamentals to polls as election approaches.
        """
        polls = self.poll_average()
        fundamentals = self.fundamentals_prediction()

        if 'error' in polls or 'error' in fundamentals:
            return polls if 'error' not in polls else fundamentals

        # Weight shifts: far out, fundamentals matter more; close in, polls dominate
        if days_until_election > 180:
            poll_weight = 0.3
        elif days_until_election > 60:
            poll_weight = 0.6
        elif days_until_election > 14:
            poll_weight = 0.8
        else:
            poll_weight = 0.9

        combined = (poll_weight * polls['weighted_average'] +
                   (1 - poll_weight) * fundamentals['incumbent_vote_share'])

        # Convert to win probability using polls uncertainty
        moe = polls.get('average_moe', 0.03)
        # Rough conversion: lead / standard error -> win probability
        from scipy.stats import norm
        lead = combined - 0.5
        win_prob = norm.cdf(lead / max(moe, 0.01))

        return {
            'vote_share_estimate': combined,
            'win_probability': win_prob,
            'poll_weight': poll_weight,
            'fundamentals_weight': 1 - poll_weight,
            'uncertainty': moe,
            'days_until_election': days_until_election
        }

Policy Impact Modeling

class PolicyImpactModel:
    """Model the cascading effects of policy changes."""

    def __init__(self):
        self.impact_chains = []

    def add_policy(self, policy: str, first_order_effects: list):
        """Define a policy and its direct effects."""
        chain = {
            'policy': policy,
            'effects': []
        }

        for effect in first_order_effects:
            chain['effects'].append({
                'description': effect['description'],
                'domain': effect['domain'],
                'magnitude': effect['magnitude'],  # -1 to 1
                'confidence': effect['confidence'],  # 0 to 1
                'time_lag_months': effect.get('lag', 6),
                'second_order': effect.get('second_order', [])
            })

        self.impact_chains.append(chain)

    def cascade_analysis(self, policy_index: int = 0) -> dict:
        """Trace the cascade of effects from a policy change."""
        chain = self.impact_chains[policy_index]
        all_effects = []

        def trace(effects, depth=0, parent=None):
            for effect in effects:
                effect_entry = {
                    'description': effect['description'],
                    'domain': effect['domain'],
                    'magnitude': effect['magnitude'],
                    'confidence': effect['confidence'],
                    'depth': depth,
                    'parent': parent,
                    'cumulative_confidence': effect['confidence'] * (0.8 ** depth)
                }
                all_effects.append(effect_entry)
                if 'second_order' in effect:
                    trace(effect['second_order'], depth + 1, effect['description'])

        trace(chain['effects'])

        return {
            'policy': chain['policy'],
            'total_effects': len(all_effects),
            'effects_by_depth': {
                d: [e for e in all_effects if e['depth'] == d]
                for d in set(e['depth'] for e in all_effects)
            },
            'effects_by_domain': self._group_by_domain(all_effects),
            'net_impact': np.mean([e['magnitude'] * e['cumulative_confidence'] for e in all_effects])
        }

    def _group_by_domain(self, effects: list) -> dict:
        domains = {}
        for e in effects:
            d = e['domain']
            if d not in domains:
                domains[d] = []
            domains[d].append(e)
        return {d: {'count': len(effects), 'avg_magnitude': np.mean([e['magnitude'] for e in effects])}
                for d, effects in domains.items()}

Geopolitical Scenario Construction

def build_geopolitical_scenarios(focal_region: str, time_horizon: int = 5) -> list:
    """
    Construct geopolitical scenarios using structured methodology.
    """
    scenarios = [
        {
            'name': 'Cooperative Stability',
            'description': 'Multilateral institutions strengthen, great power cooperation increases',
            'drivers': ['Economic interdependence', 'Shared threats', 'Diplomatic engagement'],
            'indicators': ['New trade agreements', 'Joint military exercises', 'UN resolution consensus'],
            'probability': 0.20
        },
        {
            'name': 'Competitive Coexistence',
            'description': 'Great power competition continues but within managed boundaries',
            'drivers': ['Strategic competition', 'Deterrence', 'Economic decoupling'],
            'indicators': ['Arms buildup', 'Technology restrictions', 'Alliance strengthening'],
            'probability': 0.40
        },
        {
            'name': 'Regional Fragmentation',
            'description': 'Global order fractures into competing blocs',
            'drivers': ['Nationalism', 'Technology divergence', 'Supply chain bifurcation'],
            'indicators': ['Parallel institutions', 'Currency blocs', 'Internet fragmentation'],
            'probability': 0.25
        },
        {
            'name': 'Crisis Escalation',
            'description': 'A crisis spirals beyond containment, reshaping the order',
            'drivers': ['Miscalculation', 'Domestic pressure', 'Alliance entanglement'],
            'indicators': ['Military incidents', 'Diplomatic expulsions', 'Economic warfare'],
            'probability': 0.15
        }
    ]

    return scenarios

Intelligence Analysis Structured Techniques

Key Assumptions Check

class KeyAssumptionsCheck:
    """Identify and test the assumptions underlying an assessment."""

    def __init__(self, assessment: str):
        self.assessment = assessment
        self.assumptions = []

    def add_assumption(self, assumption: str, category: str,
                       confidence: float, evidence: str,
                       impact_if_wrong: str):
        self.assumptions.append({
            'assumption': assumption,
            'category': category,  # 'factual', 'analytical', 'bridging'
            'confidence': confidence,
            'supporting_evidence': evidence,
            'impact_if_wrong': impact_if_wrong,
            'vulnerability': (1 - confidence) * self._impact_score(impact_if_wrong)
        })

    def _impact_score(self, impact: str) -> float:
        if 'fundamental' in impact.lower() or 'invalidate' in impact.lower():
            return 1.0
        if 'significant' in impact.lower():
            return 0.7
        return 0.4

    def identify_linchpin_assumptions(self) -> list:
        """
        Linchpin assumptions: if wrong, the entire assessment fails.
        These are the highest-priority items to monitor.
        """
        sorted_assumptions = sorted(self.assumptions, key=lambda x: -x['vulnerability'])
        return [a for a in sorted_assumptions if a['vulnerability'] > 0.3]

    def vulnerability_report(self) -> dict:
        return {
            'assessment': self.assessment,
            'n_assumptions': len(self.assumptions),
            'linchpin_assumptions': self.identify_linchpin_assumptions(),
            'overall_confidence': np.mean([a['confidence'] for a in self.assumptions]),
            'weakest_link': min(self.assumptions, key=lambda x: x['confidence']) if self.assumptions else None
        }

Key Takeaways

1. ACH (Analysis of Competing Hypotheses) forces consideration of alternative explanations, reducing confirmation bias
2. Structural risk factors (regime type, economic level, recent conflict history) provide a quantitative baseline for conflict prediction
3. Election forecasting combines polls (informative close to election) with fundamentals (informative far out) using time-varying weights
4. Geopolitical scenarios should cover the full range from cooperative to conflictual, with assigned probabilities that sum to 1
5. Policy impacts cascade through domains; second-order effects often matter more than first-order and are harder to predict
6. The DIME framework ensures analysis covers all instruments of national power
7. Key Assumptions Check identifies which assumptions, if wrong, would invalidate the entire assessment
8. Anocracies (mixed regimes) face the highest conflict risk; stable autocracies and stable democracies are both more peaceful