Spaces:
Sleeping
Sleeping
File size: 2,867 Bytes
990895d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """Expose authenticated server-side statistics endpoints.
Probabilities and ranks are computed in Python, never by the model.
"""
from datetime import date, datetime, timezone
from fastapi import APIRouter, Depends, Query, Request
from app.deps import require_login
from app.models import ApiEnvelope, ok
from app.stats_math import build_stats, by_remedy
router = APIRouter(
prefix="/api/stats",
tags=["stats"],
dependencies=[Depends(require_login)],
)
def _entry_dicts(request: Request) -> list[dict]:
return [e.model_dump(mode="json") for e in request.app.state.entry_store._load()]
def _daily_dicts(
request: Request,
start: date | None,
end: date | None,
) -> list[dict]:
rows = request.app.state.daily_store._load()
out = []
for row in rows:
if start and row.date < start:
continue
if end and row.date > end:
continue
out.append(row.model_dump(mode="json"))
return out
def _filter_entries(
entries: list[dict],
start: date | None,
end: date | None,
) -> list[dict]:
if start is None and end is None:
return entries
filtered = []
for entry in entries:
ts = entry.get("ts")
if isinstance(ts, str):
day = date.fromisoformat(ts[:10])
else:
continue
if start and day < start:
continue
if end and day > end:
continue
filtered.append(entry)
return filtered
@router.get("", response_model=ApiEnvelope)
def get_stats(
request: Request,
start: date | None = None,
end: date | None = None,
min_n: int | None = Query(default=None, ge=1),
) -> dict[str, object]:
"""Return aggregate statistics for entries and daily rows."""
settings = request.app.state.settings
effective_min_n = settings.min_stats_n if min_n is None else min_n
entries = _filter_entries(_entry_dicts(request), start, end)
daily = _daily_dicts(request, start, end)
payload = build_stats(
entries,
daily,
min_n=effective_min_n,
shrink_k=settings.stats_shrink_k,
generated_at=datetime.now(timezone.utc).isoformat(),
)
return ok(payload)
@router.get("/remedies", response_model=ApiEnvelope)
def get_remedy_leaderboard(
request: Request,
start: date | None = None,
end: date | None = None,
min_n: int | None = Query(default=None, ge=1),
) -> dict[str, object]:
"""Return remedies sorted by shrinkage rank."""
settings = request.app.state.settings
effective_min_n = settings.min_stats_n if min_n is None else min_n
entries = _filter_entries(_entry_dicts(request), start, end)
rows = by_remedy(
entries,
min_n=effective_min_n,
shrink_k=settings.stats_shrink_k,
)
return ok({"items": rows, "min_n": effective_min_n})
|