特征平台与训练服务一致性：时间点正确性与特征回填

定义 特征的唯一权威定义    存储 离线全量 + 在线最新值
服务 离线批量取数 + 在线点查    治理 血缘、版本、监控、权限
引入时机 —— 不需要：单模型、单团队、特征 < 50 个、离线批预测
            需要：多模型复用 / 实时特征 / 团队 > 2 个 / 上线频繁

实体（Entity）  ：特征的主体，如 user_id、item_id
特征视图（View）：一组同源特征，如「用户近 30 天行为」
特征（Feature） ：具体某一列，如 user_7d_orders

from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64
from datetime import timedelta
user = Entity(name="user_id", join_keys=["user_id"])
source = FileSource(path="data/user_stats.parquet",
                    timestamp_field="event_timestamp")   # 时间戳字段是关键
user_stats_view = FeatureView(
    name="user_stats", entities=[user], ttl=timedelta(days=30),
    schema=[Field(name="user_7d_orders", dtype=Int64),
            Field(name="user_30d_gmv", dtype=Float32)],
    source=source)

上游表 orders_raw → 加工任务 job_user_stats（每天 02:00）
  → 特征视图 user_stats.user_7d_orders
  → 模型 fraud-detector v3 / churn-model v2

FEATURE_LINEAGE = {"user_stats.user_7d_orders": {
    "upstream_tables": ["orders_raw"], "job": "job_user_stats",
    "owner": "growth-team", "sla": "每日 03:00 前就绪"}}
def impact_analysis(table):
    return [f for f, meta in FEATURE_LINEAGE.items()
            if table in meta["upstream_tables"]]
print(impact_analysis("orders_raw"))

离线存储（Offline Store）：全量历史，供训练与回填
  典型 Parquet / Hive / 数仓，大批量扫描
在线存储（Online Store）：每个实体的最新值，供推理
  典型 Redis / DynamoDB / Cassandra，单键点查

路线 A：批计算 → 写在线存储（Lambda） 简单可靠，分钟级延迟，适合大多数场景
路线 B：流计算 → 同写离线在线（Kappa） 实时最好，两路逻辑易漂移，适合风控推荐

import pandas as pd
def check_consistency(offline_df, online_df, keys, features):
    """抽样比对离线与在线取值是否一致"""
    merged = offline_df.merge(online_df, on=keys, suffixes=("_off", "_on"))
    return {f: {"mismatch_rate": float(((merged[f + "_off"] - merged[f + "_on"]).abs() > 1e-6).mean()),
                "max_diff": float((merged[f + "_off"] - merged[f + "_on"]).abs().max())}
            for f in features}

# 错误！所有历史样本都拿到了「今天」的特征
samples = labels.merge(feature_latest, on="user_id")
# 结果：模型在训练时「偷看」了未来，AUC 虚高 0.2

def point_in_time_join(labels, features, entity="user_id"):
    """对每个标签时刻，取该时刻之前最近的一条特征"""
    labels = labels.sort_values("event_timestamp")
    features = features.sort_values("event_timestamp")
    out = []
    for _, row in labels.iterrows():   # 生产环境请改用 ASOF JOIN
        hist = features[(features[entity] == row[entity]) & (features["event_timestamp"] <= row["event_timestamp"])]
        latest = hist.iloc[-1] if len(hist) else None
        out.append({**row, "f1": None if latest is None else latest["f1"]})
    return pd.DataFrame(out)

-- DuckDB / Snowflake / BigQuery 都支持 ASOF JOIN
SELECT l.user_id, l.event_timestamp, l.label, f.f1, f.f2
FROM labels l ASOF JOIN features f
  ON l.user_id = f.user_id AND l.event_timestamp >= f.event_timestamp;

def backfill(feature_view, start, end, granularity="1d"):
    """按天分片回填，每片独立可重试"""
    cur = start
    while cur < end:
        try:
            compute_and_write(feature_view, cur, cur + granularity)
        except Exception as e:
            log_failure(feature_view, cur, e)
        cur += granularity

