API Reference

This page contains the automatic documentation for the MASSIVE core logic, which is mainly located in simulator.py.

MASSIVE — Simulador híbrido de dinámica social Núcleo numérico + LLM como selector de régimen dinámico

Modelos integrados

REGLAS BASE (originales): 0: lineal — cambio proporcional suave 1: umbral — salto al cruzar punto crítico 2: memoria — inercia del estado pasado 3: backlash — propaganda refuerza posición contraria 4: polarizacion — aleja la opinión del neutro (cámara de eco)

MODELOS NUEVOS: 5: hk — Hegselmann-Krause: confianza acotada, formación natural de clusters 6: contagio_competitivo — dos narrativas compitiendo simultáneamente 7: umbral_heterogeneo — distribución de umbrales (Granovetter), cascadas sociales 8: homofilia — red co-evolutiva: los pesos de influencia cambian con la opinión

MECANISMOS TRANSVERSALES (se aplican a todas las reglas): · sesgo_confirmacion — propaganda contraria pierde peso según posición actual · homofilia dinámica — actualiza pesos de grupos en cada paso

RANGOS DE OPINIÓN

[0, 1] — Probabilístico. Neutro=0.5 [-1, 1] — Bipolar. Neutro=0.0. Rechazo activo ≠ indiferencia.

PROVEEDORES LLM

heurístico | ollama | groq | openai | openrouter

Autor: MASSIVE Research

apply_replicator_equation(population_states, payoff_matrix, dt)

Integrates one step of the replicator ODE using RK45.

Parameters:

Name	Type	Description	Default
population_states	ndarray	1-D array of strategy frequencies summing to 1.	required
payoff_matrix	ndarray	Square payoff matrix (n_strategies × n_strategies).	required
dt	float	Integration time span [0, dt].	required

Returns:

Type	Description
ndarray	Updated normalised frequency array after one step.

Source code in simulator.py

def apply_replicator_equation(
    population_states: np.ndarray,
    payoff_matrix: np.ndarray,
    dt: float,
) -> np.ndarray:
    """
    Integrates one step of the replicator ODE using RK45.

    Args:
        population_states: 1-D array of strategy frequencies summing to 1.
        payoff_matrix: Square payoff matrix (n_strategies × n_strategies).
        dt: Integration time span [0, dt].

    Returns:
        Updated normalised frequency array after one step.
    """
    pop = np.array(population_states, dtype=float)
    total = np.sum(pop)
    if total == 0.0:
        return pop
    pop = pop / total

    def replicator_rhs(t: float, x: np.ndarray) -> np.ndarray:
        x = np.clip(x, 0.0, 1.0)
        s = np.sum(x)
        if s > 0.0:
            x = x / s
        f = payoff_matrix @ x
        f_avg = x @ f
        return x * (f - f_avg)

    sol = solve_ivp(replicator_rhs, [0.0, dt], pop, method="RK45", dense_output=False)
    new_pop = sol.y[:, -1]
    new_pop = np.clip(new_pop, 0.0, 1.0)
    total_new = np.sum(new_pop)
    return new_pop / total_new if total_new > 0.0 else pop

calcular_efecto_grupos(estado, cfg)

Calculates social pressure from affin and opposing groups. Operates on differences, works for both [0,1] and [-1,1] ranges.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
float	Social influence delta.

Source code in simulator.py

def calcular_efecto_grupos(estado: dict, cfg: dict) -> float:
    """
    Calculates social pressure from affin and opposing groups.
    Operates on differences, works for both [0,1] and [-1,1] ranges.

    Args:
        estado: Current state.
        cfg: Global configuration.

    Returns:
        Social influence delta.
    """
    r      = _get_rango(cfg)
    op_a   = estado.get("opinion_grupo_a", r["ejemplo_apoyo"])
    op_b   = estado.get("opinion_grupo_b", r["ejemplo_rechazo"])
    perten = estado.get("pertenencia_grupo", 0.6)
    ref    = perten * op_a + (1.0 - perten) * op_b
    return cfg["efecto_vecinos_peso"] * (ref - estado["opinion"])

calculate_ews_metrics(window_data)

Calculates Early Warning Signal metrics over a sliding window.

Accepts a list of scalar floats (one opinion per time step) and returns per-variable arrays for variance, lag-1 autocorrelation, and skewness. The scalar time series is reshaped to (T, 1) so the return dict always contains 1-D arrays of length 1.

Parameters:

Name	Type	Description	Default
window_data	list	List of scalar opinion values, length == HISTORY_BUFFER_SIZE.	required

Returns:

Type	Description
dict	Dict with keys "variance", "autocorr", "skewness", each a np.ndarray
dict	of shape (n_vars,).

Source code in simulator.py

def calculate_ews_metrics(window_data: list) -> dict:
    """
    Calculates Early Warning Signal metrics over a sliding window.

    Accepts a list of scalar floats (one opinion per time step) and
    returns per-variable arrays for variance, lag-1 autocorrelation,
    and skewness. The scalar time series is reshaped to (T, 1) so
    the return dict always contains 1-D arrays of length 1.

    Args:
        window_data: List of scalar opinion values, length == HISTORY_BUFFER_SIZE.

    Returns:
        Dict with keys "variance", "autocorr", "skewness", each a np.ndarray
        of shape (n_vars,).
    """
    data_array = np.array(window_data, dtype=float)
    if data_array.ndim == 1:
        data_array = data_array.reshape(-1, 1)  # shape: (T, n_vars)

    variance = np.var(data_array, axis=0)

    n_vars = data_array.shape[1]
    autocorr = np.zeros(n_vars)
    for i in range(n_vars):
        if data_array.shape[0] > 1:
            cc = np.corrcoef(data_array[:-1, i], data_array[1:, i])
            val = cc[0, 1]
            autocorr[i] = val if not np.isnan(val) else 0.0

    skewness = stats.skew(data_array, axis=0)
    return {"variance": variance, "autocorr": autocorr, "skewness": skewness}

check_ews_signals(metrics, thresholds)

Checks computed EWS metrics against configurable thresholds.

Parameters:

Name	Type	Description	Default
metrics	dict	Output of calculate_ews_metrics.	required
thresholds	dict	Dict with optional keys "mean_variance_threshold" (default 0.1), "mean_autocorr_threshold" (default 0.5), "mean_skewness_threshold" (default 0.5).	required

Returns:

Type	Description
dict	Dict with bool flags "high_variance", "high_autocorr", "high_skewness".

Source code in simulator.py

def check_ews_signals(metrics: dict, thresholds: dict) -> dict:
    """
    Checks computed EWS metrics against configurable thresholds.

    Args:
        metrics: Output of calculate_ews_metrics.
        thresholds: Dict with optional keys "mean_variance_threshold"
                    (default 0.1), "mean_autocorr_threshold" (default 0.5),
                    "mean_skewness_threshold" (default 0.5).

    Returns:
        Dict with bool flags "high_variance", "high_autocorr", "high_skewness".
    """
    return {
        "high_variance": bool(
            np.mean(metrics["variance"]) > thresholds.get("mean_variance_threshold", 0.1)
        ),
        "high_autocorr": bool(
            np.mean(metrics["autocorr"]) > thresholds.get("mean_autocorr_threshold", 0.5)
        ),
        "high_skewness": bool(
            np.mean(np.abs(metrics["skewness"])) > thresholds.get("mean_skewness_threshold", 0.5)
        ),
    }

