Feature Store

# 학습 데이터 생성 - Point-in-Time Join
from feast import FeatureStore
import pandas as pd
from datetime import datetime, timedelta

fs = FeatureStore(repo_path="feature_repo/")

# 학습할 레이블 데이터 (entity + timestamp + label)
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003, 1004, 1005],
    "event_timestamp": [
        datetime(2024, 1, 15, 10, 0),
        datetime(2024, 1, 15, 11, 30),
        datetime(2024, 1, 16, 9, 0),
        datetime(2024, 1, 16, 14, 0),
        datetime(2024, 1, 17, 8, 0),
    ],
    "churned": [0, 1, 0, 1, 0]  # 레이블
})

# Point-in-Time Join: 각 시점에 맞는 피처 조회
# 데이터 누수(Data Leakage) 자동 방지!
training_df = fs.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_behavior:purchase_count_7d",
        "user_behavior:purchase_count_30d",
        "user_behavior:total_amount_30d",
        "user_behavior:avg_order_value",
        "user_behavior:days_since_last_purchase",
    ]
).to_df()

print("학습 데이터:")
print(training_df)

# 모델 학습
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

feature_cols = [
    "purchase_count_7d", "purchase_count_30d",
    "total_amount_30d", "avg_order_value",
    "days_since_last_purchase"
]

X = training_df[feature_cols].fillna(0)
y = training_df["churned"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# 온라인 스토어에 피처 동기화 (Materialization)
from datetime import datetime

fs.materialize(
    start_date=datetime(2024, 1, 1),
    end_date=datetime.now()
)
print("피처가 온라인 스토어에 동기화되었습니다.")

# 온라인 서빙 (실시간 추론)
online_features = fs.get_online_features(
    features=[
        "user_behavior:purchase_count_7d",
        "user_behavior:purchase_count_30d",
        "user_behavior:total_amount_30d",
        "user_behavior:avg_order_value",
        "user_behavior:days_since_last_purchase",
    ],
    entity_rows=[{"user_id": 1001}]
).to_dict()

# 예측
features_for_prediction = [
    online_features["purchase_count_7d"][0],
    online_features["purchase_count_30d"][0],
    online_features["total_amount_30d"][0],
    online_features["avg_order_value"][0],
    online_features["days_since_last_purchase"][0],
]
prediction = model.predict([features_for_prediction])
print(f"이탈 예측: {'이탈 위험' if prediction[0] == 1 else '정상'}")

# 스트리밍 피처 (실시간 피처 업데이트)
# Tecton 또는 Flink + Feast 통합 예제

# 방법 1: Push API를 이용한 실시간 피처 업데이트 (Feast)
from feast import FeatureStore, PushSource, FeatureView, Field
from feast.types import Float32, Int64
from datetime import timedelta

# Push Source 정의 (실시간 데이터 입력)
user_activity_push_source = PushSource(
    name="user_activity_push",
    batch_source=...,  # 배치 소스 (백필용)
)

# 실시간 Feature View
user_realtime_fv = FeatureView(
    name="user_realtime_activity",
    entities=["user_id"],
    ttl=timedelta(minutes=30),
    schema=[
        Field(name="clicks_last_5min", dtype=Int64),
        Field(name="page_views_last_5min", dtype=Int64),
        Field(name="cart_value", dtype=Float32),
    ],
    source=user_activity_push_source,
    online=True,
)

# 스트림 처리 후 Push
from feast import FeatureStore
import pandas as pd
from datetime import datetime

fs = FeatureStore(repo_path="feature_repo/")

# Kafka Consumer에서 이벤트 처리 후 피처 Push
def process_event(event):
    """Kafka 이벤트를 처리하여 Feature Store에 Push"""
    user_id = event["user_id"]

    # 집계 로직 (Redis 등에서 윈도우 집계)
    aggregated_features = aggregate_user_activity(user_id)

    feature_df = pd.DataFrame({
        "user_id": [user_id],
        "clicks_last_5min": [aggregated_features["clicks"]],
        "page_views_last_5min": [aggregated_features["page_views"]],
        "cart_value": [aggregated_features["cart_value"]],
        "event_timestamp": [datetime.now()],
    })

    # Online Store에 직접 Push
    fs.push(
        push_source_name="user_activity_push",
        df=feature_df,
        to="online"
    )

# 방법 2: Flink + Feature Store 통합 패턴
"""
Flink Job (PyFlink or Java):
1. Kafka에서 클릭스트림 데이터 소비
2. 윈도우 집계 (5분 Tumbling Window)
3. 집계 결과를 Redis/DynamoDB에 직접 Write
4. 또는 Feast Push API로 전송

┌──────────┐     ┌──────────┐     ┌──────────────┐     ┌──────────────┐
│  Kafka   │────▶│  Flink   │────▶│   Aggregator │────▶│ Online Store │
│ (Events) │     │  Window  │     │  (5min sum)  │     │   (Redis)    │
└──────────┘     └──────────┘     └──────────────┘     └──────────────┘
"""

# 방법 3: On-Demand Feature (요청 시점 계산)
from feast import on_demand_feature_view

@on_demand_feature_view(
    sources=[user_behavior_fv],
    schema=[Field(name="is_high_value_customer", dtype=bool)]
)
def compute_customer_tier(inputs: pd.DataFrame) -> pd.DataFrame:
    """요청 시점에 동적으로 계산되는 피처"""
    df = pd.DataFrame()
    df["is_high_value_customer"] = (
        inputs["total_amount_30d"] > 1000
    ) & (
        inputs["purchase_count_30d"] > 5
    )
    return df

# On-Demand 피처 사용
online_features = fs.get_online_features(
    features=[
        "user_behavior:total_amount_30d",
        "compute_customer_tier:is_high_value_customer",
    ],
    entity_rows=[{"user_id": 1001}]
).to_dict()