def compute_user_7d_orders(events, as_of):
    """events: 事件表; as_of: 计算时点。离线和在线都调这个函数"""
    window = events[(events["ts"] > as_of - timedelta(days=7)) &
                    (events["ts"] <= as_of)]
    return window.groupby("user_id").size().rename("user_7d_orders")

def shadow_compare(requests, offline_fn, online_fn, tol=1e-4):
    mismatches = [(r, offline_fn(r), online_fn(r)) for r in requests
                  if abs(offline_fn(r) - online_fn(r)) > tol]
    print(f"不一致率 {len(mismatches) / len(requests):.4%}")
    return mismatches

import numpy as np
def psi(expected, actual, bins=10):
    """群体稳定性指数：衡量两个分布的差异"""
    breakpoints = np.percentile(expected, np.linspace(0, 100, bins + 1))
    breakpoints[0], breakpoints[-1] = -np.inf, np.inf
    e = np.clip(np.histogram(expected, breakpoints)[0] / len(expected), 1e-6, None)
    a = np.clip(np.histogram(actual, breakpoints)[0] / len(actual), 1e-6, None)
    return float(np.sum((a - e) * np.log(a / e)))
# 经验阈值：< 0.1 稳定，0.1~0.2 需关注，> 0.2 显著漂移

FEATURE_CONTRACT = {"user_7d_orders": {"min": 0, "max": 10000, "null_ok": False},
                    "user_30d_gmv": {"min": 0.0, "null_ok": True}}
def validate(df, contract=FEATURE_CONTRACT):
    errors = []
    for col, rule in contract.items():
        if col not in df.columns:
            errors.append(f"缺少特征 {col}"); continue
        if not rule["null_ok"] and df[col].isna().any():
            errors.append(f"{col} 存在空值")
        if "min" in rule and (df[col] < rule["min"]).any():
            errors.append(f"{col} 存在越界小值")
    return errors

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, FeatureStore
from feast.types import Float32, Int64
driver = Entity(name="driver_id", join_keys=["driver_id"])
source = FileSource(path="data/driver_stats.parquet",
                    timestamp_field="event_timestamp",
                    created_timestamp_column="created")
driver_stats = FeatureView(
    name="driver_hourly_stats", entities=[driver], ttl=timedelta(days=1),
    schema=[Field(name="conv_rate", dtype=Float32),
            Field(name="avg_daily_trips", dtype=Int64)], source=source)
store = FeatureStore(repo_path=".")
store.apply([driver, driver_stats])          # 同步定义到注册表
store.materialize(start_date=datetime(2026, 9, 1),
                  end_date=datetime(2026, 10, 1))   # 离线 → 在线

features = store.get_online_features(
    features=["driver_hourly_stats:conv_rate", "driver_hourly_stats:avg_daily_trips"],
    entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}]).to_dict()
# {'driver_id': [1001, 1002], 'conv_rate': [0.53, 0.71], ...}

from feast import FeatureService
service = FeatureService(name="training_v1", features=[driver_stats])
training_df = store.get_historical_features(
    entity_df=entity_df,          # 必须含 entity 列 + event_timestamp 列
    features=service).to_df()

1. get_historical_features 生成训练集（自动时间点正确）
2. 训练模型，把特征列表写进模型元数据
3. materialize 同步最新特征到在线存储
4. 推理时用 get_online_features 取同样的特征名
5. 定期比对离线在线一致性

阶段一：特征逻辑集中到一个库，离线在线共用（零成本，收益最大）
阶段二：加特征契约与一致性校验任务
阶段三：引入 Feast 等平台，统一离线在线取数
阶段四：加血缘、监控与权限治理，再到实时特征与版本管理

特征定义 → 离线存储（全量历史）→ 时间点连接（防泄漏）
  → 在线存储（最新值）→ 一致性校验 + 特征监控