detect_topological_change(time_series, prev_diagram, embedding_dim=2, tau=1, wasserstein_threshold=0.3)

Detects significant topological changes via Takens-embedding + PH.

Embeds the scalar time series in R^embedding_dim using a lag of tau, computes the H1 persistence diagram via Vietoris-Rips filtration, and returns True if the Wasserstein distance to the previous diagram exceeds wasserstein_threshold.

Parameters:

Name	Type	Description	Default
time_series	ndarray	1-D numpy array of opinion values.	required
prev_diagram	ndarray | None	H1 persistence diagram from the previous call, or None.	required
embedding_dim	int	Takens embedding dimension (default 2).	2
tau	int	Delay parameter for Takens embedding (default 1).	1
wasserstein_threshold	float	Distance threshold for declaring a change.	0.3

Returns:

Type	Description
tuple[bool, ndarray | None]	(change_detected: bool, current_h1_diagram: np.ndarray | None)

Source code in simulator.py

def detect_topological_change(
    time_series: np.ndarray,
    prev_diagram: "np.ndarray | None",
    embedding_dim: int = 2,
    tau: int = 1,
    wasserstein_threshold: float = 0.3,
) -> "tuple[bool, np.ndarray | None]":
    """
    Detects significant topological changes via Takens-embedding + PH.

    Embeds the scalar time series in R^embedding_dim using a lag of tau,
    computes the H1 persistence diagram via Vietoris-Rips filtration, and
    returns True if the Wasserstein distance to the previous diagram
    exceeds wasserstein_threshold.

    Args:
        time_series: 1-D numpy array of opinion values.
        prev_diagram: H1 persistence diagram from the previous call, or None.
        embedding_dim: Takens embedding dimension (default 2).
        tau: Delay parameter for Takens embedding (default 1).
        wasserstein_threshold: Distance threshold for declaring a change.

    Returns:
        (change_detected: bool, current_h1_diagram: np.ndarray | None)
    """
    if not TDA_AVAILABLE:
        return False, prev_diagram

    min_len = embedding_dim * tau + 1
    if len(time_series) < min_len:
        return False, prev_diagram

    n = len(time_series) - (embedding_dim - 1) * tau
    embedded = np.array(
        [time_series[i: i + embedding_dim * tau: tau] for i in range(n)],
        dtype=float,
    )

    diagrams = ripser_compute(embedded, maxdim=1)["dgms"]
    h1: np.ndarray = diagrams[1] if len(diagrams) > 1 else np.empty((0, 2))

    if prev_diagram is None:
        return False, h1

    finite_prev = (
        prev_diagram[np.isfinite(prev_diagram[:, 1])]
        if len(prev_diagram) > 0
        else np.empty((0, 2))
    )
    finite_curr = (
        h1[np.isfinite(h1[:, 1])]
        if len(h1) > 0
        else np.empty((0, 2))
    )

    if len(finite_prev) == 0 and len(finite_curr) == 0:
        return False, h1

    dist = wasserstein_dist(finite_curr, finite_prev)
    return bool(dist > wasserstein_threshold), h1

get_graph_metrics(G, modo='macro', top_n=5)

Calcula centralidad de grado y betweenness sobre un grafo NetworkX y devuelve un resumen textual para el Arquitecto Social.

Parameters:

Name	Type	Description	Default
G	Graph	Grafo NetworkX (dirigido o no dirigido).	required
modo	str	'macro' o 'corporativo' — ajusta el vocabulario del resumen.	'macro'
top_n	int	Número de nodos más influyentes a listar.	5

Returns:

Type	Description
str	String con el resumen de métricas listo para inyectar en el prompt.

Source code in simulator.py

def get_graph_metrics(G: nx.Graph, modo: str = "macro", top_n: int = 5) -> str:
    """
    Calcula centralidad de grado y betweenness sobre un grafo NetworkX
    y devuelve un resumen textual para el Arquitecto Social.

    Args:
        G: Grafo NetworkX (dirigido o no dirigido).
        modo: 'macro' o 'corporativo' — ajusta el vocabulario del resumen.
        top_n: Número de nodos más influyentes a listar.

    Returns:
        String con el resumen de métricas listo para inyectar en el prompt.
    """
    if G is None or G.number_of_nodes() == 0:
        return "Red vacía — no se pudieron calcular métricas de grafo."

    # Centralidad de grado (normalizada)
    degree_cent = nx.degree_centrality(G)
    top_degree = sorted(degree_cent.items(), key=lambda x: x[1], reverse=True)[:top_n]

    # Betweenness centrality (control de flujo de información)
    try:
        between_cent = nx.betweenness_centrality(G, normalized=True)
        top_between = sorted(between_cent.items(), key=lambda x: x[1], reverse=True)[:top_n]
    except Exception:
        top_between = top_degree  # fallback si el grafo es trivial

    label_influencia = "Nodos más influyentes (grado)"
    label_puentes    = "Nodos puente (betweenness)"
    if modo == "corporativo":
        label_influencia = "Empleados/Directivos con más conexiones directas"
        label_puentes    = "Líderes informales que controlan el flujo de información"

    def fmt(items):
        return ", ".join(
            f"{nodo} ({v:.2f})" for nodo, v in items
        )

    resumen = (
        f"Red: {G.number_of_nodes()} nodos, {G.number_of_edges()} conexiones.\n"
        f"{label_influencia}: {fmt(top_degree)}.\n"
        f"{label_puentes}: {fmt(top_between)}."
    )
    return resumen

llamar_llm(estado, escenario, historial_reciente, cfg)

Main dispatcher for LLM selectors.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
escenario	str	Current scenario.	required
historial_reciente	list[dict]	History window for context.	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	A dictionary with "regla", "params", and "razon".

Source code in simulator.py

def llamar_llm(estado: dict, escenario: str,
                historial_reciente: list[dict], cfg: dict) -> dict:
    """
    Main dispatcher for LLM selectors.

    Args:
        estado: Current state.
        escenario: Current scenario.
        historial_reciente: History window for context.
        cfg: Global configuration.

    Returns:
        A dictionary with "regla", "params", and "razon".
    """
    proveedor = cfg.get("proveedor", "heurístico")

    if proveedor == "heurístico":
        return llamar_llm_heuristico(estado, escenario, historial_reciente, cfg)

    prompt = _construir_prompt(estado, escenario, historial_reciente, cfg)
    data   = None

    if proveedor == "ollama":
        data = _llamar_ollama(prompt, cfg)
    elif proveedor in PROVEEDORES:
        info    = PROVEEDORES[proveedor]
        modelo  = cfg.get("modelo", "").strip() or info["modelos_sugeridos"][0]
        if not resolve_provider_api_key(proveedor, fallback=cfg.get("api_key", "")):
            log.error(f"'{proveedor}' requiere API key. → heurístico.")
            return llamar_llm_heuristico(estado, escenario, historial_reciente, cfg)
        data = _llamar_openai_compatible(
            prompt,
            info["base_url"],
            modelo,
            cfg,
            proveedor,
        )
    else:
        log.error(f"Proveedor desconocido: '{proveedor}'. → heurístico.")
        return llamar_llm_heuristico(estado, escenario, historial_reciente, cfg)

    if data is None:
        log.warning("LLM sin respuesta → heurístico.")
        return llamar_llm_heuristico(estado, escenario, historial_reciente, cfg)

    regla_id = int(data.get("regla", 0))
    if regla_id not in REGLAS.get(escenario, {}):
        log.warning(f"Regla inválida ({regla_id}) → fallback.")
        return {"regla": 0, "params": {}, "razon": "fallback"}

    return {"regla": regla_id, "params": data.get("params", {}), "razon": data.get("razon", "")}

