feat(h4-3): modo simulación con flota ficticia (RF-12)

CHANGELOG.md

@@ -7,6 +7,17 @@ Versionado: una entrada por **entrega académica** del semestre (no SemVer estri
 
 ## [Unreleased]
 
+### Added — H4 fase 3: modo simulación (RF-12, 2026-05-21)
+- Nuevo módulo [`application/simulacion.py`](core-python/src/sentinel_dispatch/application/simulacion.py) con value object `ReporteSimulacion` (dataclass frozen+slots) que agrega: `incidentes_procesados`, tupla de `ResultadoDespacho`, porcentajes por motivo (`pct_optimo`, `pct_penalizado`, `pct_suboptimo_rn02`, `pct_saturacion`), ETA media y ETA p95.
+- Función `simular(incidentes, flota_ficticia, grafo, *, repositorio_eventos=None, factor_hora=1.0, factor_sirena=1.0) → ReporteSimulacion`. **Semántica v1**: sin evolución temporal entre incidentes (cada uno ve la flota inicial). Determinístico: el resultado depende sólo de los inputs.
+- Persistencia **opt-in**: por default NO escribe al log canónico (modo simulación ≠ operativo). Si se provee `repositorio_eventos`, se persisten eventos `despacho_creado` con `despacho_id` prefijado `SD-SIM-` para distinguirlos de despachos operativos.
+- Nuevo subcomando CLI [`interfaces/cli/simular_cmd.py`](core-python/src/sentinel_dispatch/interfaces/cli/simular_cmd.py): `sentinel simular --flota --incidentes --graph --out [--persistir-en]`. El reporte JSON de salida incluye campo `"modo": "simulacion"` como marca explícita.
+- Tests: **7 nuevos** verdes — Normal (2): resultados+métricas, ausencia de evolución temporal · Borde (2): lista vacía, flota vacía=100% saturación · ReglasNegocio (2): default no escribe, con repo escribe N eventos al archivo separado · Métricas (1): pcts suman 100.0. Suite total **256/256** verde; cobertura global **90.60 %**.
+
+### Changed — H4 fase 3
+- `docs/quality/trazabilidad.md`: **RF-12 marcado ✅** apuntando a `application/simulacion.py` + `interfaces/cli/simular_cmd.py`.
+- `interfaces/cli/app.py`: registro del subcomando `simular`.
+
 ### Added — H4 fase 2: exportador CSV/JSON (RF-11, 2026-05-21)
 - Nuevo adapter [`adapters/exportador.py`](core-python/src/sentinel_dispatch/adapters/exportador.py): funciones puras `exportar_a_csv(eventos, path)` y `exportar_a_json(eventos, path)`. CSV con flatten de `payload_*` (e.g. `payload_costo_total`) y encoding `utf-8-sig` (BOM para que Excel español abra correctamente). JSON como array indentado sin BOM. Helper `_aplanar_dict(d, prefijo)` aplana dicts recursivamente; listas (e.g. `ruta`) se serializan como JSON string en una sola celda.
 - Nuevo subcomando CLI [`interfaces/cli/export_cmd.py`](core-python/src/sentinel_dispatch/interfaces/cli/export_cmd.py): `sentinel export --formato {csv,json} --in eventos.jsonl --out reporte.{csv,json}`. Enum `FormatoExport(csv|json)` para validación de argumento. Exit 0 en éxito, 2 si `--in` no existe o el JSONL es corrupto.

core-python/src/sentinel_dispatch/application/simulacion.py

@@ -0,0 +1,197 @@
+"""Modo simulación (RF-12) — ejecuta el cálculo de despacho sobre flota ficticia.
+
+El SRS sec. RF-12 exige un modo que "ejecuta el cálculo completo sobre
+un estado de flota ficticio sin afectar el estado operativo real". Este
+módulo implementa la semántica más simple compatible con esa exigencia:
+
+- **Sin evolución temporal entre incidentes**: cada incidente ve la flota
+  ficticia en su estado inicial. Equivalente a correr ``run-dataset`` N
+  veces sobre el mismo grafo + flota. Esta interpretación literal del
+  SRS es suficiente para v1 (academic). Para v2 con reloj virtual y
+  liberación de unidades por ``eta_segundos`` se necesitaría un ADR
+  nuevo y framework de event-driven simulation.
+
+- **Persistencia opt-in**: por default NO escribe al log canónico
+  (modo simulación ≠ modo operativo). Si el caller provee
+  ``repositorio_eventos``, se escribe ahí (típicamente un archivo
+  separado tipo ``eventos_sim.jsonl`` para no contaminar el log real).
+
+- **Reporte agregado**: el output incluye los resultados crudos +
+  métricas porcentuales por motivo y ETA media/p95 para defensa
+  académica del módulo.
+"""
+
+from __future__ import annotations
+
+import math
+import statistics
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from sentinel_dispatch.application.despachar_ambulancia import despachar
+from sentinel_dispatch.application.serializacion import serializar_resultado_despacho
+from sentinel_dispatch.application.tipos import MotivoDespacho, ResultadoDespacho
+from sentinel_dispatch.ports.repositorio_eventos import EventoLog, TipoEvento
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from sentinel_dispatch.adapters.repositorio_jsonl import JsonlRepositorioEventos
+    from sentinel_dispatch.domain.dispatch.tipos import Incidente, Unidad
+    from sentinel_dispatch.domain.routing.grafo_vial import GrafoVial
+
+
+@dataclass(frozen=True, slots=True)
+class ReporteSimulacion:
+    """Resumen inmutable de una corrida de simulación.
+
+    Attributes:
+        incidentes_procesados: número de incidentes ejecutados.
+        resultados: tupla con un :class:`ResultadoDespacho` por incidente,
+            en el orden en que se procesaron.
+        pct_optimo: porcentaje de despachos en motivo ``OPTIMO`` (0-100).
+        pct_penalizado: porcentaje en motivo ``PENALIZADO``.
+        pct_suboptimo_rn02: porcentaje en motivo ``SUBOPTIMO_RN02``.
+        pct_saturacion: porcentaje en motivo ``SATURACION``.
+        eta_media_s: ETA promedio sobre los despachos no saturados.
+            ``None`` si todos los incidentes fueron saturación.
+        eta_p95_s: percentil 95 de las ETAs sobre los despachos no
+            saturados. ``None`` si N < 1 o todos saturación.
+    """
+
+    incidentes_procesados: int
+    resultados: tuple[ResultadoDespacho, ...]
+    pct_optimo: float
+    pct_penalizado: float
+    pct_suboptimo_rn02: float
+    pct_saturacion: float
+    eta_media_s: float | None
+    eta_p95_s: float | None
+
+
+def _calcular_metricas(
+    resultados: Sequence[ResultadoDespacho],
+) -> tuple[float, float, float, float, float | None, float | None]:
+    """Devuelve `(pct_optimo, pct_penalizado, pct_suboptimo_rn02, pct_saturacion, eta_media, eta_p95)`."""
+    n = len(resultados)
+    if n == 0:
+        return 0.0, 0.0, 0.0, 0.0, None, None
+
+    conteos = dict.fromkeys(MotivoDespacho, 0)
+    for r in resultados:
+        conteos[r.motivo] += 1
+
+    pct = {motivo: 100.0 * c / n for motivo, c in conteos.items()}
+
+    etas = [
+        r.costo_elegida.t_viaje_s
+        for r in resultados
+        if r.motivo is not MotivoDespacho.SATURACION
+        and r.costo_elegida is not None
+        and math.isfinite(r.costo_elegida.t_viaje_s)
+    ]
+    if etas:
+        eta_media: float | None = statistics.fmean(etas)
+        # p95 con interpolación lineal estándar; statistics no expone percentile,
+        # uso ordered + índice clamp por simplicidad (N pequeño, exacto).
+        ordenado = sorted(etas)
+        idx_p95 = max(0, math.ceil(0.95 * len(ordenado)) - 1)
+        eta_p95: float | None = ordenado[idx_p95]
+    else:
+        eta_media = None
+        eta_p95 = None
+
+    return (
+        pct[MotivoDespacho.OPTIMO],
+        pct[MotivoDespacho.PENALIZADO],
+        pct[MotivoDespacho.SUBOPTIMO_RN02],
+        pct[MotivoDespacho.SATURACION],
+        eta_media,
+        eta_p95,
+    )
+
+
+def _persistir_evento(
+    repo: JsonlRepositorioEventos,
+    resultado: ResultadoDespacho,
+    incidente: Incidente,
+) -> None:
+    """Persiste un evento ``despacho_creado`` por resultado (ADR-0018)."""
+    despacho_id = (
+        f"SD-SIM-{incidente.timestamp_iso[:10].replace('-', '')}-{incidente.id.replace('I-', '')}"
+    )
+    evento = EventoLog(
+        evento_id=repo.generar_evento_id(base_ts=datetime.now(UTC)),
+        timestamp_iso=incidente.timestamp_iso,
+        tipo=TipoEvento.DESPACHO_CREADO,
+        despacho_id=despacho_id,
+        incidente_id=incidente.id,
+        payload=serializar_resultado_despacho(resultado),
+    )
+    repo.append(evento)
+
+
+def simular(
+    incidentes: Sequence[Incidente],
+    flota_ficticia: Sequence[Unidad],
+    grafo: GrafoVial,
+    *,
+    repositorio_eventos: JsonlRepositorioEventos | None = None,
+    factor_hora: float = 1.0,
+    factor_sirena: float = 1.0,
+) -> ReporteSimulacion:
+    """Ejecuta el cálculo de despacho sobre cada incidente sin tocar el sistema real.
+
+    Args:
+        incidentes: secuencia de incidentes a procesar (deberían ya tener
+            la categoría MPDS calculada).
+        flota_ficticia: secuencia de unidades **ficticias** a usar como
+            estado inicial. Es responsabilidad del caller verificar que
+            esta flota no se cruza con la flota operativa real (por
+            convención v1, basta con ``id`` distintos).
+        grafo: :class:`GrafoVial` ya cargado (usualmente la región
+            real — la "ficción" está en la flota, no en el mapa).
+        repositorio_eventos: opcional. Si se provee, cada despacho
+            persiste un evento ``despacho_creado`` ahí. Default ``None``
+            = no persiste; **la simulación no afecta el log canónico
+            del modo operativo** (cumple "sin afectar el estado real").
+        factor_hora: multiplicador de tráfico horario para el A*.
+        factor_sirena: multiplicador de sirena.
+
+    Returns:
+        :class:`ReporteSimulacion` con los resultados + métricas.
+
+    Notas semánticas (decisiones v1):
+        - **Sin estado evolutivo entre incidentes**: cada incidente ve
+          la flota ficticia tal como llegó al inicio. Si una unidad fue
+          "despachada" al I-01, sigue ``DISPONIBLE`` al evaluar I-02.
+          Equivalente a paralelizar conceptualmente.
+        - **Determinismo**: el resultado depende sólo de los inputs,
+          no del wall-clock. Apto para tests y para reproducir corridas.
+    """
+    resultados: list[ResultadoDespacho] = []
+    for incidente in incidentes:
+        resultado = despachar(
+            incidente,
+            flota_ficticia,
+            grafo,
+            factor_hora=factor_hora,
+            factor_sirena=factor_sirena,
+        )
+        resultados.append(resultado)
+        if repositorio_eventos is not None:
+            _persistir_evento(repositorio_eventos, resultado, incidente)
+
+    pct_opt, pct_pen, pct_sub, pct_sat, eta_media, eta_p95 = _calcular_metricas(resultados)
+
+    return ReporteSimulacion(
+        incidentes_procesados=len(resultados),
+        resultados=tuple(resultados),
+        pct_optimo=pct_opt,
+        pct_penalizado=pct_pen,
+        pct_suboptimo_rn02=pct_sub,
+        pct_saturacion=pct_sat,
+        eta_media_s=eta_media,
+        eta_p95_s=eta_p95,
+    )

core-python/src/sentinel_dispatch/interfaces/cli/app.py

@@ -13,7 +13,12 @@
 
 import typer
 
-from sentinel_dispatch.interfaces.cli import export_cmd, run_dataset_cmd, triaje_cmd
+from sentinel_dispatch.interfaces.cli import (
+    export_cmd,
+    run_dataset_cmd,
+    simular_cmd,
+    triaje_cmd,
+)
 
 app = typer.Typer(
     name="sentinel",
@@ -25,6 +30,7 @@
 app.add_typer(triaje_cmd.app, name="triaje")
 app.command("run-dataset")(run_dataset_cmd.run_dataset)
 app.command("export")(export_cmd.export)
+app.command("simular")(simular_cmd.simular_cmd)
 
 
 if __name__ == "__main__":  # pragma: no cover