🏗️ 特征仓库与ML平台¶

# feature_store/feature_repo/feature_definition.py

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

# 1. 定义实体（业务主键）
user = Entity(
    name="user_id",
    join_keys=["user_id"],
    description="用户唯一标识"
)

product = Entity(
    name="product_id",
    join_keys=["product_id"],
    description="商品唯一标识"
)

# 2. 定义数据源
user_stats_source = FileSource(
    path="data/user_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp"
)

# 3. 定义特征视图
user_stats_fv = FeatureView(
    name="user_stats",
    entities=[user],
    ttl=timedelta(days=1),  # 特征有效期
    schema=[
        Field(name="total_orders", dtype=Int64, description="历史总订单数"),
        Field(name="avg_order_amount", dtype=Float32, description="平均订单金额"),
        Field(name="days_since_last_order", dtype=Int64, description="距上次下单天数"),
        Field(name="user_segment", dtype=String, description="用户分层(高/中/低)"),
    ],
    source=user_stats_source,
    online=True,  # 同步到在线存储
    tags={"team": "recommendation", "version": "v2"},
)

from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# 训练集: 用实体DataFrame获取历史特征（Point-in-Time Join）
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003, 1004],
    "event_timestamp": pd.to_datetime([
        "2026-01-15", "2026-01-16", "2026-01-17", "2026-01-18"
    ])
})

# 获取特征（自动处理时间穿越，避免数据泄露）
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_stats:total_orders",
        "user_stats:avg_order_amount",
        "user_stats:days_since_last_order",
        "user_stats:user_segment",
    ]
).to_df()

print(training_df)
# user_id | event_timestamp | total_orders | avg_order_amount | ...

# 在线推理: 低延迟获取最新特征
online_features = store.get_online_features(
    features=[
        "user_stats:total_orders",
        "user_stats:avg_order_amount",
        "user_stats:days_since_last_order",
    ],
    entity_rows=[
        {"user_id": 1001},
        {"user_id": 1002},
    ]
).to_dict()

print(online_features)
# {'user_id': [1001, 1002], 'total_orders': [42, 7], ...}

class MLPlatformClient:
    """统一ML平台客户端"""

    def __init__(self, config_path="ml-platform-config.yaml"):
        import yaml
        with open(config_path) as f:  # with自动管理资源，确保文件正确关闭
            self.config = yaml.safe_load(f)

        self.feature_store = FeatureStore(repo_path="feature_repo/")
        self.mlflow_client = mlflow.tracking.MlflowClient()

    def get_training_data(self, entity_df, feature_list):
        """从Feature Store获取训练数据"""
        return self.feature_store.get_historical_features(
            entity_df=entity_df,
            features=feature_list
        ).to_df()

    def train_and_log(self, model, X_train, y_train, params):
        """训练并记录到MLflow"""
        with mlflow.start_run():
            mlflow.log_params(params)
            model.fit(X_train, y_train)

            # 记录指标
            from sklearn.metrics import accuracy_score
            train_acc = accuracy_score(y_train, model.predict(X_train))
            mlflow.log_metric("train_accuracy", train_acc)

            # 记录模型
            mlflow.sklearn.log_model(model, "model")

            return mlflow.active_run().info.run_id

    def evaluate_and_promote(self, run_id, test_df, model_name):
        """评估并决定是否推到生产"""
        # 加载模型
        model_uri = f"runs:/{run_id}/model"
        model = mlflow.sklearn.load_model(model_uri)

        X_test = test_df.drop("label", axis=1)
        y_test = test_df["label"]

        acc = accuracy_score(y_test, model.predict(X_test))

        # 注册模型
        mv = mlflow.register_model(model_uri, model_name)

        # 质量门禁
        if acc >= 0.85:
            # 使用别名系统（MLflow 2.9.0+ 推荐，替代已废弃的 Stage 系统）
            self.mlflow_client.set_registered_model_alias(
                name=model_name,
                alias="champion",
                version=mv.version
            )
            print(f"✅ 模型 v{mv.version} 推到生产 (acc={acc:.4f})")
        else:
            print(f"❌ 模型 v{mv.version} 未通过门禁 (acc={acc:.4f})")

        return acc