hs-data/apps/api

hs-data 后端(API)

hs-data 平台的 FastAPI 只读查询服务。MVP 交付一个模块:拼团漏斗(启动 → 曝光 → 拼团详情 → 下单 → 成功),数据源是两张预聚合表:

• ads_trd_group_funnel_daily —— 单日、保留全历史(每天增量插一行 dt)。支撑 period=day。
• ads_trd_group_funnel_rolling —— 滚动 7d/30d、只有一行(每天覆盖,无历史)。支撑 period=last_7d / last_30d。

数据 T+1:今天的数据还没算出来,最大可查日始终是昨日。两表在 ads schema 下;ORM 表名不带 schema,靠连接 search_path=ads 解析(见 db/session.py)。

完整接口契约见 docs/06-接口文档;取数与口径见 docs/02 §5–6;表结构见 docs/03 §11。

技术栈

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

安装

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

配置

环境变量(放 apps/api/.env,已 gitignored;勿提交密码):

# true  → 内存假数据(无需 DB),目前默认。
# false → 读真实拼团漏斗两表。
USE_FAKE_DATA=true

# 异步 SQLAlchemy URL(仅 USE_FAKE_DATA=false 时使用)。
DATABASE_URL=postgresql+asyncpg://user:pass@host:5432/db

# 表所在 schema,默认 ads;通过连接 search_path 生效。
DB_SCHEMA=ads

假数据兜底

为了没有 PostgreSQL 也能看页面,USE_FAKE_DATA 默认 true。该模式通过与真实源相同的仓库接口提供数据,路由/服务代码完全一致:

• 日表历史 —— 约 95 天递减日行(到昨日为止;最老若干天为全 0 模拟上线前爬坡),让单日日历有历史、不同 snapshot_dt 返回不同数据,也让趋势的"裁前导 0"被覆盖到。无匹配行的 snapshot_dt → missing。
• 滚动行 —— 一行 as-of 昨日,供 7d/30d。

设 USE_FAKE_DATA=false 即查真实表。

启动

无基础设施(默认假数据):

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

接真实 PostgreSQL:在 .env 配好 DATABASE_URL + USE_FAKE_DATA=false 后直接起(真实数据由上游 ETL 写入,不需 alembic/seed):

uvicorn app.main:app --port 8000

OpenAPI 文档在 http://localhost:8000/docs,探活 /health。

线上:同一个进程用 STATIC_DIR 环境变量同源托管前端 SPA(~/hs-data 产物)+ /api,端口 8080,取代独立静态服。本地/测试不设 STATIC_DIR 则只出 /api。见 docs/02 §10。

本地建表 / 灌样例(仅本地假 PG 用)

alembic upgrade head        # 建 daily + rolling 两表
python -m scripts.seed      # 灌 ~95 天日行 + 一行滚动

导出 OpenAPI(无需 DB)

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

接口

1. 漏斗查询 POST /api/funnels/query

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

• period ∈ day | last_7d | last_30d,其它值 → 422。
• snapshot_dt(可选 YYYY-MM-DD):仅 day 有意义。省略 → 最新日行(MAX(dt));给值 → 该历史日。必须 ≤ 昨日(T+1),今天/未来 → 422;last_7d/last_30d 忽略。

响应每步含 step_index / name / event_key / uv / conversion_rate / dropoff_rate,外加 period / snapshot_dt / data_status。

2. 趋势查询 POST /api/funnels/trend(按日折线)

请求 { "start_dt": "2026-05-23", "end_dt": "2026-06-21" }(均可选)

• 都省略 → 最新可用日往前 30 天。end_dt ≤ 昨日,否则 422;start_dt > end_dt → 422。
• 响应 points[] 按 dt 升序,每点复用上面的 results(口径一致);裁前导全 0 天;范围内缺口日直接缺点(前端断线、不补零)。

口径与规则

• period=day → ads_trd_group_funnel_daily(给 dt 取该行,否则 MAX(dt));last_7d/last_30d → ads_trd_group_funnel_rolling 唯一行的 *_7d/*_30d。
• 不读 bitmap、不做 OR、不跨天聚合。
• step_index 从 1 开始;第 1 步转化率/流失率为 null。
• conversion_rate[i] = uv[i]/uv[i-1],dropoff_rate[i] = 1 - conversion_rate[i];uv[i-1]==0 时均为 null(不除零)。
• data_status ∈ {ready, missing}:目标行不存在或对应列为 NULL → missing,绝不补零。

测试

pytest

覆盖:period 校验、day vs rolling 路由、snapshot_dt 选择/校验(今天/未来 → 422;不存在 → missing)、趋势范围/裁前导0/缺口/校验、转化口径、data_status、假数据源、SQLAlchemy 仓库(内存 SQLite)、以及接口响应结构——全程无需 Postgres。