"""拼团 (group-buy) funnel query orchestration (docs/02 §5/§6 v3). Pipeline: validate the request -> route by ``period`` to the right table/row -> pick the five UV columns -> derive conversion/dropoff -> resolve data_status -> assemble response. Routing (docs/02 §6 v3): * ``day`` -> ``ads_trd_group_funnel_daily``; a given ``snapshot_dt`` selects that row, else the latest ``dt``. Columns: ``uv_{step}``. * ``last_7d`` -> ``ads_trd_group_funnel_rolling`` (single row). Columns: ``uv_{step}_7d``. * ``last_30d`` -> same rolling row. Columns: ``uv_{step}_30d``. No bitmaps, no OR, no cross-day aggregation. The repository layer is an abstract protocol so the service can be unit-tested without Postgres (and so a fake in-memory source can stand in when no DB is configured). """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from typing import Protocol from app.schemas import ( DataStatus, FunnelQueryResponse, FunnelStepResult, FunnelTrendPoint, FunnelTrendResponse, Period, ) # Default trend window (days) when the request omits a range. DEFAULT_TREND_DAYS = 30 # Fixed funnel order: 启动 -> 曝光 -> 拼团详情 -> 下单 -> 成功 (docs/03 §11). # Each entry: (event_key, Chinese display name). FUNNEL_STEPS: list[tuple[str, str]] = [ ("start", "启动"), ("show", "曝光"), ("detail", "拼团详情"), ("order", "下单"), ("paid", "成功"), ] # period -> rolling-table column suffix (for last_7d / last_30d). ROLLING_SUFFIX: dict[Period, str] = { Period.last_7d: "7d", Period.last_30d: "30d", } def daily_columns() -> list[str]: """Return the five ``uv_{step}`` daily-table column names, in funnel order.""" return [f"uv_{event_key}" for event_key, _ in FUNNEL_STEPS] def rolling_columns(period: Period) -> list[str]: """Return the five ``uv_{step}_{suffix}`` rolling-table columns for a period.""" suffix = ROLLING_SUFFIX[period] return [f"uv_{event_key}_{suffix}" for event_key, _ in FUNNEL_STEPS] # --------------------------------------------------------------------------- # # Validation # # --------------------------------------------------------------------------- # class SnapshotDateError(ValueError): """Raised when ``snapshot_dt`` is today or in the future (data is T+1).""" def validate_snapshot_dt(snapshot_dt: date) -> None: """Reject a ``snapshot_dt`` later than yesterday. Data is T+1: today's data is not computed yet, so the max queryable day is always yesterday. today or future -> :class:`SnapshotDateError`. """ yesterday = date.today() - timedelta(days=1) if snapshot_dt > yesterday: raise SnapshotDateError( f"snapshot_dt {snapshot_dt.isoformat()} is later than yesterday " f"({yesterday.isoformat()}); data is T+1 and today's data is not " "computed yet." ) class TrendRangeError(ValueError): """Raised when a trend range is invalid (end in the future, or start > end).""" def validate_trend_range(start_dt: date | None, end_dt: date | None) -> None: """Reject an out-of-range trend window (data is T+1; bounds must be ordered).""" yesterday = date.today() - timedelta(days=1) if end_dt is not None and end_dt > yesterday: raise TrendRangeError( f"end_dt {end_dt.isoformat()} is later than yesterday " f"({yesterday.isoformat()}); data is T+1." ) if start_dt is not None and end_dt is not None and start_dt > end_dt: raise TrendRangeError( f"start_dt {start_dt.isoformat()} is after end_dt {end_dt.isoformat()}." ) # --------------------------------------------------------------------------- # # Repository protocol + snapshot container # # --------------------------------------------------------------------------- # @dataclass(frozen=True) class FunnelSnapshot: """A funnel source row as returned by the repository. ``values`` maps column name to its value (NULL columns map to ``None``). ``dt`` is the row's ``dt`` (yyyyMMdd): the day for daily, the as-of day for rolling. """ dt: str values: dict[str, int | None] class FunnelRepository(Protocol): """Data access boundary. Implementations may hit Postgres or be in-memory.""" async def fetch_daily(self, dt: str | None) -> FunnelSnapshot | None: """Return a row from the daily table. ``dt`` (yyyyMMdd) selects that specific day; ``None`` returns the latest ``dt`` row. ``None`` when no matching row exists. """ ... async def fetch_rolling(self) -> FunnelSnapshot | None: """Return the single rolling row, or ``None`` when the table is empty.""" ... async def fetch_daily_range( self, start_dt: str, end_dt: str ) -> list[FunnelSnapshot]: """Return daily rows with ``start_dt <= dt <= end_dt`` (yyyyMMdd), ascending. Empty list when no rows fall in the range. """ ... # --------------------------------------------------------------------------- # # Conversion math # # --------------------------------------------------------------------------- # def build_results(uvs: list[int]) -> list[FunnelStepResult]: """Assemble per-step results with conversion/dropoff math. Step 1: conversion_rate and dropoff_rate are null. Step i: conversion = uv[i] / uv[i-1]; dropoff = 1 - conversion. If uv[i-1] == 0, conversion (and dropoff) are null. """ results: list[FunnelStepResult] = [] for idx, (uv, (event_key, name)) in enumerate(zip(uvs, FUNNEL_STEPS)): if idx == 0: conversion: float | None = None dropoff: float | None = None else: prev = uvs[idx - 1] if prev == 0: conversion = None dropoff = None else: conversion = uv / prev dropoff = 1 - conversion results.append( FunnelStepResult( step_index=idx + 1, name=name, event_key=event_key, uv=uv, conversion_rate=conversion, dropoff_rate=dropoff, ) ) return results # --------------------------------------------------------------------------- # # Orchestration # # --------------------------------------------------------------------------- # def _assemble( period: Period, snapshot: FunnelSnapshot | None, column_names: list[str], ) -> FunnelQueryResponse: """Map a fetched snapshot + period columns to the response model. Missing row or any NULL period column -> ``data_status = missing`` with no zero-fill. ``snapshot_dt`` echoes the row used (None when no row). """ if snapshot is None: return FunnelQueryResponse( period=period, snapshot_dt=None, results=[], data_status=DataStatus.missing, ) raw = [snapshot.values.get(col) for col in column_names] if any(value is None for value in raw): return FunnelQueryResponse( period=period, snapshot_dt=snapshot.dt, results=[], data_status=DataStatus.missing, ) uvs = [int(value) for value in raw] # narrow Optional -> int return FunnelQueryResponse( period=period, snapshot_dt=snapshot.dt, results=build_results(uvs), data_status=DataStatus.ready, ) async def run_funnel_query( period: Period, repo: FunnelRepository, snapshot_dt: date | None = None, ) -> FunnelQueryResponse: """Run the group-buy funnel query pipeline and return the response model. Routes by ``period``: * ``day`` -> daily table; validates ``snapshot_dt`` (<= yesterday) when given, then selects that day or the latest row. * ``last_7d`` / ``last_30d`` -> the single rolling row; ``snapshot_dt`` is ignored. """ if period is Period.day: dt_str: str | None = None if snapshot_dt is not None: validate_snapshot_dt(snapshot_dt) # may raise SnapshotDateError dt_str = snapshot_dt.strftime("%Y%m%d") snapshot = await repo.fetch_daily(dt_str) return _assemble(period, snapshot, daily_columns()) # last_7d / last_30d -> rolling row; snapshot_dt ignored. snapshot = await repo.fetch_rolling() return _assemble(period, snapshot, rolling_columns(period)) # --------------------------------------------------------------------------- # # Trend orchestration (按日折线) # # --------------------------------------------------------------------------- # def _parse_dt(dt: str) -> date: """Parse a yyyyMMdd string to a ``date``.""" return date(int(dt[:4]), int(dt[4:6]), int(dt[6:8])) def _empty_trend() -> FunnelTrendResponse: return FunnelTrendResponse( start_dt=None, end_dt=None, data_status=DataStatus.missing, points=[] ) def _build_trend_points(snapshots: list[FunnelSnapshot]) -> list[FunnelTrendPoint]: """Map daily snapshots -> trend points, dropping null-bearing days, then trimming LEADING all-zero (pre-launch) days. Conversion math reuses :func:`build_results` so the口径 matches the funnel query exactly. Days with any NULL uv column are omitted (a calendar gap the client renders as a line break); the daily table has no NULLs in practice. Middle/trailing zero days are kept (real zero activity, not pre-launch). """ cols = daily_columns() points: list[FunnelTrendPoint] = [] for snap in sorted(snapshots, key=lambda s: s.dt): raw = [snap.values.get(col) for col in cols] if any(value is None for value in raw): continue points.append( FunnelTrendPoint(dt=snap.dt, results=build_results([int(v) for v in raw])) ) first = 0 while first < len(points) and points[first].results[0].uv == 0: first += 1 return points[first:] async def run_funnel_trend( repo: FunnelRepository, start_dt: date | None = None, end_dt: date | None = None, ) -> FunnelTrendResponse: """Run the按日 trend pipeline over the daily table and return the series. Range resolution: * ``end`` = ``end_dt`` (capped at yesterday) if given, else the latest available daily ``dt``. * ``start`` = ``start_dt`` if given, else ``end`` - (DEFAULT_TREND_DAYS - 1). Leading pre-launch (all-zero) days are trimmed; the response bounds reflect the kept series. No rows / all-zero -> ``missing``. """ validate_trend_range(start_dt, end_dt) # may raise TrendRangeError yesterday = date.today() - timedelta(days=1) if end_dt is not None: end = min(end_dt, yesterday) else: latest = await repo.fetch_daily(None) if latest is None: return _empty_trend() end = _parse_dt(latest.dt) start = start_dt if start_dt is not None else end - timedelta(days=DEFAULT_TREND_DAYS - 1) if start > end: return _empty_trend() snapshots = await repo.fetch_daily_range( start.strftime("%Y%m%d"), end.strftime("%Y%m%d") ) points = _build_trend_points(snapshots) if not points: return _empty_trend() return FunnelTrendResponse( start_dt=points[0].dt, end_dt=points[-1].dt, data_status=DataStatus.ready, points=points, )