| """Reproducible resolution summaries and period comparisons. |
| |
| Pandas computes statistics; independent fixed SQL templates must agree before |
| a result is returned. No SQL or Python supplied by a model is executed. |
| """ |
| from contextlib import closing |
| from datetime import date |
| from hashlib import sha256 |
| import json |
| from math import isclose |
| import sqlite3 |
| from typing import Literal |
| import pandas as pd |
| from pydantic import ConfigDict, Field, field_validator, model_validator |
| from service_analysis.quality import PreparedDataset |
| from service_analysis.schemas import Contract, parse_date |
|
|
| Grouping = Literal["overall", "category", "region"] |
| METHOD = "resolution-summary-v1" |
| REL_TOL = 1e-10 |
| ABS_TOL = 1e-9 |
|
|
|
|
| class ResultContract(Contract): |
| model_config = ConfigDict(allow_inf_nan=False) |
|
|
|
|
| class Period(Contract): |
| start: date |
| end: date |
|
|
| @field_validator("start", "end", mode="before") |
| @classmethod |
| def dates(cls, value): |
| return parse_date(value) |
|
|
| @model_validator(mode="after") |
| def ordered(self): |
| if self.end < self.start: |
| raise ValueError("Period end precedes start") |
| return self |
|
|
| @property |
| def days(self): |
| return (self.end - self.start).days + 1 |
|
|
|
|
| class StatsRow(ResultContract): |
| group: str |
| sample_size: int = Field(ge=0) |
| mean_hours: float | None |
| median_hours: float | None |
|
|
| @model_validator(mode="after") |
| def population(self): |
| if self.sample_size == 0: |
| if self.mean_hours is not None or self.median_hours is not None: |
| raise ValueError("Empty population cannot have duration statistics") |
| elif self.mean_hours is None or self.median_hours is None: |
| raise ValueError("Nonempty population requires duration statistics") |
| if any(value is not None and value < 0 for value in (self.mean_hours, self.median_hours)): |
| raise ValueError("Duration statistics cannot be negative") |
| return self |
|
|
|
|
| class SummaryResult(ResultContract): |
| result_id: str |
| dataset_id: str |
| method: str = METHOD |
| grouping: Grouping |
| period: Period |
| rows: tuple[StatsRow, ...] |
| population_size: int |
| excluded_classification_count: int |
| included_count: int |
| dataset_excluded_count: int |
| dataset_duplicate_count: int |
| status: Literal["complete", "partial", "no_observations"] |
| warnings: tuple[str, ...] |
| sql_verified: Literal[True] = True |
| units: Literal["elapsed_calendar_hours"] = "elapsed_calendar_hours" |
|
|
|
|
| class ChangeRow(ResultContract): |
| group: str |
| baseline_count: int |
| comparison_count: int |
| count_delta: int |
| mean_change_hours: float | None |
| median_change_hours: float | None |
| mean_change_percent: float | None |
| median_change_percent: float | None |
|
|
|
|
| class ComparisonResult(ResultContract): |
| result_id: str |
| dataset_id: str |
| method: str = "period-comparison-v1" |
| baseline: SummaryResult |
| comparison: SummaryResult |
| rows: tuple[ChangeRow, ...] |
| status: Literal["complete", "partial", "no_observations"] |
| warnings: tuple[str, ...] |
|
|
|
|
| class CalculationMismatch(RuntimeError): |
| """Independent calculation engines disagreed; publication is blocked.""" |
|
|
|
|
| def _identity(prefix, payload): |
| encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False) |
| return prefix + sha256(encoded.encode()).hexdigest() |
|
|
|
|
| def _dataset_id(dataset): |
| return _identity("dataset-", { |
| "coverage": dataset.coverage.model_dump(mode="json"), |
| "profile": dataset.profile.model_dump(mode="json"), |
| "policy": dataset.policy.model_dump(mode="json"), |
| "requests": [row.model_dump(mode="json") for row in sorted(dataset.requests, key=lambda r: r.request_id)], |
| "operations": [row.model_dump(mode="json") for row in sorted(dataset.operations, key=lambda r: (r.date, r.region))], |
| }) |
|
|
|
|
| def _check_arguments(dataset, period, grouping): |
| if not isinstance(dataset, PreparedDataset) or not isinstance(period, Period): |
| raise ValueError("Validated dataset and Period are required") |
| if grouping not in ("overall", "category", "region"): |
| raise ValueError("Unsupported grouping") |
| dataset.require("resolution" if grouping == "overall" else grouping) |
| if period.start < dataset.coverage.resolution_start or period.end > dataset.coverage.resolution_end: |
| raise ValueError("Period lies outside declared resolution coverage") |
|
|
|
|
| def _records(dataset): |
| return [(row.resolved_at.date().isoformat(), row.category, row.region, row.resolution_hours) |
| for row in dataset.requests] |
|
|
|
|
| def _pandas_rows(records, period, grouping): |
| frame = pd.DataFrame(records, columns=["resolved_date", "category", "region", "hours"]) |
| selected = frame[(frame.resolved_date >= period.start.isoformat()) & |
| (frame.resolved_date <= period.end.isoformat())].copy() |
| population = len(selected) |
| excluded = 0 |
| if grouping == "overall": |
| selected["bucket"] = "All" |
| else: |
| excluded = int(selected[grouping].isna().sum()) |
| selected = selected[selected[grouping].notna()].copy() |
| selected["bucket"] = selected[grouping] |
| rows = tuple(StatsRow(group=str(group), sample_size=len(values), |
| mean_hours=float(values.mean()), median_hours=float(values.median())) |
| for group, values in selected.groupby("bucket", sort=True)["hours"]) |
| if grouping == "overall" and not rows: |
| rows = (StatsRow(group="All", sample_size=0, mean_hours=None, median_hours=None),) |
| return rows, population, excluded |
|
|
|
|
| def _sql_rows(records, period, grouping): |
| |
| expression = {"overall": "'All'", "category": "category", "region": "region"}[grouping] |
| nonnull = "" if grouping == "overall" else f" AND {expression} IS NOT NULL" |
| query = f""" |
| WITH ranked AS ( |
| SELECT {expression} AS bucket, hours, |
| ROW_NUMBER() OVER (PARTITION BY {expression} ORDER BY hours) AS rn, |
| COUNT(*) OVER (PARTITION BY {expression}) AS n |
| FROM requests |
| WHERE resolved_date BETWEEN ? AND ? {nonnull} |
| ) |
| SELECT bucket, COUNT(*), AVG(hours), |
| AVG(CASE WHEN rn IN ((n + 1) / 2, (n + 2) / 2) THEN hours END) |
| FROM ranked GROUP BY bucket ORDER BY bucket |
| """ |
| with closing(sqlite3.connect(":memory:")) as connection: |
| connection.execute("CREATE TABLE requests (resolved_date TEXT, category TEXT, region TEXT, hours REAL)") |
| connection.executemany("INSERT INTO requests VALUES (?, ?, ?, ?)", records) |
| connection.commit() |
| connection.execute("PRAGMA query_only = ON") |
| values = connection.execute(query, (period.start.isoformat(), period.end.isoformat())).fetchall() |
| rows = tuple(StatsRow(group=row[0], sample_size=row[1], mean_hours=row[2], median_hours=row[3]) |
| for row in values) |
| if grouping == "overall" and not rows: |
| rows = (StatsRow(group="All", sample_size=0, mean_hours=None, median_hours=None),) |
| return rows |
|
|
|
|
| def _verify(left, right): |
| if len(left) != len(right): |
| raise CalculationMismatch("Group populations disagree") |
| for a, b in zip(left, right): |
| if (a.group, a.sample_size) != (b.group, b.sample_size): |
| raise CalculationMismatch("Group populations disagree") |
| for field in ("mean_hours", "median_hours"): |
| x, y = getattr(a, field), getattr(b, field) |
| if x is None or y is None: |
| if x != y: |
| raise CalculationMismatch("Undefined statistics disagree") |
| elif not isclose(x, y, rel_tol=REL_TOL, abs_tol=ABS_TOL): |
| raise CalculationMismatch("Duration statistics disagree") |
|
|
|
|
| def summarize_resolution(dataset: PreparedDataset, period: Period, |
| grouping: Grouping = "overall") -> SummaryResult: |
| _check_arguments(dataset, period, grouping) |
| records = _records(dataset) |
| rows, population, excluded = _pandas_rows(records, period, grouping) |
| _verify(rows, _sql_rows(records, period, grouping)) |
| included = sum(row.sample_size for row in rows) |
| warnings = ["COMPLETED_REQUESTS_ONLY"] |
| if included == 0: |
| warnings.append("NO_OBSERVATIONS") |
| if excluded: |
| warnings.append("MISSING_CLASSIFICATION_EXCLUDED") |
| if any(0 < row.sample_size < dataset.policy.tiny_group_threshold for row in rows): |
| warnings.append("TINY_PERIOD_GROUP") |
| if dataset.profile.requests.duplicate_rows: |
| warnings.append("EXACT_DUPLICATES_REMOVED") |
| dataset_id = _dataset_id(dataset) |
| result_id = _identity("summary-", {"dataset_id": dataset_id, "method": METHOD, |
| "period": period.model_dump(mode="json"), "grouping": grouping}) |
| return SummaryResult(result_id=result_id, dataset_id=dataset_id, grouping=grouping, |
| period=period, rows=rows, population_size=population, |
| excluded_classification_count=excluded, included_count=included, |
| dataset_excluded_count=dataset.profile.requests.excluded_rows, |
| dataset_duplicate_count=dataset.profile.requests.duplicate_rows, |
| status="no_observations" if not included else ("partial" if excluded else "complete"), |
| warnings=tuple(warnings)) |
|
|
|
|
| def _delta(baseline, comparison): |
| return None if baseline is None or comparison is None else comparison - baseline |
|
|
|
|
| def _percent(baseline, comparison): |
| return None if baseline in (None, 0) or comparison is None else (comparison - baseline) / baseline * 100 |
|
|
|
|
| def compare_periods(dataset: PreparedDataset, baseline: Period, comparison: Period, |
| grouping: Grouping = "overall") -> ComparisonResult: |
| _check_arguments(dataset, baseline, grouping) |
| _check_arguments(dataset, comparison, grouping) |
| if baseline.end >= comparison.start: |
| raise ValueError("Baseline must precede a nonoverlapping comparison period") |
| first = summarize_resolution(dataset, baseline, grouping) |
| second = summarize_resolution(dataset, comparison, grouping) |
| a = {row.group: row for row in first.rows} |
| b = {row.group: row for row in second.rows} |
| rows = [] |
| warnings = list(dict.fromkeys(first.warnings + second.warnings)) |
| missing_group = False |
| for group in sorted(a.keys() | b.keys()): |
| empty = StatsRow(group=group, sample_size=0, mean_hours=None, median_hours=None) |
| x, y = a.get(group, empty), b.get(group, empty) |
| if not x.sample_size or not y.sample_size: |
| missing_group = True |
| elif x.mean_hours == 0 or x.median_hours == 0: |
| warnings.append("ZERO_BASELINE_PERCENT_UNDEFINED") |
| rows.append(ChangeRow(group=group, baseline_count=x.sample_size, comparison_count=y.sample_size, |
| count_delta=y.sample_size - x.sample_size, |
| mean_change_hours=_delta(x.mean_hours, y.mean_hours), |
| median_change_hours=_delta(x.median_hours, y.median_hours), |
| mean_change_percent=_percent(x.mean_hours, y.mean_hours), |
| median_change_percent=_percent(x.median_hours, y.median_hours))) |
| if missing_group: |
| warnings.append("MISSING_PERIOD_GROUP") |
| if baseline.days != comparison.days: |
| warnings.append("UNEQUAL_PERIOD_LENGTHS_COUNTS_NOT_RATES") |
| if first.status == second.status == "no_observations": |
| status = "no_observations" |
| else: |
| status = "partial" if missing_group or first.status != "complete" or second.status != "complete" else "complete" |
| result_id = _identity("comparison-", {"baseline": first.result_id, "comparison": second.result_id, |
| "method": "period-comparison-v1"}) |
| return ComparisonResult(result_id=result_id, dataset_id=first.dataset_id, baseline=first, |
| comparison=second, rows=tuple(rows), status=status, warnings=tuple(dict.fromkeys(warnings))) |
|
|
