affordability_score_feature

affordability_score_feature.py

customer_demographics_fv = fs_banking.get_feature_view("customer_demographics_fv")

from feast import Field
from feast.on_demand_feature_view import on_demand_feature_view
from feast.types import Float64
import pandas as pd

@on_demand_feature_view(
    sources=[customer_demographics_fv],
    schema=[Field(name="affordability_score", dtype=Float64)],
    description="Affordability score based on income",
)
def calculate_affordability_score(customer_features: pd.DataFrame) -> pd.DataFrame:
    df = pd.DataFrame()
    income = customer_features.get('income', pd.Series([0.0])).fillna(0.0)
    affordability_score = (income / 500000.0) * 100
    affordability_score = affordability_score.clip(upper=100.0)
    df["affordability_score"] = affordability_score.astype(float)
    return df

fs_banking.apply([calculate_affordability_score])