特征平台：Feast 实战

# 没有特征平台时的典型混乱场景
# 团队 A 的模型训练脚本
import pandas as pd

def compute_user_avg_spend(user_id: str) -> float:
    """计算用户平均消费 - 团队 A 的实现"""
    orders = pd.read_csv("data/orders.csv")
    return orders[orders["user_id"] == user_id]["amount"].mean()

# 团队 B 的推理服务 - 同样的逻辑重新写了一遍
def get_user_avg_spend(user_id: str) -> float:
    """获取用户平均消费 - 团队 B 的实现"""
    import sqlite3
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(
        "SELECT AVG(amount) FROM orders WHERE user_id = ?",
        (user_id,)
    )
    return cursor.fetchone()[0]

# 使用 Feast 统一管理特征后的场景
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# 训练和推理使用完全相同的 API
feature_vector = store.get_online_features(
    features=[
        "driver_stats:conv_rate",
        "driver_stats:acc_rate",
        "driver_stats:avg_daily_trips",
    ],
    entity_rows=[{"driver_id": "1001"}],
).to_dict()

print(feature_vector)
# {'driver_id': ['1001'], 'conv_rate': [0.85],
#  'acc_rate': [0.92], 'avg_daily_trips': [42]}

from feast import FeatureStore
from datetime import datetime, timedelta

store = FeatureStore(repo_path="./feature_repo")

# 离线历史数据查询 - 用于训练
entity_df = pd.DataFrame({
    "driver_id": ["1001", "1002", "1003"],
    "event_timestamp": [
        datetime(2026, 4, 1, 10, 0, 0),
        datetime(2026, 4, 1, 11, 0, 0),
        datetime(2026, 4, 1, 12, 0, 0),
    ],
})

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "driver_stats:conv_rate",
        "driver_stats:acc_rate",
    ],
).to_df()

print(training_df.shape)
# (3, 5) - 包含 entity key, timestamp 和特征值

from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# 在线特征查询 - 用于实时推理
feature_vector = store.get_online_features(
    features=[
        "driver_stats:conv_rate",
        "driver_stats:acc_rate",
        "driver_stats:avg_daily_trips",
    ],
    entity_rows=[
        {"driver_id": "1001"},
        {"driver_id": "1002"},
    ],
).to_dict()

# 在推理服务中使用
def predict(driver_id: str) -> float:
    features = store.get_online_features(
        features=["driver_stats:conv_rate", "driver_stats:acc_rate"],
        entity_rows=[{"driver_id": driver_id}],
    ).to_dict()
    return model.predict(features)[0]

from feast import Entity, FeatureView, Field
from feast.types import Float32, Int64, String
from feast.data_source import FileSource
from datetime import timedelta

# 定义 Entity - 司机
driver = Entity(
    name="driver",
    join_keys=["driver_id"],
    description="网约车司机",
)

# 定义数据源
driver_stats_source = FileSource(
    path="data/driver_stats.parquet",
    timestamp_field="event_timestamp",
)

# 定义 Feature View
driver_stats_fv = FeatureView(
    name="driver_stats",
    entities=[driver],
    ttl=timedelta(days=30),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
    ],
    source=driver_stats_source,
    online=True,
)

from feast import OnDemandFeatureView, FeatureService
from feast.on_demand_feature_view import on_demand_feature_view
from typing import Dict

# On-Demand Feature View - 实时计算派生特征
@on_demand_feature_view(
    sources=[driver_stats_fv],
    schema=[Field(name="trip_efficiency", dtype=Float32)],
)
def trip_efficiency(features: Dict[str, float]) -> Dict[str, float]:
    """计算行程效率 = 成交率 * 活跃率 * 日均行程归一化"""
    conv = features.get("conv_rate", 0.0)
    acc = features.get("acc_rate", 0.0)
    trips = features.get("avg_daily_trips", 0.0)
    normalized_trips = min(trips / 100.0, 1.0)
    return {"trip_efficiency": conv * acc * normalized_trips}

