From cdabc77ccdeb97c0477545533c88e7170a47cd84 Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 13:55:03 -0400 Subject: [PATCH] =?UTF-8?q?feat(forecast):=20pr=C3=A9visions=20au=20d?= =?UTF-8?q?=C3=A9bit=20r=C3=A9el=20du=20journal=20(issue=20#7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit app/forecast.py: fermetures hebdo (8 sem.), débit moy/min/max sur 4 semaines complètes, ETA en fourchette, silence sans historique. Dashboard (épuisement du backlog), bandeau milestone (alerte échéance), page /forecast (sparkline SVG, verdicts), /api/metrics étendu (weekly_closures, throughput_4w). --- app/forecast.py | 119 +++++++++++++++++++++++++++++++++++ app/templates/dashboard.html | 5 ++ app/templates/forecast.html | 76 ++++++++++++++++++++++ 3 files changed, 200 insertions(+) create mode 100644 app/forecast.py create mode 100644 app/templates/forecast.html diff --git a/app/forecast.py b/app/forecast.py new file mode 100644 index 0000000..e3cac4f --- /dev/null +++ b/app/forecast.py @@ -0,0 +1,119 @@ +"""Forecasts (roadmap #7): honest ETAs from the event journal. + +Throughput = issues closed per week, counted from board_events. +Projections are deliberately given as a RANGE (best/worst weekly +rate over the last 4 complete weeks) rather than a fake-precise +date, and are withheld entirely when the journal has no closures +to learn from — "not enough history" is an answer, too. +""" + +from datetime import date, datetime, timedelta + +MONTHS_FR = [ + "janvier", "février", "mars", "avril", "mai", "juin", + "juillet", "août", "septembre", "octobre", "novembre", "décembre", +] +MONTHS_FR_SHORT = [ + "janv.", "févr.", "mars", "avr.", "mai", "juin", + "juil.", "août", "sept.", "oct.", "nov.", "déc.", +] + + +def french_date(d: date, short: bool = False) -> str: + months = MONTHS_FR_SHORT if short else MONTHS_FR + return f"{d.day} {months[d.month - 1]}" + + +def _event_date(event: dict) -> date | None: + raw = str(event.get("created_at") or "") + try: + return datetime.strptime(raw[:10], "%Y-%m-%d").date() + except ValueError: + return None + + +def weekly_closures(events: list[dict], weeks: int = 8, + today: date | None = None) -> list[dict]: + """Closure counts per ISO week, oldest first, current week last.""" + today = today or date.today() + current_monday = today - timedelta(days=today.weekday()) + buckets: list[dict] = [] + for i in range(weeks - 1, -1, -1): + start = current_monday - timedelta(weeks=i) + buckets.append({"start": start, "count": 0}) + by_start = {b["start"]: b for b in buckets} + for event in events: + if event.get("event_type") != "closed": + continue + day = _event_date(event) + if day is None: + continue + monday = day - timedelta(days=day.weekday()) + if monday in by_start: + by_start[monday]["count"] += 1 + return buckets + + +def throughput_stats(weekly: list[dict], last: int = 4) -> dict | None: + """avg/min/max closures per week over the last COMPLETE weeks. + + The current (partial) week is excluded: on Monday morning it + would crush the average for no reason. Returns None when the + complete weeks contain no closure at all. + """ + complete = weekly[:-1][-last:] if len(weekly) > 1 else [] + if not complete: + return None + counts = [b["count"] for b in complete] + if sum(counts) == 0: + return None + return { + "avg": sum(counts) / len(counts), + "min": min(counts), + "max": max(counts), + "weeks": len(counts), + } + + +def estimate(remaining: int, weekly: list[dict], + today: date | None = None) -> dict | None: + """Projected completion for `remaining` items at the real pace. + + Returns {"done": True} when nothing remains, None when there is + no throughput to project from, else dates: central (avg rate), + earliest (best week), latest (worst nonzero week, if any). + """ + today = today or date.today() + if remaining <= 0: + return {"done": True} + stats = throughput_stats(weekly) + if stats is None: + return None + + def at_rate(rate: float) -> date: + days = remaining / rate * 7 + return today + timedelta(days=max(1, round(days))) + + result = { + "done": False, + "avg_per_week": round(stats["avg"], 1), + "central": at_rate(stats["avg"]), + "earliest": at_rate(stats["max"]) if stats["max"] > 0 else None, + "latest": at_rate(stats["min"]) if stats["min"] > 0 else None, + } + return result + + +def eta_label(est: dict | None, done_text: str = "terminé") -> str | None: + """One-line French label for banners and cards (or None).""" + if est is None: + return None + if est.get("done"): + return f"✅ {done_text}" + label = f"à ce rythme ≈ {french_date(est['central'], short=True)}" + if est.get("earliest") and est.get("latest"): + label += ( + f" (fourchette {french_date(est['earliest'], short=True)}" + f" – {french_date(est['latest'], short=True)})" + ) + return label diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html index 9332110..68f1162 100644 --- a/app/templates/dashboard.html +++ b/app/templates/dashboard.html @@ -16,12 +16,14 @@ @@ -86,6 +88,9 @@ {% endif %}