import streamlit as st import pandas as pd import numpy as np from dataset_loader import ( load_territories, load_parties, load_indicators, load_geo_44, ) EXPECTED_MUNICIPALITIES = 44 EPS = 1e-9 APP_VERSION = "v0.9.0-beta" APP_STATUS = "Prototipo jugable con eventos multi-ronda, memoria contextual y tablero con deltas visibles" LAST_MAJOR_UPDATE = "Memoria de shocks, deltas por ronda, centro de mando reorganizado y resolución de turno más clara" COLUMN_MAP = { "population": [ "population_projected_2023", "population_est", "population", "poblacion_est", "poblacion", "population_2023", "poblacion_2023", "total_population", "total_poblacion", ], "registry": [ "electoral_registry", "registro_electoral", "registered_voters", "padron", "padron_electoral", ], "rural": [ "rural_population_pct", "rural_pct", "ruralidad_pct", ], "education": [ "schooling_years_avg", "schooling_years", "educacion_promedio", ], "poverty": [ "monetary_poverty_households_pct", "poverty_pct", "pobreza_monetaria_pct", ], "internet": [ "households_with_internet_pct", "internet_pct", ], "remittances": [ "remittance_receiving_households_pct", "remittances_pct", ], } CHANNELS = { "digital": {"label": "Campaña digital", "base_effect": 0.075}, "territorial": {"label": "Territorio y brigadas", "base_effect": 0.095}, "media": {"label": "Publicidad masiva", "base_effect": 0.065}, } PARTY_STRATEGY = { "n": { "budget": 95000, "digital_share": 0.40, "territorial_share": 0.38, "media_share": 0.22, "digital_bonus": 1.12, "territorial_bonus": 1.08, "media_bonus": 0.96, }, "arena": { "budget": 90000, "digital_share": 0.34, "territorial_share": 0.24, "media_share": 0.42, "digital_bonus": 1.04, "territorial_bonus": 0.95, "media_bonus": 1.14, }, "fmln": { "budget": 82000, "digital_share": 0.18, "territorial_share": 0.60, "media_share": 0.22, "digital_bonus": 0.92, "territorial_bonus": 1.15, "media_bonus": 0.94, }, "_default": { "budget": 80000, "digital_share": 0.33, "territorial_share": 0.34, "media_share": 0.33, "digital_bonus": 1.00, "territorial_bonus": 1.00, "media_bonus": 1.00, }, } PARTY_COLORS = { "n": "#00C2FF", "arena": "#1E4DFF", "fmln": "#E53935", "pcn": "#FFB300", "pdc": "#43A047", "gana": "#FB8C00", "ds": "#8E24AA", "fs": "#607D8B", "_default": "#607D8B", } EVENT_LIBRARY = [ { "event_type": "economy", "label": "Presión por costo de vida", "severity_range": (0.35, 0.75), "scope": "national", "frame": "economy", "target_sector": None, "blame_mode": "incumbent", "opportunity_mode": "opposition", "default_duration": 3, }, { "event_type": "security", "label": "Tensión por seguridad local", "severity_range": (0.30, 0.70), "scope": "territorial", "frame": "security", "target_sector": None, "blame_mode": "mixed", "opportunity_mode": "mixed", "default_duration": 2, }, { "event_type": "services", "label": "Fallo en servicios públicos", "severity_range": (0.25, 0.65), "scope": "territorial", "frame": "services", "target_sector": "Servicios", "blame_mode": "incumbent", "opportunity_mode": "opposition", "default_duration": 3, }, { "event_type": "agriculture", "label": "Presión sobre actividad agrícola", "severity_range": (0.25, 0.65), "scope": "sectoral", "frame": "agriculture", "target_sector": "Agricultura", "blame_mode": "mixed", "opportunity_mode": "rural_parties", "default_duration": 2, }, { "event_type": "tourism", "label": "Impulso o tensión turística", "severity_range": (0.20, 0.60), "scope": "sectoral", "frame": "tourism", "target_sector": "Turismo", "blame_mode": "mixed", "opportunity_mode": "urban_opposition", "default_duration": 2, }, ] PARTY_EVENT_PROFILE = { "n": { "economy": 0.56, "security": 0.74, "services": 0.58, "agriculture": 0.66, "tourism": 0.62, }, "arena": { "economy": 0.72, "security": 0.55, "services": 0.60, "agriculture": 0.48, "tourism": 0.64, }, "fmln": { "economy": 0.46, "security": 0.42, "services": 0.61, "agriculture": 0.58, "tourism": 0.43, }, "_default": { "economy": 0.50, "security": 0.50, "services": 0.50, "agriculture": 0.50, "tourism": 0.50, }, } DEFAULT_TUNING = { "channel_effect_multiplier": 1.15, "vote_update_multiplier": 0.85, "saturation_factor": 0.0000007, "turnout_spend_scale": 0.020, "turnout_competition_scale": 0.028, "event_effect_multiplier": 1.10, "player_focus_multiplier": 1.15, "npc_budget_aggression": 1.00, "manual_order_override_mode": "additive", "event_memory_decay": 0.68, "event_probability": 0.72, } PRESET_PROFILES = { "Balanceado": { "channel_effect_multiplier": 1.15, "vote_update_multiplier": 0.85, "saturation_factor": 0.0000007, "turnout_spend_scale": 0.020, "turnout_competition_scale": 0.028, "event_effect_multiplier": 1.10, "player_focus_multiplier": 1.15, "npc_budget_aggression": 1.00, "manual_order_override_mode": "additive", "event_memory_decay": 0.68, "event_probability": 0.72, }, "Más competitivo": { "channel_effect_multiplier": 1.35, "vote_update_multiplier": 1.00, "saturation_factor": 0.00000055, "turnout_spend_scale": 0.024, "turnout_competition_scale": 0.032, "event_effect_multiplier": 1.20, "player_focus_multiplier": 1.25, "npc_budget_aggression": 1.08, "manual_order_override_mode": "additive", "event_memory_decay": 0.72, "event_probability": 0.78, }, "Volátil": { "channel_effect_multiplier": 1.55, "vote_update_multiplier": 1.18, "saturation_factor": 0.00000045, "turnout_spend_scale": 0.027, "turnout_competition_scale": 0.036, "event_effect_multiplier": 1.35, "player_focus_multiplier": 1.32, "npc_budget_aggression": 1.12, "manual_order_override_mode": "additive", "event_memory_decay": 0.78, "event_probability": 0.82, }, } def clamp(value: float, min_value: float, max_value: float) -> float: return max(min_value, min(max_value, value)) def normalize_text(s: str) -> str: if s is None: return "" return str(s).strip().lower() def softmax_dict(scores: dict[str, float]) -> dict[str, float]: keys = list(scores.keys()) values = np.array([scores[k] for k in keys], dtype=float) values = values - np.max(values) exps = np.exp(values) probs = exps / exps.sum() return {k: float(probs[i]) for i, k in enumerate(keys)} def find_best_column(df: pd.DataFrame, candidates: list[str]) -> str | None: normalized = {normalize_text(c): c for c in df.columns} for cand in candidates: if normalize_text(cand) in normalized: return normalized[normalize_text(cand)] return None def safe_numeric(row: pd.Series, col: str | None, fallback: float) -> float: if not col or col not in row.index: return fallback value = pd.to_numeric(pd.Series([row[col]]), errors="coerce").iloc[0] if pd.isna(value): return fallback return float(value) def get_party_strategy(party_id: str) -> dict: return PARTY_STRATEGY.get(normalize_text(party_id), PARTY_STRATEGY["_default"]) def get_party_color(party_id: str) -> str: return PARTY_COLORS.get(normalize_text(party_id), PARTY_COLORS["_default"]) def get_party_event_strength(party_id: str, frame: str) -> float: profile = PARTY_EVENT_PROFILE.get(normalize_text(party_id), PARTY_EVENT_PROFILE["_default"]) return float(profile.get(frame, 0.50)) def validate_territories(territories: pd.DataFrame, geo: dict) -> dict: geo_ids = set([f["properties"]["id"] for f in geo["features"]]) csv_ids = set(territories["territory_id"]) missing_in_csv = geo_ids - csv_ids missing_in_geo = csv_ids - geo_ids duplicates = territories["territory_id"].duplicated().sum() return { "geo_count": len(geo_ids), "csv_count": len(csv_ids), "missing_in_csv": missing_in_csv, "missing_in_geo": missing_in_geo, "duplicates": int(duplicates), } def estimate_registry(row: pd.Series, aux_cols: dict) -> float: pop = safe_numeric(row, aux_cols["population_col"], np.nan) if pd.isna(pop): return np.nan rural = safe_numeric(row, aux_cols["rural_col"], 50.0) education = safe_numeric(row, aux_cols["education_col"], 6.0) poverty = safe_numeric(row, aux_cols["poverty_col"], 30.0) base_rate = 0.55 rural_penalty = (rural / 100.0) * 0.08 poverty_penalty = (poverty / 100.0) * 0.05 education_boost = (education / 10.0) * 0.05 rate = base_rate - rural_penalty - poverty_penalty + education_boost rate = clamp(rate, 0.45, 0.75) return pop * rate def build_master_territories(territories: pd.DataFrame, indicators: pd.DataFrame) -> tuple[pd.DataFrame, dict]: if "territory_id" not in indicators.columns: raise ValueError("indicators.csv debe tener columna territory_id") population_col = find_best_column(indicators, COLUMN_MAP["population"]) registry_col = find_best_column(indicators, COLUMN_MAP["registry"]) rural_col = find_best_column(indicators, COLUMN_MAP["rural"]) education_col = find_best_column(indicators, COLUMN_MAP["education"]) poverty_col = find_best_column(indicators, COLUMN_MAP["poverty"]) internet_col = find_best_column(indicators, COLUMN_MAP["internet"]) remittances_col = find_best_column(indicators, COLUMN_MAP["remittances"]) merged = territories.merge( indicators, on="territory_id", how="left", suffixes=("", "_ind"), ) diagnostics = { "population_col": population_col, "registry_col": registry_col, "rural_col": rural_col, "education_col": education_col, "poverty_col": poverty_col, "internet_col": internet_col, "remittances_col": remittances_col, "population_non_null": 0, "registry_non_null": 0, } if population_col: merged["weight_population_raw"] = pd.to_numeric(merged[population_col], errors="coerce") diagnostics["population_non_null"] = int(merged["weight_population_raw"].notna().sum()) else: merged["weight_population_raw"] = np.nan aux_cols = { "population_col": population_col, "rural_col": rural_col, "education_col": education_col, "poverty_col": poverty_col, } if registry_col: merged["weight_registry_raw"] = pd.to_numeric(merged[registry_col], errors="coerce") diagnostics["registry_non_null"] = int(merged["weight_registry_raw"].notna().sum()) diagnostics["registry_mode"] = "real_data" else: merged["weight_registry_raw"] = merged.apply(lambda row: estimate_registry(row, aux_cols), axis=1) diagnostics["registry_non_null"] = int(merged["weight_registry_raw"].notna().sum()) diagnostics["registry_col"] = "estimated_from_indicators" diagnostics["registry_mode"] = "estimated_from_indicators" pop_sum = merged["weight_population_raw"].sum(skipna=True) reg_sum = merged["weight_registry_raw"].sum(skipna=True) if pop_sum and pop_sum > 0: merged["weight_population"] = merged["weight_population_raw"] / pop_sum diagnostics["population_mode"] = "real_data" else: merged["weight_population"] = 1 / len(merged) diagnostics["population_mode"] = "uniform_fallback" if reg_sum and reg_sum > 0: merged["weight_registry"] = merged["weight_registry_raw"] / reg_sum if diagnostics["registry_mode"] != "estimated_from_indicators": diagnostics["registry_mode"] = "real_data" else: merged["weight_registry"] = 1 / len(merged) diagnostics["registry_mode"] = "uniform_fallback" return merged, diagnostics def compute_initial_preferences(row: pd.Series, party_ids: list[str], diagnostics: dict) -> dict: poverty = safe_numeric(row, diagnostics["poverty_col"], 30.0) rural = safe_numeric(row, diagnostics["rural_col"], 50.0) education = safe_numeric(row, diagnostics["education_col"], 6.0) internet = safe_numeric(row, diagnostics["internet_col"], 20.0) remittances = safe_numeric(row, diagnostics["remittances_col"], 15.0) prefs = {} for p in party_ids: p_norm = normalize_text(p) if p_norm in {"n", "ni", "nuevas_ideas"}: score = 0.42 + (rural / 100.0) * 0.16 + (poverty / 100.0) * 0.08 + (internet / 100.0) * 0.04 elif p_norm == "arena": score = 0.30 + (education / 10.0) * 0.18 + (internet / 100.0) * 0.06 elif p_norm == "fmln": score = 0.22 + (poverty / 100.0) * 0.18 + (rural / 100.0) * 0.04 elif p_norm in {"gana", "pcn", "pdc"}: score = 0.16 + (rural / 100.0) * 0.08 + (remittances / 100.0) * 0.04 else: score = 0.12 + (education / 10.0) * 0.03 + (internet / 100.0) * 0.03 prefs[p] = max(score, 0.01) return softmax_dict(prefs) def territory_channel_fit(row: pd.Series, channel: str, diagnostics: dict) -> float: rural = safe_numeric(row, diagnostics["rural_col"], 50.0) education = safe_numeric(row, diagnostics["education_col"], 6.0) internet = safe_numeric(row, diagnostics["internet_col"], 20.0) poverty = safe_numeric(row, diagnostics["poverty_col"], 30.0) if channel == "digital": fit = 0.8 + (internet / 100.0) * 0.9 + (education / 10.0) * 0.2 elif channel == "territorial": fit = 0.8 + (rural / 100.0) * 1.0 + (poverty / 100.0) * 0.25 elif channel == "media": fit = 0.8 + ((100.0 - rural) / 100.0) * 0.6 + (internet / 100.0) * 0.1 else: fit = 1.0 return clamp(fit, 0.5, 2.2) def compute_target_score(row: pd.Series, current_votes: dict, party_id: str, diagnostics: dict) -> float: rural = safe_numeric(row, diagnostics["rural_col"], 50.0) education = safe_numeric(row, diagnostics["education_col"], 6.0) poverty = safe_numeric(row, diagnostics["poverty_col"], 30.0) internet = safe_numeric(row, diagnostics["internet_col"], 20.0) current_share = current_votes.get(party_id, 0.0) leader_share = max(current_votes.values()) if current_votes else 0.0 gap = max(0.0, leader_share - current_share) swing_bonus = 1.0 + max(0.0, 0.20 - gap) * 3.5 hold_bonus = 1.0 + current_share * 0.22 p_norm = normalize_text(party_id) if p_norm in {"n", "ni", "nuevas_ideas"}: structural = 1.0 + (rural / 100.0) * 0.30 + (poverty / 100.0) * 0.18 + (internet / 100.0) * 0.10 elif p_norm == "arena": structural = 1.0 + (education / 10.0) * 0.22 + ((100.0 - rural) / 100.0) * 0.18 + (internet / 100.0) * 0.08 elif p_norm == "fmln": structural = 1.0 + (poverty / 100.0) * 0.24 + (rural / 100.0) * 0.10 else: structural = 1.0 + (education / 10.0) * 0.05 return max(0.05, structural * swing_bonus * hold_bonus) def initialize_simulation( master_territories: pd.DataFrame, parties: pd.DataFrame, use_exterior: bool, diagnostics: dict, ) -> list[dict]: party_ids = list(parties["party_id"]) state = [] for _, t in master_territories.iterrows(): state.append( { "territory_id": t["territory_id"], "name": t["name"], "turnout": 0.6, "weight_population": float(t["weight_population"]), "weight_registry": float(t["weight_registry"]), "votes": compute_initial_preferences(t, party_ids, diagnostics), } ) if use_exterior: state.append( { "territory_id": "exterior", "name": "Exterior", "turnout": 0.5, "weight_population": 0.0, "weight_registry": 0.0, "votes": {p: 1 / len(party_ids) for p in party_ids}, } ) return state def get_total_campaign_budget(party_id: str, total_rounds: int) -> float: return float(get_party_strategy(party_id)["budget"] * total_rounds) def build_campaign_allocations( state: list[dict], selected_party_ids: list[str], master_territories: pd.DataFrame, diagnostics: dict, available_cash: dict, tuning: dict, ) -> dict: allocations = {} territory_lookup = master_territories.set_index("territory_id").to_dict("index") npc_aggression = float(tuning["npc_budget_aggression"]) for p in selected_party_ids: strategy = get_party_strategy(p) round_budget_cap = float(strategy["budget"]) * npc_aggression liquid_cash = float(available_cash.get(p, 0.0)) total_budget = min(round_budget_cap, liquid_cash) raw_scores = {} for territory in state: t_id = territory["territory_id"] if t_id == "exterior": raw_scores[t_id] = 0.25 continue row = pd.Series(territory_lookup.get(t_id, {})) raw_scores[t_id] = compute_target_score(row, territory["votes"], p, diagnostics) score_sum = sum(raw_scores.values()) if score_sum <= 0: score_sum = 1.0 allocations[p] = {} for t_id, score in raw_scores.items(): territory_budget = total_budget * (score / score_sum) if t_id == "exterior": allocations[p][t_id] = { "digital": territory_budget * 0.60, "territorial": territory_budget * 0.10, "media": territory_budget * 0.30, } else: allocations[p][t_id] = { "digital": territory_budget * strategy["digital_share"], "territorial": territory_budget * strategy["territorial_share"], "media": territory_budget * strategy["media_share"], } return allocations def compute_channel_party_multiplier(party_id: str, channel: str) -> float: strategy = get_party_strategy(party_id) if channel == "digital": return strategy["digital_bonus"] if channel == "territorial": return strategy["territorial_bonus"] if channel == "media": return strategy["media_bonus"] return 1.0 def estimate_turnout_shift( current_turnout: float, votes: dict, territory_spend_total: float, turnout_spend_scale: float, turnout_competition_scale: float, event_turnout_shift: float = 0.0, ) -> float: if not votes: return current_turnout shares = sorted(votes.values(), reverse=True) competition = 1.0 - (shares[0] - shares[1] if len(shares) > 1 else shares[0]) spend_factor = np.log1p(territory_spend_total / 40000.0) * turnout_spend_scale competition_factor = competition * turnout_competition_scale new_turnout = current_turnout + spend_factor + competition_factor + event_turnout_shift return clamp(new_turnout, 0.45, 0.82) def pick_event_territories(master_territories: pd.DataFrame, event_def: dict, rng: np.random.Generator) -> list[str]: df = master_territories.copy() if event_def["scope"] == "national": return list(df["territory_id"]) if event_def["scope"] == "sectoral" and event_def["target_sector"]: mask = df["cluster"].astype(str).str.lower() == str(event_def["target_sector"]).lower() chosen = df.loc[mask, "territory_id"].tolist() if chosen: return chosen sample_size = int(clamp(rng.integers(4, 10), 4, 10)) choices = df["territory_id"].tolist() if len(choices) <= sample_size: return choices return list(rng.choice(choices, size=sample_size, replace=False)) def build_event_for_round( round_number: int, master_territories: pd.DataFrame, selected_party_ids: list[str], seed: int, tuning: dict, ) -> dict | None: rng = np.random.default_rng(seed + round_number * 97) event_probability = float(tuning["event_probability"]) if rng.random() > event_probability: return None event_def = EVENT_LIBRARY[int(rng.integers(0, len(EVENT_LIBRARY)))] severity = float(rng.uniform(event_def["severity_range"][0], event_def["severity_range"][1])) affected = pick_event_territories(master_territories, event_def, rng) blame_targets = [] opportunity_parties = [] for p in selected_party_ids: p_norm = normalize_text(p) if event_def["blame_mode"] == "incumbent" and p_norm in {"n", "ni", "nuevas_ideas"}: blame_targets.append(p) elif event_def["blame_mode"] == "mixed" and rng.random() < 0.30: blame_targets.append(p) if event_def["opportunity_mode"] == "opposition" and p_norm not in {"n", "ni", "nuevas_ideas"}: opportunity_parties.append(p) elif event_def["opportunity_mode"] == "rural_parties" and p_norm in {"fmln", "pcn", "gana"}: opportunity_parties.append(p) elif event_def["opportunity_mode"] == "urban_opposition" and p_norm in {"arena"}: opportunity_parties.append(p) elif event_def["opportunity_mode"] == "mixed" and rng.random() < 0.30: opportunity_parties.append(p) return { "event_id": f"evt_r{round_number}", "round": round_number, "label": event_def["label"], "event_type": event_def["event_type"], "frame": event_def["frame"], "scope": event_def["scope"], "severity": severity, "base_severity": severity, "remaining_rounds": int(event_def["default_duration"]), "target_sector": event_def["target_sector"], "affected_territories": affected, "blame_targets": blame_targets, "opportunity_parties": opportunity_parties, "turnout_shift": float(np.interp(severity, [0.2, 0.8], [-0.012, 0.018])), } def decay_active_events(active_events: list[dict], decay_factor: float) -> list[dict]: updated = [] for event in active_events: new_event = dict(event) new_event["remaining_rounds"] = int(new_event["remaining_rounds"]) - 1 new_event["severity"] = float(new_event["severity"]) * decay_factor new_event["turnout_shift"] = float(new_event["turnout_shift"]) * decay_factor if new_event["remaining_rounds"] > 0 and new_event["severity"] >= 0.05: updated.append(new_event) return updated def evaluate_event_effect_for_party(party_id: str, event: dict, event_effect_multiplier: float) -> float: frame = event["frame"] severity = float(event["severity"]) issue_strength = get_party_event_strength(party_id, frame) effect = (issue_strength - 0.50) * severity * 0.24 if party_id in event["blame_targets"]: effect -= severity * 0.11 if party_id in event["opportunity_parties"]: effect += severity * 0.08 return effect * event_effect_multiplier def build_event_map(events: list[dict], selected_party_ids: list[str], event_effect_multiplier: float) -> dict: event_map = {} for event in events: for t_id in event["affected_territories"]: if t_id not in event_map: event_map[t_id] = {} for p in selected_party_ids: event_map[t_id][p] = event_map[t_id].get(p, 0.0) + evaluate_event_effect_for_party( p, event, event_effect_multiplier, ) return event_map def get_player_focus_multiplier(player_party: str | None, current_party: str, spend_bundle: dict, tuning: dict) -> float: if player_party != current_party: return 1.0 total_spend = float(sum(spend_bundle.values())) if total_spend <= 0: return 1.0 focus_bonus = np.log1p(total_spend / 12000.0) * 0.10 * float(tuning["player_focus_multiplier"]) return 1.0 + focus_bonus def snapshot_territory_state(state: list[dict], selected_party_ids: list[str]) -> pd.DataFrame: rows = [] for territory in state: if territory["territory_id"] == "exterior": continue leader_party = max(territory["votes"].items(), key=lambda x: x[1])[0] rows.append( { "territory_id": territory["territory_id"], "name": territory["name"], "leader_party": leader_party, "leader_share_pct": territory["votes"][leader_party] * 100, **{f"vote_{p}": territory["votes"].get(p, 0.0) * 100 for p in selected_party_ids}, } ) return pd.DataFrame(rows) def apply_one_round( state: list[dict], allocations: dict, selected_party_ids: list[str], master_territories: pd.DataFrame, diagnostics: dict, tuning: dict, noise_scale: float = 0.0015, round_number: int = 1, cumulative_spend: dict | None = None, seed: int = 1234, player_party: str | None = None, active_events: list[dict] | None = None, ) -> tuple[list[dict], dict, dict | None, list[dict]]: updated_state = [] territory_lookup = master_territories.set_index("territory_id").to_dict("index") if cumulative_spend is None: cumulative_spend = { p: {t["territory_id"]: 0.0 for t in state} for p in selected_party_ids } if active_events is None: active_events = [] decayed_events = decay_active_events( active_events=active_events, decay_factor=float(tuning["event_memory_decay"]), ) new_event = build_event_for_round( round_number=round_number, master_territories=master_territories, selected_party_ids=selected_party_ids, seed=seed, tuning=tuning, ) all_events = list(decayed_events) if new_event is not None: all_events.append(new_event) for territory in state: updated_state.append( { "territory_id": territory["territory_id"], "name": territory["name"], "turnout": territory["turnout"], "weight_population": territory["weight_population"], "weight_registry": territory["weight_registry"], "votes": dict(territory["votes"]), } ) event_map = build_event_map( all_events, selected_party_ids, event_effect_multiplier=float(tuning["event_effect_multiplier"]), ) turnout_event_map = {} for event in all_events: for t_id in event["affected_territories"]: turnout_event_map[t_id] = turnout_event_map.get(t_id, 0.0) + float(event["turnout_shift"]) state_lookup = {t["territory_id"]: t for t in updated_state} rng = np.random.default_rng(seed + round_number * 211) for t_id, territory in state_lookup.items(): if t_id == "exterior": row = pd.Series(dtype=float) else: row = pd.Series(territory_lookup.get(t_id, {})) effect_scores = {} territory_total_spend = 0.0 for p in selected_party_ids: spend_bundle = allocations[p][t_id] total_effect = 0.0 for channel, amount in spend_bundle.items(): territory_total_spend += amount base_effect = CHANNELS[channel]["base_effect"] * float(tuning["channel_effect_multiplier"]) if t_id == "exterior": fit = 1.25 if channel == "digital" else 0.85 else: fit = territory_channel_fit(row, channel, diagnostics) party_bonus = compute_channel_party_multiplier(p, channel) focus_bonus = get_player_focus_multiplier(player_party, p, spend_bundle, tuning) cumulative_spend[p][t_id] += amount saturation = 1.0 / ( 1.0 + float(tuning["saturation_factor"]) * cumulative_spend[p][t_id] ) impact = ( base_effect * np.log1p(amount / 7000.0) * fit * party_bonus * focus_bonus * saturation ) total_effect += impact event_impact = event_map.get(t_id, {}).get(p, 0.0) noise = rng.normal(0, noise_scale) effect_scores[p] = total_effect + event_impact + noise score_space = {} for p in selected_party_ids: current_vote = max(territory["votes"].get(p, 0.0001), 0.0001) score_space[p] = np.log(current_vote + EPS) + effect_scores[p] * float(tuning["vote_update_multiplier"]) turnout_shift = turnout_event_map.get(t_id, 0.0) territory["votes"] = softmax_dict(score_space) territory["turnout"] = estimate_turnout_shift( territory["turnout"], territory["votes"], territory_total_spend, turnout_spend_scale=float(tuning["turnout_spend_scale"]), turnout_competition_scale=float(tuning["turnout_competition_scale"]), event_turnout_shift=turnout_shift, ) return updated_state, cumulative_spend, new_event, all_events def config_panel(parties: pd.DataFrame) -> dict: st.sidebar.header("Configuración") mode = st.sidebar.selectbox( "Modo", ["2026 (default)", "Histórico", "Laboratorio"], ) selected_parties = st.sidebar.multiselect( "Seleccionar partidos", options=list(parties["party_id"]), default=list(parties["party_id"])[:3], max_selections=4, ) use_exterior = st.sidebar.toggle("Activar Exterior", True) base = st.sidebar.selectbox( "Base de ponderación", ["Población", "Registro electoral"], ) rounds = st.sidebar.slider("Número de rondas", 5, 30, 10) noise = st.sidebar.slider("Ruido", 0.0005, 0.0100, 0.0015, 0.0005) return { "mode": mode, "parties": selected_parties, "exterior": use_exterior, "base": base, "rounds": rounds, "noise": noise, } def compute_national_shares(state: list[dict], selected_parties: list[str], base: str) -> pd.DataFrame: rows = [] weight_key = "weight_population" if base == "Población" else "weight_registry" for party_id in selected_parties: total = 0.0 for t in state: total += t["votes"].get(party_id, 0.0) * t.get(weight_key, 0.0) rows.append({"party_id": party_id, "national_share": total}) df = pd.DataFrame(rows) if not df.empty and df["national_share"].sum() > 0: df["national_share_pct"] = df["national_share"] / df["national_share"].sum() * 100 else: df["national_share_pct"] = 0.0 return df.sort_values("national_share_pct", ascending=False) def build_state_preview(state: list[dict]) -> pd.DataFrame: rows = [] for t in state[:10]: rows.append( { "territory": t["name"], "turnout": round(t["turnout"], 4), "weight_population": round(t["weight_population"], 6), "weight_registry": round(t["weight_registry"], 6), } ) return pd.DataFrame(rows) def build_turn_summary(state: list[dict], selected_parties_df: pd.DataFrame, base: str) -> pd.DataFrame: weight_key = "weight_population" if base == "Población" else "weight_registry" rows = [] for territory in state: if territory["territory_id"] == "exterior": continue leader_party = max(territory["votes"].items(), key=lambda x: x[1])[0] leader_share = territory["votes"][leader_party] * 100 rows.append( { "territory": territory["name"], "leader_party": leader_party, "leader_share_pct": leader_share, "weight": territory[weight_key], } ) df = pd.DataFrame(rows) df = df.merge( selected_parties_df[["party_id", "sigla"]], left_on="leader_party", right_on="party_id", how="left", ) df = df.sort_values(["leader_share_pct", "weight"], ascending=[False, False]) return df[["territory", "sigla", "leader_share_pct", "weight"]].rename( columns={ "sigla": "Líder", "leader_share_pct": "% líder", "weight": "Peso", } ) def build_national_delta_df(current_df: pd.DataFrame, previous_df: pd.DataFrame | None, selected_parties_df: pd.DataFrame) -> pd.DataFrame: merged = current_df.merge( selected_parties_df[["party_id", "sigla"]], on="party_id", how="left", ) if previous_df is None or previous_df.empty: merged["delta_pct"] = 0.0 return merged prev = previous_df[["party_id", "national_share_pct"]].rename(columns={"national_share_pct": "prev_pct"}) merged = merged.merge(prev, on="party_id", how="left") merged["prev_pct"] = merged["prev_pct"].fillna(0.0) merged["delta_pct"] = merged["national_share_pct"] - merged["prev_pct"] return merged def build_territory_change_df( current_snapshot: pd.DataFrame, previous_snapshot: pd.DataFrame | None, selected_parties_df: pd.DataFrame, ) -> pd.DataFrame: if previous_snapshot is None or previous_snapshot.empty: base = current_snapshot.copy() base["leader_changed"] = False base["delta_leader_share"] = 0.0 else: prev = previous_snapshot[["territory_id", "leader_party", "leader_share_pct"]].rename( columns={ "leader_party": "prev_leader_party", "leader_share_pct": "prev_leader_share_pct", } ) base = current_snapshot.merge(prev, on="territory_id", how="left") base["leader_changed"] = base["leader_party"] != base["prev_leader_party"] base["delta_leader_share"] = base["leader_share_pct"] - base["prev_leader_share_pct"].fillna(base["leader_share_pct"]) base = base.merge( selected_parties_df[["party_id", "sigla"]], left_on="leader_party", right_on="party_id", how="left", ) return base[["name", "sigla", "leader_share_pct", "delta_leader_share", "leader_changed"]].rename( columns={ "name": "Territorio", "sigla": "Líder", "leader_share_pct": "% líder", "delta_leader_share": "Δ % líder", "leader_changed": "Cambio de líder", } ) def render_candidate_cards(delta_df: pd.DataFrame): cols = st.columns(len(delta_df)) if len(delta_df) > 0 else [] for i, (_, row) in enumerate(delta_df.iterrows()): color = get_party_color(row["party_id"]) pct = float(row["national_share_pct"]) if pd.notna(row["national_share_pct"]) else 0.0 delta = float(row["delta_pct"]) if pd.notna(row["delta_pct"]) else 0.0 delta_sign = "+" if delta >= 0 else "" with cols[i]: st.markdown( f"""