피처 스토어

from feast import FeatureStore, Entity, Feature, FeatureView, FileSource
from feast.types import Float32, Int64, String
from datetime import timedelta
import pandas as pd

# 1. 피처 정의 (feature_repo/features.py)
# 엔티티 정의 - 피처의 기본 키
user = Entity(
    name="user_id",
    description="사용자 고유 ID",
    value_type=String,
)

# 데이터 소스 정의
user_daily_stats_source = FileSource(
    path="data/user_daily_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# 피처 뷰 정의
user_daily_stats_fv = FeatureView(
    name="user_daily_stats",
    entities=[user],
    ttl=timedelta(days=1),  # Time-to-live
    schema=[
        Feature(name="daily_transactions", dtype=Int64),
        Feature(name="daily_amount", dtype=Float32),
        Feature(name="avg_transaction_value", dtype=Float32),
        Feature(name="days_since_last_login", dtype=Int64),
    ],
    online=True,  # 온라인 스토어에 동기화
    source=user_daily_stats_source,
    tags={"team": "fraud-detection"},
)

# 2. 피처 스토어 초기화 및 데이터 적재
def setup_feature_store():
    """피처 스토어 설정 및 오프라인/온라인 스토어 동기화"""
    store = FeatureStore(repo_path="feature_repo/")

    # 피처 정의 적용
    store.apply([user, user_daily_stats_fv])

    # 오프라인 스토어에서 온라인 스토어로 최신 피처 로드
    store.materialize_incremental(end_date=pd.Timestamp.now())

    return store

# 3. 학습용 피처 데이터 조회 (오프라인)
def get_training_features(store: FeatureStore, entity_df: pd.DataFrame):
    """배치 학습을 위한 과거 피처 조회"""
    # entity_df에는 user_id와 event_timestamp가 포함되어야 함
    training_df = store.get_historical_features(
        entity_df=entity_df,
        features=[
            "user_daily_stats:daily_transactions",
            "user_daily_stats:daily_amount",
            "user_daily_stats:avg_transaction_value",
            "user_daily_stats:days_since_last_login",
        ],
    ).to_df()

    return training_df

# 4. 실시간 추론용 피처 조회 (온라인)
def get_online_features(store: FeatureStore, user_ids: list):
    """실시간 추론을 위한 최신 피처 조회 (저지연)"""
    feature_vector = store.get_online_features(
        features=[
            "user_daily_stats:daily_transactions",
            "user_daily_stats:daily_amount",
            "user_daily_stats:avg_transaction_value",
            "user_daily_stats:days_since_last_login",
        ],
        entity_rows=[{"user_id": uid} for uid in user_ids],
    ).to_dict()

    return feature_vector

# 5. 온디맨드 피처 (실시간 계산)
from feast import on_demand_feature_view, RequestSource

@on_demand_feature_view(
    sources=[user_daily_stats_fv],
    schema=[Feature(name="risk_score", dtype=Float32)],
)
def risk_features(inputs: pd.DataFrame) -> pd.DataFrame:
    """실시간으로 계산되는 파생 피처"""
    df = pd.DataFrame()
    # 간단한 규칙 기반 리스크 스코어
    df["risk_score"] = (
        inputs["daily_transactions"] * 0.3 +
        inputs["days_since_last_login"] * 0.2 +
        (inputs["daily_amount"] / 1000) * 0.5
    ).clip(0, 100)
    return df

# 사용 예시
if __name__ == "__main__":
    store = setup_feature_store()

    # 온라인 피처 조회 (실시간 추론)
    features = get_online_features(store, ["user_123", "user_456"])
    print(f"온라인 피처: {features}")

    # 학습 데이터 생성
    entity_df = pd.DataFrame({
        "user_id": ["user_123", "user_456"],
        "event_timestamp": [pd.Timestamp("2025-01-20"), pd.Timestamp("2025-01-20")]
    })
    training_data = get_training_features(store, entity_df)
    print(f"학습 데이터:\n{training_data}")