Fix historical data identity and weekly ranking quality

ai/tasks/done/TASK-032-historical-data-quality-and-ranking-validation.md

@@ -0,0 +1,119 @@
+# TASK-032-historical-data-quality-and-ranking-validation
+
+## Goal
+Validar la calidad del historico ingerido y la correccion del ranking semanal de kills para los 2 servidores reales de la comunidad, detectando y corrigiendo problemas de integridad, duplicados, naming o rango temporal antes de construir UI historica propia.
+
+## Context
+El proyecto ya dispone de:
+- estado actual en vivo via A2S
+- capa historica separada via CRCON scoreboard JSON
+- persistencia historica propia
+- ingesta historica funcional
+- endpoint `GET /api/historical/weekly-top-kills`
+
+Antes de exponer estos datos en una UI propia del proyecto, hay que asegurar que la base historica tiene calidad suficiente y que el ranking semanal devuelve resultados coherentes, consistentes y trazables.
+
+## Steps
+1. Revisar la implementacion actual de:
+   - almacenamiento historico
+   - ingesta historica
+   - modelos historicos
+   - endpoint `GET /api/historical/weekly-top-kills`
+2. Verificar la calidad de los datos historicos persistidos para ambos servidores. Comprobar al menos:
+   - numero de partidas ingeridas
+   - numero de jugadores ingeridos
+   - presencia de datos relevantes por partida
+   - distribucion por servidor
+3. Detectar posibles duplicados o inconsistencias en:
+   - partidas
+   - jugadores
+   - estadisticas por jugador y partida
+4. Revisar la estrategia actual de identidad y deduplicacion:
+   - ids de partida
+   - ids de jugador o claves degradadas
+   - relacion entre servidor y match
+5. Validar que el calculo de "ultima semana" usado por el ranking sea correcto y consistente.
+6. Validar que el ranking de kills:
+   - sume correctamente kills por jugador
+   - no mezcle datos entre servidores
+   - no use partidas fuera del rango temporal esperado
+   - no devuelva duplicados de jugador por mala consolidacion
+7. Revisar si los nombres de jugador y nombres de mapa se almacenan y devuelven de forma suficientemente limpia.
+8. Si se detectan problemas, corregir unicamente lo necesario en:
+   - almacenamiento
+   - consultas
+   - normalizacion
+   - endpoint historico
+9. Anadir validaciones o utilidades minimas si ayudan a reforzar la fiabilidad del historico.
+10. Documentar brevemente los hallazgos y el estado final de calidad historica.
+11. No crear todavia paginas o bloques UI historicos.
+12. Al completar la implementacion:
+   - dejar el repositorio consistente
+   - hacer commit
+   - hacer push al remoto si el entorno lo permite
+
+## Files to Read First
+- AGENTS.md
+- ai/repo-context.md
+- ai/architecture-index.md
+- docs/decisions.md
+- docs/historical-crcon-source-discovery.md
+- docs/historical-domain-model.md
+- backend/README.md
+- backend/app/historical_models.py
+- backend/app/historical_storage.py
+- backend/app/historical_ingestion.py
+- backend/app/routes.py
+- backend/app/payloads.py
+- cualquier consulta o helper historico adicional ya creado
+
+## Expected Files to Modify
+- backend/README.md
+- backend/app/historical_storage.py
+- backend/app/historical_ingestion.py
+- backend/app/routes.py
+- backend/app/payloads.py
+- opcionalmente nuevos modulos auxiliares si mejoran validacion o normalizacion, por ejemplo:
+  - backend/app/historical_queries.py
+  - backend/app/historical_validation.py
+- opcionalmente un documento tecnico breve, por ejemplo:
+  - docs/historical-data-quality-notes.md
+
+## Constraints
+- No crear todavia UI historica.
+- No basar correcciones historicas en A2S.
+- No acoplar el frontend a URLs de la comunidad.
+- No introducir complejidad innecesaria.
+- No romper el flujo actual de live status.
+- No hacer cambios destructivos.
+- Mantener el trabajo centrado en calidad, consistencia y fiabilidad del historico.
+
+## Validation
+- Se ha revisado la calidad del historico para ambos servidores.
+- Se han detectado y corregido, si existen, duplicados o inconsistencias relevantes.
+- El endpoint `GET /api/historical/weekly-top-kills` devuelve resultados coherentes.
+- El rango temporal de "ultima semana" queda validado.
+- Los datos no se mezclan entre servidores.
+- Los cambios quedan committeados y se hace push al remoto si el entorno lo permite.
+
+## Change Budget
+- Preferir menos de 7 archivos modificados o creados.
+- Preferir menos de 260 lineas cambiadas.
+
+## Outcome
+- Se corrigio la estrategia de identidad historica en `backend/app/historical_storage.py` para priorizar SteamID real y usar `player_id` como clave `crcon-player:*` cuando no existe SteamID.
+- La inicializacion del storage ahora fusiona jugadores duplicados y partidas duplicadas persistidas con id sintetico frente a id CRCON final.
+- El ranking `GET /api/historical/weekly-top-kills` paso a usar solo partidas cerradas con `ended_at`, evitando contar sesiones en curso o duplicadas.
+- Se documentaron los hallazgos y el estado final en `docs/historical-data-quality-notes.md` y se alineo `backend/README.md`.
+
+## Validation Result
+- Validado con inicializacion real sobre `backend/data/hll_vietnam_dev.sqlite3`.
+- Tras la correccion, el dataset local quedo en `12` partidas, `510` jugadores y `914` filas de estadisticas por jugador y partida.
+- Comprobado que no quedan duplicados por `steam_id`, `source_player_id`, nombre normalizado ni por la combinacion `(servidor, started_at, mapa)`.
+- Comprobado que ya no quedan partidas abiertas (`ended_at IS NULL`) en el dataset local actual.
+- Validado con `list_weekly_top_kills()` para ambos servidores, confirmando separacion por servidor y uso exclusivo de partidas cerradas dentro de la ventana movil de 7 dias.
+
+## Decision Notes
+- `steaminfo.id` deja de tratarse como `steam_id` real porque en los datos observados funcionaba como identificador corto auxiliar y fragmentaba la identidad del jugador.
+- Para resolver duplicados de partida se usa una heuristica conservadora por `(historical_server_id, started_at, mapa normalizado)`, priorizando la fila cerrada, numerica y con mas jugadores.
+- No se eliminaron partidas con muy pocos jugadores porque eso es una decision de calidad de producto futura, no un problema de integridad estructural.

