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"""
{str(row["sigla"])[:2]}
Partido
{row["sigla"]}
{pct:.2f}%
Share nacional estimado
{delta_sign}{delta:.2f} pts
""", unsafe_allow_html=True, ) def render_map_placeholder(turn_summary_df: pd.DataFrame, geo: dict): territories_count = len(turn_summary_df) geometry_ready = False try: geometry_ready = any( f.get("geometry", {}).get("type") not in {"GeometryCollection", None} for f in geo.get("features", []) ) except Exception: geometry_ready = False st.markdown( """
""", unsafe_allow_html=True, ) if geometry_ready: st.success("La geometría real ya está disponible para pintar el mapa.") else: st.markdown( """
Tablero territorial
La lógica ya está lista. Falta enchufar polígonos reales para pintar el mapa municipal.
""", unsafe_allow_html=True, ) c1, c2, c3 = st.columns(3) with c1: st.metric("Territorios jugables", territories_count) with c2: st.metric("Features geo", len(geo.get("features", []))) with c3: st.metric("Mapa pintable", "No") st.dataframe(turn_summary_df.head(12), width="stretch", hide_index=True) st.markdown("
", unsafe_allow_html=True) def render_about_tab(): st.markdown("### Descripción general") st.write( "Pulso Electoral SV es un simulador estratégico político-electoral en desarrollo, concebido como un juego de campaña " "territorial y competencia partidaria. El sistema modela la disputa entre partidos a través de rondas, uso de presupuesto, " "ponderación territorial, turnout, liderazgo local, estrategia de campaña y eventos contextuales." ) st.markdown("### Alcance actual") st.write( "La versión actual permite ejecutar simulaciones base con partidos seleccionables, ponderación por población o registro electoral " "estimado, estrategias diferenciadas por partido, rondas de simulación, ruido controlado, turno manual, presupuesto persistente, " "eventos multi-ronda estructurados, memoria de shocks y parámetros tácticos ajustables desde el Centro de mando." ) st.markdown("### Enfoque metodológico") st.write( "La lógica del sistema combina datos territoriales estructurados con reglas de simulación orientadas a un entorno jugable y auditable. " "El objetivo de esta etapa no es producir predicción electoral real, sino construir una base robusta para un simulador político interactivo." ) st.markdown("### Estado de versiones") st.write( f"Versión actual {APP_VERSION}. Estado {APP_STATUS}. Última actualización relevante {LAST_MAJOR_UPDATE}." ) st.markdown("### Advertencia de uso") st.warning( "Este espacio corresponde a un simulador-juego en desarrollo. No constituye encuesta, forecast, proyección oficial ni estimación verificable " "de resultados electorales reales. Toda salida debe interpretarse exclusivamente como resultado interno del modelo de simulación." ) def render_roadmap_tab(): st.markdown("### Hoja de ruta del proyecto") st.write( "La hoja de ruta resume las capacidades implementadas, los módulos en fase de diseño y las siguientes iteraciones previstas." ) completed = pd.DataFrame( [ {"Estado": "Implementado", "Componente": "Conexión del Space al dataset remoto"}, {"Estado": "Implementado", "Componente": "Bootstrap de inicialización"}, {"Estado": "Implementado", "Componente": "Carga estructurada de territorios, partidos e indicadores"}, {"Estado": "Implementado", "Componente": "Validación territorial contra los 44 municipios"}, {"Estado": "Implementado", "Componente": "Ponderación por población"}, {"Estado": "Implementado", "Componente": "Estimación de registro electoral desde indicadores"}, {"Estado": "Implementado", "Componente": "Preferencias iniciales diferenciadas por partido"}, {"Estado": "Implementado", "Componente": "Estrategias base diferenciadas por partido"}, {"Estado": "Implementado", "Componente": "Simulación por rondas con ruido y saturación"}, {"Estado": "Implementado", "Componente": "Presupuesto persistente entre rondas"}, {"Estado": "Implementado", "Componente": "Turno manual con órdenes por territorio"}, {"Estado": "Implementado", "Componente": "Eventos estructurados por ronda"}, {"Estado": "Implementado", "Componente": "Centro de mando con tuning del motor"}, {"Estado": "Implementado", "Componente": "Eventos multi-ronda con decaimiento"}, {"Estado": "Implementado", "Componente": "Deltas nacionales visibles por ronda"}, {"Estado": "Implementado", "Componente": "Tabla de movimiento territorial"}, ] ) in_progress = pd.DataFrame( [ {"Estado": "En diseño", "Componente": "NPC reactivos a la conducta del jugador"}, {"Estado": "En diseño", "Componente": "Cierre de ronda con comparación antes y después"}, {"Estado": "En diseño", "Componente": "Métricas por territorio para partido controlado"}, ] ) pending = pd.DataFrame( [ {"Estado": "Pendiente", "Componente": "Mapa municipal real con geometría utilizable"}, {"Estado": "Pendiente", "Componente": "Tablero clickeable estilo juego de estrategia"}, {"Estado": "Pendiente", "Componente": "NPC adaptativos con sofisticación táctica configurable"}, {"Estado": "Pendiente", "Componente": "Panel de control para edición de partidos y atributos"}, {"Estado": "Pendiente", "Componente": "Persistencia de partidas y escenarios"}, {"Estado": "Pendiente", "Componente": "Eventos encadenados con narrativa estructurada"}, ] ) st.markdown("#### Capacidades implementadas") st.dataframe(completed, width="stretch", hide_index=True) st.markdown("#### Componentes en diseño") st.dataframe(in_progress, width="stretch", hide_index=True) st.markdown("#### Próximas capacidades") st.dataframe(pending, width="stretch", hide_index=True) def render_versions_tab(): st.markdown("### Versionado del simulador") st.write( "El sistema utiliza versionado funcional para distinguir cambios de motor, interfaz y alcance metodológico. " "Este esquema permite documentar el estado del proyecto y mantener trazabilidad entre iteraciones." ) versions = pd.DataFrame( [ {"Versión": "v0.1.0", "Estado": "Base inicial", "Contenido": "Conexión al dataset, lectura estructurada y validaciones básicas"}, {"Versión": "v0.2.0", "Estado": "Motor base", "Contenido": "Pesos territoriales, preferencias iniciales y simulación por rondas"}, {"Versión": "v0.3.0", "Estado": "Motor estable", "Contenido": "Estrategias por partido, ruido controlado, saturación y resumen nacional"}, {"Versión": "v0.4.0-alpha", "Estado": "UI funcional", "Contenido": "Tabs informativas, tablero principal, hoja de ruta y presentación formal"}, {"Versión": "v0.5.0-alpha", "Estado": "Juego básico", "Contenido": "Turno manual, órdenes por territorio y control inicial de ronda"}, {"Versión": "v0.6.0-alpha", "Estado": "Eventos iniciales", "Contenido": "Eventos estructurados por ronda e historial visible"}, {"Versión": "v0.7.0-alpha", "Estado": "Elasticidad", "Contenido": "Presupuesto persistente y mayor sensibilidad del motor"}, {"Versión": "v0.8.0-alpha", "Estado": "Tuning táctico", "Contenido": "Centro de mando con parámetros visibles"}, {"Versión": "v0.9.0-beta", "Estado": "Iteración actual", "Contenido": "Eventos multi-ronda, memoria de shocks, deltas nacionales y movimiento territorial"}, {"Versión": "v1.0.0", "Estado": "Objetivo siguiente", "Contenido": "NPC reactivos, tablero territorial clickeable y sensación completa de campaña"}, ] ) st.dataframe(versions, width="stretch", hide_index=True) st.info( "Cambios de interfaz, documentación y flujo de juego incrementan la subversión funcional. Cambios de motor, reglas y simulación incrementan la versión principal de desarrollo." ) def build_orders_dataframe(orders: list[dict]) -> pd.DataFrame: if not orders: return pd.DataFrame(columns=["Partido", "Territorio", "Canal", "Monto"]) return pd.DataFrame(orders) def sum_orders_for_party(orders: list[dict], party_id: str) -> float: return float(sum(o["Monto"] for o in orders if o["Partido"] == party_id)) def render_event_card(event: dict | None): st.markdown("### Evento de la ronda") if event is None: st.info("En esta ronda no se activó un evento nuevo. Pueden seguir activos shocks previos.") return sev_pct = int(round(event["base_severity"] * 100)) affected_count = len(event["affected_territories"]) blame = ", ".join(event["blame_targets"]) if event["blame_targets"] else "Ninguno" opp = ", ".join(event["opportunity_parties"]) if event["opportunity_parties"] else "Ninguno" st.markdown( f"""
{event["label"]}
Tipo {event["event_type"]} | Alcance {event["scope"]} | Severidad inicial {sev_pct}%
Territorios afectados {affected_count}
Partidos más expuestos {blame}
Partidos con oportunidad {opp}
Duración base {event["remaining_rounds"]} rondas
""", unsafe_allow_html=True, ) def render_active_shocks(): st.markdown("### Shocks activos") if not st.session_state.sim_active_events: st.info("No hay shocks activos acumulados.") return rows = [] for event in st.session_state.sim_active_events: rows.append( { "Evento": event["label"], "Tipo": event["event_type"], "Severidad vigente": round(event["severity"], 3), "Rondas restantes": event["remaining_rounds"], "Territorios": len(event["affected_territories"]), } ) st.dataframe(pd.DataFrame(rows), width="stretch", hide_index=True) def render_event_history(): st.markdown("### Historial de rondas") if not st.session_state.sim_event_history: st.info("Todavía no hay rondas cerradas en esta partida.") return rows = [] for item in st.session_state.sim_event_history: event = item["event"] rows.append( { "Ronda": item["round"], "Evento": event["label"] if event else "Sin evento nuevo", "Tipo": event["event_type"] if event else "-", "Severidad": round(event["base_severity"], 3) if event else 0.0, "Territorios afectados": len(event["affected_territories"]) if event else 0, } ) st.dataframe(pd.DataFrame(rows), width="stretch", hide_index=True) def render_round_summary(): st.markdown("### Resumen de gasto por ronda") if not st.session_state.sim_round_summary: st.info("Todavía no hay gasto registrado en rondas cerradas.") return rows = [] for item in st.session_state.sim_round_summary: for party_id, spent in item["spent"].items(): rows.append( { "Ronda": item["round"], "Partido": party_id, "Gastado": round(spent, 2), } ) st.dataframe(pd.DataFrame(rows), width="stretch", hide_index=True) def ensure_session_defaults(): defaults = { "sim_initialized": False, "sim_round": 1, "sim_state": None, "sim_orders": [], "player_party": None, "sim_cumulative_spend": None, "sim_last_key": None, "sim_finished": False, "sim_seed": 1234, "sim_event_history": [], "sim_active_events": [], "master_territories": None, "diagnostics": None, "current_config": None, "sim_party_cash": None, "sim_round_summary": [], "sim_tuning": DEFAULT_TUNING.copy(), "sim_previous_national_df": None, "sim_previous_snapshot": None, "sim_last_territory_changes": None, "sim_last_national_delta": None, "sim_profile_name": "Balanceado", } for key, value in defaults.items(): if key not in st.session_state: st.session_state[key] = value.copy() if isinstance(value, dict) else value def reset_simulation_session(config_key): st.session_state.sim_initialized = False st.session_state.sim_round = 1 st.session_state.sim_state = None st.session_state.sim_orders = [] st.session_state.player_party = None st.session_state.sim_cumulative_spend = None st.session_state.sim_finished = False st.session_state.sim_last_key = config_key st.session_state.sim_seed = 1234 st.session_state.sim_event_history = [] st.session_state.sim_active_events = [] st.session_state.master_territories = None st.session_state.diagnostics = None st.session_state.current_config = None st.session_state.sim_party_cash = None st.session_state.sim_round_summary = [] st.session_state.sim_tuning = DEFAULT_TUNING.copy() st.session_state.sim_previous_national_df = None st.session_state.sim_previous_snapshot = None st.session_state.sim_last_territory_changes = None st.session_state.sim_last_national_delta = None st.session_state.sim_profile_name = "Balanceado" def load_profile_into_tuning(profile_name: str): st.session_state.sim_tuning = PRESET_PROFILES[profile_name].copy() st.session_state.sim_profile_name = profile_name def render_tuning_panel(): st.markdown("#### Ajustes tácticos del motor") profile_name = st.selectbox( "Perfil táctico", options=list(PRESET_PROFILES.keys()), index=list(PRESET_PROFILES.keys()).index(st.session_state.sim_profile_name), ) if profile_name != st.session_state.sim_profile_name: load_profile_into_tuning(profile_name) st.rerun() t1, t2 = st.columns(2) with t1: st.session_state.sim_tuning["channel_effect_multiplier"] = st.slider( "Potencia del gasto", min_value=0.50, max_value=2.50, value=float(st.session_state.sim_tuning["channel_effect_multiplier"]), step=0.05, ) st.session_state.sim_tuning["vote_update_multiplier"] = st.slider( "Elasticidad del voto", min_value=0.20, max_value=1.60, value=float(st.session_state.sim_tuning["vote_update_multiplier"]), step=0.05, ) st.session_state.sim_tuning["event_effect_multiplier"] = st.slider( "Impacto de eventos", min_value=0.25, max_value=2.50, value=float(st.session_state.sim_tuning["event_effect_multiplier"]), step=0.05, ) st.session_state.sim_tuning["npc_budget_aggression"] = st.slider( "Agresividad presupuestaria NPC", min_value=0.40, max_value=1.50, value=float(st.session_state.sim_tuning["npc_budget_aggression"]), step=0.05, ) st.session_state.sim_tuning["event_probability"] = st.slider( "Probabilidad de evento nuevo", min_value=0.10, max_value=1.00, value=float(st.session_state.sim_tuning["event_probability"]), step=0.02, ) with t2: st.session_state.sim_tuning["player_focus_multiplier"] = st.slider( "Ventaja por concentración del jugador", min_value=0.50, max_value=2.50, value=float(st.session_state.sim_tuning["player_focus_multiplier"]), step=0.05, ) st.session_state.sim_tuning["saturation_factor"] = st.slider( "Fatiga por saturación", min_value=0.0000002, max_value=0.0000030, value=float(st.session_state.sim_tuning["saturation_factor"]), step=0.0000001, format="%.7f", ) st.session_state.sim_tuning["turnout_spend_scale"] = st.slider( "Efecto del gasto sobre turnout", min_value=0.005, max_value=0.050, value=float(st.session_state.sim_tuning["turnout_spend_scale"]), step=0.001, ) st.session_state.sim_tuning["turnout_competition_scale"] = st.slider( "Efecto de competencia sobre turnout", min_value=0.005, max_value=0.060, value=float(st.session_state.sim_tuning["turnout_competition_scale"]), step=0.001, ) st.session_state.sim_tuning["event_memory_decay"] = st.slider( "Persistencia de shocks", min_value=0.40, max_value=0.95, value=float(st.session_state.sim_tuning["event_memory_decay"]), step=0.01, ) st.session_state.sim_tuning["manual_order_override_mode"] = st.selectbox( "Modo de órdenes manuales", options=["additive", "override"], index=0 if st.session_state.sim_tuning["manual_order_override_mode"] == "additive" else 1, help="additive suma tus órdenes a la estrategia base del partido. override reemplaza completamente la asignación automática del partido que controlas.", ) if st.button("Restablecer ajustes tácticos"): st.session_state.sim_tuning = DEFAULT_TUNING.copy() st.session_state.sim_profile_name = "Balanceado" st.rerun() def apply_manual_orders_to_allocations( base_allocations: dict, manual_orders: list[dict], player_party: str, state: list[dict], tuning: dict, ): if player_party not in base_allocations: return base_allocations mode = tuning["manual_order_override_mode"] if mode == "override": for territory in state: t_id = territory["territory_id"] base_allocations[player_party][t_id] = { "digital": 0.0, "territorial": 0.0, "media": 0.0, } for order in manual_orders: if order["Partido"] != player_party: continue t_id = order["territory_id"] ch = order["channel"] amount = float(order["Monto"]) if t_id in base_allocations[player_party]: base_allocations[player_party][t_id][ch] += amount return base_allocations def render_control_panel( state: list[dict], parties_df: pd.DataFrame, selected_party_ids: list[str], total_rounds: int, ): st.markdown("### Centro de mando") if st.session_state.player_party not in selected_party_ids: st.session_state.player_party = selected_party_ids[0] player_party = st.selectbox( "Partido controlado por el jugador", options=selected_party_ids, index=selected_party_ids.index(st.session_state.player_party), format_func=lambda x: parties_df.loc[parties_df["party_id"] == x, "sigla"].iloc[0] if not parties_df.loc[parties_df["party_id"] == x].empty else x, key="player_party_selector", ) st.session_state.player_party = player_party round_budget_cap = float(get_party_strategy(player_party)["budget"]) total_cash_left = float(st.session_state.sim_party_cash.get(player_party, 0.0)) effective_round_cap = min(round_budget_cap, total_cash_left) spent_this_round = sum_orders_for_party(st.session_state.sim_orders, player_party) remaining_this_round = max(0.0, effective_round_cap - spent_this_round) m1, m2, m3, m4 = st.columns(4) with m1: st.metric("Ronda actual", f"{st.session_state.sim_round} / {total_rounds}") with m2: st.metric("Tope por ronda", f"${round_budget_cap:,.0f}") with m3: st.metric("Caja restante", f"${total_cash_left:,.0f}") with m4: st.metric("Disponible esta ronda", f"${remaining_this_round:,.0f}") control_tab, tuning_tab = st.tabs(["Órdenes", "Ajustes tácticos"]) territory_options = [t["territory_id"] for t in state if t["territory_id"] != "exterior"] territory_names = {t["territory_id"]: t["name"] for t in state} with control_tab: c1, c2 = st.columns([1.35, 1.0]) with c1: with st.form("manual_order_form", clear_on_submit=True): territory_selected = st.selectbox( "Territorio objetivo", options=territory_options, format_func=lambda x: territory_names.get(x, x), ) channel_selected = st.selectbox( "Canal", options=list(CHANNELS.keys()), format_func=lambda x: CHANNELS[x]["label"], ) amount_selected = st.number_input( "Monto a asignar", min_value=0.0, max_value=float(remaining_this_round), value=min(10000.0, float(remaining_this_round)), step=1000.0, ) submitted = st.form_submit_button("Agregar jugada") if submitted: if remaining_this_round <= 0: st.warning("Ese partido ya no tiene presupuesto utilizable en esta ronda.") elif amount_selected <= 0: st.warning("Ingresa un monto mayor que cero.") else: st.session_state.sim_orders.append( { "Partido": player_party, "Territorio": territory_names[territory_selected], "territory_id": territory_selected, "Canal": CHANNELS[channel_selected]["label"], "channel": channel_selected, "Monto": float(amount_selected), } ) st.success("Jugada agregada a la ronda actual.") with c2: st.markdown("#### Estado del partido controlado") st.write(f"**Partido** {player_party}") st.write(f"**Caja total restante** ${total_cash_left:,.0f}") st.write(f"**Gasto ya ordenado esta ronda** ${spent_this_round:,.0f}") st.write(f"**Modo manual** {st.session_state.sim_tuning['manual_order_override_mode']}") st.write(f"**Perfil táctico** {st.session_state.sim_profile_name}") st.markdown("#### Órdenes cargadas") st.dataframe(build_orders_dataframe(st.session_state.sim_orders), width="stretch", hide_index=True) clear_col, close_col = st.columns(2) with clear_col: if st.button("Limpiar órdenes de la ronda"): st.session_state.sim_orders = [] st.rerun() with close_col: if st.button("Cerrar ronda y resolver turno"): if st.session_state.sim_finished: st.warning("La simulación ya terminó.") else: pre_national_df = compute_national_shares( state=st.session_state.sim_state, selected_parties=selected_party_ids, base=st.session_state.current_config["base"], ) pre_snapshot = snapshot_territory_state( st.session_state.sim_state, selected_party_ids, ) heuristic_allocations = build_campaign_allocations( state=st.session_state.sim_state, selected_party_ids=selected_party_ids, master_territories=st.session_state.master_territories, diagnostics=st.session_state.diagnostics, available_cash=st.session_state.sim_party_cash, tuning=st.session_state.sim_tuning, ) final_allocations = apply_manual_orders_to_allocations( heuristic_allocations, st.session_state.sim_orders, player_party, st.session_state.sim_state, st.session_state.sim_tuning, ) spent_map = {} for p in selected_party_ids: party_total_spent = 0.0 for t_id in final_allocations[p]: party_total_spent += sum(final_allocations[p][t_id].values()) spent_map[p] = round(party_total_spent, 2) st.session_state.sim_party_cash[p] = max( 0.0, float(st.session_state.sim_party_cash.get(p, 0.0)) - party_total_spent ) new_state, new_cumulative, round_event, new_active_events = apply_one_round( state=st.session_state.sim_state, allocations=final_allocations, selected_party_ids=selected_party_ids, master_territories=st.session_state.master_territories, diagnostics=st.session_state.diagnostics, tuning=st.session_state.sim_tuning, noise_scale=st.session_state.current_config["noise"], round_number=st.session_state.sim_round, cumulative_spend=st.session_state.sim_cumulative_spend, seed=st.session_state.sim_seed, player_party=player_party, active_events=st.session_state.sim_active_events, ) post_national_df = compute_national_shares( state=new_state, selected_parties=selected_party_ids, base=st.session_state.current_config["base"], ) post_snapshot = snapshot_territory_state( new_state, selected_party_ids, ) st.session_state.sim_last_national_delta = build_national_delta_df( current_df=post_national_df, previous_df=pre_national_df, selected_parties_df=parties_df[parties_df["party_id"].isin(selected_party_ids)].copy(), ) st.session_state.sim_last_territory_changes = build_territory_change_df( current_snapshot=post_snapshot, previous_snapshot=pre_snapshot, selected_parties_df=parties_df[parties_df["party_id"].isin(selected_party_ids)].copy(), ) st.session_state.sim_previous_national_df = pre_national_df st.session_state.sim_previous_snapshot = pre_snapshot st.session_state.sim_event_history.append( { "round": st.session_state.sim_round, "event": round_event, } ) st.session_state.sim_active_events = new_active_events st.session_state.sim_round_summary.append( { "round": st.session_state.sim_round, "spent": spent_map, } ) st.session_state.sim_state = new_state st.session_state.sim_cumulative_spend = new_cumulative st.session_state.sim_orders = [] if st.session_state.sim_round >= total_rounds: st.session_state.sim_finished = True else: st.session_state.sim_round += 1 st.rerun() with tuning_tab: render_tuning_panel() def render_game_state( selected_parties_df: pd.DataFrame, state: list[dict], geo: dict, base: str, ): national_df = compute_national_shares( state=state, selected_parties=list(selected_parties_df["party_id"]), base=base, ) delta_df = build_national_delta_df( current_df=national_df, previous_df=st.session_state.sim_previous_national_df, selected_parties_df=selected_parties_df, ) turn_summary_df = build_turn_summary( state=state, selected_parties_df=selected_parties_df, base=base, ) st.subheader("Resumen nacional") render_candidate_cards(delta_df) st.subheader("Tablero principal") c_map, c_side = st.columns([1.7, 1.0], gap="large") with c_map: render_map_placeholder(turn_summary_df, geo) with c_side: latest_event = st.session_state.sim_event_history[-1]["event"] if st.session_state.sim_event_history else None render_event_card(latest_event) st.markdown("### Panel de turno") st.dataframe(build_state_preview(state), width="stretch", hide_index=True) st.markdown("### Estrategia por partido") strategy_rows = [] for p in selected_parties_df["party_id"]: strat = get_party_strategy(p) cash_left = float(st.session_state.sim_party_cash.get(p, 0.0)) strategy_rows.append( { "party_id": p, "budget_round": strat["budget"], "cash_left": round(cash_left, 2), "digital_share": strat["digital_share"], "territorial_share": strat["territorial_share"], "media_share": strat["media_share"], } ) st.dataframe(pd.DataFrame(strategy_rows), width="stretch", hide_index=True) st.subheader("Marcador nacional") st.dataframe( delta_df[["sigla", "national_share_pct", "delta_pct"]].rename( columns={ "sigla": "Partido", "national_share_pct": "% nacional estimado", "delta_pct": "Δ ronda", } ), width="stretch", hide_index=True, ) render_active_shocks() render_event_history() render_round_summary() st.subheader("Movimiento territorial") if st.session_state.sim_last_territory_changes is None: st.info("Todavía no hay cambios territoriales comparables. Cierra al menos una ronda.") else: st.dataframe( st.session_state.sim_last_territory_changes.sort_values( ["Cambio de líder", "Δ % líder", "% líder"], ascending=[False, False, False], ), width="stretch", hide_index=True, ) def run_game_tab(): ensure_session_defaults() territories = load_territories() parties = load_parties() indicators = load_indicators() geo = load_geo_44() validation = validate_territories(territories, geo) master_territories, diagnostics = build_master_territories(territories, indicators) config = config_panel(parties) config_key = ( config["mode"], tuple(config["parties"]), config["exterior"], config["base"], config["rounds"], config["noise"], ) if st.session_state.sim_last_key != config_key: reset_simulation_session(config_key) st.subheader("Validación territorial") c1, c2, c3 = st.columns(3) with c1: st.metric("Geo municipios", validation["geo_count"]) with c2: st.metric("CSV territorios", validation["csv_count"]) with c3: st.metric("Duplicados", validation["duplicates"]) if validation["missing_in_csv"]: st.error(f"Faltan en CSV {list(validation['missing_in_csv'])}") if validation["missing_in_geo"]: st.warning(f"No están en GEO {list(validation['missing_in_geo'])}") if validation["csv_count"] != EXPECTED_MUNICIPALITIES: st.warning(f"Territorios esperados {EXPECTED_MUNICIPALITIES}") st.divider() st.subheader("Configuración seleccionada") st.json(config) st.subheader("Diagnóstico de pesos") d1, d2 = st.columns(2) with d1: st.write("**Columna detectada para población**") st.code(str(diagnostics["population_col"])) st.write(f"Valores no nulos {diagnostics['population_non_null']}") st.write(f"Modo {diagnostics['population_mode']}") with d2: st.write("**Columna detectada para registro electoral**") st.code(str(diagnostics["registry_col"])) st.write(f"Valores no nulos {diagnostics['registry_non_null']}") st.write(f"Modo {diagnostics['registry_mode']}") if diagnostics["population_mode"] == "uniform_fallback": st.warning("No hay datos numéricos de población utilizables en indicators.csv. Se está usando ponderación uniforme.") if config["base"] == "Registro electoral" and diagnostics["registry_mode"] == "uniform_fallback": st.warning("No hay datos numéricos de registro electoral utilizables en indicators.csv. Se está usando ponderación uniforme.") if diagnostics["registry_mode"] == "estimated_from_indicators": st.info("El registro electoral está siendo estimado a partir de población, ruralidad, educación y pobreza.") if not config["parties"]: st.warning("Selecciona al menos un partido para continuar.") return selected_parties_df = parties[parties["party_id"].isin(config["parties"])].copy() if not st.session_state.sim_initialized: if st.button("Iniciar simulación"): st.session_state.master_territories = master_territories st.session_state.diagnostics = diagnostics st.session_state.current_config = config st.session_state.sim_seed = 1234 st.session_state.sim_state = initialize_simulation( master_territories=master_territories, parties=selected_parties_df, use_exterior=config["exterior"], diagnostics=diagnostics, ) st.session_state.sim_cumulative_spend = { p: {t["territory_id"]: 0.0 for t in st.session_state.sim_state} for p in list(selected_parties_df["party_id"]) } st.session_state.sim_party_cash = { p: get_total_campaign_budget(p, config["rounds"]) for p in list(selected_parties_df["party_id"]) } st.session_state.sim_initialized = True st.session_state.sim_round = 1 st.session_state.sim_orders = [] st.session_state.player_party = list(selected_parties_df["party_id"])[0] st.session_state.sim_event_history = [] st.session_state.sim_active_events = [] st.session_state.sim_round_summary = [] st.session_state.sim_previous_national_df = None st.session_state.sim_previous_snapshot = None st.session_state.sim_last_territory_changes = None st.session_state.sim_last_national_delta = None st.success("Simulación inicializada.") st.rerun() return st.success("Simulación activa.") if st.session_state.sim_finished: st.info("La simulación terminó. Puedes cambiar configuración para reiniciar una nueva partida.") render_game_state( selected_parties_df=selected_parties_df, state=st.session_state.sim_state, geo=geo, base=config["base"], ) st.divider() render_control_panel( state=st.session_state.sim_state, parties_df=selected_parties_df, selected_party_ids=list(selected_parties_df["party_id"]), total_rounds=config["rounds"], ) st.info( "La siguiente iteración buena ya es NPC reactivo, tablero territorial clickeable y resolución visual de antes y después por ronda." ) def run_app(): st.title("Pulso Electoral SV") st.caption( f"Simulador-juego político-electoral en desarrollo. Versión {APP_VERSION}. " f"Esta versión corresponde a una base jugable experimental y no representa una proyección oficial." ) tab_game, tab_about, tab_roadmap, tab_versions = st.tabs( [ "Simulador", "Acerca del juego", "Hoja de ruta", "Versiones", ] ) with tab_game: run_game_tab() with tab_about: render_about_tab() with tab_roadmap: render_roadmap_tab() with tab_versions: render_versions_tab()