پیادهسازی پایپلاین استخراج ویژگی و Feature Store با Feast و ردیس

from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List
import pandas as pd
from feast import Entity, FeatureView, Field, FileSource, FeatureStore, ValueType
from feast.types import Float32, Int64

# 1. Define Entity and Offline Parquet Source
user_entity = Entity(
    name='user_id',
    join_keys=['user_id'],
    value_type=ValueType.INT64,
    description='Unique identifier for banking customer'
)

transaction_stats_source = FileSource(
    name='user_tx_aggregates_source',
    path='s3://ml-lake/features/user_tx_rolling_stats.parquet',
    timestamp_field='event_timestamp',
    created_timestamp_column='created_timestamp'
)

# 2. Define Unified FeatureView with strict TTL
user_risk_fv = FeatureView(
    name='user_risk_features',
    entities=[user_entity],
    ttl=timedelta(hours=48),
    schema=[
        Field(name='tx_count_24h', dtype=Int64),
        Field(name='avg_amount_24h', dtype=Float32),
        Field(name='failed_pin_attempts_1h', dtype=Int64),
        Field(name='velocity_score', dtype=Float32),
    ],
    online=True,
    source=transaction_stats_source
)

class RealTimeFeatureService:
    def __init__(self, repo_path: str = './feature_repo'):
        self.store = FeatureStore(repo_path=repo_path)
        self.feature_refs = [
            'user_risk_features:tx_count_24h',
            'user_risk_features:avg_amount_24h',
            'user_risk_features:failed_pin_attempts_1h',
            'user_risk_features:velocity_score'
        ]

    def build_point_in_time_training_set(self, entity_df: pd.DataFrame) -> pd.DataFrame:
        """
        Performs an ASOF Point-in-Time Join to prevent future data leakage.
        entity_df must contain ['user_id', 'event_timestamp', 'is_fraud'].
        """
        training_data = self.store.get_historical_features(
            entity_df=entity_df,
            features=self.feature_refs
        ).to_df()
        return training_data

    def sync_offline_to_redis(self) -> None:
        """Materializes latest feature values from Parquet into Redis Online Store."""
        now_utc = datetime.now(timezone.utc)
        self.store.materialize_incremental(end_date=now_utc)

    def fetch_online_vector(self, user_ids: List[int]) -> Dict[str, Any]:
        """Sub-10ms Redis lookup via pipelined HGETALL/HMGET under the hood."""
        entity_rows = [{'user_id': uid} for uid in user_ids]
        response = self.store.get_online_features(
            features=self.feature_refs,
            entity_rows=entity_rows
        )
        return response.to_dict()