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