Realtime Feature View Examples

Using Request Data

from tecton import RequestSource, realtime_feature_view, Attribute
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.
@realtime_feature_view(
    sources=[transaction_request],
    mode="python",
    features=[Attribute("transaction_amount_is_high", Bool)],
    description="The transaction amount is higher than $100.",
)
def transaction_amount_is_high(transaction_request):
    return {"transaction_amount_is_high": transaction_request["amt"] > 100}

Using Feature View Dependencies

from tecton import RequestSource, realtime_feature_view, Attribute
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)


@realtime_feature_view(
    sources=[transaction_request, user_transaction_amount_metrics],
    mode="python",
    features=[Attribute("transaction_amount_is_higher_than_average", Bool)],
    description="The transaction amount is higher than the 1 day average.",
)
def transaction_amount_is_higher_than_average(transaction_request, user_transaction_amount_metrics):
    amount_mean = (
        0
        if user_transaction_amount_metrics["amt_mean_1d_10m"] is None
        else user_transaction_amount_metrics["amt_mean_1d_10m"]
    )
    return {"transaction_amount_is_higher_than_average": transaction_request["amt"] > amount_mean}

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,
        }
    }