""" User feature views for recommendation system. IMPORTANT: FEAST_SNOWFLAKE_DATABASE must be set BEFORE importing this module. The SnowflakeSource objects call get_snowflake_database() at module import time, baking in the database value. If the env var isn't set, you'll get an error (preventing accidental use of wrong database). """ from datetime import timedelta from feast import FeatureView, Field from feast.infra.offline_stores.snowflake_source import SnowflakeSource from feast.types import Float32, Int64 from .entities import user from .config import get_snowflake_database, get_snowflake_schema # Snowflake data source for user features user_features_source = SnowflakeSource( name="user_features_snowflake", database=get_snowflake_database(), schema=get_snowflake_schema(), table="RECS_HOOKS_RANKER_USER_FEATURE_STORE", timestamp_field="EVENT_TIMESTAMP", created_timestamp_column="CREATED_TIMESTAMP", ) # User features user_features = FeatureView( name="user_features", description="User features metrics including average watch time", entities=[user], ttl=timedelta(days=30), # Keep features for 30 days schema=[ # Note: Don't include user_id here - it's specified via entities=[user] Field( name="AVERAGE_WATCH_TIME_30D", dtype=Float32, description="Average watch time in seconds across all hooks watched", ), Field( name="MEDIAN_WATCH_TIME_30D", dtype=Float32, description="Median watch time in seconds across all hooks watched", ), Field( name="P95_WATCH_TIME_30D", dtype=Float32, description="95th percentile watch time in seconds across all hooks watched", ), Field( name="AVERAGE_WATCH_TIME_7D", dtype=Float32, description="Average watch time in seconds across all hooks watched in the last 7 days", ), Field( name="MEDIAN_WATCH_TIME_7D", dtype=Float32, description="Median watch time in seconds across all hooks watched in the last 7 days", ), Field( name="P95_WATCH_TIME_7D", dtype=Float32, description="95th percentile watch time in seconds across all hooks watched in the last 7 days", ), Field( name="NUM_ACTIVE_DAYS_30D", dtype=Int64, description="Number of active days watching hooks in the last 30 days", ), Field( name="NUM_ACTIVE_DAYS_7D", dtype=Int64, description="Number of active days watching hooks in the last 7 days", ), Field(name="FRACTION_OVER_2S_7D", dtype=Float32, description="Fraction of hooks watched by the user for more than 2 seconds in the last 7 days"), Field(name="FRACTION_OVER_5S_7D", dtype=Float32, description="Fraction of hooks watched by the user for more than 5 seconds in the last 7 days"), Field(name="FRACTION_OVER_2S_30D", dtype=Float32, description="Fraction of hooks watched by the user for more than 2 seconds in the last 30 days"), Field(name="FRACTION_OVER_5S_30D", dtype=Float32, description="Fraction of hooks watched by the user for more than 5 seconds in the last 30 days"), ], source=user_features_source, online=True, # Available for online serving tags={"team": "recs"}, ) user_genre_features_source = SnowflakeSource( name="user_genre_features_snowflake", database=get_snowflake_database(), schema=get_snowflake_schema(), table="RECS_HOOKS_RANKER_USER_GENRE_FEATURE_STORE", timestamp_field="EVENT_TIMESTAMP", created_timestamp_column="CREATED_TIMESTAMP", ) user_genre_features = FeatureView( name="user_genre_features", description="User genre features metrics including average watch time of the user for hooks of each genre", entities=[user], ttl=timedelta(days=60), # Keep features for 60 days schema=[ Field(name="GENRE_AMBIENT_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre ambient"), Field(name="GENRE_CLASSICAL_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre classical"), Field(name="GENRE_COUNTRY_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre country"), Field(name="GENRE_DANCE_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre dance"), Field(name="GENRE_ELECTRONIC_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre electronic"), Field(name="GENRE_EXPERIMENTAL_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre experimental"), Field(name="GENRE_FOLK_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre folk"), Field(name="GENRE_FUNK_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre funk"), Field(name="GENRE_HIP_HOP_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre hip hop"), Field(name="GENRE_JAZZ_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre jazz"), Field(name="GENRE_LATIN_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre latin"), Field(name="GENRE_METAL_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre metal"), Field(name="GENRE_POP_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre pop"), Field(name="GENRE_RANDB_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre R&B"), Field(name="GENRE_REGGAE_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre reggae"), Field(name="GENRE_ROCK_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre rock"), Field(name="GENRE_SOUL_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre soul"), Field(name="GENRE_SOUNDTRACK_AVG_WATCH_TIME_30D", dtype=Float32, description="Average watch time of the user for hooks of genre soundtrack"), ], source=user_genre_features_source, online=True, # Available for online serving tags={"team": "recs"}, )