Spaces:
Runtime error
Runtime error
| 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""" | |
| <div style=" | |
| border: 2px solid {color}; | |
| border-radius: 16px; | |
| padding: 14px; | |
| background: linear-gradient(180deg, #ffffff 0%, #f7f9fc 100%); | |
| box-shadow: 0 4px 12px rgba(0,0,0,0.05); | |
| min-height: 178px; | |
| "> | |
| <div style="display:flex; align-items:center; gap:10px; margin-bottom:10px;"> | |
| <div style=" | |
| width:44px; | |
| height:44px; | |
| border-radius:50%; | |
| background:{color}; | |
| color:white; | |
| display:flex; | |
| align-items:center; | |
| justify-content:center; | |
| font-weight:700; | |
| font-size:16px; | |
| "> | |
| {str(row["sigla"])[:2]} | |
| </div> | |
| <div> | |
| <div style="font-size:14px; color:#6b7280;">Partido</div> | |
| <div style="font-size:20px; font-weight:700;">{row["sigla"]}</div> | |
| </div> | |
| </div> | |
| <div style="font-size:32px; font-weight:800; color:{color}; line-height:1;"> | |
| {pct:.2f}% | |
| </div> | |
| <div style="font-size:13px; color:#6b7280; margin-top:6px;"> | |
| Share nacional estimado | |
| </div> | |
| <div style="margin-top:10px; font-size:14px; font-weight:700; color:{color};"> | |
| {delta_sign}{delta:.2f} pts | |
| </div> | |
| </div> | |
| """, | |
| 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( | |
| """ | |
| <div style=" | |
| border: 2px dashed #cbd5e1; | |
| border-radius: 18px; | |
| background: linear-gradient(180deg, #f8fbff 0%, #eef4fb 100%); | |
| padding: 24px; | |
| min-height: 340px; | |
| "> | |
| """, | |
| unsafe_allow_html=True, | |
| ) | |
| if geometry_ready: | |
| st.success("La geometría real ya está disponible para pintar el mapa.") | |
| else: | |
| st.markdown( | |
| """ | |
| <div style="font-size:20px; font-weight:700; margin-bottom:6px;"> | |
| Tablero territorial | |
| </div> | |
| <div style="color:#64748b; margin-bottom:18px;"> | |
| La lógica ya está lista. Falta enchufar polígonos reales para pintar el mapa municipal. | |
| </div> | |
| """, | |
| 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("</div>", 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""" | |
| <div style=" | |
| border: 1px solid #dbeafe; | |
| border-radius: 14px; | |
| padding: 16px; | |
| background: linear-gradient(180deg, #f8fbff 0%, #eef6ff 100%); | |
| "> | |
| <div style="font-size:18px; font-weight:700;">{event["label"]}</div> | |
| <div style="font-size:13px; color:#64748b; margin-top:4px;"> | |
| Tipo {event["event_type"]} | Alcance {event["scope"]} | Severidad inicial {sev_pct}% | |
| </div> | |
| <div style="margin-top:10px; font-size:14px;"> | |
| Territorios afectados {affected_count} | |
| </div> | |
| <div style="margin-top:6px; font-size:14px;"> | |
| Partidos más expuestos {blame} | |
| </div> | |
| <div style="margin-top:6px; font-size:14px;"> | |
| Partidos con oportunidad {opp} | |
| </div> | |
| <div style="margin-top:6px; font-size:14px;"> | |
| Duración base {event["remaining_rounds"]} rondas | |
| </div> | |
| </div> | |
| """, | |
| 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() | |