ENI-projet-piscine/docs/architecture/40-data.md

# Données

PostgreSQL 17 avec l'extension TimescaleDB. Le choix, ses alternatives et ses conséquences sont
dans l'[ADR 0001](../adr/0001-postgresql-timescaledb.md), qui fait foi. Ce document décrit le
système qui en découle.

## Ce que couvre ce document

**Douze tables applicatives existent** : six pour l'authentification et six pour les données
d'énergie, dont l'hypertable `reading`.

Les sections marquées `Fait` relèvent du code déjà implémenté. Les sections marquées `Cible`
décrivent les éléments prévus mais pas encore réalisés.

L'ingestion des **mesures** est implémentée pour les deux sources du MVP, le dataset CSV/JSON et
l'API Mock. Celle des **alertes** de l'API Mock, `/alerts`, reste à faire : voir
l'[ADR 0006](../adr/0006-moteur-de-regles-dans-le-backend.md). Les alertes `source='enervision'`,
elles, sont produites par la détection interne, désormais ordonnancée par le DAG Airflow `alertes`
(issue #116). L'orchestration de l'ingestion, les agrégats continus et la compression restent
des cibles. La rétention de `reading` est faite : chaque chunk plus vieux que la borne est exporté
vers Garage puis supprimé (issue #36, section « Rétention et archivage » ci-dessous).

## Trois emplacements, trois rôles

C'est la règle que l'ADR 0001 existe surtout pour fixer. La confondre coûte cher : un script placé
au mauvais endroit ne s'exécute jamais, ou s'exécute deux fois.

| Emplacement | Contenu | Quand ça s'exécute |
|---|---|---|
| `db/init/` | Extensions, bases annexes | **Une seule fois**, à la première initialisation du conteneur, quand `PGDATA` est vide. Ne rejoue jamais |
| `db/migrations/` | SQL versionné qui ne découle pas du schéma applicatif : compression. La rétention de `reading` n'y est pas : une politique TimescaleDB ignorerait l'export, elle vit dans `apps/backend/app/etl/reading_retention.py`, ordonnancée par le DAG `retention` ([ADR 0019](../adr/0019-stockage-objet-garage-et-cycle-de-vie-des-mesures.md)) | À la main, aujourd'hui vide |
| `apps/backend/alembic/` | Le schéma exposé par l'API, et lui seul | `alembic upgrade head`, c'est `Base.metadata` qui fait foi |

Une hypertable relève des deux derniers : **Alembic crée la table, et le `create_hypertable()`
vit dans la même révision**. Les séparer rendrait le schéma irreproductible depuis un seul
`alembic upgrade head`.

Détail de `db/init/` et du piège de montage : [`db/README.md`](../../db/README.md).

## Ce qui existe

Statut : `Fait`.

- `db/init/100-extensions.sql` crée l'extension `timescaledb`.
- `db/init/110-test-database.sql` crée `enervision_test`, dont le nom est attendu en dur par
  `apps/backend/tests/conftest.py`.
- Six révisions Alembic sont actuellement appliquées.
- La première, `5353c0e4f094`, **ne crée aucune table** : elle établit `alembic_version`
  et refuse de s'appliquer si l'extension TimescaleDB manque.
- Les révisions suivantes créent les tables liées à l'authentification :
  `app_user`, `login_attempt`, `audit_log` et `refresh_token`.
- La révision `e6d2026091501` crée six des sept tables Data et déclare l'hypertable `reading`.
- La révision `d3f1a2b7c904` ajoute `drift_report`, la septième.
- La révision `c0adab96238c` ajoute les tables `password_reset_attempt`
  et `password_reset_token`.

La garde de la première migration est :

```sql
IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb') THEN
    RAISE EXCEPTION 'extension timescaledb absente, voir db/init et db/README.md';
END IF;
```

Cette garde forme paire avec le 503 de `/api/v1/health/ready`. Un bootstrap sauté ne se voit pas
au démarrage de l'API : ces deux gardes le rendent visible tôt, des deux côtés.

## Cycle de vie d'une mesure

Statut : `Partiellement fait`.

Les mécanismes d'ingestion sont maintenant implémentés pour les deux sources de données du MVP :

- le dataset historique CSV/JSON avec `historical_import.py` ;
- l'API Mock avec `mock_api_import.py`.

Les traitements sont actuellement exécutables directement depuis le backend.

L'orchestration avec Apache Airflow reste une cible, tout comme les agrégats continus et la
compression. La rétention est faite : le DAG `retention` exporte chaque chunk de `reading` plus
vieux que `READING_RETENTION_DAYS` vers Garage, puis le supprime.

```mermaid
flowchart LR
  csv["CSV + JSON"] --> hist["historical_import.py"]
  mock["API Mock"] --> api["mock_api_import.py"]

  hist --> hy[("Hypertable reading")]
  api --> hy

  airflow["Airflow"] -.-> hist
  airflow -.-> api

  hy -.-> agg[("Agrégat continu")]
  hy -.-> comp["Compression"]
  hy --> ret["Rétention : export CSV gzip vers Garage, puis drop_chunks"]
  ret --> garage[("Garage S3")]

  agg -.-> backend["API FastAPI"]
  agg -.-> graf["Grafana"]
```

Les flèches pleines représentent les traitements actuellement implémentés.

Les flèches pointillées représentent les éléments encore prévus comme cibles.

Les lectures de l'API et de Grafana viseront l'agrégat continu, pas la table brute : c'est tout
l'intérêt de TimescaleDB, et cela doit rester vrai quand les volumes augmenteront.

### Rétention et archivage (issue #36)

Statut : `Fait`.

`apps/backend/app/etl/reading_retention.py`, ordonnancé chaque nuit à 03:20 UTC par le DAG
`retention`, sélectionne dans `timescaledb_information.chunks` les chunks de `reading` dont
`range_end` est antérieur ou égal à `now() - READING_RETENTION_DAYS` : seul un chunk entièrement
plus vieux que la borne est éligible. Chaque chunk est lu via l'hypertable (`WHERE timestamp >=
range_start AND timestamp < range_end`), jamais via la table interne, sérialisé en CSV gzip
reproductible (jsonb et tableaux en JSON trié), puis écrit chiffré SSE-C sous la clé
`reading/<année>/reading_<début>_<fin>.csv.gz`, bornes UTC compactes. L'objet est relu et son
sha256 comparé à celui du corps envoyé ; en cas d'écart le chunk est conservé. Seulement alors
`drop_chunks('reading', older_than => range_end, newer_than => range_start)` supprime ce chunk et
lui seul, dans une transaction dédiée et courte : `drop_chunks` pose un verrou exclusif sur
`reading`, `site` et `dataset` jusqu'au COMMIT. Un objet déjà présent avec le même sha256 n'est pas
réécrit et un chunk supprimé n'est plus éligible : rejouer le DAG est sans effet, `--dry-run` liste
et mesure sans rien écrire. Le premier passage en production archive les chunks de janvier à
septembre 2023 ; la démo, ancrée au 31/12/2024, n'est pas touchée. Restauration manuelle :
télécharger l'objet avec la clé SSE-C, `gunzip`, `COPY` dans `reading` ; aucune commande fournie.

## Tables d'authentification

Statut : `Fait`.

Elles ne sont pas des séries temporelles et n'ont donc rien à voir avec les hypertables ;
elles vivent dans `apps/backend/alembic/`, qui porte le schéma exposé par l'API.

```mermaid
erDiagram
  APP_USER ||--o{ REFRESH_TOKEN : ouvre
  APP_USER ||--o{ PASSWORD_RESET_TOKEN : recoit

  APP_USER {
    uuid id PK
    string email UK
    text password_hash
    text role
    text kind
    bool is_active
    bool must_change_password
    timestamptz credentials_changed_at
  }

  REFRESH_TOKEN {
    uuid id PK
    uuid family_id
    uuid user_id FK
    bytea token_hash UK
    timestamptz expires_at
    timestamptz rotated_at
    timestamptz revoked_at
    text revoked_reason
    uuid replaced_by
  }

  LOGIN_ATTEMPT {
    bigint id PK
    timestamptz occurred_at
    string email_tried
    inet client_ip
    text outcome
  }

  AUDIT_LOG {
    bigint id PK
    timestamptz occurred_at
    uuid actor_id
    text actor_email
    text action
    jsonb detail
  }

  PASSWORD_RESET_ATTEMPT {
    bigint id PK
    timestamptz occurred_at
    string email_tried
    inet client_ip
  }

  PASSWORD_RESET_TOKEN {
    uuid id PK
    uuid user_id FK
    bytea token_hash UK
    timestamptz issued_at
    timestamptz expires_at
    timestamptz consumed_at
    inet client_ip
    text user_agent
  }
```

Six choix de modélisation portent une intention et se défendent seuls :

- **`app_user` et non `user`** : `user` est un mot réservé PostgreSQL, raccourci de
  `CURRENT_USER`. Le nom rappelle en prime qu'il s'agit d'un compte applicatif, par opposition
  au rôle PostgreSQL qui portera le cantonnement de l'ETL.
- **`credentials_changed_at`, une seule colonne**, couvre le changement de mot de passe, le
  changement de rôle et la désactivation. Un compteur de version ne dirait rien à un humain qui
  lit un audit.
- **`refresh_token.expires_at` est absolu et hérité** du prédécesseur à chaque rotation. S'il
  glissait, la promesse de sept jours serait fictive et une session active ne finirait jamais.
- **`audit_log.actor_id` n'a aucune clé étrangère**, et `actor_email` comme `actor_role` sont
  dénormalisés. Une contrainte `ON DELETE SET NULL` déclencherait un `UPDATE` que le déclencheur
  d'ajout seul refuserait. Voir l'[ADR 0004](../adr/0004-journal-d-audit-en-ajout-seul.md).
- **`password_reset_token` ne stocke que l'empreinte du jeton**, jamais sa valeur. Une fuite de
  la table ne donne donc rien à rejouer.
- **`password_reset_attempt` est séparée de `audit_log`** : son volume est piloté par le
  demandeur, comme celui de `login_attempt`, donc elle doit pouvoir se purger.

`audit_log` porte deux déclencheurs qui refusent `UPDATE`, `DELETE` et `TRUNCATE`. Elle n'est
donc **pas** une hypertable : une politique de rétention émettrait des `DELETE` qu'ils
refuseraient. `login_attempt`, à l'inverse, est faite pour se purger, puisque son volume est
piloté par l'attaquant.

## Gabarit de révision créant une hypertable

Conforme à la règle de l'ADR 0001 : table et hypertable dans la même révision.

La révision `e6d2026091501` en est l'exemple réel, réduit ici à l'essentiel.

```python
def upgrade() -> None:
    op.create_table(
        "reading",
        sa.Column("reading_id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("site_id", sa.Text(), nullable=False),
        sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False),
        sa.PrimaryKeyConstraint("reading_id", "timestamp"),
    )
    op.execute(
        "SELECT create_hypertable('reading', by_range('timestamp'), "
        "create_default_indexes => FALSE)"
    )


def downgrade() -> None:
    op.drop_table("reading")
```

La clé primaire inclut la colonne de temps parce que TimescaleDB l'exige : toute contrainte
unique d'une hypertable doit porter la colonne de partitionnement, et une clé sur le seul
`reading_id` serait refusée par `create_hypertable`.

`create_default_indexes => FALSE` écarte l'index que TimescaleDB pose d'office sur la seule
colonne de temps : les index déclarés dans la révision le couvrent déjà.

`drop_table` suffit au retour arrière : supprimer la table supprime l'hypertable et ses partitions.

## Conventions

- **Noms au singulier**, en minuscules, sans préfixe de table : `app_user`, `reading`.
- **Toute colonne de temps en `timestamptz`.** Jamais de `timestamp` nu : une mesure sans fuseau
  devient ininterprétable dès le premier changement d'heure.
- **La colonne de partitionnement entre dans la clé primaire.** Dans `reading` elle s'appelle
  `timestamp` : c'est un nom de colonne, son type reste `timestamptz`.
- **Les politiques de compression** vont dans `db/migrations/`, pas dans Alembic : elles ne
  découlent pas du schéma applicatif. La rétention de `reading` est un traitement ETL
  (`reading_retention.py`), pas une politique TimescaleDB : elle doit exporter avant de supprimer.
- **Tout modèle doit être importé dans `app/models/__init__.py`**, sans quoi
  `alembic revision --autogenerate` ne le voit pas et génère un `drop` de sa table.

## Questions ouvertes

Elles relèvent du jalon J2, « valider le périmètre retenu ». Le schéma et l'ingestion sont
livrés : ce qui suit porte sur leur exploitation, plus sur leur forme.

- **Quelle granularité** conserver à long terme à l'ingestion : seconde, minute ou quart d'heure.
- **Quels agrégats continus** créer et sur quelles fenêtres.
- **Quelle profondeur de rétention** : répondu par l'issue #36. Trois ans en base chaude par
  défaut (`READING_RETENTION_DAYS`, 1095 jours) ; au-delà, les chunks sont archivés en CSV gzip
  sur Garage, chiffrés SSE-C, puis supprimés. Reste ouvert : à partir de quand compresser.
- **Multi-tenant ou non** : un site appartient-il à un client et faut-il cloisonner les lectures.

## Modélisation détaillée des données

Cette modélisation prend en compte :

- les fichiers CSV historiques ;
- leurs métadonnées JSON ;
- les données de l'API Mock.

Elle comprend sept tables Data, depuis le stockage des mesures jusqu'aux recommandations
proposées à l'utilisateur, et jusqu'au suivi de la dérive du modèle.

### Schéma de données

Le diagramme ci-dessous présente les tables et leurs relations.

La révision `e6d2026091501` les crée.

![Schéma de données EnerVision](images/EnerVision-schema-donnees.png)

*Figure : Modélisation des données EnerVision.*

### Description des tables

Chaque table remplit un rôle précis dans le traitement et l'exploitation des données.

| Table | Rôle | Origine des informations |
|---|---|---|
| `dataset` | Identifier les jeux historiques, retrouver leurs fichiers et conserver leurs métadonnées | Archive CSV/JSON et informations ajoutées lors de l'import |
| `site` | Regrouper les informations des sites : identifiant, nom, type et caractéristiques disponibles | CSV et API Mock `/api/v1/sites` |
| `reading` | Stocker les mesures, leur provenance, leur qualité et les éventuelles valeurs imputées | CSV et API Mock `/current` et `/readings` |
| `prediction` | Conserver les prévisions, leur période cible et la référence du modèle utilisé | Traitements ML d'EnerVision |
| `alert` | Enregistrer les alertes, leur type, leur gravité et leur message | API Mock `/alerts` et détections EnerVision |
| `recommendation` | Proposer des actions et expliquer la règle qui les motive | Règles métier d'EnerVision |
| `drift_report` | Suivre l'écart entre prévisions et réalisé, par site et tous sites confondus | Surveillance de dérive d'EnerVision |

Le scoring (`ml_score`) charge le modèle depuis un fichier local (`models/lightgbm-consumption.txt`)
et trace son empreinte SHA-256 dans `prediction.model_reference`. Il ne lit aucune version depuis
le Model Registry MLflow (`ml/`) : ce registre sert aujourd'hui à la traçabilité des
entraînements, pas au déploiement du modèle de scoring.
Les anomalies historiques décrites dans les JSON sont conservées dans `dataset.metadata`.

Elles servent à l'analyse des données et ne sont pas considérées comme des alertes actuelles.

Les lignes de `drift_report` sont écrites par `app.monitoring.drift`, ordonnancé par le DAG
`derive`. Une ligne dont le `site_id` est `NULL` porte le résultat global, tous sites confondus :
c'est pourquoi l'unicité passe par un index sur `coalesce(site_id, '')` et non par une contrainte,
qui ne dédoublonnerait jamais deux lignes globales. Le calcul, ses seuils et ce qu'il refuse de
comparer sont dans l'[ADR 0013](../adr/0013-surveillance-de-derive-dans-le-backend.md).

Les lignes de `recommendation` sont écrites par le moteur de règles du backend
(`app/services/recommendation_rules.py`), déclenché par `POST /api/v1/recommendations/generate`,
par `make recommendations`, ou par la seconde tâche du DAG `alertes`, à partir des alertes déjà en
base. Le couple `(alert_id, rule_reference)` est unique : rejouer le moteur sur les mêmes alertes
n'ajoute aucune ligne.

### Relations entre les tables

- Un site possède plusieurs mesures, prévisions et alertes.
- Un jeu de données historique contient plusieurs mesures CSV.
- Les mesures API ne sont pas rattachées à un dataset historique.
- Une alerte peut être associée à une prévision du même site.
- Une alerte peut donner lieu à plusieurs recommandations.
- Un site possède plusieurs rapports de dérive ; un rapport global n'est rattaché à aucun site.

## Ingestion des données historiques

Statut : `Fait`.

Le MVP EnerVision initialise les données énergétiques à partir du dataset fourni dans le cadre
du projet.

Le dataset de référence contient 122 647 mesures issues de 7 sites et couvre la période
du 1er janvier 2023 au 31 décembre 2024.

Les fichiers sources CSV et JSON sont nécessaires uniquement pour l'initialisation des données.

Ils ne sont pas versionnés dans Git et sont placés localement dans `data/raw/`.

### Architecture du flux historique

```text
Dataset CSV + métadonnées JSON
              |
              v
     historical_import.py
              |
       +------+------+
       |             |
       v             v
   Validation     SHA-256
       |          Traçabilité
       +------+------+
              |
              v
      Normalisation
      + qualité data
              |
              v
    Chargement par batches
              |
              v
 PostgreSQL / TimescaleDB
       |      |       |
       v      v       v
    dataset  site   reading
```

Le pipeline est développé en Python.

Pandas est utilisé pour l'extraction, la validation et la préparation des données.

SQLAlchemy Async assure le chargement transactionnel dans PostgreSQL/TimescaleDB.

Une empreinte SHA-256 permet d'identifier le dataset utilisé et d'assurer sa traçabilité.

Les valeurs manquantes sont conservées pendant l'ingestion afin de préserver les données sources.

Aucune imputation n'est réalisée à cette étape.

Le chargement des mesures est effectué par batches de 1 000 lignes.

Les données provenant du dataset CSV sont identifiées par :

```text
source = "csv"
dataset_id = identifiant du dataset
```

### Résultats validés pour l'historique

Le chargement de référence a permis d'obtenir :

- 1 dataset ;
- 7 sites ;
- 122 647 mesures ;
- 0 doublon détecté dans le dataset source.

L'idempotence a également été vérifiée par une deuxième exécution du pipeline :
aucune nouvelle mesure n'a été créée et le nombre de `reading` est resté à 122 647.

La procédure détaillée d'installation, d'exécution, de validation et de contrôle du pipeline
est disponible dans `etl/README.md`.

## Ingestion depuis l'API Mock

Statut : `Fait`.

La deuxième source du pipeline Data est l'API Mock EnerVision.

Le traitement est implémenté dans :

```text
apps/backend/app/etl/mock_api_import.py
```

### Endpoints utilisés

Le pipeline récupère les informations des sites depuis :

```text
GET /api/v1/sites
```

puis les mesures historiques simulées depuis :

```text
GET /api/v1/readings
```

Pour `/api/v1/readings`, les informations suivantes sont envoyées :

```text
site_id
start_time
end_time
limit
```

Les paramètres de ligne de commande disponibles pour l'import sont :

```text
--start-time
--end-time
--dry-run
```

**Piège sur `limit`, corrigé dans le code plutôt que documenté** : 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, la première au tout début de la fenêtre (vérifié
empiriquement en interrogeant directement l'API). Une fenêtre d'une heure avec `limit=1000`, le
réglage d'origine, renvoyait donc 1000 lectures espacées de 3,6 secondes à l'intérieur de cette
heure, pas une lecture horaire, incompatible avec les lags positionnels de `build_features`.
Plutôt que documenter la règle « `limit` = nombre d'heures de la fenêtre » et compter sur chaque
appelant pour la respecter, `limit_for_window()` la porte : `import_mock_api_history()` calcule
`limit` depuis la fenêtre reçue, refuse une fenêtre dont `start_time` ne tombe pas pile sur
l'heure (c'est elle qui ancre l'alignement), et refuse un intervalle de plus de 1000 heures (le
plafond `limit` de l'API). `--limit` n'existe donc plus côté CLI. Deux formes de fenêtre sont
gérées : un multiple entier d'heures (`limit` = ce nombre d'heures, une lecture par heure
espacée d'1h pile, chemin du backfill manuel) ou une fenêtre plus courte qu'une heure, ou qui
n'en est pas un multiple entier (`limit=1`, seule valeur qui reste alignée quand l'espacement
`durée / limit` ne peut valoir 1h pile). Le DAG `mock_api_import` est dans ce second cas : il
demande la fenêtre `[heure pile précédant le déclenchement, instant du déclenchement)`, plus
courte qu'une heure, plutôt que l'intervalle Airflow `[data_interval_start, data_interval_end)`
tel quel (`[:45, :45)`) qui aurait placé l'unique lecture à :45, hors de la grille horaire du
reste du schéma.

### Flux d'ingestion API Mock

```text
        API Mock
           |
     +-----+------+
     |            |
     v            v
   /sites      /readings
     |            |
     +-----+------+
           |
           v
 mock_api_import.py
           |
           v
 Transformation
 + qualité data
           |
           v
PostgreSQL / TimescaleDB
     |          |
     v          v
   site       reading
```

Les informations des sites sont insérées ou mises à jour dans `site`.

Les mesures sont enregistrées dans l'hypertable `reading` avec :

```text
source = "api_history"
dataset_id = NULL
```

Les données provenant de l'API Mock ne sont donc pas associées à un enregistrement de la table
`dataset`.

La réponse source reçue depuis l'API est conservée dans :

```text
raw_data
```

### Frontière de confiance avec l'API Mock

L'API Mock de l'école n'a aucune authentification et expose un endpoint mutatif à quiconque. Sa
réponse est donc traitée comme une entrée hostile, conformément à API10 dans
[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.

Cinq garde-fous, tous dans `mock_api_import.py` :

| Garde-fou | Mise en œuvre |
|---|---|
| Timeout | `APP_MOCK_API_TIMEOUT_SECONDS`, dix secondes par défaut |
| 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`
descend à `degraded`. Une `data_quality` que `ck_reading_quality` refuserait devient `NULL`
plutôt que de faire échouer le lot entier. Dans tous les cas `raw_data` conserve la réponse
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. Conséquence pour le ML, pas nouvelle mais que ce trou rend
  définitive : `build_features()` calcule ses lags par `shift(n)` positionnel, et `train.py`
  n'écarte que les lignes où `lag_168h` est `NaN`. Pour un site présent dans les deux sources, les
  168 premières lectures `api_history` qui suivent le trou héritent donc de lags et de moyennes
  glissantes calculés sur décembre 2024 (et tant que l'ingestion a moins de 7 jours, c'est le cas
  de toutes les lectures). Même effet, plus ponctuel, pour chaque heure que le DAG manque
  (`mock_api_import` en échec, Airflow arrêté). Aucun garde-fou ne détecte aujourd'hui un lag
  calculé sur un écart réel différent de celui attendu ; issue de suivi à ouvrir.
- **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, y compris en `--dry-run` (le
  contrôle est en lecture seule) et avant le moindre appel à l'API Mock : 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 contrôle ne porte que sur la fenêtre demandée,
  pas sur les lectures reçues : `fetch_readings()` écarte donc toute lecture dont le `timestamp`
  déborde de `[start_time, end_time)`, pour qu'une réponse hors fenêtre (bug du mock, ou hostile)
  ne puisse pas le contourner. Ce contrôle compare des instants, pas des chaînes : `parse_datetime()`
  pose `tzinfo=UTC` sur une entrée sans fuseau (même pattern que `_vers_utc()` dans
  `app/services/reading.py`), sans quoi l'encodeur `timestamptz` d'asyncpg lirait un datetime naïf
  dans le fuseau local du **processus**, correct dans le conteneur Airflow (UTC) mais décalé pour
  un import manuel lancé depuis un poste en Europe/Paris.
- **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). **Cette préférence est spécifique au chargeur
  ML.** `GET /readings` renvoie les deux lignes sans les fusionner, et `DriftRepository` /
  `ReadingRepository.latest_by_site()` / `.latest_for_site()` départagent par `reading_id` le plus
  grand (en pratique la ligne insérée en dernier, pas forcément `csv`) : en cas de recouvrement, la
  dérive comparerait alors une prévision à une valeur différente de celle sur laquelle le modèle a
  appris. Pas d'incohérence aujourd'hui tant que le recouvrement reste refusé à l'ingestion ; à
  aligner si ce garde-fou devait un jour être contourné.

### Qualité des données de l'API Mock

Les valeurs `NULL` ne sont pas remplacées pendant l'ingestion.

Les informations suivantes fournies par l'API sont conservées :

```text
data_quality
null_reasons
```

Cette conservation permet de distinguer une valeur manquante d'une valeur réelle égale à zéro
et de garder les informations liées aux éventuelles défaillances de capteurs.

Aucune imputation n'est réalisée pendant cette phase :

```text
imputed_values = NULL
imputation_method = NULL
```

### Validation de l'import API Mock

Un scénario de validation a été exécuté pour les 7 sites sur la période :

```text
15/06/2024 12:00 UTC
à
15/06/2024 13:00 UTC
```

avec :

```text
limit = 60
```

Résultat :

```text
7 sites
60 lectures par site
420 lectures récupérées
```

Les données ont été chargées dans PostgreSQL/TimescaleDB puis contrôlées directement en base.

Les contrôles ont confirmé :

- `source = "api_history"` ;
- `dataset_id = NULL` ;
- la conservation des valeurs `NULL` ;
- la conservation de `data_quality` ;
- la conservation de `null_reasons` ;
- la conservation de `raw_data`.

L'idempotence a été vérifiée en rejouant le même import.

Une mesure déjà présente n'est pas ajoutée une seconde fois.

Les tests automatisés couvrent également :

- la récupération des sites ;
- les paramètres envoyés à `/api/v1/readings` ;
- les réponses HTTP en erreur ;
- le format de la réponse ;
- la transformation des mesures ;
- les valeurs manquantes ;
- la qualité des données ;
- la conservation des données sources ;
- l'idempotence en base.

## Évolution prévue

La prochaine étape consiste à orchestrer les deux mécanismes d'ingestion avec Apache Airflow.

```text
CSV / JSON ----------------+
                           |
                           v
                 +------------------+
                 |     Airflow      |
                 +------------------+
                           |
          +----------------+----------------+
          |                                 |
          v                                 v
historical_import.py               mock_api_import.py
          |                                 |
          +----------------+----------------+
                           |
                           v
                PostgreSQL / TimescaleDB
```

Airflow servira à :

- planifier les traitements ;
- définir leur ordre d'exécution ;
- suivre leur état ;
- gérer et remonter les erreurs ;
- faciliter les exécutions récurrentes.

Airflow ne remplacera pas la logique ETL déjà implémentée.

Les scripts Python resteront responsables de l'extraction, de la validation, de la transformation
et du chargement des données.

Le pipeline servira ensuite de base à la préparation des données nécessaires au modèle
de Machine Learning.