aligned.model_store.ModelFeatureStore

class documentation

class ModelFeatureStore: (source)

View In Hierarchy

Undocumented

Method all_predictions Undocumented
Method cached_at Loads the model features from a pre computed location
Method depends_on Returns the views and models that the model depend on to compute it's output.
Method feature_references_for Undocumented
Method features_for Returns the features for the given entities
Async Method freshness Undocumented
Method has_exposed_model Undocumented
Method has_one_source_for_input_features If the input features are from the same source.
Method input_features The input features to load, based on the input features in a model.
Async Method input_freshness Undocumented
Method input_request Undocumented
Async Method insert_predictions Writes data to a source defined as a prediction source
Method needed_entities Undocumented
Method predict_over Undocumented
Async Method prediction_freshness Undocumented
Method prediction_request Undocumented
Method predictions_between Undocumented
Method predictions_for Undocumented
Method process_features Undocumented
Method raw_string_features Undocumented
Method request Undocumented
Async Method upsert_predictions Upserts data to a source defined as a prediction source
Method using_model Makes it possible to define which model to use for predictions.
Method using_source Undocumented
Method using_version Undocumented
Method with_labels Will also load the labels for the model
Class Variable model Undocumented
Class Variable model_to_use Undocumented
Class Variable selected_version Undocumented
Class Variable store Undocumented
Property dataset_store Undocumented
Property location Undocumented
Property source Undocumented
def all_predictions(self, limit: int | None = None) -> RetrievalJob: (source)

Undocumented

def cached_at(self, location: DataFileReference) -> RetrievalJob: (source)

Loads the model features from a pre computed location

```python from aligned import FileSource

store = await FileSource.json_at("features-latest.json").feature_store()

cached_features = FileSource.parquet_at("titanic_features.parquet")

df = store.model("titanic")
.cached_at(cached_features).to_polars()

print(df.collect()) >>> ┌──────────────┬───────┬─────────┬─────────────────────┬──────────────┐ >>> │ passenger_id ┆ is_mr ┆ is_male ┆ constant_filled_age ┆ has_siblings │ >>> │ --- ┆ --- ┆ --- ┆ --- ┆ --- │ >>> │ i32 ┆ bool ┆ bool ┆ f64 ┆ bool │ >>> ╞══════════════╪═══════╪═════════╪═════════════════════╪══════════════╡ >>> │ 1 ┆ true ┆ true ┆ 22.0 ┆ true │ >>> │ 2 ┆ false ┆ false ┆ 38.0 ┆ true │ >>> │ 3 ┆ false ┆ false ┆ 26.0 ┆ false │ >>> └──────────────┴───────┴─────────┴─────────────────────┴──────────────┘ ```

Args:
location (DataFileReference): _description_
Returns:
RetrievalJob: _description_
def depends_on(self) -> set[FeatureLocation]: (source)

Returns the views and models that the model depend on to compute it's output.

Examples:

```python @feature_view(name="passenger", ...) class Passenger:

passenger_id = Int32().as_entity()

age = Float()

@feature_view(name="location", ...) class Location:

location_id = String().as_entity()

location_area = Float()

@model_contract(name="some_model", ...) class SomeModel:

some_id = String().as_entity()

some_computed_metric = Int32()

@model_contract(

name="new_model", features=[

Passenger().age, Location().location_area, SomeModel().some_computed_metric

]

) class NewModel:

...

print(store.model("new_model").depends_on()) >>> { >>> FeatureLocation(location="feature_view", name="passenger"), >>> FeatureLocation(location="feature_view", name="location"), >>> FeatureLocation(location="model", name="some_model") >>> } ```

def feature_references_for(self, version: str) -> list[FeatureReference]: (source)

Undocumented

def features_for(self, entities: ConvertableToRetrievalJob | RetrievalJob, event_timestamp_column: str | None = None) -> RetrievalJob: (source)

Returns the features for the given entities

```python store = await FileSource.json_at("features-latest.json").feature_store()

df = store.model("titanic")
.features_for({"passenger_id": [1, 2, 3]}).to_polars()

print(df.collect()) >>> ┌──────────────┬───────┬─────────┬─────────────────────┬──────────────┐ >>> │ passenger_id ┆ is_mr ┆ is_male ┆ constant_filled_age ┆ has_siblings │ >>> │ --- ┆ --- ┆ --- ┆ --- ┆ --- │ >>> │ i32 ┆ bool ┆ bool ┆ f64 ┆ bool │ >>> ╞══════════════╪═══════╪═════════╪═════════════════════╪══════════════╡ >>> │ 1 ┆ true ┆ true ┆ 22.0 ┆ true │ >>> │ 2 ┆ false ┆ false ┆ 38.0 ┆ true │ >>> │ 3 ┆ false ┆ false ┆ 26.0 ┆ false │ >>> └──────────────┴───────┴─────────┴─────────────────────┴──────────────┘ ```