# Feature Service - 组合特征供模型使用
ride_prediction_service = FeatureService(
    name="ride_prediction_features",
    features=[
        driver_stats_fv,
        trip_efficiency,
    ],
)

# 查看注册表中的所有特征定义
from feast import FeatureStore
from feast.infra.registry.sql import SqlRegistry

store = FeatureStore(repo_path="./feature_repo")

# 列出所有 Entity
for entity in store.list_entities():
    print(f"Entity: {entity.name}, keys: {entity.join_keys}")

# 列出所有 Feature View
for fv in store.list_feature_views():
    print(f"FeatureView: {fv.name}")
    for field in fv.schema:
        print(f"  {field.name}: {field.dtype}")

# 列出所有 Feature Service
for fs in store.list_feature_services():
    print(f"FeatureService: {fs.name}")
    for fv in fs.features:
        print(f"  includes: {fv.name}")

# feature_store.yaml - 生产环境注册表配置
project: ride_prediction

# 使用远程注册表（S3 后端）
registry:
  path: s3://my-feast-registry/registry.db
  cache_ttl_seconds: 600

# 离线存储使用 BigQuery
offline_store:
  type: bigquery
  project_id: my-gcp-project
  dataset: feast_offline

# 在线存储使用 Redis
online_store:
  type: redis
  connection_string: "redis://redis-cluster:6379"

provider: gcp

# 监控配置
feature_server:
  enabled: true
  auth:
    type: mtls

import pandas as pd
from datetime import datetime
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# 训练数据 - 使用 point-in-time join
entity_df = pd.DataFrame({
    "driver_id": ["1001", "1001", "1002", "1003"],
    "event_timestamp": [
        datetime(2026, 3, 1, 10, 0, 0),
        datetime(2026, 3, 15, 10, 0, 0),
        datetime(2026, 3, 1, 11, 0, 0),
        datetime(2026, 3, 1, 12, 0, 0),
    ],
    "label": [1, 0, 1, 0],
})

# Feast 确保每个 timestamp 只使用该时间点之前的特征
training_data = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "driver_stats:conv_rate",
        "driver_stats:acc_rate",
    ],
).to_df()

# 验证：同一 driver 在不同时间点的特征值不同
print(training_data.groupby("driver_id")["conv_rate"].nunique())

# 检测 training-serving skew
import pandas as pd
import numpy as np
from scipy import stats

def detect_training_serving_skew(
    training_features: pd.DataFrame,
    serving_features: pd.DataFrame,
    feature_name: str,
    threshold: float = 0.1,
) -> dict:
    """
    使用 PSI (Population Stability Index) 检测特征分布差异
    PSI < 0.1: 无显著变化
    0.1 <= PSI < 0.2: 轻微变化
    PSI >= 0.2: 显著变化，需要调查
    """
    def calculate_psi(expected: np.ndarray, actual: np.ndarray, bins: int = 10):
        breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
        breakpoints[-1] = float("inf")
        expected_counts = np.histogram(expected, breakpoints)[0]
        actual_counts = np.histogram(actual, breakpoints)[0]
        expected_pct = (expected_counts + 1e-6) / len(expected)
        actual_pct = (actual_counts + 1e-6) / len(actual)
        psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
        return psi

    psi = calculate_psi(
        training_features[feature_name].values,
        serving_features[feature_name].values,
    )
    return {"feature": feature_name, "psi": psi, "needs_attention": psi >= threshold}

from feast import FeatureStore
from datetime import datetime, timedelta
import pandas as pd

