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