backend/README.md

@@ -423,8 +423,10 @@ Parametros opcionales:
 - `server` con slug historico como `comunidad-hispana-01`
 
 La ventana temporal es siempre la ultima semana movil respecto al momento de la
-request. El payload devuelve servidor, rango temporal, jugador, kills semanales,
+request y solo considera partidas cerradas con `ended_at` para no mezclar
-posicion y numero de partidas consideradas.
+partidas aun en curso ni filas historicas transitorias. El payload devuelve
+servidor, rango temporal, jugador, kills semanales, posicion y numero de
+partidas consideradas.
 
 ## Ingesta historica CRCON
 
@@ -455,6 +457,18 @@ La ejecucion `refresh` usa una ventana de solape sobre la ultima partida
 persistida por servidor para releer solo paginas recientes y absorber updates
 tardios sin reimportar todo el historico.
 
+Al inicializar la persistencia local, el backend normaliza tambien la identidad
+historica ya guardada:
+
+- prioriza `steaminfo.profile.steamid` cuando existe
+- si `player_id` ya parece un SteamID real, lo promueve igualmente a `steam:*`
+- si no hay SteamID, usa `player_id` como clave `crcon-player:*`
+- deja `steaminfo.id` como ultimo fallback cuando faltan las claves anteriores
+
+La misma inicializacion fusiona filas duplicadas si una partida abierta quedo
+guardada con un id sintetico y mas tarde CRCON la expone con un id numerico
+definitivo. Esto evita que el ranking semanal cuente dos veces la misma sesion.
+
 ## CORS local minimo
 
 El backend responde con `Access-Control-Allow-Origin` solo si la peticion llega

