Realtime Feature View Examples

Using Request Data

from tecton import RequestSource, RealtimeFeatureView, Calculation
from tecton.types import Float64, Field, Bool

request_schema = [Field("amt", Float64)]
transaction_request = RequestSource(schema=request_schema)

# An example of a realtime feature view that depends only on a request source.
transaction_amount_is_high = RealtimeFeatureView(
    name="transaction_amount_is_high",
    sources=[transaction_request],
    features=[Calculation(name="transaction_amount_is_high", expr="transaction_request.amt > 100")],
    description="The transaction amount is higher than $100.",
)

Using Feature View Dependencies

from tecton import RequestSource, RealtimeFeatureView, Calculation
from tecton.types import String, Timestamp, Float64, Field, Bool
from fraud.features.stream_features.user_transaction_amount_metrics import (
    user_transaction_amount_metrics,
)

request_schema = [Field("amt", Float64)]
transaction_request = RequestSource(schema=request_schema)

transaction_amount_is_higher_than_average = RealtimeFeatureView(
    name="transaction_amount_is_higher_than_average",
    sources=[transaction_request, user_transaction_amount_metrics],
    features=[
        Calculation(
            name="transaction_amount_is_higher_than_average",
            expr="transaction_request.amt > COALESCE(user_transaction_amount_metrics.amt_mean_1d_10m, 0)",
        )
    ],
    description="The transaction amount is higher than the 1 day average.",
)

Embedding Cosine Similarity

Cosine similarity between embeddings is a common and powerful cross feature that can be implemented with Realtime Feature Views. In this example, the feature view is computing the similarity between a pre-computed user embedding and a query embedding passed in the request to the Tecton Feature Service.

from tecton import realtime_feature_view, Attribute
from tecton.types import Float64
from ads.features.feature_tables.user_embeddings import user_embeddings
from ads.features.feature_tables.ad_embeddings import ad_embeddings
import pandas


@realtime_feature_view(
    sources=[ad_embeddings, user_embeddings],
    mode="python",
    features=[Attribute("cosine_similarity", Float64)],
    description="Computes the cosine similarity between a precomputed ad embedding and a precomputed user embedding.",
)
def user_ad_embedding_similarity(ad_embedding, user_embedding):
    import numpy as np
    from numpy.linalg import norm

    def cosine_similarity(a: np.ndarray, b: np.ndarray):
        # Handle the case where one or both entities do not have a precomputed embedding.
        if a is None or b is None:
            return -1.0

        return np.dot(a, b) / (norm(a) * norm(b))

    result = {}
    result["cosine_similarity"] = cosine_similarity(
        user_embedding["user_embedding"], ad_embedding["ad_embedding"]
    ).astype("float64")
    return result

Returning Struct Features

from tecton import realtime_feature_view, RequestSource, Attribute
from tecton.types import Array, Field, Float64, String, Struct

request_source = RequestSource([Field("input_float", Float64)])

output_schema = (Struct([Field("string_field", String), Field("float64_field", Float64)]),)


@realtime_feature_view(
    mode="python",
    sources=[request_source],
    features=[Attribute("output_struct", output_schema)],
    description="Output a struct with two fields.",
)
def simple_struct_example_rtfv(request):
    input_float = request["input_float"]
    return {
        "output_struct": {
            "string_field": str(input_float * 2),
            "float64_field": input_float * 2,
        }
    }

RTFVs Processing Data from Batch and Streaming Feature Views

These examples demonstrate how to create Realtime Feature Views that process data from pre-materialized Batch Feature Views (BFVs) and Stream Feature Views (SFVs).

RTFV with Batch Feature View

from tecton import RequestSource, realtime_feature_view, Attribute
from tecton.types import String, Timestamp, Int64, Field
from fraud.features.batch_features.user_date_of_birth import user_date_of_birth

request_schema = [Field("timestamp", String)]
request = RequestSource(schema=request_schema)
features = [Attribute("user_age", Int64)]


@realtime_feature_view(
    sources=[request, user_date_of_birth], mode="python", features=features, description="The user's age in days."
)
def user_age(request, user_date_of_birth):
    from datetime import datetime, date

    request_datetime = datetime.fromisoformat(request["timestamp"]).replace(tzinfo=None)
    dob_datetime = datetime.fromisoformat(user_date_of_birth["user_date_of_birth"])

    td = request_datetime - dob_datetime

    return {"user_age": td.days}

RTFV with Stream Feature View

from tecton import realtime_feature_view, RequestSource, Attribute
from tecton.types import String, Int64, Field
from Personalization.features.user_last_game_played import user_last_game_played

request = RequestSource(schema=[Field("TIMESTAMP", String)])


@realtime_feature_view(
    description="""Number of minutes elapsed between current time (coming from the request payload)
    and the time of the user's last game (fetched from a Streaming Feature View).""",
    sources=[request, user_last_game_played],
    mode="python",
    features=[Attribute("user_time_since_last_game", Int64)],
)
def user_time_since_install(request, user_last_game_played):
    from datetime import datetime, date
    import pandas

    request_datetime = pandas.to_datetime(request["TIMESTAMP"]).replace(tzinfo=None)
    last_game_datetime = pandas.to_datetime(user_last_game_played["TIME_GAME_PLAYED"])
    td = request_datetime - last_game_datetime

    return {"user_time_since_last_game": td.minute}

RTFV with Stream Feature View for Product Recommendations

from tecton import RequestSource, Attribute, realtime_feature_view
from tecton.types import Field, String, Bool

# Assume user_products_viewed is a streaming feature view
@realtime_feature_view(
    description="""This features verifies whether the current candidate product has been visited
   by the user in the last hour, it is computed in real-time and depends on a streaming feature view""",
    sources=[search_query, user_products_viewed],
    features=[Attribute("user_viewed_product_in_last_10_pages", Bool)],
    mode="python",
)
def user_viewed_product(search_query, user_products_viewed):
    product_id = search_query["product_uid"]
    last_products_viewed = user_products_viewed["product_uid_last_distinct_10_1h_continuous"]
    return {"user_viewed_product_in_last_10_pages": product_id in last_products_viewed}