llamar_llm_heuristico(estado, escenario, historial_reciente, cfg)

Deterministic selector with expanded logic for all rules. Works as a baseline or fallback when no LLM is available.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
escenario	str	Current scenario.	required
historial_reciente	list[dict]	History window.	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Rule decision dictionary.

Source code in simulator.py

def llamar_llm_heuristico(estado: dict, escenario: str,
                           historial_reciente: list[dict], cfg: dict) -> dict:
    """
    Deterministic selector with expanded logic for all rules.
    Works as a baseline or fallback when no LLM is available.

    Args:
        estado: Current state.
        escenario: Current scenario.
        historial_reciente: History window.
        cfg: Global configuration.

    Returns:
        Rule decision dictionary.
    """
    opinion    = estado["opinion"]
    propaganda = estado["propaganda"]
    confianza  = estado.get("confianza", 0.5)
    neutro     = _neutro(cfg)
    amp        = _amplitud(cfg)
    op_a       = estado.get("opinion_grupo_a", neutro + 0.3 * amp)
    op_b       = estado.get("opinion_grupo_b", neutro - 0.3 * amp)

    tendencia  = [h["opinion"] for h in historial_reciente]
    delta      = tendencia[-1] - tendencia[0] if len(tendencia) > 1 else 0.0

    zona_rechazo = neutro - 0.35 * amp
    umbral_prop  = neutro + 0.15 * amp
    distancia_grupos = abs(op_a - op_b)

    # Narrativa B activa → contagio competitivo
    if "narrativa_b" in estado and abs(estado.get("narrativa_b", 0)) > 0.2:
        return {"regla": 6,
                "params": {"competencia": cfg.get("competencia_peso", 0.4)},
                "razon": "contagio_competitivo: narrativa B activa y relevante"}

    # Capa estratégica activa + alta polarización → Replicador EGT
    # Los agentes ya están bajo presión de juego; el modelo evolutivo
    # captura mejor la dinámica de estrategias enfrentadas.
    if cfg.get("strategic", {}).get("enabled", False) and distancia_grupos > _STRATEGIC_POLARIZATION_THRESHOLD * amp:
        return {"regla": 9,
                "params": {"dt": 0.1},
                "razon": "replicador: capa estratégica activa con alta polarización entre grupos"}

    # Grupos muy distantes → HK (solo escucha a similares)
    if distancia_grupos > 0.6 * amp:
        return {"regla": 5,
                "params": {"epsilon": cfg.get("hk_epsilon", 0.3)},
                "razon": f"hk: grupos muy distantes ({distancia_grupos:.2f})"}

    # Rechazo establecido + propaganda → backlash
    if opinion < zona_rechazo and abs(propaganda) > 0.3:
        return {"regla": 3,
                "params": {"penalizacion": 0.12},
                "razon": f"backlash: rechazo establecido (op={opinion:.2f})"}

    # Tendencia fuerte ya iniciada → polarización
    if abs(delta) > 0.05 * amp:
        return {"regla": 4,
                "params": {"fuerza": 0.08},
                "razon": f"polarizacion: tendencia {'positiva' if delta>0 else 'negativa'} fuerte"}

    # Propaganda intensa + baja confianza → umbral
    if abs(propaganda) > abs(umbral_prop) and confianza < 0.5:
        return {"regla": 1,
                "params": {"umbral": round(abs(umbral_prop), 2), "incremento": 0.12},
                "razon": "umbral: propaganda intensa + baja confianza"}

    # Sistema cerca del neutro + grupos similares → homofilia
    if abs(opinion - neutro) < 0.1 * amp and distancia_grupos < 0.4 * amp:
        return {"regla": 8,
                "params": {"tasa": cfg.get("homofilia_tasa", 0.05)},
                "razon": "homofilia: sistema cerca del neutro, grupos convergentes"}

    # Sistema estable → memoria
    if abs(delta) < 0.01 * amp:
        return {"regla": 2,
                "params": {"alpha": 0.75, "beta": 0.18, "gamma": 0.07},
                "razon": "memoria: sistema estable, inercia dominante"}

    # Sistema estable con grupos muy similares → Nash equilibrium
    if EXTENDED_MODELS_AVAILABLE and distancia_grupos < 0.25 * amp and abs(delta) < 0.02 * amp:
        return {"regla": 10,
                "params": {"c_same": 2.0, "c_diff": 0.5},
                "razon": "nash: grupos próximos, equilibrio de coordinación"}

    return {"regla": 0,
            "params": {"a": 0.72, "b": 0.28},
            "razon": "lineal: condiciones moderadas"}

load_checkpoint(filepath)

Loads a simulation history from a JSON checkpoint file.

Parameters:

Name	Type	Description	Default
filepath	str | Path	Path to the checkpoint file previously saved by :func:save_checkpoint.	required

Returns:

Name	Type	Description
	list[dict]	List of state dictionaries (historial) that can be passed directly
to	list[dict]	func:resumen_historial or used as the base for continued
	list[dict]	simulation.

Raises:

Type	Description
FileNotFoundError	If filepath does not exist.
ValueError	If the file format is unrecognised or corrupted.

Source code in simulator.py

def load_checkpoint(filepath: str | Path) -> list[dict]:
    """
    Loads a simulation history from a JSON checkpoint file.

    Args:
        filepath: Path to the checkpoint file previously saved by
            :func:`save_checkpoint`.

    Returns:
        List of state dictionaries (historial) that can be passed directly
        to :func:`resumen_historial` or used as the base for continued
        simulation.

    Raises:
        FileNotFoundError: If *filepath* does not exist.
        ValueError: If the file format is unrecognised or corrupted.
    """
    filepath = Path(filepath)
    if not filepath.exists():
        raise FileNotFoundError(f"Checkpoint no encontrado: {filepath}")
    with filepath.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict) or "historial" not in data:
        raise ValueError(f"Formato de checkpoint inválido: {filepath}")
    historial = data["historial"]
    if not isinstance(historial, list):
        raise ValueError(f"Campo 'historial' debe ser una lista: {filepath}")
    log.info(f"Checkpoint cargado: {filepath} ({len(historial)} entradas)")
    return historial

regla_backlash(estado, params, cfg)

