hs-data API

FastAPI backend for the hs-data platform. MVP delivers one module: the 拼团 (group-buy) funnel (启动 → 曝光 → 拼团详情 → 下单 → 成功) backed by two pre-aggregated tables:

ads_trd_group_funnel_daily — single day, keeps full history (daily incremental insert of a new dt). Backs period=day.
ads_trd_group_funnel_rolling — rolling 7d/30d, only one row (daily overwrite, no history). Backs period=last_7d / last_30d.

Data is T+1: today's data is not computed yet, so the max queryable day is always yesterday.

Tech stack

Python 3.11+, FastAPI, Pydantic v2, SQLAlchemy 2.x (async) + asyncpg, Alembic, pytest.

Setup

cd apps/api
python -m venv .venv
. .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -e ".[dev]"

Configuration

Two environment variables (see .env.example):

# true  -> serve realistic in-memory data (no DB needed). DEFAULT for now.
# false -> read the real group-buy funnel tables from Postgres.
USE_FAKE_DATA=true

# Async SQLAlchemy URL (only used when USE_FAKE_DATA=false).
DATABASE_URL=postgresql+asyncpg://hsdata:hsdata@localhost:5432/hsdata

Fake-data fallback

So the page can be seen before any Postgres exists, USE_FAKE_DATA defaults to true. In that mode the funnel API serves realistic data through the same repository interface as the real source, so route and service code is identical:

Daily history — ~10 descending daily rows ending yesterday, so the single-day date picker has history and different snapshot_dt values return different data. A snapshot_dt with no matching row → missing.
Rolling — one as-of-yesterday row for the 7d/30d windows.

Set USE_FAKE_DATA=false to query the real tables.

Run the server

No infrastructure (fake data, default):

uvicorn app.main:app --reload --port 8000

Against a real Postgres:

export USE_FAKE_DATA=false                  # Windows: $env:USE_FAKE_DATA="false"
export DATABASE_URL=postgresql+asyncpg://hsdata:hsdata@localhost:5432/hsdata
alembic upgrade head
python -m scripts.seed                      # insert daily history + rolling row
uvicorn app.main:app --reload --port 8000

OpenAPI docs at http://localhost:8000/docs. Health probe at /health.

Database migration

alembic upgrade head    # creates the daily + rolling tables

Seed sample data

Inserts ~10 historical daily rows (ending yesterday) into the daily table plus one rolling row as-of yesterday, all with clean descending group-buy funnels:

python -m scripts.seed

Export OpenAPI schema (no DB needed)

python -m scripts.export_openapi   # writes apps/api/openapi.json

API

POST /api/funnels/query

Request:

{ "period": "day", "snapshot_dt": "2026-06-20" }

period ∈ day | last_7d | last_30d. Any other value → HTTP 422.
snapshot_dt (optional ISO YYYY-MM-DD): only meaningful for day. Omitted → latest daily row (yesterday); given → that historical day. Must be ≤ yesterday (T+1); today/future → HTTP 422. Ignored for last_7d / last_30d.

Response:

{
  "period": "day",
  "snapshot_dt": "20260620",
  "results": [
    { "step_index": 1, "name": "启动",     "event_key": "start",  "uv": 10000, "conversion_rate": null, "dropoff_rate": null },
    { "step_index": 2, "name": "曝光",     "event_key": "show",   "uv": 8200,  "conversion_rate": 0.82, "dropoff_rate": 0.18 },
    { "step_index": 3, "name": "拼团详情", "event_key": "detail", "uv": 5100,  "conversion_rate": 0.62, "dropoff_rate": 0.38 },
    { "step_index": 4, "name": "下单",     "event_key": "order",  "uv": 2200,  "conversion_rate": 0.43, "dropoff_rate": 0.57 },
    { "step_index": 5, "name": "成功",     "event_key": "paid",   "uv": 1800,  "conversion_rate": 0.82, "dropoff_rate": 0.18 }
  ],
  "data_status": "ready"
}

Routing & rules

period=day → ads_trd_group_funnel_daily. Given snapshot_dt → WHERE dt=:dt; otherwise latest ORDER BY dt DESC LIMIT 1. Columns uv_start/show/detail/order/paid.
period=last_7d → ads_trd_group_funnel_rolling (single row), columns uv_*_7d.
period=last_30d → same rolling row, columns uv_*_30d.
No bitmaps, no OR, no cross-day aggregation.
snapshot_dt (response) is the dt (yyyyMMdd) of the row actually used: the day for day, the rolling row's as-of dt for 7d/30d. null when missing.
step_index starts at 1. Step 1 has null conversion/dropoff rates.
conversion_rate[i] = uv[i] / uv[i-1]; dropoff_rate[i] = 1 - conversion_rate[i]. If uv[i-1] == 0, both are null (no division by zero).
data_status ∈ {ready, missing}: ready when the target row exists and the period's columns are non-null; missing when there is no row OR the period columns are NULL. Missing data is never silently zero-filled.

Tests

pytest

Period validation, day vs rolling routing, snapshot_dt selection/validation (today/future → 422; non-existent dt → missing), conversion math, data_status, the fake data source (daily history + rolling), the SQLAlchemy repo against in-memory SQLite, and the exact API response shape are all tested without Postgres.

README.md 5.2 KB تاريخچه خام