Args:
entities (dict[str, list] | RetrievalJob): The entities to fetch features for
Returns:
RetrievalJob: A retrieval job that can be used to fetch the features
async def freshness(self) -> dict[FeatureLocation, datetime | None]: (source)

Undocumented

def has_exposed_model(self) -> bool: (source)

Undocumented

def has_one_source_for_input_features(self) -> bool: (source)

If the input features are from the same source.

This can be interesting to know in order to automatically predict over the input. E.g. predict over all data in the source.

def input_features(self) -> list[FeatureReference]: (source)

The input features to load, based on the input features in a model.

async def input_freshness(self) -> dict[FeatureLocation, datetime | None]: (source)

Undocumented

def input_request(self, except_features: set[str] | None = None, event_timestamp_column: str | None = None) -> FeatureRequest: (source)

Undocumented

async def insert_predictions(self, predictions: ConvertableToRetrievalJob | RetrievalJob): (source)

Writes data to a source defined as a prediction source

```python @model_contract(

name="taxi_eta", features=[...] predictions_source=FileSource.parquet_at("predictions.parquet")

) class TaxiEta:

trip_id = Int32().as_entity()

duration = Int32()

...

store = FeatureStore.from_dir(".")

await store.model("taxi_eta").insert_predictions({
"trip_id": [1, 2, 3, ...], "duration": [20, 33, 42, ...]

})

def needed_entities(self) -> set[Feature]: (source)

Undocumented

def predict_over(self, entities: ConvertableToRetrievalJob | RetrievalJob) -> PredictionJob: (source)

Undocumented

async def prediction_freshness(self) -> datetime | None: (source)

Undocumented

def prediction_request(self, exclude_features: set[str] | None = None, model_version_as_entity: bool = False) -> RetrievalRequest: (source)

Undocumented

def predictions_between(self, start_date: datetime, end_date: datetime) -> RetrievalJob: (source)

Undocumented

def predictions_for(self, entities: ConvertableToRetrievalJob | RetrievalJob, event_timestamp_column: str | None = None, model_version_as_entity: bool | None = None) -> RetrievalJob: (source)

Undocumented

def process_features(self, input: RetrievalJob | ConvertableToRetrievalJob) -> RetrievalJob: (source)

Undocumented

def raw_string_features(self, except_features: set[str]) -> set[str]: (source)

Undocumented

def request(self, except_features: set[str] | None = None, event_timestamp_column: str | None = None) -> FeatureRequest: (source)

Undocumented

async def upsert_predictions(self, predictions: ConvertableToRetrievalJob | RetrievalJob): (source)

Upserts data to a source defined as a prediction source

```python @model_contract(

name="taxi_eta", features=[...] predictions_source=FileSource.parquet_at("predictions.parquet")

) class TaxiEta:

trip_id = Int32().as_entity()

duration = Int32()

...

store = FeatureStore.from_dir(".")

await store.model("taxi_eta").upsert_predictions({
"trip_id": [1, 2, 3, ...], "duration": [20, 33, 42, ...]

})

def using_model(self, model_to_use: ExposedModel) -> ModelFeatureStore: (source)

Makes it possible to define which model to use for predictions.

def using_source(self, source: BatchDataSource) -> ModelFeatureStore: (source)

Undocumented

def using_version(self, version: str) -> ModelFeatureStore: (source)

Undocumented

def with_labels(self, label_refs: set[FeatureReference] | None = None) -> SupervisedModelFeatureStore: (source)

Will also load the labels for the model

```python store = await FileSource.json_at("features-latest.json").feature_store()

data = store.model("titanic")
.with_labels().features_for({"passenger_id": [1, 2, 3]}).to_polars()

print(data.labels.collect(), data.input.collect()) >>> ┌──────────┐ ┌───────┬─────────┬─────────────────────┬──────────────┐ >>> │ survived │ │ is_mr ┆ is_male ┆ constant_filled_age ┆ has_siblings │ >>> │ --- │ │ --- ┆ --- ┆ --- ┆ --- │ >>> │ bool │ │ bool ┆ bool ┆ f64 ┆ bool │ >>> ╞══════════╡ ╞═══════╪═════════╪═════════════════════╪══════════════╡ >>> │ false │ │ true ┆ true ┆ 22.0 ┆ true │ >>> │ true │ │ false ┆ false ┆ 38.0 ┆ true │ >>> │ true │ │ false ┆ false ┆ 26.0 ┆ false │ >>> └──────────┘ └───────┴─────────┴─────────────────────┴──────────────┘ ```

Returns:
SupervisedModelFeatureStore: A new queryable feature store
model: ModelSchema = (source)

Undocumented

model_to_use: ExposedModel | None = (source)

Undocumented

selected_version: str | None = (source)

Undocumented

store: ContractStore = (source)

Undocumented

@property
dataset_store: DatasetStore | None = (source)

Undocumented

@property
location: FeatureLocation = (source)

Undocumented

@property
source: BatchDataSource = (source)

Undocumented