fix(backend,airflow,ml): cloture la reconciliation entre les deux sources de lectures (#15)

apps/backend/app/etl/mock_api_import.py

@@ -19,6 +19,7 @@ from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncConnection, create_async_engine
 
 from app.core.config import get_settings
+from app.etl.historical_import import SOURCE_NAME as SOURCE_CSV
 
 SOURCE_HISTORY = "api_history"
 
@@ -244,6 +245,35 @@ def build_reading_row(
     }
 
 
+# `uq_reading_source` autorise deux lignes au même (site_id, timestamp) dès que `source` diffère :
+# sans ce garde-fou, importer une fenêtre déjà couverte par le dataset historique (source='csv')
+# dupliquerait silencieusement chaque point plutôt que de lever une erreur, et rien côté lecture
+# (pipeline ML, GET /readings) ne saurait laquelle des deux lectures retenir.
+OVERLAP_CHECK = text(
+    "SELECT count(*) FROM reading WHERE source = :source_csv "
+    "AND timestamp >= :start_time AND timestamp < :end_time"
+)
+
+
+async def refuse_if_overlaps_historical_dataset(
+    connection: AsyncConnection,
+    start_time: datetime,
+    end_time: datetime,
+) -> None:
+    resultat = await connection.execute(
+        OVERLAP_CHECK,
+        {"source_csv": SOURCE_CSV, "start_time": start_time, "end_time": end_time},
+    )
+    nombre = resultat.scalar_one()
+
+    if nombre > 0:
+        raise ValueError(
+            f"La fenêtre [{start_time.isoformat()}, {end_time.isoformat()}) recouvre "
+            f"{nombre} lecture(s) déjà importée(s) du dataset historique (source='{SOURCE_CSV}') : "
+            "import refusé pour éviter un doublon inter-source."
+        )
+
+
 # Le conflit vise l'index unique uq_reading_source plutôt que la table entière : sans cible
 # nommée, DO NOTHING avalerait aussi une violation de clé primaire.
 READING_INSERT = text(
@@ -345,6 +375,8 @@ async def import_mock_api_history(
 
     try:
         async with engine.begin() as connection:
+            await refuse_if_overlaps_historical_dataset(connection, start_time, end_time)
+
             await upsert_sites(
                 connection,
                 sites,

apps/backend/tests/etl/test_mock_api_import.py

@@ -357,6 +357,31 @@ async def test_upsert_sites_with_empty_list_does_nothing() -> None:
     connection.execute.assert_not_awaited()
 
 
+async def test_refuse_if_overlaps_historical_dataset_lets_a_clear_window_through() -> None:
+    connection = AsyncMock()
+    connection.execute.return_value.scalar_one = MagicMock(return_value=0)
+
+    await mock_api_import.refuse_if_overlaps_historical_dataset(
+        connection,
+        datetime.fromisoformat("2026-01-01T00:00:00"),
+        datetime.fromisoformat("2026-01-01T01:00:00"),
+    )
+
+    connection.execute.assert_awaited_once()
+
+
+async def test_refuse_if_overlaps_historical_dataset_rejects_a_window_already_in_the_csv() -> None:
+    connection = AsyncMock()
+    connection.execute.return_value.scalar_one = MagicMock(return_value=5)
+
+    with pytest.raises(ValueError, match="doublon inter-source"):
+        await mock_api_import.refuse_if_overlaps_historical_dataset(
+            connection,
+            datetime.fromisoformat("2023-06-15T12:00:00"),
+            datetime.fromisoformat("2023-06-15T13:00:00"),
+        )
+
+
 async def test_import_mock_api_history_dry_run_does_not_write(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -454,6 +479,7 @@ async def test_import_mock_api_history_loads_data(
     )
 
     connection = AsyncMock()
+    connection.execute.return_value.scalar_one = MagicMock(return_value=0)
 
     transaction_context = MagicMock()
     transaction_context.__aenter__ = AsyncMock(
@@ -502,7 +528,58 @@ async def test_import_mock_api_history_loads_data(
         [make_site()],
     )
 
+    assert connection.execute.await_count == 2
+    dernier_appel = connection.execute.await_args_list[-1]
+    assert dernier_appel.args[0] is READING_INSERT
+    engine.dispose.assert_awaited_once()
+
+
+async def test_import_mock_api_history_refuses_when_it_overlaps_the_historical_dataset(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def handler(request: Request) -> Response:
+        if request.url.path == "/api/v1/sites":
+            return Response(status_code=200, json=[make_site()])
+
+        if request.url.path == "/api/v1/readings":
+            return Response(status_code=200, json=[make_reading()])
+
+        return Response(status_code=404)
+
+    client = AsyncClient(transport=MockTransport(handler), base_url="https://mock.test")
+
+    monkeypatch.setattr(mock_api_import, "create_mock_api_client", lambda: client)
+    monkeypatch.setattr(
+        mock_api_import,
+        "get_settings",
+        lambda: SimpleNamespace(database_url="postgresql+asyncpg://test:test@localhost/test"),
+    )
+
+    connection = AsyncMock()
+    connection.execute.return_value.scalar_one = MagicMock(return_value=3)
+
+    transaction_context = MagicMock()
+    transaction_context.__aenter__ = AsyncMock(return_value=connection)
+    transaction_context.__aexit__ = AsyncMock(return_value=None)
+
+    engine = MagicMock()
+    engine.begin.return_value = transaction_context
+    engine.dispose = AsyncMock()
+
+    monkeypatch.setattr(mock_api_import, "create_async_engine", MagicMock(return_value=engine))
+    upsert_sites_mock = AsyncMock()
+    monkeypatch.setattr(mock_api_import, "upsert_sites", upsert_sites_mock)
+
+    with pytest.raises(ValueError, match="doublon inter-source"):
+        await mock_api_import.import_mock_api_history(
+            start_time=datetime.fromisoformat("2023-06-15T12:00:00"),
+            end_time=datetime.fromisoformat("2023-06-15T13:00:00"),
+            limit=60,
+            dry_run=False,
+        )
+
     connection.execute.assert_awaited_once()
+    upsert_sites_mock.assert_not_awaited()
     engine.dispose.assert_awaited_once()
 
 

docs/architecture/00-vue-ensemble.md

@@ -70,11 +70,17 @@ Le lien `front -.-> api` reste en pointillé : le frontend appelle bien une API,
 intercepteur répond à sa place tant que les endpoints n'existent pas. Voir
 [30-frontend.md](30-frontend.md).
 
-Le lien `airflow --> db` est maintenant en trait plein : cinq DAGs tournent, deux pour
+Le lien `airflow --> db` est maintenant en trait plein : six DAGs tournent, deux pour
 l'entraînement et le scoring du modèle ML (issue #115), un pour la détection d'alertes et la
-génération des recommandations (issue #116), `historical_import` pour le dataset historique
-(issue #119) et `mock_api_import` pour l'ingestion horaire de l'API Mock (issue #15).
-La réconciliation globale des données provenant des deux sources reste à compléter dans l'issue #15.
+génération des recommandations (issue #116), un pour la surveillance de dérive (issue #45),
+`historical_import` pour le dataset historique (issue #119) et `mock_api_import` pour l'ingestion
+horaire de l'API Mock (issue #15). La réconciliation entre les deux sources de lectures (issue
+#15) est tranchée : le trou entre la fin de l'historique (31/12/2024) et le début de l'ingestion
+API Mock est accepté comme définitivement perdu, aucune mesure réelle n'existant pour cette
+période. `mock_api_import` refuse toute fenêtre qui recouvrirait des lectures déjà importées du
+CSV plutôt que de laisser les deux sources dupliquer silencieusement un même instant, et le
+pipeline ML déduplique par construction (`DISTINCT ON`, source `csv` préférée) au cas où un
+recouvrement se produirait malgré tout — voir [40-data.md](40-data.md).
 
 Le lien `prom -.-> api` de même : l'API expose bien `/metrics` au format Prometheus, mais aucun
 collecteur ne vient le lire.
@@ -89,7 +95,7 @@ collecteur ne vient le lire.
 | ML | LightGBM, MLflow | `ml` | `En cours` | Pipeline d'entraînement et de scoring (`enervision_ml.train`/`.score`, features par lags/moyennes glissantes partagées entre les deux, baseline de persistance saisonnière, suivi MLflow local), exposé en lecture via `GET /predictions`, orchestré par Airflow (`ml_train`/`ml_score`). Voir [ADR 0005](../adr/0005-modele-prediction-lightgbm.md) et [ML-START.md](../ML-START.md). Surveillance de dérive livrée côté backend (`app.monitoring.drift`, table `drift_report`, `GET /monitoring/drift`, DAG `derive`), voir [ADR 0013](../adr/0013-surveillance-de-derive-dans-le-backend.md) |
 | Infra | Docker Compose, Nginx, Terraform, k3s single-node | `infra`, `docker-compose.prod.yml` | `En cours` | Reverse proxy et overlay de déploiement écrits et validés, jamais lancés sur le serveur ([ADR 0007](../adr/0007-terminaison-tls-et-reverse-proxy-nginx.md)). Provisionnement de la VM par Terraform, qui installe Docker, prépare les deux environnements et enregistre le runner, jamais appliqué ([ADR 0010](../adr/0010-terraform-provisionne-github-actions-deploie.md)). Module d'installation k3s jamais appliqué, aucune ressource Kubernetes déclarée |
 | Monitoring | Prometheus, Grafana, Alertmanager | `monitoring` | `Cible` | Rien, hors le `/metrics` exposé par l'API |
-| ETL | Apache Airflow | `etl/airflow` | `En cours` | Webserver et scheduler avec LocalExecutor via Docker Compose, sur une base PostgreSQL dédiée. Six DAGs en sous-processus `uv run` : `ml_train`, `ml_score`, `alertes`, `historical_import`, `mock_api_import` et `derive` (quotidien, surveillance de dérive). L'import historique reste manuel et l'import API Mock s'exécute chaque heure. La réconciliation globale des deux sources reste à compléter dans l'issue #15. |
+| ETL | Apache Airflow | `etl/airflow` | `En cours` | Webserver et scheduler avec LocalExecutor via Docker Compose, sur une base PostgreSQL dédiée. Six DAGs en sous-processus `uv run` : `ml_train`, `ml_score`, `alertes`, `historical_import`, `mock_api_import` et `derive` (quotidien, surveillance de dérive). L'import historique reste manuel et l'import API Mock s'exécute chaque heure. Réconciliation entre les deux sources (issue #15) : trou temporel accepté, recouvrement refusé à l'ingestion et dédupliqué en défense côté ML, voir [40-data.md](40-data.md). |
 | CI/CD | GitHub Actions | `.github/workflows` | `En cours` | 7 workflows, 19 jobs : lint, typage, tests avec seuil de couverture bloquant, tests d'intégration sur TimescaleDB réel, audit de dépendances, SAST Bandit, quality gate SonarCloud, intégrité des DAGs Airflow, formatage et validation du Terraform. Déploiement continu vers la VM ENI écrit par `deploy.yml`, `dev` en recette et `main` en production après approbation ([ADR 0009](../adr/0009-deux-environnements-compose-sur-la-vm-eni.md)), mais jamais exécuté : la machine n'est pas provisionnée et le runner n'y est pas enregistré. Détail dans [50-cicd.md](50-cicd.md) |
 
 ## Flux bout en bout
@@ -99,7 +105,8 @@ Statut : `En cours`. **Le chemin de lecture tourne** entre la base, l'API et le
 dataset CSV/JSON sur déclenchement manuel et `mock_api_import` collecte chaque heure les mesures
 de l'API Mock. Les DAGs `ml_train` et `ml_score` (issue #115), `alertes` (issue #116) et `derive`
 (issue #45) portent le pipeline ML, la détection d'alertes et la surveillance de dérive. La
-réconciliation globale des données provenant des deux sources reste à compléter dans l'issue #15.
+réconciliation entre les deux sources de lectures (issue #15) est close : voir
+[40-data.md](40-data.md) pour le détail du garde-fou d'ingestion et de la déduplication ML.
 
 ```mermaid
 sequenceDiagram

docs/architecture/40-data.md

@@ -441,6 +441,16 @@ Les paramètres de ligne de commande disponibles pour l'import sont :
 --dry-run
 ```
 
+**Piège sur `--limit`** : l'API ne renvoie pas un flux à un rythme naturel, elle répartit
+exactement `limit` lectures, espacées uniformément, sur toute la fenêtre `[start_time, end_time)`
+demandée. Une fenêtre d'une heure avec `limit=1000` renvoie donc 1000 lectures espacées de 3,6
+secondes à l'intérieur de cette heure, pas une lecture horaire — vérifié empiriquement en
+interrogeant directement l'API. Le seul réglage qui produise une lecture par heure, alignée sur
+l'heure et cohérente avec le grain horaire du reste du schéma (`period_minutes=60`, historique
+CSV à une ligne par heure), est `limit` = nombre d'heures de la fenêtre. Le DAG `mock_api_import`
+interroge toujours une fenêtre d'1h (`interval=timedelta(hours=1)`, voir
+[10-infra.md](10-infra.md)), donc `limit=1`.
+
 ### Flux d'ingestion API Mock
 
 ```text
@@ -492,7 +502,7 @@ réponse est donc traitée comme une entrée hostile, conformément à API10 dan
 [la traçabilité OWASP](owasp-traceabilite.md). Le risque premier n'est pas la fausse alerte,
 c'est l'empoisonnement du jeu d'entraînement du modèle de prédiction.
 
-Quatre garde-fous, tous dans `mock_api_import.py` :
+Cinq garde-fous, tous dans `mock_api_import.py` :
 
 | Garde-fou | Mise en œuvre |
 |---|---|
@@ -500,6 +510,7 @@ Quatre garde-fous, tous dans `mock_api_import.py` :
 | Taille de tableau plafonnée | `MAX_SITES` sites, et au plus `--limit` mesures par site |
 | Bornes physiques | `PHYSICAL_BOUNDS`, une plage par grandeur |
 | Frontière d'anti-corruption | `build_site_row()` et `build_reading_row()`, qui ne recopient que les champs attendus |
+| Refus de recouvrir l'historique | `refuse_if_overlaps_historical_dataset()`, voir ci-dessous |
 
 Une valeur hors bornes, d'un type inattendu, `NaN` ou infinie devient `NULL`. Elle laisse sa
 trace dans `null_reasons` sous la forme `out_of_physical_bounds:<colonne>`, et `data_quality`
@@ -510,6 +521,30 @@ d'origine intacte : rien n'est perdu, seule son exploitation est bornée.
 Le plafond de taille s'applique après désérialisation de la réponse. Borner le corps HTTP
 lui-même demanderait une lecture en flux, et reste à faire.
 
+### Réconciliation entre les deux sources (issue #15)
+
+`historical_import` (source `csv`) et `mock_api_import` (source `api_history`) écrivent toutes
+deux dans `reading`. Trois décisions ferment cette réconciliation :
+
+- **Le trou temporel est accepté.** Le dataset historique s'arrête au 31/12/2024, et
+  `mock_api_import` n'importe que l'heure précédant chaque déclenchement : rien ne comble
+  automatiquement la période intermédiaire, et rien ne le pourra jamais — aucune mesure réelle
+  n'existe pour ces instants.
+- **Le recouvrement est refusé à l'ingestion.** `uq_reading_source` autorise deux lignes au même
+  `(site_id, timestamp)` dès que `source` diffère : rien dans le schéma n'empêche donc un import
+  Mock API manuel avec une fenêtre passée (le script accepte `--start-time`/`--end-time`
+  arbitraires) de dupliquer un point déjà couvert par le CSV. `import_mock_api_history()` appelle
+  `refuse_if_overlaps_historical_dataset()` avant toute écriture : si la fenêtre demandée recouvre
+  au moins une lecture `source='csv'`, l'import est refusé (`ValueError`) plutôt que d'écrire un
+  doublon inter-source silencieux.
+- **Le pipeline ML déduplique en défense.** Le garde-fou ci-dessus protège l'ingestion, pas
+  la lecture : si un recouvrement se produisait malgré tout (import direct en base, contournement
+  du script), `ml/enervision_ml/data.py` ne doit pas casser silencieusement l'hypothèse de
+  `build_features` (« une ligne par `(site_id, timestamp)` »). `load_from_database()` et
+  `load_recent_from_database()` utilisent donc `SELECT DISTINCT ON (site_id, timestamp)`, `source
+  = 'csv'` gagnant sur `'api_history'` en cas d'égalité — l'historique étant une source vérifiée,
+  l'API Mock une entrée hostile (cf. ci-dessus).
+
 ### Qualité des données de l'API Mock
 
 Les valeurs `NULL` ne sont pas remplacées pendant l'ingestion.

etl/airflow/dags/mock_api_import.py

@@ -18,9 +18,15 @@ from airflow.timetables.trigger import CronTriggerTimetable
 # Le backend possède son propre environnement uv dans l'image Airflow (ADR 0008).
 COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m"
 
-# Le pipeline backend et l'API acceptent au maximum 1 000 lectures par site.
-# Cette marge évite de perdre silencieusement une lecture si une heure en contient plus de 60.
-LIMITE_LECTURES = 1000
+# L'API Mock ne renvoie pas un flux au rythme naturel : elle répartit exactement `limit`
+# lectures, espacées uniformément, sur toute la fenêtre demandée (vérifié empiriquement :
+# une fenêtre d'1h avec `limit=1000` renvoie 1000 lectures espacées de 3,6s à l'intérieur de
+# cette heure, pas une lecture horaire). Comme la fenêtre de ce DAG est toujours 1h
+# (`interval=timedelta(hours=1)` ci-dessous), `limit=1` est ce qui produit une lecture par
+# heure, alignée sur l'heure, cohérente avec le grain horaire du reste du schéma
+# (`period_minutes=60`, historique CSV à 1 ligne/heure). Un `limit` plus grand ici fabriquerait
+# des lectures infra-horaires, incompatibles avec les lags positionnels de `build_features`.
+LIMITE_LECTURES = 1
 
 # Deux reprises donnent trois tentatives au total. Même dans le pire cas, l'exécution reste
 # inférieure au pas horaire du DAG.

etl/airflow/tests/test_dags.py

@@ -118,7 +118,7 @@ def test_mock_api_import_uses_the_airflow_data_interval(dagbag: DagBag) -> None:
 
     assert "--start-time \"{{ data_interval_start.strftime('%Y-%m-%dT%H:%M:%S') }}\"" in commande
     assert "--end-time \"{{ data_interval_end.strftime('%Y-%m-%dT%H:%M:%S') }}\"" in commande
-    assert "--limit 1000" in commande
+    assert "--limit 1" in commande
 
 
 @pytest.mark.parametrize("task_id", ["detection", "recommandations"])

ml/enervision_ml/data.py

@@ -47,9 +47,15 @@ NUMERIC_COLUMNS = [
 # `bool` : `astype(bool)` ferait un `True` d'une absence, et les deux chargeurs divergeraient.
 FLAG_COLUMNS = ["is_working_hours"]
 
+# `uq_reading_source` autorise deux lignes au meme (site_id, timestamp) des que `source` differe
+# (cf. `app/etl/mock_api_import.py`, qui refuse desormais d'importer une fenetre deja couverte par
+# le CSV, mais ne protege pas le sens inverse). `build_features` suppose une ligne par
+# (site_id, timestamp) sans doublon : le `DISTINCT ON` l'impose plutot que de la supposer.
+# 'csv' gagne sur 'api_history' en cas de recouvrement, l'historique etant une source verifiee
+# alors que l'API Mock est traitee comme une entree hostile (cf. OWASP API10).
 _READING_QUERY = text(
     """
-    SELECT
+    SELECT DISTINCT ON (r.site_id, r.timestamp)
         r.site_id,
         r.timestamp,
         r.consumption_kwh,
@@ -61,14 +67,14 @@ _READING_QUERY = text(
         s.capacity_kw
     FROM reading r
     JOIN site s ON s.site_id = r.site_id
-    ORDER BY r.site_id, r.timestamp
+    ORDER BY r.site_id, r.timestamp, (r.source = 'csv') DESC, r.reading_id DESC
     """
 )
 
 
 _RECENT_READING_QUERY = text(
     """
-    SELECT
+    SELECT DISTINCT ON (r.site_id, r.timestamp)
         r.site_id,
         r.timestamp,
         r.consumption_kwh,
@@ -81,7 +87,7 @@ _RECENT_READING_QUERY = text(
     FROM reading r
     JOIN site s ON s.site_id = r.site_id
     WHERE r.timestamp >= :since AND r.timestamp <= :until
-    ORDER BY r.site_id, r.timestamp
+    ORDER BY r.site_id, r.timestamp, (r.source = 'csv') DESC, r.reading_id DESC
     """
 )
 

ml/tests/conftest.py

@@ -16,7 +16,7 @@ from collections.abc import Iterator
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 from uuid import uuid4
 
 import lightgbm as lgb
@@ -44,20 +44,30 @@ _INSERT_SITE = text(
     """
 )
 
-# `source = 'api_history'` impose `dataset_id IS NULL` (ck_reading_dataset_source), ce qui evite
-# de creer une ligne `dataset`. `raw_data` est NOT NULL, d'ou le litteral jsonb.
+# `source = 'api_history'` impose `dataset_id IS NULL` (ck_reading_dataset_source) : le defaut
+# `dataset_id=None` evite de creer une ligne `dataset` pour la plupart des tests. `source='csv'`
+# impose l'inverse, d'ou `insere_dataset()` quand un test a besoin de cette source precise.
+# `raw_data` est NOT NULL, d'ou le litteral jsonb.
 _INSERT_READING = text(
     """
     INSERT INTO reading (
-        site_id, timestamp, source, consumption_kwh, temperature_celsius,
+        site_id, timestamp, source, dataset_id, consumption_kwh, temperature_celsius,
         humidity_percent, solar_irradiance_wm2, is_working_hours, raw_data
     ) VALUES (
-        :site_id, :timestamp, :source, :consumption_kwh, :temperature_celsius,
+        :site_id, :timestamp, :source, :dataset_id, :consumption_kwh, :temperature_celsius,
         :humidity_percent, :solar_irradiance_wm2, :is_working_hours, '{}'::jsonb
     )
     """
 )
 
+_INSERT_DATASET = text(
+    """
+    INSERT INTO dataset (dataset_name, archive_sha256, storage_uri, source_timezone, metadata)
+    VALUES (:dataset_name, :archive_sha256, :storage_uri, 'UTC', '{}'::jsonb)
+    RETURNING dataset_id
+    """
+)
+
 _SELECT_PREDICTIONS = text(
     """
     SELECT target_at, predicted_value, status, failure_reason, model_reference
@@ -111,6 +121,25 @@ def insere_site(
     return site_id
 
 
+def insere_dataset(connexion: Connection) -> int:
+    """Ligne `dataset` minimale, requise pour inserer une lecture `source='csv'`
+
+    (`ck_reading_dataset_source` impose `dataset_id IS NOT NULL` pour cette seule source).
+    """
+    marque = uuid4().hex
+    return cast(
+        int,
+        connexion.execute(
+            _INSERT_DATASET,
+            {
+                "dataset_name": f"jeu de test {marque}",
+                "archive_sha256": marque.rjust(64, "0"),
+                "storage_uri": f"file:///test/{marque}.csv",
+            },
+        ).scalar_one(),
+    )
+
+
 def insere_lectures(
     connexion: Connection,
     site_id: str,
@@ -119,6 +148,7 @@ def insere_lectures(
     fin: datetime,
     valeur: float = 50.0,
     source: str = "api_history",
+    dataset_id: int | None = None,
     is_working_hours: bool | None = True,
 ) -> list[datetime]:
     """Grille horaire contigue finissant a `fin`, incluse.
@@ -134,6 +164,7 @@ def insere_lectures(
                 "site_id": site_id,
                 "timestamp": instant,
                 "source": source,
+                "dataset_id": dataset_id,
                 "consumption_kwh": valeur + math.sin(rang / 12.0) * 10.0,
                 "temperature_celsius": 15.0,
                 "humidity_percent": 50.0,
@@ -153,6 +184,7 @@ def insere_lecture(
     instant: datetime,
     consumption_kwh: float | None = 50.0,
     source: str = "api_history",
+    dataset_id: int | None = None,
     is_working_hours: bool | None = True,
 ) -> None:
     """Une lecture isolee, quand le test pilote sa valeur plutot que sa forme."""
@@ -162,6 +194,7 @@ def insere_lecture(
             "site_id": site_id,
             "timestamp": instant,
             "source": source,
+            "dataset_id": dataset_id,
             "consumption_kwh": consumption_kwh,
             "temperature_celsius": 15.0,
             "humidity_percent": 50.0,

ml/tests/test_data_integration.py

@@ -11,7 +11,7 @@ from enervision_ml.data import (
     load_from_database,
     load_recent_from_database,
 )
-from tests.conftest import ANCRAGE, insere_lecture, insere_lectures, insere_site
+from tests.conftest import ANCRAGE, insere_dataset, insere_lecture, insere_lectures, insere_site
 
 pytestmark = pytest.mark.integration
 
@@ -39,6 +39,58 @@ def test_load_from_database_joins_the_site_attributes_to_every_reading(
     assert set(mien["capacity_kw"]) == {250.0}
 
 
+def test_load_from_database_deduplicates_two_sources_at_the_same_instant(
+    connexion_ml: Connection,
+) -> None:
+    # `uq_reading_source` autorise deux lignes au meme (site_id, timestamp) des que `source`
+    # differe : le garde-fou vit dans `mock_api_import.py`, pas dans le schema. Le chargeur ML
+    # doit donc imposer lui-meme "une ligne par (site_id, timestamp)", pas la supposer.
+    site_id = insere_site(connexion_ml)
+    dataset_id = insere_dataset(connexion_ml)
+    insere_lecture(
+        connexion_ml, site_id, instant=ANCRAGE, consumption_kwh=10.0, source="api_history"
+    )
+    insere_lecture(
+        connexion_ml,
+        site_id,
+        instant=ANCRAGE,
+        consumption_kwh=99.0,
+        source="csv",
+        dataset_id=dataset_id,
+    )
+
+    frame = load_from_database(connexion_ml)
+
+    mien = frame[frame["site_id"] == site_id]
+    assert len(mien) == 1
+    assert mien["consumption_kwh"].iloc[0] == 99.0
+
+
+def test_load_recent_from_database_prefers_csv_when_two_sources_share_an_instant(
+    connexion_ml: Connection,
+) -> None:
+    site_id = insere_site(connexion_ml)
+    dataset_id = insere_dataset(connexion_ml)
+    insere_lecture(
+        connexion_ml, site_id, instant=ANCRAGE, consumption_kwh=10.0, source="api_history"
+    )
+    insere_lecture(
+        connexion_ml,
+        site_id,
+        instant=ANCRAGE,
+        consumption_kwh=99.0,
+        source="csv",
+        dataset_id=dataset_id,
+    )
+
+    frame = load_recent_from_database(
+        connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
+    )
+
+    assert len(frame) == 1
+    assert frame["consumption_kwh"].iloc[0] == 99.0
+
+
 def test_load_recent_from_database_excludes_readings_before_the_since_bound(
     connexion_ml: Connection,
 ) -> None: