diff --git a/dashboard/backend/api/routers/admin_analytics.py b/dashboard/backend/api/routers/admin_analytics.py index f45bb3a3..866a50ec 100644 --- a/dashboard/backend/api/routers/admin_analytics.py +++ b/dashboard/backend/api/routers/admin_analytics.py @@ -26,6 +26,7 @@ from dashboard.backend.domain.analytics.value_queries import ( CommercialAnalyticsResponse, LifecycleAnalyticsResponse, + MAX_VALUE_RANGE_DAYS, OperationalAnalyticsResponse, PaginatedValueUsers, RetentionAnalyticsResponse, @@ -54,7 +55,7 @@ _LIFECYCLE_SEGMENTS = {"new", "onboarding", "growing", "core", "at_risk", "dormant"} _OPERATIONAL_STATES = {"blocked", "needs_attention", "healthy"} _COMMERCIAL_TIERS = {"unpaid", "starter", "invested", "high_value"} -_MAX_VALUE_RANGE_DAYS = 180 +_LIFECYCLE_MOVEMENT_RANGES = {"5d", "1w", "1m", "1y"} def _invalid_query() -> Never: @@ -175,7 +176,7 @@ def _value_range_from_values(values: dict[str, str]) -> tuple[date, date, bool]: start = from_date or end - timedelta(days=30) except OverflowError: _invalid_query() - if end <= start or (end - start).days > _MAX_VALUE_RANGE_DAYS: + if end <= start or (end - start).days > MAX_VALUE_RANGE_DAYS: _invalid_query() include_internal = ( _parse_bool(values["include_internal"]) @@ -406,12 +407,19 @@ def get_lifecycle( request: Request, service: ValueAnalyticsQueryService = Depends(get_value_analytics_query_service), ): - start, end, include_internal, _values = _value_range(request) + start, end, include_internal, values = _value_range( + request, + additional={"movement_range"}, + ) + movement_range = values.get("movement_range", "5d") + if movement_range not in _LIFECYCLE_MOVEMENT_RANGES: + _invalid_query() try: return service.get_lifecycle( start=start, end=end, include_internal=include_internal, + movement_range=movement_range, ) except Exception as exc: _raise_service_error(exc) diff --git a/dashboard/backend/domain/analytics/value_queries.py b/dashboard/backend/domain/analytics/value_queries.py index 6853ee3d..66b4d0be 100644 --- a/dashboard/backend/domain/analytics/value_queries.py +++ b/dashboard/backend/domain/analytics/value_queries.py @@ -54,6 +54,31 @@ "high_value", ) _TIER_RANK = {tier: rank for rank, tier in enumerate(_COMMERCIAL_TIERS)} +_MOVEMENT_WINDOWS: dict[str, tuple[int, Literal["day", "week", "month"]]] = { + "5d": (5, "day"), + "1w": (7, "day"), + "1m": (31, "week"), + "1y": (365, "month"), +} +MAX_VALUE_RANGE_DAYS = 180 +"""Longest filter range a caller may request, in days. + +The router enforces the same bound on the query string; this is the copy that +holds for every caller, and both now read it from here. It was duplicated as a +bare literal, which is how the two could drift apart unnoticed. +""" + +_MAX_HISTORY_SCAN_DAYS = max( + MAX_VALUE_RANGE_DAYS, *(days for days, _granularity in _MOVEMENT_WINDOWS.values()) +) +"""Widest span of per-user daily rows one lifecycle request may scan. + +`MAX_VALUE_RANGE_DAYS` bounds the *filter* range, but the movement chart reads a +window of its own, so the scan is the wider of the two -- 365 days today, double +the filter cap. Derived rather than written down so that adding a longer +movement range cannot silently widen every user's scan. +""" + _PRIORITY_RANK = { "blocked": 0, "needs_attention": 1, @@ -77,6 +102,14 @@ def _week_start(value: date) -> date: return value - timedelta(days=value.weekday()) +def _period_start(value: date, granularity: Literal["day", "week", "month"]) -> date: + if granularity == "day": + return value + if granularity == "week": + return _week_start(value) + return value.replace(day=1) + + def _parse_timestamp(value: object) -> datetime: parsed = datetime.fromisoformat(str(value)) if parsed.tzinfo is None or parsed.utcoffset() is None: @@ -91,8 +124,10 @@ def _validate_dates(start: date, end: date) -> tuple[date, date]: raise ValueError("end must be a date") if end <= start: raise ValueError("end must be later than start") - if (end - start).days > 180: - raise ValueError("date range must contain at most 180 days") + if (end - start).days > MAX_VALUE_RANGE_DAYS: + raise ValueError( + f"date range must contain at most {MAX_VALUE_RANGE_DAYS} days" + ) return start, end @@ -123,6 +158,16 @@ class WeeklyLifecycleCount(BaseModel): data_quality: Literal["complete", "partial"] +class LifecycleMovementPoint(BaseModel): + """A display-safe lifecycle snapshot at the selected chart granularity.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + period_start: date + segment_counts: dict[LifecycleSegment, int] + data_quality: Literal["complete", "partial"] + + class LifecycleTransition(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) @@ -141,6 +186,9 @@ class LifecycleAnalyticsResponse(BaseModel): headline: LifecycleHeadline segment_counts: dict[LifecycleSegment, int] weekly_segments: Sequence[WeeklyLifecycleCount] + movement_range: Literal["5d", "1w", "1m", "1y"] = "5d" + movement_granularity: Literal["day", "week", "month"] = "day" + movement_segments: Sequence[LifecycleMovementPoint] = Field(default_factory=tuple) transitions: Sequence[LifecycleTransition] availability: dict[str, SectionAvailability] @@ -452,19 +500,34 @@ def _history( start: date, end: date, use_anonymous_rollups: bool, + movement_start: date | None = None, + movement_range: str = "5d", ) -> tuple[ - list[WeeklyLifecycleCount], list[LifecycleTransition], SectionAvailability + list[WeeklyLifecycleCount], + list[LifecycleMovementPoint], + list[LifecycleTransition], + SectionAvailability, ]: + if movement_range not in _MOVEMENT_WINDOWS: + raise ValueError("unsupported lifecycle movement range") + window_days, granularity = _MOVEMENT_WINDOWS[movement_range] + selected_movement_start = movement_start or end - timedelta(days=window_days) + if selected_movement_start >= end: + raise ValueError("lifecycle movement range is empty") + history_start = max( + min(start, selected_movement_start), + end - timedelta(days=_MAX_HISTORY_SCAN_DAYS), + ) rows = self._daily( user_ids, - start=start - timedelta(days=1), + start=history_start - timedelta(days=1), end=end, ) by_date: dict[date, list[UserLifecycleDailySnapshot]] = defaultdict(list) for row in rows: by_date[row.snapshot_date].append(row) rollups = ( - self.query_store.rollups.list_rollups(start=start, end=end) + self.query_store.rollups.list_rollups(start=history_start, end=end) if use_anonymous_rollups else [] ) @@ -481,7 +544,7 @@ def _history( daily_counts: dict[date, dict[LifecycleSegment, int]] = {} daily_quality: dict[date, str] = {} - direct_dates = {day for day in by_date if start <= day < end} + direct_dates = {day for day in by_date if history_start <= day < end} for day in direct_dates: counts = Counter(row.lifecycle_segment for row in by_date[day]) daily_counts[day] = { @@ -503,6 +566,8 @@ def _history( weekly: list[WeeklyLifecycleCount] = [] by_week: dict[date, list[date]] = defaultdict(list) for day in daily_counts: + if not start <= day < end: + continue by_week[_week_start(day)].append(day) for week, dates in sorted(by_week.items()): latest = max(dates) @@ -514,6 +579,21 @@ def _history( ) ) + movement: list[LifecycleMovementPoint] = [] + by_period: dict[date, list[date]] = defaultdict(list) + for day in daily_counts: + if selected_movement_start <= day < end: + by_period[_period_start(day, granularity)].append(day) + for period, dates in sorted(by_period.items()): + latest = max(dates) + movement.append( + LifecycleMovementPoint( + period_start=max(period, selected_movement_start), + segment_counts=daily_counts[latest], + data_quality=daily_quality[latest], + ) + ) + transition_counts: Counter[tuple[str, str]] = Counter() transition_partial: set[tuple[str, str]] = set() by_user_date = {(row.user_id, row.snapshot_date): row for row in rows} @@ -532,6 +612,7 @@ def _history( for row in rollups: if ( row.metric_name != "lifecycle_transition" + or not start <= row.rollup_date < end or row.rollup_date in direct_dates ): continue @@ -557,21 +638,25 @@ def _history( key=lambda item: (-item[1], item[0]), ) ] - coverage = sorted(daily_counts) + coverage = sorted(day for day in daily_counts if start <= day < end) if not coverage: availability = SectionAvailability( available=False, status="building", ) else: - partial = any(value == "partial" for value in daily_quality.values()) + partial = ( + any(daily_quality[day] == "partial" for day in coverage) + or coverage[0] > start + or coverage[-1] < end - timedelta(days=1) + ) availability = SectionAvailability( available=True, status="partial" if partial else "ready", coverage_start=coverage[0], coverage_end=coverage[-1], ) - return weekly, transitions, availability + return weekly, movement, transitions, availability def get_lifecycle( self, @@ -579,9 +664,14 @@ def get_lifecycle( start: date, end: date, include_internal: bool = False, + movement_range: str = "5d", now: datetime | None = None, ) -> LifecycleAnalyticsResponse: start, end = _validate_dates(start, end) + if movement_range not in _MOVEMENT_WINDOWS: + raise ValueError("unsupported lifecycle movement range") + window_days, _granularity = _MOVEMENT_WINDOWS[movement_range] + movement_start = end - timedelta(days=window_days) current_time = _utc(now or datetime.now(UTC), "now") users = self._eligible_users(include_internal=include_internal) current = self._current(users) @@ -611,14 +701,16 @@ def get_lifecycle( status="unavailable", ) try: - weekly, transitions, history_availability = self._history( + weekly, movement, transitions, history_availability = self._history( self._ids(users), start=start, end=end, use_anonymous_rollups=not include_internal, + movement_start=movement_start, + movement_range=movement_range, ) except Exception: - weekly, transitions = [], [] + weekly, movement, transitions = [], [], [] history_availability = SectionAvailability( available=False, status="unavailable", @@ -636,6 +728,9 @@ def get_lifecycle( ), segment_counts=segment_counts, weekly_segments=weekly, + movement_range=movement_range, + movement_granularity=_MOVEMENT_WINDOWS[movement_range][1], + movement_segments=movement, transitions=transitions, availability=availability, ) @@ -1046,7 +1141,7 @@ def get_user_profile( start=_day_start(start), end=_day_start(end), ) - _weekly, transitions, _availability = self._history( + _weekly, _movement, transitions, _availability = self._history( [subject_id], start=start, end=end, @@ -1069,6 +1164,7 @@ def get_user_profile( "CommercialPeriodSummary", "LifecycleAnalyticsResponse", "LifecycleHeadline", + "LifecycleMovementPoint", "LifecycleTransition", "OperationalAnalyticsResponse", "PaginatedValueUsers", diff --git a/dashboard/backend/tests/_frontend_source.py b/dashboard/backend/tests/_frontend_source.py index 59079737..1f065af8 100644 --- a/dashboard/backend/tests/_frontend_source.py +++ b/dashboard/backend/tests/_frontend_source.py @@ -131,9 +131,12 @@ def _match_brace(source: str, index: int) -> int: index += 1 -def fn_body(signature: str) -> str: +def fn_body(signature: str, source: str | None = None) -> str: """The named function's source, brace-matched to its real closing brace. + `source` defaults to app.js; pass one of the `js/*.js` modules to slice a + function out of it with the same brace matching. + Brace-matching rather than a fixed-width slice: a `[start:start + 900]` window over-reads into whatever unrelated top-level code happens to follow, so an assertion can pass on a neighbour's source instead of the function @@ -145,9 +148,10 @@ def fn_body(signature: str) -> str: returns that parameter block instead of the body: a short, plausible-looking string in which every `assert "..." in body` fails, or worse, passes. """ - start = APP_JS.index(signature) - open_brace = APP_JS.index("{", match_paren(APP_JS, APP_JS.index("(", start))) - return APP_JS[start : _match_brace(APP_JS, open_brace) + 1] + text = APP_JS if source is None else source + start = text.index(signature) + open_brace = text.index("{", match_paren(text, text.index("(", start))) + return text[start : _match_brace(text, open_brace) + 1] def js_const(name: str) -> str: diff --git a/dashboard/backend/tests/domain/analytics/test_value_queries.py b/dashboard/backend/tests/domain/analytics/test_value_queries.py index 6affc220..66835036 100644 --- a/dashboard/backend/tests/domain/analytics/test_value_queries.py +++ b/dashboard/backend/tests/domain/analytics/test_value_queries.py @@ -13,6 +13,8 @@ AnalyticsUserProfile, ) from dashboard.backend.domain.analytics.value_queries import ( + _MAX_HISTORY_SCAN_DAYS, + _MOVEMENT_WINDOWS, UserValueFilters, ValueAnalyticsQueryService, ) @@ -129,6 +131,7 @@ def __init__(self, *, snapshots, commercial, daily=(), credit_activity=None): self.daily = list(daily) self.credit_activity = dict(credit_activity or {}) self.commercial_windows = [] + self.daily_windows = [] def list_current_snapshots(self, user_ids): return { @@ -149,6 +152,7 @@ def list_commercial_values(self, user_ids, *, start, end): } def list_daily_snapshots(self, *, start, end, user_ids=None): + self.daily_windows.append((start, end)) selected = None if user_ids is None else set(user_ids) return [ row @@ -303,6 +307,44 @@ def test_date_filter_changes_history_not_current_lifecycle_identity(): assert short.weekly_segments != long.weekly_segments +@pytest.mark.parametrize( + ("movement_range", "granularity"), + [("5d", "day"), ("1w", "day"), ("1m", "week"), ("1y", "month")], +) +def test_lifecycle_movement_returns_selected_range_and_granularity( + movement_range, granularity +): + snapshots = {1: _snapshot(1)} + daily = [ + _daily(1, date(2025, 10, 1) + timedelta(days=offset), "core") + for offset in range(365) + ] + service, _value_store, _legacy = _service(snapshots=snapshots, daily=daily) + + response = service.get_lifecycle( + start=date(2026, 4, 5), + end=date(2026, 10, 1), + movement_range=movement_range, + now=datetime(2026, 10, 1, 12, tzinfo=UTC), + ) + + assert response.movement_range == movement_range + assert response.movement_granularity == granularity + starts = [point.period_start for point in response.movement_segments] + assert starts + assert starts == sorted(set(starts)) + # Derived from the window rather than written down: a calendar bucket count + # depends on where the window falls in the calendar, so a hard-coded ceiling + # holds only for the dates this case happens to pick. + window_days = _MOVEMENT_WINDOWS[movement_range][0] + period_days = {"day": 1, "week": 7, "month": 28}[granularity] + assert len(starts) <= window_days // period_days + 2 + assert all( + date(2026, 10, 1) - timedelta(days=window_days) <= day < date(2026, 10, 1) + for day in starts + ) + + def test_retention_uses_nulls_for_immature_cells_and_weighted_mature_summary(): first_week = date(2026, 7, 6) second_week = date(2026, 7, 13) @@ -573,3 +615,122 @@ def test_internal_accounts_are_excluded_unless_explicitly_included(): assert [item.user_id for item in external.items] == [1] assert [item.user_id for item in all_users.items] == [1, 2] + + +def _transition_rollup(day: date, count: int, outcome: str = "complete"): + return SimpleNamespace( + rollup_date=day, + metric_name="lifecycle_transition", + event_name="growing", + user_state="core", + value_count=count, + outcome=outcome, + ) + + +def test_lifecycle_transitions_ignore_rollups_outside_the_requested_window(): + """A long movement range widens the scan; it must not widen the totals. + + `transitions` is stamped with the requested period, so a rollup from before + `start` that is summed in reports itself as having happened inside a window + it predates. + """ + start = date(2026, 9, 1) + end = date(2026, 10, 1) + daily = [ + _daily(1, start + timedelta(days=offset)) + for offset in range(30) + if start + timedelta(days=offset) != date(2026, 9, 10) + ] + service, _value_store, _legacy = _service( + snapshots={1: _snapshot(1)}, + daily=daily, + rollups=[ + _transition_rollup(date(2026, 9, 10), 2), + _transition_rollup(date(2026, 1, 15), 97), + ], + ) + + response = service.get_lifecycle( + start=start, + end=end, + movement_range="1y", + now=datetime(2026, 10, 1, 12, tzinfo=UTC), + ) + + assert [ + (row.from_segment, row.to_segment, row.users) for row in response.transitions + ] == [("growing", "core", 2)] + + +def test_lifecycle_coverage_reports_the_requested_window_not_the_movement_history(): + """Coverage describes the window the admin asked for. + + The movement chart needs a wider scan than the filter range, but + `availability.history` is what the Lifecycle distribution card renders, so + it must keep describing `start..end`. + """ + start = date(2026, 9, 1) + end = date(2026, 10, 1) + daily = [_daily(1, start + timedelta(days=offset)) for offset in range(30)] + daily.append(_daily(1, date(2025, 12, 1), quality="partial")) + service, _value_store, _legacy = _service(snapshots={1: _snapshot(1)}, daily=daily) + + response = service.get_lifecycle( + start=start, + end=end, + movement_range="1y", + now=datetime(2026, 10, 1, 12, tzinfo=UTC), + ) + + history = response.availability["history"] + assert history.coverage_start == start + assert history.coverage_end == end - timedelta(days=1) + assert history.status == "ready" + + +def test_lifecycle_movement_buckets_never_start_before_the_selected_window(): + """Calendar bucketing must not label a point outside the chart's own range.""" + start = date(2026, 9, 15) + end = date(2026, 10, 15) + window_start = end - timedelta(days=365) + daily = [_daily(1, window_start + timedelta(days=offset)) for offset in range(366)] + service, _value_store, _legacy = _service(snapshots={1: _snapshot(1)}, daily=daily) + + response = service.get_lifecycle( + start=start, + end=end, + movement_range="1y", + now=datetime(2026, 10, 15, 12, tzinfo=UTC), + ) + + assert response.movement_segments + assert min(point.period_start for point in response.movement_segments) >= window_start + + +@pytest.mark.parametrize("movement_range", sorted(_MOVEMENT_WINDOWS)) +def test_lifecycle_daily_scan_stays_within_the_derived_history_bound(movement_range): + """Pins the widest span of per-user rows one request may read. + + Not a regression test -- `_validate_dates` already bounds the filter range, + so the scan is bounded today. It is a drift guard: the movement window is + read from a table, and a longer entry added to that table would widen this + scan for every eligible user with nothing else noticing. + """ + start = date(2026, 9, 1) + end = date(2026, 10, 1) + service, value_store, _legacy = _service( + snapshots={1: _snapshot(1)}, + daily=[_daily(1, start + timedelta(days=offset)) for offset in range(30)], + ) + + service.get_lifecycle( + start=start, + end=end, + movement_range=movement_range, + now=datetime(2026, 10, 1, 12, tzinfo=UTC), + ) + + scan_start, scan_end = value_store.daily_windows[-1] + assert scan_end == end + assert scan_start >= end - timedelta(days=_MAX_HISTORY_SCAN_DAYS + 1) diff --git a/dashboard/backend/tests/fixtures/admin_analytics/lifecycle.json b/dashboard/backend/tests/fixtures/admin_analytics/lifecycle.json index 416c4828..34ec8db1 100644 --- a/dashboard/backend/tests/fixtures/admin_analytics/lifecycle.json +++ b/dashboard/backend/tests/fixtures/admin_analytics/lifecycle.json @@ -40,6 +40,34 @@ "data_quality": "partial" } ], + "movement_range": "5d", + "movement_granularity": "day", + "movement_segments": [ + { + "period_start": "2026-08-30", + "segment_counts": { + "new": 3, + "onboarding": 6, + "growing": 7, + "core": 8, + "at_risk": 5, + "dormant": 4 + }, + "data_quality": "complete" + }, + { + "period_start": "2026-08-31", + "segment_counts": { + "new": 3, + "onboarding": 6, + "growing": 7, + "core": 8, + "at_risk": 5, + "dormant": 4 + }, + "data_quality": "partial" + } + ], "transitions": [ { "from_segment": "growing", diff --git a/dashboard/backend/tests/test_admin_analytics_api.py b/dashboard/backend/tests/test_admin_analytics_api.py index d9b082e1..196006bb 100644 --- a/dashboard/backend/tests/test_admin_analytics_api.py +++ b/dashboard/backend/tests/test_admin_analytics_api.py @@ -539,6 +539,31 @@ def test_admin_value_sections_have_independent_contracts( assert call["billing_mode"] == "platform_credits" +@pytest.mark.parametrize("movement_range", ["5d", "1w", "1m", "1y"]) +def test_lifecycle_accepts_documented_movement_ranges(admin_analytics_api, movement_range): + api = admin_analytics_api + response = api["client"].get( + "/api/admin/analytics/lifecycle", + params={"from": "2026-08-01", "to": "2026-08-31", "movement_range": movement_range}, + headers=api["admin_headers"], + ) + + assert response.status_code == 200, response.text + name, call = api["value_query_service"].calls[-1] + assert name == "lifecycle" + assert call["movement_range"] == movement_range + + +def test_lifecycle_rejects_unknown_movement_range(admin_analytics_api): + response = admin_analytics_api["client"].get( + "/api/admin/analytics/lifecycle", + params={"movement_range": "2q"}, + headers=admin_analytics_api["admin_headers"], + ) + assert response.status_code == 422 + assert response.json() == {"detail": "Invalid Analytics query."} + + def test_admin_overview_accepts_documented_filters(admin_analytics_api): api = admin_analytics_api response = api["client"].get( diff --git a/dashboard/backend/tests/test_admin_analytics_frontend.py b/dashboard/backend/tests/test_admin_analytics_frontend.py index 0510715e..84f29114 100644 --- a/dashboard/backend/tests/test_admin_analytics_frontend.py +++ b/dashboard/backend/tests/test_admin_analytics_frontend.py @@ -15,7 +15,12 @@ RetentionAnalyticsResponse, ValueUserProfile, ) -from dashboard.backend.tests._frontend_source import APP_HTML, APP_JS, STYLES +from dashboard.backend.tests._frontend_source import ( + APP_HTML, + APP_JS, + STYLES, + fn_body, +) ROOT = Path(__file__).resolve().parents[2] @@ -109,7 +114,7 @@ def test_admin_analytics_surface_and_module_exist(): assert 'id="adminPanelAnalytics"' in APP_HTML assert 'id="adminAnalyticsOverview"' in APP_HTML assert 'id="adminAnalyticsProfile"' in APP_HTML - assert 'js/admin-analytics.js?v=5' in APP_HTML + assert 'js/admin-analytics.js?v=6' in APP_HTML assert ANALYTICS_JS_PATH.exists() assert ".admin-analytics-overview" in STYLES assert ".admin-analytics-profile" in STYLES @@ -140,6 +145,19 @@ def test_admin_rail_is_accessible_default_and_url_backed(): assert "openAccountManagement" in tabs +def test_account_management_clears_every_analytics_profile_param(): + """Leaving analytics for a user's account must not leave a profile behind. + + `analyticsProfile` was added alongside `analyticsUser` but not to this list, + so the only thing clearing it was the caller happening to close the profile + first. A stale one here lands on a URL the overview guard refuses to load. + """ + tabs = ADMIN_TABS_JS_PATH.read_text(encoding="utf-8") + body = fn_body("function openAccountManagement(", tabs) + for key in ("analyticsUser", "analyticsProfile", "analyticsSection"): + assert f"searchParams.delete('{key}')" in body + + def test_credits_navigation_remains_horizontal(): start = APP_HTML.index('