def check_feature_freshness(
    store: FeatureStore,
    feature_view_name: str,
    max_age_hours: int = 24,
) -> dict:
    """检查特征新鲜度""" 
    fv = store.get_feature_view(feature_view_name)
    registry = store.registry
    project = store.project

    # 获取注册表中的特征元数据
    fv_meta = registry.get_feature_view(feature_view_name, project)

    # 检查 data_source 中的最新时间戳
    if hasattr(fv_meta.source, "get_latest_timestamp"):
        latest_ts = fv_meta.source.get_latest_timestamp(
            start_date=datetime(2020, 1, 1),
            end_date=datetime.now(),
        )
        age = datetime.now() - latest_ts
        is_fresh = age < timedelta(hours=max_age_hours)
        return {
            "feature_view": feature_view_name,
            "latest_timestamp": latest_ts,
            "age_hours": age.total_seconds() / 3600,
            "is_fresh": is_fresh,
        }
    return {"feature_view": feature_view_name, "status": "unknown"}

# 特征质量监控仪表盘数据采集
def collect_feature_quality_metrics(
    store: FeatureStore,
    feature_view_name: str,
    entity_ids: list,
) -> dict:
    """收集特征质量指标""" 
    results = {
        "missing_rate": {},
        "out_of_range": {},
        "query_latency_ms": [],
    }

    range_checks = {
        "conv_rate": (0.0, 1.0),
        "acc_rate": (0.0, 1.0),
        "avg_daily_trips": (0, 1000),
    }

    import time
    for eid in entity_ids:
        start = time.time()
        features = store.get_online_features(
            features=[
                f"{feature_view_name}:conv_rate",
                f"{feature_view_name}:acc_rate",
            ],
            entity_rows=[{"driver_id": eid}],
        ).to_dict()
        elapsed = (time.time() - start) * 1000
        results["query_latency_ms"].append(elapsed)

        for fname, (min_val, max_val) in range_checks.items():
            val = features.get(fname, [None])[0]
            if val is None:
                results["missing_rate"][fname] = results["missing_rate"].get(fname, 0) + 1
            elif val < min_val or val > max_val:
                results["out_of_range"][fname] = results["out_of_range"].get(fname, 0) + 1

    return results

# feature_repo/driver_repo.py - 完整的特征定义
from feast import Entity, FeatureView, Field, FeatureStore
from feast.types import Float32, Int64, String
from feast.data_source import FileSource
from datetime import timedelta

# 1. Entity
driver = Entity(name="driver", join_keys=["driver_id"])

# 2. Data Source
driver_stats_source = FileSource(
    name="driver_stats_source",
    path="data/driver_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# 3. Feature View
driver_stats_fv = FeatureView(
    name="driver_stats",
    entities=[driver],
    ttl=timedelta(days=30),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
        Field(name="rating", dtype=Float32),
    ],
    source=driver_stats_source,
    online=True,
)

# 4. Feature Service
driver_score_service = FeatureService(
    name="driver_score_service",
    features=[driver_stats_fv],
)

# train_and_serve.py - 训练与推理全流程
import pandas as pd
import joblib
from datetime import datetime
from sklearn.ensemble import GradientBoostingRegressor
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# === 阶段 1: 获取训练数据 ===
entity_df = pd.read_csv("data/training_entity_df.csv")
entity_df["event_timestamp"] = pd.to_datetime(entity_df["event_timestamp"])

training_data = store.get_historical_features(
    entity_df=entity_df,
    features=["driver_stats:conv_rate", "driver_stats:acc_rate",
              "driver_stats:avg_daily_trips"],
).to_df()

X = training_data[["conv_rate", "acc_rate", "avg_daily_trips"]]
y = training_data["label"]
model = GradientBoostingRegressor(n_estimators=100)
model.fit(X, y)
joblib.dump(model, "models/driver_score_model.pkl")

# === 阶段 2: 填充在线存储 ===
store.materialize(
    start_date=datetime(2026, 1, 1),
    end_date=datetime.now(),
)

# === 阶段 3: 在线推理 ===
def predict_driver_score(driver_id: str) -> float:
    features = store.get_online_features(
        features=["driver_stats:conv_rate", "driver_stats:acc_rate",
                  "driver_stats:avg_daily_trips"],
        entity_rows=[{"driver_id": driver_id}],
    ).to_dict()
    return model.predict([features])[0]