Fetch historical feature values for a given entity ID
(
ctx: click.Context,
dataframe: str,
features: List[str],
start_date: str,
end_date: str,
)
| 163 | ) |
| 164 | @click.pass_context |
| 165 | def get_historical_features( |
| 166 | ctx: click.Context, |
| 167 | dataframe: str, |
| 168 | features: List[str], |
| 169 | start_date: str, |
| 170 | end_date: str, |
| 171 | ): |
| 172 | """ |
| 173 | Fetch historical feature values for a given entity ID |
| 174 | """ |
| 175 | store = create_feature_store(ctx) |
| 176 | if not dataframe and not start_date and not end_date: |
| 177 | click.echo( |
| 178 | "Either --dataframe or --start-date and/or --end-date must be provided." |
| 179 | ) |
| 180 | return |
| 181 | |
| 182 | if dataframe and (start_date or end_date): |
| 183 | click.echo("Cannot specify both --dataframe and --start-date/--end-date.") |
| 184 | return |
| 185 | |
| 186 | entity_df = None |
| 187 | if dataframe: |
| 188 | try: |
| 189 | entity_list = json.loads(dataframe) |
| 190 | if not isinstance(entity_list, list): |
| 191 | raise ValueError("Entities must be a list of dictionaries.") |
| 192 | |
| 193 | entity_df = pd.DataFrame(entity_list) |
| 194 | entity_df["event_timestamp"] = pd.to_datetime(entity_df["event_timestamp"]) |
| 195 | |
| 196 | except Exception as e: |
| 197 | click.echo(f"Error parsing entities JSON: {e}", err=True) |
| 198 | return |
| 199 | |
| 200 | feature_vector = store.get_historical_features( |
| 201 | entity_df=entity_df, |
| 202 | features=list(features), |
| 203 | start_date=datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") |
| 204 | if start_date |
| 205 | else None, |
| 206 | end_date=datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S") if end_date else None, |
| 207 | ).to_df() |
| 208 | |
| 209 | click.echo(feature_vector.to_json(orient="records", indent=4)) |
nothing calls this directly
no test coverage detected