# meteo/gaps.py from __future__ import annotations from dataclasses import dataclass from typing import List import pandas as pd @dataclass(frozen=True) class TimeGap: """ Représente une période pendant laquelle il manque des points dans une série temporelle censée être échantillonnée à intervalle régulier. """ # Timestamp du dernier point avant le trou before: pd.Timestamp # Timestamp du premier point après le trou after: pd.Timestamp # Premier timestamp "attendu" manquant missing_start: pd.Timestamp # Dernier timestamp "attendu" manquant missing_end: pd.Timestamp # Nombre d'intervalles manquants (par ex. 3 => 3 minutes manquantes) missing_intervals: int # Durée totale du gap (after - before) duration: pd.Timedelta def find_time_gaps( df: pd.DataFrame, expected_freq: pd.Timedelta = pd.Timedelta(minutes=1), ) -> List[TimeGap]: """ Détecte les gaps temporels dans un DataFrame indexé par le temps. Un "gap" est un intervalle entre deux timestamps successifs strictement supérieur à `expected_freq`. Exemple : si expected_freq = 1 minute et qu'on passe de 10:00 à 10:05, on détecte un gap avec 4 minutes manquantes (10:01, 10:02, 10:03, 10:04). """ if not isinstance(df.index, pd.DatetimeIndex): raise TypeError( "find_time_gaps nécessite un DataFrame avec un DatetimeIndex." ) index = df.index.sort_values() diffs = index.to_series().diff() gaps: list[TimeGap] = [] for i, delta in enumerate(diffs.iloc[1:], start=1): if delta <= expected_freq: continue before_ts = index[i - 1] after_ts = index[i] # Nombre d'intervalles manquants (ex: 5min / 1min => 4 intervalles manquants) missing_intervals = int(delta // expected_freq) - 1 missing_start = before_ts + expected_freq missing_end = after_ts - expected_freq gap = TimeGap( before=before_ts, after=after_ts, missing_start=missing_start, missing_end=missing_end, missing_intervals=missing_intervals, duration=delta, ) gaps.append(gap) return gaps