Hamilton项目中的漏斗指标计算实践指南

2025-07-04 18:44:09作者：冯梦姬Eddie

Your single tool to express data, ML, and LLM pipelines with simple python functions. Runs anywhere that python runs, E.G. spark, airflow, jupyter, fastapi, etc. Incrementally adoptable. Use Hamilton to build testable, reusable, and self-documenting dataflows with lineage and metadata out of the box.

项目地址：https://gitcode.com/gh_mirrors/ha/hamilton

在数据分析领域，漏斗指标（如日活跃用户DAU、月活跃用户MAU、转化率和留存率）是衡量产品健康度和用户行为的关键指标。本文将介绍如何在Hamilton项目中实现这些核心业务指标的计算，并探讨如何通过开源协作构建可复用的分析组件。

为什么需要标准化的漏斗指标计算

传统上，企业需要为每个数据源重复编写类似的指标计算逻辑，这不仅效率低下，还容易产生不一致的结果。Hamilton作为一个数据流框架，为解决这个问题提供了优雅的方案——通过模块化的数据转换管道和可共享的计算组件。

技术实现方案

1. 数据源接入与标准化

以Zendesk为例，首先需要建立数据接入层。Hamilton支持通过装饰器定义数据源，建议采用以下标准化字段：

用户ID
事件类型（如注册、登录、购买）
时间戳
附加属性（可选）

@extract_columns("user_id", "event_type", "timestamp")
def raw_zendesk_data(zendesk_conn: Connection) -> pd.DataFrame:
    """从Zendesk提取原始事件数据"""
    query = "SELECT user_id, event_type, created_at FROM events"
    return pd.read_sql(query, zendesk_conn)

2. 数据转换层

建立统一的事件模式转换器，将不同来源的数据映射到标准格式：

def standardized_events(
    raw_data: pd.DataFrame,
    timezone: str = "UTC"
) -> pd.DataFrame:
    """将原始数据转换为标准事件格式"""
    return (
        raw_data
        .rename(columns={"created_at": "timestamp"})
        .assign(timestamp=lambda df: pd.to_datetime(df.timestamp).dt.tz_convert(timezone))
        .pipe(filter_invalid_events)
    )

3. 核心指标计算模块

活跃用户计算

def daily_active_users(events: pd.DataFrame, date: dt.date) -> int:
    """计算指定日期的DAU"""
    target_day = pd.Timestamp(date, tz=events.timestamp.dt.tz)
    return (
        events
        .loc[events.timestamp.dt.floor("D") == target_day]
        .user_id.nunique()
    )

def monthly_active_users(events: pd.DataFrame, month: dt.date) -> int:
    """计算指定月份的MAU"""
    target_month = pd.Timestamp(month).strftime("%Y-%m")
    return (
        events
        .loc[events.timestamp.dt.to_period("M").astype(str) == target_month]
        .user_id.nunique()
    )

转化率分析

def conversion_funnel(
    events: pd.DataFrame,
    steps: List[str],
    time_window: str = "7D"
) -> Dict[str, float]:
    """计算多步骤转化漏斗"""
    first_actions = (
        events
        .sort_values("timestamp")
        .groupby("user_id")
        .first()
        .reset_index()
    )
    
    results = {}
    for i, step in enumerate(steps[:-1]):
        next_step = steps[i+1]
        converted = first_actions[
            (first_actions.event_type == step) & 
            (first_actions.timestamp + pd.Timedelta(time_window) >= 
             first_actions.shift(-1).timestamp)
            )
        ].shape[0]
        rate = converted / first_actions[first_actions.event_type == step].shape[0]
        results[f"{step}_to_{next_step}"] = rate
    return results

留存率计算

def retention_rates(
    cohort_events: pd.DataFrame,
    initial_event: str,
    followup_event: str,
    periods: List[str] = ["1D", "7D", "30D"]
) -> Dict[str, float]:
    """计算多周期留存率"""
    cohort_users = set(
        cohort_events[cohort_events.event_type == initial_event]
        .user_id.unique()
    )
    
    results = {}
    for period in periods:
        delta = pd.Timedelta(period)
        retained_users = set(
            cohort_events[
                (cohort_events.event_type == followup_event) &
                (cohort_events.timestamp >= cohort_events.timestamp.min() + delta)
            ].user_id.unique()
        )
        results[f"retention_{period}"] = len(retained_users & cohort_users) / len(cohort_users)
    return results