hs-data/apps/api/app/services/funnel.py

"""拼团 (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,
    Period,
)

# 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."
        )


# --------------------------------------------------------------------------- #
# 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."""
        ...


# --------------------------------------------------------------------------- #
# 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))