geobot/simulation/hawkes.py

"""
Hawkes Process Wrapper for GeoBotv1

Provides convenient access to Hawkes processes for conflict dynamics
and event contagion modeling. This module re-exports functionality from
geobot.timeseries.point_processes for easier discoverability.

Hawkes processes are self-exciting point processes that model how
events cluster in time and how they can trigger subsequent events,
making them ideal for modeling:
- Conflict escalation dynamics
- Contagion between countries/regions
- Cascading geopolitical events
"""

# Import from timeseries.point_processes
from ..timeseries.point_processes import (
    UnivariateHawkesProcess,
    MultivariateHawkesProcess,
    ConflictContagionModel,
    HawkesParameters,
    HawkesFitResult,
    estimate_branching_ratio,
    detect_explosive_regime
)

__all__ = [
    "UnivariateHawkesProcess",
    "MultivariateHawkesProcess",
    "ConflictContagionModel",
    "HawkesParameters",
    "HawkesFitResult",
    "estimate_branching_ratio",
    "detect_explosive_regime",
    "HawkesSimulator",  # Additional convenience class below
]


class HawkesSimulator:
    """
    High-level interface for Hawkes process simulation.

    Provides simplified API for common Hawkes process use cases
    in geopolitical modeling.
    """

    def __init__(self, n_dimensions: int = 1):
        """
        Initialize Hawkes simulator.

        Parameters
        ----------
        n_dimensions : int
            Number of dimensions (countries, regions, etc.)
        """
        self.n_dimensions = n_dimensions

        if n_dimensions == 1:
            self.process = UnivariateHawkesProcess()
        else:
            self.process = MultivariateHawkesProcess(n_dimensions=n_dimensions)

    def fit(self, events, T: float = None, **kwargs):
        """
        Fit Hawkes process to event data.

        Parameters
        ----------
        events : Union[np.ndarray, Dict[str, np.ndarray], List[np.ndarray]]
            Event times
        T : float
            Observation window
        **kwargs
            Additional arguments passed to fit method

        Returns
        -------
        HawkesFitResult
            Fit results
        """
        return self.process.fit(events, T=T, **kwargs)

    def simulate(
        self,
        T: float,
        params: HawkesParameters,
        random_state: int = None,
        **kwargs
    ):
        """
        Simulate Hawkes process.

        Parameters
        ----------
        T : float
            Simulation time horizon
        params : HawkesParameters
            Process parameters
        random_state : int
            Random seed
        **kwargs
            Additional arguments

        Returns
        -------
        Union[np.ndarray, List[np.ndarray]]
            Simulated event times
        """
        return self.process.simulate(T, params, random_state=random_state, **kwargs)

    def predict_intensity(self, events, t: float, params: HawkesParameters):
        """
        Predict intensity at time t given past events.

        Parameters
        ----------
        events : Union[np.ndarray, List[np.ndarray]]
            Past event times
        t : float
            Time at which to predict
        params : HawkesParameters
            Process parameters

        Returns
        -------
        Union[float, np.ndarray]
            Predicted intensity
        """
        return self.process.intensity(events, t, params)

    def assess_stability(self, params: HawkesParameters) -> dict:
        """
        Assess stability of Hawkes process.

        Parameters
        ----------
        params : HawkesParameters
            Process parameters

        Returns
        -------
        dict
            Stability assessment with branching ratio and regime
        """
        branching_ratio = estimate_branching_ratio(params)
        is_explosive = detect_explosive_regime(params)

        return {
            'branching_ratio': branching_ratio,
            'is_explosive': is_explosive,
            'is_stable': branching_ratio < 1.0,
            'regime': 'supercritical (explosive)' if is_explosive else 'subcritical (stable)',
            'interpretation': (
                'Process is stable - events will not cascade indefinitely'
                if branching_ratio < 1.0
                else 'Process is explosive - events can trigger cascading escalation'
            )
        }


# ============================================================================
# Convenience Functions
# ============================================================================

def quick_conflict_contagion_analysis(
    events_by_country: dict,
    T: float,
    country_names: list = None
) -> dict:
    """
    Quick analysis of conflict contagion between countries.

    Parameters
    ----------
    events_by_country : dict
        Dictionary mapping country names to event times
    T : float
        Observation window
    country_names : list
        List of country names (if None, use dict keys)

    Returns
    -------
    dict
        Contagion analysis results
    """
    if country_names is None:
        country_names = list(events_by_country.keys())

    model = ConflictContagionModel(countries=country_names)
    result = model.fit(events_by_country, T=T)

    return {
        'fit_result': result,
        'most_contagious': result['most_contagious_source'],
        'most_vulnerable': result['most_vulnerable_target'],
        'total_events': {
            country: len(events_by_country[country])
            for country in country_names
        },
        'contagion_matrix': result['alpha_matrix'],
        'baseline_rates': result['mu'],
        'branching_ratio': estimate_branching_ratio(result['parameters']),
    }


def simulate_conflict_scenario(
    n_countries: int,
    baseline_rates: list,
    contagion_strength: float,
    T: float,
    random_state: int = None
) -> dict:
    """
    Simulate conflict scenario with specified parameters.

    Parameters
    ----------
    n_countries : int
        Number of countries
    baseline_rates : list
        Baseline conflict rates for each country
    contagion_strength : float
        Strength of cross-country contagion
    T : float
        Simulation horizon
    random_state : int
        Random seed

    Returns
    -------
    dict
        Simulation results
    """
    import numpy as np

    # Create contagion matrix
    alpha_matrix = np.full((n_countries, n_countries), contagion_strength)
    np.fill_diagonal(alpha_matrix, contagion_strength * 1.5)  # Self-excitation stronger

    # Create parameters
    params = HawkesParameters(
        mu=np.array(baseline_rates),
        alpha=alpha_matrix,
        beta=np.ones((n_countries, n_countries)) * 2.0  # Decay rate
    )

    # Simulate
    simulator = HawkesSimulator(n_dimensions=n_countries)
    events = simulator.simulate(T, params, random_state=random_state)

    # Assess stability
    stability = simulator.assess_stability(params)

    return {
        'events': events,
        'parameters': params,
        'stability': stability,
        'event_counts': [len(e) for e in events],
        'total_events': sum(len(e) for e in events),
        'average_rate': sum(len(e) for e in events) / T / n_countries,
    }