基于Feast实现指定历史日期的30天交易数量离线特征查询

Feast完全支持你描述的离线场景需求——传入指定历史日期，计算该日期前30天的交易数量特征，核心是通过滑动窗口聚合定义和历史时间戳参数传递来实现，无需依赖NOW()这类动态时间函数。

具体实现步骤 #

1. 重新定义数据源与FeatureView（推荐方式）

放弃硬编码NOW()的聚合查询，改用Feast内置的滑动窗口聚合功能，确保离线/在线逻辑一致：

from feast import FeatureView, Field, aggregation
from feast.infra.offline_stores.contrib.postgres_offline_store.postgres_source import PostgreSQLSource
from datetime import timedelta

# 定义基础数据源：仅获取交易明细的核心字段
order_source = PostgreSQLSource(
    name="client_placed_order_source",
    query="SELECT client_id, event_timestamp FROM events.client_placed_order",
    timestamp_field="event_timestamp",
)

# 定义FeatureView，绑定30天滑动窗口聚合
client_order_features = FeatureView(
    name="client_order_features",
    entities=["client_id"],
    schema=[Field(name="cnt_last_30days", dtype="int64")],
    online=True,
    source=order_source,
    aggregations=[
        aggregation(
            function="count",
            name="cnt_last_30days",
            window=timedelta(days=30),
        )
    ],
)

2. 传入历史日期获取离线特征

构造包含目标历史时间戳的实体DataFrame，调用get_historical_features即可自动计算对应窗口的特征：

import pandas as pd
from feast import FeatureStore

# 初始化Feast Store
store = FeatureStore(repo_path="./your_feast_repo")

# 构造实体请求：指定每个客户需要计算特征的历史时间点（2022-02-01）
entity_df = pd.DataFrame({
    "client_id": ["client_001", "client_002", "client_003"],
    "event_timestamp": pd.to_datetime(["2022-02-01"] * 3)
})

# 获取历史特征：Feast将自动计算每个客户在2022-02-01前30天的交易数量
historical_features = store.get_historical_features(
    entity_df=entity_df,
    features=["client_order_features:cnt_last_30days"]
).to_df()

print(historical_features)

关键逻辑说明 #

• 在线场景下，Feast会自动使用当前时间作为event_timestamp计算窗口特征，与你之前用NOW()的逻辑完全一致，保证特征一致性。
• 离线场景中，Feast会根据你传入的event_timestamp（即历史日期），过滤该时间点之前的交易数据，并计算过去30天的聚合值，完全匹配你需要的2022-01-01至2022-02-01时间范围。

可选：自定义SQL参数化查询（不推荐） #

如果需要手动控制SQL逻辑，也可以使用Feast的参数化查询（通过@event_timestamp占位符），但内置窗口聚合更简洁且不易出错：

order_source = PostgreSQLSource(
    name="client_placed_order_source",
    query="""
        SELECT client_id,
               @event_timestamp AS event_timestamp,
               SUM(CASE WHEN event_timestamp >= @event_timestamp - INTERVAL '30 days' THEN 1 ELSE 0 END) AS cnt_last_30days
        FROM events.client_placed_order
        WHERE event_timestamp <= @event_timestamp
        GROUP BY client_id
    """,
    timestamp_field="event_timestamp",
)

内容的提问来源于stack exchange，提问作者Vladislav Zhilmanov