feat(analysis): _detect_gaps su timestamp consecutivi (> 20 min)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
root
2026-05-13 09:29:48 +00:00
parent 9e2216d202
commit ea5c612446
2 changed files with 50 additions and 0 deletions
+16
View File
@@ -124,3 +124,19 @@ def _expected_ticks(since: datetime, until: datetime) -> int:
# aligned multiples and would mis-count non-aligned spans. # aligned multiples and would mis-count non-aligned spans.
span_seconds = (until - first_tick).total_seconds() span_seconds = (until - first_tick).total_seconds()
return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60)) return math.ceil(span_seconds / (_TICK_INTERVAL_MIN * 60))
def _detect_gaps(timestamps: list[datetime]) -> tuple[GapRecord, ...]:
"""Return gaps where consecutive timestamps differ by > threshold."""
out: list[GapRecord] = []
for prev, nxt in zip(timestamps, timestamps[1:], strict=False):
delta_min = int((nxt - prev).total_seconds() // 60)
if delta_min > _GAP_THRESHOLD_MIN:
out.append(
GapRecord(
prev_timestamp=prev,
next_timestamp=nxt,
gap_minutes=delta_min,
)
)
return tuple(out)
+34
View File
@@ -43,3 +43,37 @@ def test_expected_ticks_exact_quarter_boundary_is_excluded() -> None:
until = datetime(2026, 5, 12, 12, 45, tzinfo=UTC) until = datetime(2026, 5, 12, 12, 45, tzinfo=UTC)
# Ticks at 12:00, 12:15, 12:30 → 3 (12:45 excluded) # Ticks at 12:00, 12:15, 12:30 → 3 (12:45 excluded)
assert _expected_ticks(since, until) == 3 assert _expected_ticks(since, until) == 3
from cerbero_bite.analysis.data_audit import _detect_gaps # noqa: PLC2701
def test_detect_gaps_returns_empty_when_no_gap() -> None:
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 15, tzinfo=UTC),
datetime(2026, 5, 12, 12, 30, tzinfo=UTC),
]
assert _detect_gaps(ts) == ()
def test_detect_gaps_flags_above_threshold() -> None:
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 45, tzinfo=UTC), # 45-min gap
datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
]
gaps = _detect_gaps(ts)
assert len(gaps) == 1
assert gaps[0].gap_minutes == 45
assert gaps[0].prev_timestamp == ts[0]
assert gaps[0].next_timestamp == ts[1]
def test_detect_gaps_ignores_threshold_boundary() -> None:
# 20-min gap is exactly the threshold → NOT flagged (strict >)
ts = [
datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
datetime(2026, 5, 12, 12, 20, tzinfo=UTC),
]
assert _detect_gaps(ts) == ()