backend/app/historical_storage.py

@@ -153,6 +153,7 @@ def initialize_historical_storage(*, db_path: Path | None = None) -> Path:
         if legacy_historical_schema:
             _migrate_legacy_historical_data(connection)
         _normalize_historical_player_identities(connection)
+        _normalize_historical_match_identities(connection)
 
     return resolved_path
 
@@ -488,8 +489,9 @@ def list_weekly_top_kills(
     window_start = window_end - timedelta(days=DEFAULT_WEEKLY_WINDOW_DAYS)
 
     where_clauses = [
-        "COALESCE(historical_matches.ended_at, historical_matches.started_at, historical_matches.created_at_source) >= ?",
+        "historical_matches.ended_at IS NOT NULL",
-        "COALESCE(historical_matches.ended_at, historical_matches.started_at, historical_matches.created_at_source) <= ?",
+        "historical_matches.ended_at >= ?",
+        "historical_matches.ended_at <= ?",
     ]
     params: list[object] = [
         window_start.isoformat().replace("+00:00", "Z"),
@@ -778,26 +780,89 @@ def _migrate_legacy_historical_data(connection: sqlite3.Connection) -> None:
 def _normalize_historical_player_identities(connection: sqlite3.Connection) -> None:
     rows = connection.execute(
         """
-        SELECT id, stable_player_key, steam_id
+        SELECT id, stable_player_key, display_name, steam_id, source_player_id
         FROM historical_players
         ORDER BY id ASC
         """
     ).fetchall()
     for row in rows:
-        stable_player_key = _stringify(row["stable_player_key"])
+        player_id = int(row["id"])
-        if not stable_player_key or ":" in stable_player_key:
+        canonical_key, steam_id, source_player_id, display_name = _canonicalize_stored_player_row(row)
+        existing = connection.execute(
+            """
+            SELECT id
+            FROM historical_players
+            WHERE stable_player_key = ?
+            """,
+            (canonical_key,),
+        ).fetchone()
+        if existing is not None and int(existing["id"]) != player_id:
+            _merge_historical_player_rows(
+                connection,
+                source_player_id=player_id,
+                target_player_id=int(existing["id"]),
+                display_name=display_name,
+                steam_id=steam_id,
+                source_ref=source_player_id,
+            )
             continue
-        if stable_player_key.isdigit() and len(stable_player_key) >= 16:
+
-            normalized_key = f"steam:{stable_player_key}"
+        connection.execute(
-            connection.execute(
+            """
-                """
+            UPDATE historical_players
-                UPDATE historical_players
+            SET stable_player_key = ?,
-                SET stable_player_key = ?,
+                display_name = ?,
-                    steam_id = COALESCE(steam_id, ?),
+                steam_id = ?,
-                    updated_at = CURRENT_TIMESTAMP
+                source_player_id = ?,
-                WHERE id = ?
+                updated_at = CURRENT_TIMESTAMP
-                """,
+            WHERE id = ?
-                (normalized_key, stable_player_key, row["id"]),
+            """,
+            (canonical_key, display_name, steam_id, source_player_id, player_id),
+        )
+
+
+def _normalize_historical_match_identities(connection: sqlite3.Connection) -> None:
+    rows = connection.execute(
+        """
+        SELECT
+            historical_matches.id,
+            historical_matches.historical_server_id,
+            historical_matches.external_match_id,
+            historical_matches.started_at,
+            historical_matches.ended_at,
+            historical_matches.created_at_source,
+            historical_matches.map_name,
+            historical_matches.map_pretty_name,
+            COUNT(historical_player_match_stats.id) AS player_count
+        FROM historical_matches
+        LEFT JOIN historical_player_match_stats
+            ON historical_player_match_stats.historical_match_id = historical_matches.id
+        WHERE historical_matches.started_at IS NOT NULL
+        GROUP BY historical_matches.id
+        ORDER BY historical_matches.historical_server_id ASC, historical_matches.started_at ASC, historical_matches.id ASC
+        """
+    ).fetchall()
+
+    grouped_matches: dict[tuple[int, str, str], list[sqlite3.Row]] = {}
+    for row in rows:
+        group_key = (
+            int(row["historical_server_id"]),
+            str(row["started_at"]),
+            _normalize_match_identity_label(row["map_pretty_name"] or row["map_name"]),
+        )
+        grouped_matches.setdefault(group_key, []).append(row)
+
+    for grouped_rows in grouped_matches.values():
+        if len(grouped_rows) < 2:
+            continue
+        target_row = max(grouped_rows, key=_match_identity_preference)
+        for source_row in grouped_rows:
+            if int(source_row["id"]) == int(target_row["id"]):
+                continue
+            _merge_historical_match_rows(
+                connection,
+                source_match_id=int(source_row["id"]),
+                target_match_id=int(target_row["id"]),
             )
 
 
@@ -889,12 +954,8 @@ def _upsert_historical_player(
     connection: sqlite3.Connection,
     player_payload: Mapping[str, object],
 ) -> int:
-    stable_player_key = _build_stable_player_key(player_payload)
+    stable_player_key, steam_id, source_player_id = _derive_player_identity(player_payload)
-    display_name = _stringify(player_payload.get("player")) or "Unknown player"
+    display_name = _normalize_player_display_name(player_payload.get("player")) or "Unknown player"
-    steam_id = _stringify(_get_nested(player_payload, "steaminfo", "profile", "steamid"))
-    if not steam_id:
-        steam_id = _stringify(_get_nested(player_payload, "steaminfo", "id"))
-    source_player_id = _stringify(player_payload.get("player_id"))
     seen_at = _utc_now_iso()
 
     connection.execute(
@@ -933,25 +994,26 @@ def _upsert_historical_player(
 
 
 def _build_stable_player_key(player_payload: Mapping[str, object]) -> str:
+    stable_player_key, _, _ = _derive_player_identity(player_payload)
+    return stable_player_key
+
+
+def _derive_player_identity(player_payload: Mapping[str, object]) -> tuple[str, str | None, str | None]:
     steam_id = _stringify(_get_nested(player_payload, "steaminfo", "profile", "steamid"))
-    if steam_id:
-        return f"steam:{steam_id}"
-
-    steaminfo_id = _stringify(_get_nested(player_payload, "steaminfo", "id"))
-    if steaminfo_id:
-        return f"steaminfo:{steaminfo_id}"
-
     source_player_id = _stringify(player_payload.get("player_id"))
-    if source_player_id:
+    steaminfo_id = _stringify(_get_nested(player_payload, "steaminfo", "id"))
-        return f"crcon-player:{source_player_id}"
 
-    player_name = _stringify(player_payload.get("player")) or "unknown-player"
+    if steam_id:
-    normalized_name = "".join(
+        return f"steam:{steam_id}", steam_id, source_player_id
-        character.lower() if character.isalnum() else "-"
+    if _is_probable_steam_id(source_player_id):
-        for character in player_name
+        return f"steam:{source_player_id}", source_player_id, source_player_id
-    )
+    if source_player_id:
-    compact_name = "-".join(part for part in normalized_name.split("-") if part)
+        return f"crcon-player:{source_player_id}", None, source_player_id
-    return f"name:{compact_name or 'unknown-player'}"
+    if steaminfo_id:
+        return f"steaminfo:{steaminfo_id}", None, None
+
+    player_name = _normalize_player_display_name(player_payload.get("player")) or "unknown-player"
+    return f"name:{_normalize_name_key(player_name)}", None, None
 
 
 def _extract_map_name(match_payload: Mapping[str, object]) -> str | None:
@@ -1001,6 +1063,357 @@ def _stringify(value: object) -> str | None:
     return text or None
 
 
+def _normalize_player_display_name(value: object) -> str | None:
+    text = _stringify(value)
+    if not text:
+        return None
+    return " ".join(text.split())
+
+
+def _normalize_name_key(player_name: str) -> str:
+    normalized_name = "".join(
+        character.lower() if character.isalnum() else "-"
+        for character in player_name
+    )
+    compact_name = "-".join(part for part in normalized_name.split("-") if part)
+    return compact_name or "unknown-player"
+
+
+def _is_probable_steam_id(value: object) -> bool:
+    text = _stringify(value)
+    return bool(text and text.isdigit() and len(text) >= 16)
+
+
+def _canonicalize_stored_player_row(
+    row: sqlite3.Row,
+) -> tuple[str, str | None, str | None, str]:
+    stable_player_key = _stringify(row["stable_player_key"])
+    display_name = _normalize_player_display_name(row["display_name"]) or "Unknown player"
+    steam_id = _stringify(row["steam_id"])
+    source_player_id = _stringify(row["source_player_id"])
+
+    if _is_probable_steam_id(steam_id):
+        return f"steam:{steam_id}", steam_id, source_player_id, display_name
+    if _is_probable_steam_id(source_player_id):
+        return f"steam:{source_player_id}", source_player_id, source_player_id, display_name
+    if source_player_id:
+        return f"crcon-player:{source_player_id}", None, source_player_id, display_name
+    if stable_player_key and stable_player_key.startswith("steaminfo:"):
+        return stable_player_key, None, None, display_name
+    if stable_player_key and stable_player_key.startswith("name:"):
+        return stable_player_key, None, None, display_name
+    if stable_player_key and stable_player_key.startswith("steam:"):
+        return stable_player_key, steam_id, source_player_id, display_name
+    if stable_player_key and stable_player_key.startswith("crcon-player:"):
+        source_ref = stable_player_key.removeprefix("crcon-player:")
+        return stable_player_key, None, source_player_id or source_ref, display_name
+    if stable_player_key:
+        if _is_probable_steam_id(stable_player_key):
+            return f"steam:{stable_player_key}", stable_player_key, source_player_id, display_name
+        return f"crcon-player:{stable_player_key}", None, source_player_id or stable_player_key, display_name
+    return f"name:{_normalize_name_key(display_name)}", None, None, display_name
+
+
+def _merge_historical_player_rows(
+    connection: sqlite3.Connection,
+    *,
+    source_player_id: int,
+    target_player_id: int,
+    display_name: str,
+    steam_id: str | None,
+    source_ref: str | None,
+) -> None:
+    target_row = connection.execute(
+        """
+        SELECT display_name, steam_id, source_player_id, first_seen_at, last_seen_at
+        FROM historical_players
+        WHERE id = ?
+        """,
+        (target_player_id,),
+    ).fetchone()
+    if target_row is None:
+        return
+
+    connection.execute(
+        """
+        UPDATE historical_players
+        SET display_name = ?,
+            steam_id = ?,
+            source_player_id = ?,
+            first_seen_at = MIN(first_seen_at, ?),
+            last_seen_at = MAX(last_seen_at, ?),
+            updated_at = CURRENT_TIMESTAMP
+        WHERE id = ?
+        """,
+        (
+            _pick_preferred_display_name(target_row["display_name"], display_name),
+            _pick_preferred_steam_id(target_row["steam_id"], steam_id),
+            _pick_preferred_source_player_id(target_row["source_player_id"], source_ref),
+            connection.execute(
+                "SELECT first_seen_at FROM historical_players WHERE id = ?",
+                (source_player_id,),
+            ).fetchone()["first_seen_at"],
+            connection.execute(
+                "SELECT last_seen_at FROM historical_players WHERE id = ?",
+                (source_player_id,),
+            ).fetchone()["last_seen_at"],
+            target_player_id,
+        ),
+    )
+
+    stats_rows = connection.execute(
+        """
+        SELECT *
+        FROM historical_player_match_stats
+        WHERE historical_player_id = ?
+        ORDER BY id ASC
+        """,
+        (source_player_id,),
+    ).fetchall()
+    for stat_row in stats_rows:
+        existing = connection.execute(
+            """
+            SELECT *
+            FROM historical_player_match_stats
+            WHERE historical_match_id = ? AND historical_player_id = ?
+            """,
+            (stat_row["historical_match_id"], target_player_id),
+        ).fetchone()
+        if existing is None:
+            connection.execute(
+                """
+                UPDATE historical_player_match_stats
+                SET historical_player_id = ?,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = ?
+                """,
+                (target_player_id, stat_row["id"]),
+            )
+            continue
+
+        _merge_player_match_stats_row(connection, existing["id"], stat_row)
+        connection.execute(
+            "DELETE FROM historical_player_match_stats WHERE id = ?",
+            (stat_row["id"],),
+        )
+
+    connection.execute(
+        "DELETE FROM historical_players WHERE id = ?",
+        (source_player_id,),
+    )
+
+
+def _normalize_match_identity_label(value: object) -> str:
+    text = _stringify(value) or "unknown-map"
+    return " ".join(text.lower().split())
+
+
+def _match_identity_preference(row: sqlite3.Row) -> tuple[int, int, int, str, int]:
+    return (
+        1 if _stringify(row["ended_at"]) else 0,
+        1 if (_stringify(row["external_match_id"]) or "").isdigit() else 0,
+        int(row["player_count"] or 0),
+        _stringify(row["created_at_source"]) or "",
+        int(row["id"]),
+    )
+
+
+def _merge_historical_match_rows(
+    connection: sqlite3.Connection,
+    *,
+    source_match_id: int,
+    target_match_id: int,
+) -> None:
+    source_row = connection.execute(
+        "SELECT * FROM historical_matches WHERE id = ?",
+        (source_match_id,),
+    ).fetchone()
+    target_row = connection.execute(
+        "SELECT * FROM historical_matches WHERE id = ?",
+        (target_match_id,),
+    ).fetchone()
+    if source_row is None or target_row is None:
+        return
+
+    connection.execute(
+        """
+        UPDATE historical_matches
+        SET historical_map_id = COALESCE(historical_map_id, ?),
+            created_at_source = COALESCE(created_at_source, ?),
+            started_at = COALESCE(started_at, ?),
+            ended_at = COALESCE(ended_at, ?),
+            map_name = COALESCE(map_name, ?),
+            map_pretty_name = COALESCE(map_pretty_name, ?),
+            game_mode = COALESCE(game_mode, ?),
+            image_name = COALESCE(image_name, ?),
+            allied_score = COALESCE(allied_score, ?),
+            axis_score = COALESCE(axis_score, ?),
+            raw_payload_ref = COALESCE(raw_payload_ref, ?),
+            last_seen_at = MAX(last_seen_at, ?),
+            updated_at = CURRENT_TIMESTAMP
+        WHERE id = ?
+        """,
+        (
+            source_row["historical_map_id"],
+            source_row["created_at_source"],
+            source_row["started_at"],
+            source_row["ended_at"],
+            source_row["map_name"],
+            source_row["map_pretty_name"],
+            source_row["game_mode"],
+            source_row["image_name"],
+            source_row["allied_score"],
+            source_row["axis_score"],
+            source_row["raw_payload_ref"],
+            source_row["last_seen_at"],
+            target_match_id,
+        ),
+    )
+
+    stats_rows = connection.execute(
+        """
+        SELECT *
+        FROM historical_player_match_stats
+        WHERE historical_match_id = ?
+        ORDER BY id ASC
+        """,
+        (source_match_id,),
+    ).fetchall()
+    for stat_row in stats_rows:
+        existing = connection.execute(
+            """
+            SELECT *
+            FROM historical_player_match_stats
+            WHERE historical_match_id = ? AND historical_player_id = ?
+            """,
+            (target_match_id, stat_row["historical_player_id"]),
+        ).fetchone()
+        if existing is None:
+            connection.execute(
+                """
+                UPDATE historical_player_match_stats
+                SET historical_match_id = ?,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = ?
+                """,
+                (target_match_id, stat_row["id"]),
+            )
+            continue
+
+        _merge_player_match_stats_row(connection, existing["id"], stat_row)
+        connection.execute(
+            "DELETE FROM historical_player_match_stats WHERE id = ?",
+            (stat_row["id"],),
+        )
+
+    connection.execute(
+        "DELETE FROM historical_matches WHERE id = ?",
+        (source_match_id,),
+    )
+
+
+def _merge_player_match_stats_row(
+    connection: sqlite3.Connection,
+    target_stat_id: int,
+    source_row: sqlite3.Row,
+) -> None:
+    target_row = connection.execute(
+        "SELECT * FROM historical_player_match_stats WHERE id = ?",
+        (target_stat_id,),
+    ).fetchone()
+    if target_row is None:
+        return
+
+    connection.execute(
+        """
+        UPDATE historical_player_match_stats
+        SET match_player_ref = COALESCE(match_player_ref, ?),
+            team_side = COALESCE(team_side, ?),
+            level = ?,
+            kills = ?,
+            deaths = ?,
+            teamkills = ?,
+            time_seconds = ?,
+            kills_per_minute = ?,
+            deaths_per_minute = ?,
+            kill_death_ratio = ?,
+            combat = ?,
+            offense = ?,
+            defense = ?,
+            support = ?,
+            updated_at = CURRENT_TIMESTAMP
+        WHERE id = ?
+        """,
+        (
+            source_row["match_player_ref"],
+            source_row["team_side"],
+            _max_int_value(target_row["level"], source_row["level"]),
+            _max_int_value(target_row["kills"], source_row["kills"]),
+            _max_int_value(target_row["deaths"], source_row["deaths"]),
+            _max_int_value(target_row["teamkills"], source_row["teamkills"]),
+            _max_int_value(target_row["time_seconds"], source_row["time_seconds"]),
+            _max_float_value(target_row["kills_per_minute"], source_row["kills_per_minute"]),
+            _max_float_value(target_row["deaths_per_minute"], source_row["deaths_per_minute"]),
+            _max_float_value(target_row["kill_death_ratio"], source_row["kill_death_ratio"]),
+            _max_int_value(target_row["combat"], source_row["combat"]),
+            _max_int_value(target_row["offense"], source_row["offense"]),
+            _max_int_value(target_row["defense"], source_row["defense"]),
+            _max_int_value(target_row["support"], source_row["support"]),
+            target_stat_id,
+        ),
+    )
+
+
+def _pick_preferred_display_name(current_value: object, incoming_value: object) -> str:
+    current_name = _normalize_player_display_name(current_value)
+    incoming_name = _normalize_player_display_name(incoming_value)
+    if not current_name:
+        return incoming_name or "Unknown player"
+    if not incoming_name:
+        return current_name
+    if len(incoming_name) > len(current_name):
+        return incoming_name
+    return current_name
+
+
+def _pick_preferred_steam_id(current_value: object, incoming_value: object) -> str | None:
+    current_id = _stringify(current_value)
+    incoming_id = _stringify(incoming_value)
+    if _is_probable_steam_id(current_id):
+        return current_id
+    if _is_probable_steam_id(incoming_id):
+        return incoming_id
+    return None
+
+
+def _pick_preferred_source_player_id(current_value: object, incoming_value: object) -> str | None:
+    current_id = _stringify(current_value)
+    incoming_id = _stringify(incoming_value)
+    if current_id:
+        return current_id
+    return incoming_id
+
+
+def _max_int_value(current_value: object, incoming_value: object) -> int | None:
+    current_number = _coerce_int(current_value)
+    incoming_number = _coerce_int(incoming_value)
+    if current_number is None:
+        return incoming_number
+    if incoming_number is None:
+        return current_number
+    return max(current_number, incoming_number)
+
+
+def _max_float_value(current_value: object, incoming_value: object) -> float | None:
+    current_number = _coerce_float(current_value)
+    incoming_number = _coerce_float(incoming_value)
+    if current_number is None:
+        return incoming_number
+    if incoming_number is None:
+        return current_number
+    return max(current_number, incoming_number)
+
+
 def _normalize_timestamp(value: object) -> str | None:
     text = _stringify(value)
     if not text:

docs/historical-data-quality-notes.md

@@ -0,0 +1,66 @@
+# Historical Data Quality Notes
+
+## Validation Date
+
+- 2026-03-20
+
+## Scope
+
+Validacion local del historico CRCON persistido en `backend/data/hll_vietnam_dev.sqlite3`
+para los servidores:
+
+- `comunidad-hispana-01`
+- `comunidad-hispana-02`
+
+## Findings Before Correction
+
+- habia jugadores fragmentados entre claves `steam:*`, `steaminfo:*`,
+  `crcon-player:*` e incluso claves legacy sin prefijo
+- algunas filas usaban `steaminfo.id` corto como si fuera `steam_id`, lo que no
+  representaba un SteamID real
+- existian partidas duplicadas por sesion cuando una partida en curso quedaba
+  persistida con id sintetico y luego aparecia cerrada con id CRCON numerico
+- el ranking semanal podia contar esas partidas transitorias porque aceptaba
+  filas sin `ended_at`
+
+## Corrections Applied
+
+- la identidad de jugador ahora prioriza:
+  - `steaminfo.profile.steamid`
+  - `player_id` cuando ya parece un SteamID real
+  - `player_id` como `crcon-player:*`
+  - `steaminfo.id` solo como ultimo fallback
+- la inicializacion del storage fusiona jugadores duplicados y reasigna sus
+  estadisticas por partida
+- la inicializacion del storage fusiona partidas duplicadas por
+  `(servidor, started_at, mapa)` cuando la fila mas completa ya representa la
+  partida final cerrada
+- `weekly-top-kills` filtra solo partidas cerradas con `ended_at`
+
+## Final Local Snapshot After Correction
+
+- partidas historicas: `12`
+- jugadores historicos: `510`
+- filas `historical_player_match_stats`: `914`
+- distribucion:
+  - `comunidad-hispana-01`: `7` partidas, `487` jugadores unicos, `859` filas
+  - `comunidad-hispana-02`: `5` partidas, `44` jugadores unicos, `55` filas
+
+## Checks Performed
+
+- sin duplicados por `steam_id`
+- sin duplicados por `source_player_id`
+- sin duplicados de nombre normalizado en el dataset local actual
+- sin partidas abiertas restantes (`ended_at IS NULL`)
+- sin duplicados por misma combinacion de servidor, `started_at` y mapa
+- el ranking semanal devuelve resultados separados por servidor y basados solo
+  en partidas cerradas dentro de la ventana movil de 7 dias
+
+## Notes
+
+- el volumen actual sigue siendo pequeno y claramente parcial; la calidad
+  estructural queda validada, pero no sustituye un bootstrap historico mas
+  profundo cuando se quiera construir UI historica propia
+- siguen existiendo partidas con muy pocos jugadores en el dataset local
+  actual; por ahora se conservan porque no son un problema de integridad, sino
+  una caracteristica del muestreo ingerido hasta hoy