Backlash/Boomerang effect rule. Propaganda reinforces the opposite position when negative sentiment is established.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (penalizacion).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_backlash(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Backlash/Boomerang effect rule.
    Propaganda reinforces the opposite position when negative sentiment is established.

    Args:
        estado: Current state.
        params: Rule parameters (penalizacion).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    penalizacion = params.get("penalizacion", 0.15)
    prop         = _aplicar_sesgo_confirmacion(estado["propaganda"], estado["opinion"], cfg)
    nuevo = estado.copy()
    neutro = _neutro(cfg)
    if _es_bipolar(cfg):
        if estado["opinion"] < neutro:
            val = estado["opinion"] - penalizacion * abs(prop)
        else:
            val = estado["opinion"]
    else:
        umbral_inf = params.get("umbral_inferior", 0.35)
        if estado["opinion"] < umbral_inf:
            val = estado["opinion"] - penalizacion * prop
        else:
            val = estado["opinion"]
    nuevo["opinion"] = _clip(val, cfg)
    return nuevo

regla_contagio_competitivo(estado, params, cfg)

Competitive Contagion model based on Beutel et al. (2012). Models competition between two simultaneous narratives.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (competencia).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_contagio_competitivo(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Competitive Contagion model based on Beutel et al. (2012).
    Models competition between two simultaneous narratives.

    Args:
        estado: Current state.
        params: Rule parameters (competencia).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    competencia  = params.get("competencia", cfg.get("competencia_peso", 0.4))
    opinion      = estado["opinion"]
    narrativa_a  = _aplicar_sesgo_confirmacion(estado["propaganda"], opinion, cfg)
    # narrativa_b puede estar en el estado o inferirse como la opuesta
    narrativa_b  = estado.get("narrativa_b", -narrativa_a if _es_bipolar(cfg) else 1.0 - narrativa_a)

    # Influencia neta: A gana si es más fuerte que B
    influencia_neta = narrativa_a - competencia * narrativa_b
    neutro          = _neutro(cfg)

    # La influencia neta empuja la opinión hacia o desde el neutro
    nuevo = estado.copy()
    val   = opinion + 0.15 * influencia_neta * (1.0 - abs(opinion - neutro) / _amplitud(cfg))
    nuevo["opinion"] = _clip(val, cfg)
    return nuevo

regla_hk(estado, params, cfg)

Hegselmann-Krause (2002) - Bounded Confidence model. Agents only interact with groups whose opinion is within a radius ε.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (epsilon).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_hk(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Hegselmann-Krause (2002) - Bounded Confidence model.
    Agents only interact with groups whose opinion is within a radius ε.

    Args:
        estado: Current state.
        params: Rule parameters (epsilon).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    epsilon = params.get("epsilon", cfg.get("hk_epsilon", 0.3))
    opinion = estado["opinion"]
    op_a    = estado.get("opinion_grupo_a", _get_rango(cfg)["ejemplo_apoyo"])
    op_b    = estado.get("opinion_grupo_b", _get_rango(cfg)["ejemplo_rechazo"])
    perten  = estado.get("pertenencia_grupo", 0.6)
    prop    = _aplicar_sesgo_confirmacion(estado["propaganda"], opinion, cfg)

    # Determinar qué grupos están dentro del radio de confianza
    grupos_validos = []
    pesos_validos  = []

    if abs(opinion - op_a) <= epsilon:
        grupos_validos.append(op_a)
        pesos_validos.append(perten)

    if abs(opinion - op_b) <= epsilon:
        grupos_validos.append(op_b)
        pesos_validos.append(1.0 - perten)

    nuevo = estado.copy()
    if grupos_validos:
        # Promedio ponderado solo de grupos dentro del radio
        total_peso   = sum(pesos_validos)
        opinion_ref  = sum(g * p for g, p in zip(grupos_validos, pesos_validos)) / total_peso
        # Convergencia gradual hacia la referencia de confianza
        alpha        = params.get("alpha", 0.3)
        val          = opinion + alpha * (opinion_ref - opinion) + 0.05 * prop
    else:
        # Nadie dentro del radio → fragmentación, opinión casi estática
        val = opinion + 0.01 * prop  # influencia mínima de propaganda

    nuevo["opinion"] = _clip(val, cfg)
    return nuevo

regla_homofilia(estado, params, cfg)

Axelrod (1997) - Co-evolutionary Network / Homophily. Influence weights change based on opinion similarity.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (tasa).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_homofilia(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Axelrod (1997) - Co-evolutionary Network / Homophily.
    Influence weights change based on opinion similarity.

    Args:
        estado: Current state.
        params: Rule parameters (tasa).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    tasa    = params.get("tasa", cfg.get("homofilia_tasa", 0.05))
    opinion = estado["opinion"]
    op_a    = estado.get("opinion_grupo_a", _get_rango(cfg)["ejemplo_apoyo"])
    op_b    = estado.get("opinion_grupo_b", _get_rango(cfg)["ejemplo_rechazo"])
    perten  = estado.get("pertenencia_grupo", 0.6)
    prop    = _aplicar_sesgo_confirmacion(estado["propaganda"], opinion, cfg)

    amp    = _amplitud(cfg)
    # Similitud normalizada al rango
    sim_a  = 1.0 - abs(opinion - op_a) / amp
    sim_b  = 1.0 - abs(opinion - op_b) / amp

    # Actualizar pertenencia según similitud (homofilia)
    nuevo_perten = float(np.clip(perten + tasa * (sim_a - sim_b), 0.1, 0.9))

    # Calcular referencia social con nuevos pesos
    ref_social   = nuevo_perten * op_a + (1.0 - nuevo_perten) * op_b
    peso_social  = cfg.get("efecto_vecinos_peso", 0.05)

    val  = opinion + peso_social * (ref_social - opinion) + 0.08 * prop

    nuevo = estado.copy()
    nuevo["opinion"]           = _clip(val, cfg)
    nuevo["pertenencia_grupo"] = nuevo_perten  # persiste al próximo paso
    nuevo["_sim_grupo_a"]      = round(sim_a, 3)
    nuevo["_sim_grupo_b"]      = round(sim_b, 3)
    return nuevo

regla_lineal(estado, params, cfg)

Linear transition rule based on Friedkin-Johnsen model. Opinion changes proportionally to current opinion and propaganda.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (a: resistance, b: influence).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_lineal(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Linear transition rule based on Friedkin-Johnsen model.
    Opinion changes proportionally to current opinion and propaganda.

    Args:
        estado: Current state.
        params: Rule parameters (a: resistance, b: influence).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    a, b  = params.get("a", 0.7), params.get("b", 0.3)
    prop  = _aplicar_sesgo_confirmacion(estado["propaganda"], estado["opinion"], cfg)
    nuevo = estado.copy()
    nuevo["opinion"] = _clip(a * estado["opinion"] + b * prop, cfg)
    return nuevo

regla_memoria(estado, params, cfg)

Inertia rule based on DeGroot with lag. The current state depends on the previous state and history.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (alpha, beta, gamma).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_memoria(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Inertia rule based on DeGroot with lag.
    The current state depends on the previous state and history.

    Args:
        estado: Current state.
        params: Rule parameters (alpha, beta, gamma).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    alpha = params.get("alpha", 0.7)
    beta  = params.get("beta",  0.2)
    gamma = params.get("gamma", 0.1)
    prev  = estado.get("opinion_prev", estado["opinion"])
    prop  = _aplicar_sesgo_confirmacion(estado["propaganda"], estado["opinion"], cfg)
    nuevo = estado.copy()
    nuevo["opinion"] = _clip(
        alpha * estado["opinion"] + beta * prev + gamma * prop, cfg
    )
    return nuevo

regla_polarizacion(estado, params, cfg)

Polarization/Echo chamber rule. Moves opinion further away from the neutral point.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (fuerza).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_polarizacion(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Polarization/Echo chamber rule.
    Moves opinion further away from the neutral point.

    Args:
        estado: Current state.
        params: Rule parameters (fuerza).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    fuerza  = params.get("fuerza", 0.1)
    opinion = estado["opinion"]
    neutro  = _neutro(cfg)
    r       = _get_rango(cfg)
    nuevo   = estado.copy()
    if opinion >= neutro:
        val = opinion + fuerza * (r["max"] - opinion)
    else:
        val = opinion - fuerza * (opinion - r["min"])
    nuevo["opinion"] = _clip(val, cfg)
    return nuevo

regla_replicador(estado, params, cfg)

Replicator Equation (EGT) — two-strategy evolutionary game theory.

Models the evolutionary pressure between two group alignments

Strategy 0 → align with group A (opinion_grupo_a) Strategy 1 → align with group B (opinion_grupo_b)

pertenencia_grupo tracks the frequency of Strategy 0. After integrating the replicator ODE, opinion shifts to the payoff- weighted group average.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (payoff_matrix, dt).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state with new opinion and pertenencia_grupo.

Source code in simulator.py

def regla_replicador(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Replicator Equation (EGT) — two-strategy evolutionary game theory.

    Models the evolutionary pressure between two group alignments:
      Strategy 0 → align with group A (opinion_grupo_a)
      Strategy 1 → align with group B (opinion_grupo_b)

    pertenencia_grupo tracks the frequency of Strategy 0.  After
    integrating the replicator ODE, opinion shifts to the payoff-
    weighted group average.

    Args:
        estado: Current state.
        params: Rule parameters (payoff_matrix, dt).
        cfg: Global configuration.

    Returns:
        Updated state with new opinion and pertenencia_grupo.
    """
    pertenencia = estado.get("pertenencia_grupo", 0.6)
    pop = np.array([pertenencia, 1.0 - pertenencia], dtype=float)

    raw_payoff = params.get(
        "payoff_matrix",
        cfg.get("payoff_matrix", DEFAULT_PAYOFF_MATRIX),
    )
    payoff_matrix = np.array(raw_payoff, dtype=float)
    if payoff_matrix.shape != (2, 2):
        payoff_matrix = np.eye(2)

    dt = float(params.get("dt", cfg.get("dt", 0.1)))
    updated = apply_replicator_equation(pop, payoff_matrix, dt)

    nuevo_perten = float(np.clip(updated[0], 0.1, 0.9))
    op_a = estado.get("opinion_grupo_a", _get_rango(cfg)["ejemplo_apoyo"])
    op_b = estado.get("opinion_grupo_b", _get_rango(cfg)["ejemplo_rechazo"])

    nuevo = estado.copy()
    nuevo["pertenencia_grupo"] = nuevo_perten
    nuevo["opinion"] = _clip(
        nuevo_perten * op_a + (1.0 - nuevo_perten) * op_b,
        cfg,
    )
    return nuevo

regla_umbral(estado, params, cfg)

Threshold/Tipping point rule based on Granovetter (Simple). A non-linear jump occurs when propaganda exceeds a critical threshold.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (umbral, incremento).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_umbral(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Threshold/Tipping point rule based on Granovetter (Simple).
    A non-linear jump occurs when propaganda exceeds a critical threshold.

    Args:
        estado: Current state.
        params: Rule parameters (umbral, incremento).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    r          = _get_rango(cfg)
    umbral     = params.get("umbral", 0.65 if not _es_bipolar(cfg) else 0.4)
    incremento = params.get("incremento", 0.15)
    prop       = _aplicar_sesgo_confirmacion(estado["propaganda"], estado["opinion"], cfg)
    nuevo = estado.copy()
    if abs(prop) > abs(umbral):
        signo = np.sign(prop) if _es_bipolar(cfg) else 1.0
        val   = estado["opinion"] + signo * incremento * (r["max"] - abs(estado["opinion"]))
    else:
        val = estado["opinion"]
    nuevo["opinion"] = _clip(val, cfg)
    return nuevo

regla_umbral_heterogeneo(estado, params, cfg)

Heterogeneous Threshold model based on Granovetter (1978). Thresholds are normally distributed, enabling social cascades.

Parameters:

Name	Type	Description	Default
estado	dict	Current state.	required
params	dict	Rule parameters (media, std).	required
cfg	dict	Global configuration.	required

Returns:

Type	Description
dict	Updated state.

Source code in simulator.py

def regla_umbral_heterogeneo(estado: dict, params: dict, cfg: dict) -> dict:
    """
    Heterogeneous Threshold model based on Granovetter (1978).
    Thresholds are normally distributed, enabling social cascades.

    Args:
        estado: Current state.
        params: Rule parameters (media, std).
        cfg: Global configuration.

    Returns:
        Updated state.
    """
    media   = params.get("media", cfg.get("umbral_media", 0.5))
    std     = params.get("std",   cfg.get("umbral_std",   0.15))
    opinion = estado["opinion"]
    neutro  = _neutro(cfg)
    prop    = _aplicar_sesgo_confirmacion(estado["propaganda"], opinion, cfg)

    # Fracción de la población que ya superó su umbral personal
    # (modelado con CDF de la normal)
    fraccion_adoptantes = 0.5 * (1 + erf((opinion - neutro - media + 0.5) / (std * np.sqrt(2))))
    fraccion_adoptantes = float(np.clip(fraccion_adoptantes, 0.0, 1.0))

    # La fracción de adoptantes genera presión social adicional
    r    = _get_rango(cfg)
    val  = opinion + 0.2 * fraccion_adoptantes * (r["max"] - opinion) + 0.05 * prop

    nuevo = estado.copy()
    nuevo["opinion"] = _clip(val, cfg)
    # Guardar fracción para análisis
    nuevo["_fraccion_adoptantes"] = round(fraccion_adoptantes, 3)
    return nuevo

resumen_historial(historial, config=None)

Calculates descriptive statistics for a simulation history.

Parameters:

Name	Type	Description	Default
historial	list[dict]	List of state dictionaries.	required
config	dict | None	Configuration for range/neutral reference.	None

Returns:

Type	Description
dict	Dictionary with mean, std, delta, and dominant regime.

Source code in simulator.py

def resumen_historial(historial: list[dict], config: dict | None = None) -> dict:
    """
    Calculates descriptive statistics for a simulation history.

    Args:
        historial: List of state dictionaries.
        config: Configuration for range/neutral reference.

    Returns:
        Dictionary with mean, std, delta, and dominant regime.
    """
    cfg       = {**DEFAULT_CONFIG, **(config or {})}
    neutro    = _neutro(cfg)
    opiniones = np.array([h["opinion"] for h in historial])
    reglas    = [h["_regla_nombre"] for h in historial if "_regla_nombre" in h]
    return {
        "opinion_inicial":    float(opiniones[0]),
        "opinion_final":      float(opiniones[-1]),
        "delta_total":        float(opiniones[-1] - opiniones[0]),
        "media":              float(opiniones.mean()),
        "desviacion":         float(opiniones.std()),
        "minimo":             float(opiniones.min()),
        "maximo":             float(opiniones.max()),
        "polarizacion_media": float(np.mean(np.abs(opiniones - neutro))),
        "pasos":              len(historial) - 1,
        "regla_dominante":    Counter(reglas).most_common(1)[0][0] if reglas else "—",
        "neutro":             neutro,
        "rango":              cfg.get("rango", "—"),
    }

run_with_schedule(estado_inicial, strategy_schedule, escenario='campana', config=None, verbose=True)

Ejecuta la simulación siguiendo estrictamente un schedule de intervenciones generado por el LLM en Modo Inverso.

Source code in simulator.py

def run_with_schedule(
    estado_inicial: dict,
    strategy_schedule: dict,
    escenario: str = "campana",
    config: dict | None = None,
    verbose: bool = True,
) -> list[dict]:
    """
    Ejecuta la simulación siguiendo estrictamente un schedule de intervenciones
    generado por el LLM en Modo Inverso.
    """
    cfg = {**DEFAULT_CONFIG, **(config or {})}
    r = _get_rango(cfg)
    alpha_blend = cfg["alpha_blend"]

    estado = estado_inicial.copy()
    estado.setdefault("opinion_prev", estado["opinion"])
    estado.setdefault("confianza", 0.5)
    estado.setdefault("opinion_grupo_a", min(estado["opinion"] + 0.2 * _amplitud(cfg), r["max"]))
    estado.setdefault("opinion_grupo_b", max(estado["opinion"] - 0.2 * _amplitud(cfg), r["min"]))
    estado.setdefault("pertenencia_grupo", 0.6)

    historial: list[dict] = [estado.copy()]

    name_to_id = {v: k for k, v in NOMBRES_REGLAS.items()}

    paso_actual = 1
    for fase in strategy_schedule.get("interventions", []):
        start = max(paso_actual, fase["time_start"])
        end = fase["time_end"]
        # Parsing fallback for manual LLM responses
        regla_nombre = fase["model_name"].lower().strip()
        if "degroot" in regla_nombre: regla_nombre = "memoria"
        if "granovetter" in regla_nombre: regla_nombre = "umbral_heterogeneo"
        if "hegselmann" in regla_nombre or "hk" in regla_nombre: regla_nombre = "hk"

        regla_id = name_to_id.get(regla_nombre, 0) # Fallback a lineal(0)
        params = _validar_params(NOMBRES_REGLAS.get(regla_id, "lineal"), fase.get("parameters", {}))
        razon = fase.get("fase_rationale", "")

        # Extraer target_nodes desde parameters o desde la fase directamente
        target_nodes = params.pop("target_nodes", None) or fase.get("target_nodes", None)

        for paso in range(start, end + 1):
            if verbose and paso == start:
                log.info(
                    f"Fase Inversa [{start}-{end}]: {NOMBRES_REGLAS[regla_id]} | {razon}"
                    + (f" | target_nodes={target_nodes}" if target_nodes else "")
                )

            regla_func = REGLAS[escenario].get(regla_id, regla_lineal)
            estado_regla = regla_func(estado, params, cfg)
            opinion_regla = _clip(estado_regla["opinion"], cfg)

            # ── Aplicar boosting de target_nodes (modo corporativo) ──────
            # Si se especifican nodos líderes, su „voto" fuerza la propaganda
            # hacia el centro de sus posiciones, aumentando la convergencia.
            if target_nodes:
                proporcion_target = min(1.0, len(target_nodes) / max(1, cfg.get("_n_nodos", 20)))
                # La opinión de los nodos target jala la opinión global con extra peso
                opinion_regla = _clip(
                    opinion_regla + 0.12 * proporcion_target * (
                        estado.get("propaganda", 0.5) - opinion_regla
                    ),
                    cfg,
                )

            tendencia_base = 0.92 * estado["opinion"] + 0.08 * estado["propaganda"]
            opinion_blend = alpha_blend * opinion_regla + (1.0 - alpha_blend) * tendencia_base
            ruido_std = cfg["ruido_base"] + cfg["ruido_desconfianza"] * (1.0 - estado["confianza"])

            opinion_final = _clip(
                opinion_blend
                + calcular_efecto_grupos(estado, cfg)
                + _calcular_fuerza_estrategica(estado, cfg)
                + np.random.normal(0.0, ruido_std),
                cfg,
            )

            nuevo = estado.copy()
            if "pertenencia_grupo" in estado_regla:
                nuevo["pertenencia_grupo"] = estado_regla["pertenencia_grupo"]
            for k in ("_fraccion_adoptantes", "_sim_grupo_a", "_sim_grupo_b",
                      "_nash_sigma_a", "_nash_sigma_b", "_bayes_uncertainty",
                      "_sir_S", "_sir_I", "_sir_R"):
                if k in estado_regla:
                    nuevo[k] = estado_regla[k]

            nuevo["opinion_prev"]   = estado["opinion"]
            nuevo["opinion"]        = opinion_final
            nuevo["_paso"]          = paso
            nuevo["_regla"]         = regla_id
            nuevo["_regla_nombre"]  = NOMBRES_REGLAS.get(regla_id, regla_nombre)
            nuevo["_razon"]         = razon
            nuevo["_rango"]         = cfg["rango"]
            nuevo["_target_nodes"]  = target_nodes  # trazabilidad

            estado = nuevo
            historial.append(estado.copy())
            paso_actual = paso + 1

    return historial

save_checkpoint(historial, filepath)

Saves simulation history to a JSON checkpoint file for later recovery.

Only JSON-serializable fields are preserved; non-serializable values (numpy arrays, etc.) are converted to Python native types where possible or dropped with a warning.

Parameters:

Name	Type	Description	Default
historial	list[dict]	List of state dictionaries from :func:simular.	required
filepath	str | Path	Destination path for the checkpoint file.	required

Source code in simulator.py

def save_checkpoint(historial: list[dict], filepath: str | Path) -> None:
    """
    Saves simulation history to a JSON checkpoint file for later recovery.

    Only JSON-serializable fields are preserved; non-serializable values
    (numpy arrays, etc.) are converted to Python native types where possible
    or dropped with a warning.

    Args:
        historial: List of state dictionaries from :func:`simular`.
        filepath: Destination path for the checkpoint file.
    """
    filepath = Path(filepath)
    filepath.parent.mkdir(parents=True, exist_ok=True)

    def _make_serializable(obj: Any) -> Any:
        if isinstance(obj, np.ndarray):
            return obj.tolist()
        if isinstance(obj, np.integer):
            return int(obj)
        if isinstance(obj, np.floating):
            return float(obj)
        if isinstance(obj, dict):
            return {k: _make_serializable(v) for k, v in obj.items()}
        if isinstance(obj, (list, tuple)):
            return [_make_serializable(v) for v in obj]
        if isinstance(obj, set):
            return sorted(_make_serializable(v) for v in obj)
        if not isinstance(obj, (str, int, float, bool, type(None))):
            log.warning(
                "save_checkpoint: tipo no serializable ignorado: %s=%r",
                type(obj).__name__,
                obj,
            )
            return None
        return obj

    serializable = _make_serializable(historial)
    with filepath.open("w", encoding="utf-8") as fh:
        json.dump({"version": 1, "historial": serializable}, fh, ensure_ascii=False, indent=2)
    log.info(f"Checkpoint guardado: {filepath} ({len(historial)} pasos)")

simular(estado_inicial, escenario='campana', pasos=50, cada_n_pasos=5, config=None, verbose=True)

Executes the hybrid simulation with all available rules.

Parameters:

Name	Type	Description	Default
estado_inicial	dict	Dictionary with at least "opinion" and "propaganda".	required
escenario	str	Scenario key in REGLAS.	'campana'
pasos	int	Number of time steps.	50
cada_n_pasos	int	Frequency of LLM rule selection updates.	5
config	dict | None	Override dictionary for DEFAULT_CONFIG.	None
verbose	bool	If true, logs step details.	True

Returns:

Type	Description
list[dict]	A list of state dictionaries (including t=0).

Source code in simulator.py

def simular(
    estado_inicial: dict,
    escenario: str = "campana",
    pasos: int = 50,
    cada_n_pasos: int = 5,
    config: dict | None = None,
    verbose: bool = True,
) -> list[dict]:
    """
    Executes the hybrid simulation with all available rules.

    Args:
        estado_inicial: Dictionary with at least "opinion" and "propaganda".
        escenario: Scenario key in REGLAS.
        pasos: Number of time steps.
        cada_n_pasos: Frequency of LLM rule selection updates.
        config: Override dictionary for DEFAULT_CONFIG.
        verbose: If true, logs step details.

    Returns:
        A list of state dictionaries (including t=0).
    """
    cfg         = {**DEFAULT_CONFIG, **(config or {})}
    # EMPIRICAL INTEGRATION — aplicar parámetros empíricos como defaults antes que el usuario los sobreescriba
    # Los parámetros del usuario en config tienen prioridad; los valores 0.0 se tratan como neutralidad activa.
    if EMPIRICAL_AVAILABLE and BEYONDSIGHT_RUNTIME_PARAMS:
        cultural_profile = str((config or {}).get("cultural_profile", "mixed"))
        empirical_defaults = build_empirical_engine_config(cultural_profile)
        # Only set keys NOT already overridden by the caller's config argument
        user_keys = set((config or {}).keys())
        for k, v in empirical_defaults.items():
            # strategic requires nested merging; cultural_profile and
            # validation_flags are metadata, not simulator control knobs.
            if k in ENGINE_METADATA_KEYS:
                continue
            if k not in user_keys:
                cfg[k] = v
        if "strategic" not in user_keys:
            # strategic needs nested merging so empirical payoffs enrich the
            # default matrix without discarding other simulator defaults.
            strategic_defaults = empirical_defaults.get("strategic", {})
            payoff_defaults = strategic_defaults.get("payoff_matrix", {})
            cfg["strategic"] = {
                **cfg.get("strategic", {}),
                **{k: v for k, v in strategic_defaults.items() if k != "payoff_matrix"},
                "payoff_matrix": {
                    **cfg.get("strategic", {}).get("payoff_matrix", {}),
                    **payoff_defaults,
                },
            }
    r           = _get_rango(cfg)
    alpha_blend = cfg["alpha_blend"]
    proveedor   = cfg.get("proveedor", "heurístico")

    estado = estado_inicial.copy()
    estado.setdefault("opinion_prev",     estado["opinion"])
    estado.setdefault("confianza",        0.5)
    estado.setdefault("opinion_grupo_a",  min(estado["opinion"] + 0.2 * _amplitud(cfg), r["max"]))
    estado.setdefault("opinion_grupo_b",  max(estado["opinion"] - 0.2 * _amplitud(cfg), r["min"]))
    estado.setdefault("pertenencia_grupo", 0.6)

    historial: list[dict] = [estado.copy()]
    regla_actual   = 0
    params_actuales: dict = {}
    razon_actual   = "inicial"

    # EWS: sliding window of scalar opinion values
    opinion_history: deque = deque(maxlen=HISTORY_BUFFER_SIZE)
    # Mutable runtime state for non-serialisable objects (e.g. TDA diagram)
    cfg_runtime: dict = {}

    def _seleccionar(est, hist):
        # EGT Replicator forced mode: bypass LLM and lock to rule 9
        if cfg.get("modelo_matematico") == "Replicator":
            payoff = np.array(
                cfg.get("payoff_matrix", DEFAULT_PAYOFF_MATRIX),
                dtype=float,
            )
            dt = float(cfg.get("dt", 0.1))
            return 9, {"payoff_matrix": payoff.tolist(), "dt": dt}, "replicador: EGT mode active"

        # CfC fast path — evita llamada LLM cuando hay modelo entrenado y confianza alta
        if CFC_AVAILABLE and len(hist) >= 6:
            opinion_window = [h["opinion"] for h in hist[-6:]]
            rid, source, conf = _cfc.select_regime(
                history=opinion_window, state=est
            )
            if source == "cfc":
                p = _validar_params(NOMBRES_REGLAS[rid], {})
                return rid, p, f"cfc (conf={conf:.2f})"

        ventana = hist[-cfg["ventana_historial_llm"]:]
        dec     = llamar_llm(est, escenario, ventana, cfg)
        r_id    = dec["regla"]
        p       = _validar_params(NOMBRES_REGLAS[r_id], dec.get("params", {}))
        return r_id, p, dec.get("razon", "")

    regla_actual, params_actuales, razon_actual = _seleccionar(estado, historial)
    if verbose:
        log.info(
            f"t=0 | [{proveedor}] rango={r['min']},{r['max']} "
            f"| {NOMBRES_REGLAS[regla_actual]} | {razon_actual}"
        )

    for paso in range(1, pasos + 1):

        if paso % cada_n_pasos == 0:
            regla_actual, params_actuales, razon_actual = _seleccionar(estado, historial)
            if verbose:
                log.info(
                    f"t={paso:3d} | [{proveedor}] {NOMBRES_REGLAS[regla_actual]:22s} "
                    f"op={estado['opinion']:+.3f} | {razon_actual}"
                )

        # Aplicar regla elegida
        regla_func    = REGLAS[escenario].get(regla_actual, regla_lineal)
        estado_regla  = regla_func(estado, params_actuales, cfg)
        opinion_regla = _clip(estado_regla["opinion"], cfg)

        # Tendencia base + blending
        tendencia_base = 0.92 * estado["opinion"] + 0.08 * estado["propaganda"]
        opinion_blend  = alpha_blend * opinion_regla + (1.0 - alpha_blend) * tendencia_base

        # Efecto de grupos + fuerza estratégica + ruido adaptativo
        ruido_std     = cfg["ruido_base"] + cfg["ruido_desconfianza"] * (1.0 - estado["confianza"])
        opinion_final = _clip(
            opinion_blend
            + calcular_efecto_grupos(estado, cfg)
            + _calcular_fuerza_estrategica(estado, cfg)
            + np.random.normal(0.0, ruido_std),
            cfg
        )

        # Construir nuevo estado
        nuevo = estado.copy()
        # Si la regla actualizó pertenencia_grupo (homofilia), persiste
        if "pertenencia_grupo" in estado_regla:
            nuevo["pertenencia_grupo"] = estado_regla["pertenencia_grupo"]
        # Métricas auxiliares de reglas avanzadas
        for k in ("_fraccion_adoptantes", "_sim_grupo_a", "_sim_grupo_b",
                  "_nash_sigma_a", "_nash_sigma_b", "_bayes_uncertainty",
                  "_sir_S", "_sir_I", "_sir_R"):
            if k in estado_regla:
                nuevo[k] = estado_regla[k]

        nuevo["opinion_prev"]  = estado["opinion"]
        nuevo["opinion"]       = opinion_final
        nuevo["_paso"]         = paso
        nuevo["_regla"]        = regla_actual
        nuevo["_regla_nombre"] = NOMBRES_REGLAS[regla_actual]
        nuevo["_razon"]        = razon_actual
        nuevo["_rango"]        = cfg["rango"]

        estado = nuevo
        historial.append(estado.copy())

        # ── EWS: collect opinion, compute CSD metrics ─────────────────
        opinion_history.append(estado["opinion"])
        if len(opinion_history) == HISTORY_BUFFER_SIZE:
            ews_metrics = calculate_ews_metrics(list(opinion_history))
            ews_flags   = check_ews_signals(ews_metrics, cfg.get("ews_thresholds", {}))
            estado["_ews_flags"] = ews_flags
            historial[-1]["ews"] = {
                "metrics": {k: v.tolist() for k, v in ews_metrics.items()},
                "flags":   ews_flags,
            }

        # ── TDA: topological change detection every 5 steps ───────────
        if TDA_AVAILABLE and paso % 5 == 0 and len(opinion_history) >= HISTORY_BUFFER_SIZE:
            mean_opinions = np.array(list(opinion_history), dtype=float)
            tda_change, prev_diagram = detect_topological_change(
                mean_opinions,
                prev_diagram=cfg_runtime.get("prev_tda_diagram"),
                wasserstein_threshold=cfg.get("tda_wasserstein_threshold", 0.3),
            )
            cfg_runtime["prev_tda_diagram"] = prev_diagram
            historial[-1]["tda_change"] = tda_change

    if verbose:
        log.info(
            f"Completo: {pasos} pasos | "
            f"{historial[0]['opinion']:+.3f} → {historial[-1]['opinion']:+.3f} "
            f"(neutro={_neutro(cfg)})"
        )
    return historial

simular_multiples(estado_inicial, escenario='campana', pasos=50, cada_n_pasos=5, config=None, n_simulaciones=100)

Runs N simulations with variations in the initial state to return a distribution.

Parameters:

Name	Type	Description	Default
estado_inicial	dict	Base state for all simulations.	required
escenario	str	Scenario key.	'campana'
pasos	int	Steps per simulation.	50
cada_n_pasos	int	LLM update frequency.	5
config	dict | None	Override config.	None
n_simulaciones	int	Number of runs.	100

Returns:

Type	Description
dict	Statistics dictionary of the final opinion distribution.

Source code in simulator.py

def simular_multiples(
    estado_inicial: dict,
    escenario: str = "campana",
    pasos: int = 50,
    cada_n_pasos: int = 5,
    config: dict | None = None,
    n_simulaciones: int = 100,
) -> dict:
    """
    Runs N simulations with variations in the initial state to return a distribution.

    Args:
        estado_inicial: Base state for all simulations.
        escenario: Scenario key.
        pasos: Steps per simulation.
        cada_n_pasos: LLM update frequency.
        config: Override config.
        n_simulaciones: Number of runs.

    Returns:
        Statistics dictionary of the final opinion distribution.
    """
    cfg        = {**DEFAULT_CONFIG, **(config or {})}
    r          = _get_rango(cfg)
    ruido_ini  = cfg["ruido_estado_inicial"]
    resultados = np.zeros(n_simulaciones)

    # Keys whose valid range is [0, 1] regardless of the opinion range.
    # Clipping these to the opinion range (e.g. [-1, 1] in bipolar mode) is
    # incorrect: with low values and non-zero noise, they can go negative,
    # which inflates ruido_std beyond its intended maximum and corrupts the
    # noise model for the entire Monte Carlo run.
    _UNIT_INTERVAL_KEYS = {"confianza", "pertenencia_grupo"}

    for i in range(n_simulaciones):
        estado_ruido = {}
        for k, v in estado_inicial.items():
            if isinstance(v, float):
                noisy = v + np.random.normal(0, ruido_ini)
                if k in _UNIT_INTERVAL_KEYS:
                    estado_ruido[k] = float(np.clip(noisy, 0.0, 1.0))
                else:
                    estado_ruido[k] = float(np.clip(noisy, r["min"], r["max"]))
            else:
                estado_ruido[k] = v
        hist = simular(estado_ruido, escenario=escenario, pasos=pasos,
                       cada_n_pasos=cada_n_pasos, config=config, verbose=False)
        resultados[i] = hist[-1]["opinion"]

    neutro = _neutro(cfg)
    p10, p25, p50, p75, p90 = np.percentile(resultados, [10, 25, 50, 75, 90])
    return {
        "media":          float(resultados.mean()),
        "std":            float(resultados.std()),
        "p_sobre_neutro": float((resultados > neutro).mean()),
        "percentiles":    {"p10": float(p10), "p25": float(p25), "p50": float(p50),
                           "p75": float(p75), "p90": float(p90)},
        "escenarios":     {"optimista": float(p90), "mediano": float(p50),
                           "pesimista":  float(p10)},
        "neutro":         neutro,
        "n_simulaciones": n_simulaciones,
        "rango":          cfg["rango"],
    }

simular_multiples_dask(estado_inicial, escenario='campana', pasos=50, cada_n_pasos=5, config=None, n_simulaciones=100, seed=None)

Runs N simulations in parallel using Dask delayed computation. Falls back to sequential simular_multiples if Dask is unavailable.

Parameters:

Name	Type	Description	Default
Same as simular_multiples, plus			required
seed	int | None	Optional RNG seed for reproducibility (default: None = random).	None

Returns:

Type	Description
dict	Same statistics dictionary as simular_multiples, with "parallel" key.

Source code in simulator.py

def simular_multiples_dask(
    estado_inicial: dict,
    escenario: str = "campana",
    pasos: int = 50,
    cada_n_pasos: int = 5,
    config: dict | None = None,
    n_simulaciones: int = 100,
    seed: int | None = None,
) -> dict:
    """
    Runs N simulations in parallel using Dask delayed computation.
    Falls back to sequential simular_multiples if Dask is unavailable.

    Args:
        Same as simular_multiples, plus:
        seed: Optional RNG seed for reproducibility (default: None = random).

    Returns:
        Same statistics dictionary as simular_multiples, with "parallel" key.
    """
    try:
        from dask import delayed, compute as dask_compute
    except ImportError:
        log.warning("Dask no disponible — usando modo secuencial.")
        return simular_multiples(estado_inicial, escenario, pasos,
                                 cada_n_pasos, config, n_simulaciones)

    cfg       = {**DEFAULT_CONFIG, **(config or {})}
    r         = _get_rango(cfg)
    ruido_ini = cfg["ruido_estado_inicial"]
    rng       = np.random.default_rng(seed)
    noises    = rng.normal(0, ruido_ini, size=(n_simulaciones, len(estado_inicial)))

    @delayed
    def _run_one(noise_row: np.ndarray) -> float:
        keys = list(estado_inicial.keys())
        estado_ruido = {}
        for idx, k in enumerate(keys):
            v = estado_inicial[k]
            if isinstance(v, float):
                estado_ruido[k] = float(np.clip(v + noise_row[idx], r["min"], r["max"]))
            else:
                estado_ruido[k] = v
        hist = simular(estado_ruido, escenario=escenario, pasos=pasos,
                       cada_n_pasos=cada_n_pasos, config=config, verbose=False)
        return hist[-1]["opinion"]

    tasks      = [_run_one(noises[i]) for i in range(n_simulaciones)]
    resultados = np.array(dask_compute(*tasks))

    neutro = _neutro(cfg)
    p10, p25, p50, p75, p90 = np.percentile(resultados, [10, 25, 50, 75, 90])
    return {
        "media":          float(resultados.mean()),
        "std":            float(resultados.std()),
        "p_sobre_neutro": float((resultados > neutro).mean()),
        "percentiles":    {"p10": float(p10), "p25": float(p25), "p50": float(p50),
                           "p75": float(p75), "p90": float(p90)},
        "escenarios":     {"optimista": float(p90), "mediano": float(p50),
                           "pesimista":  float(p10)},
        "neutro":         neutro,
        "n_simulaciones": n_simulaciones,
        "rango":          cfg["rango"],
        "parallel":       True,
    }