验证 Feast 特征存储检索与时点连接的一致性

# Feature Store with Feast: Live Integration


# Feature Store with Feast: Live Integration

**Chapter 26: MLOps and Governance**
**Docker image**: `ml4t`
**Book Reference**: Chapter 26, Section 26.6
**Companion to**: [`05_feast_feature_store`](05_feast_feature_store.ipynb), which does the
same work in Polars

**Learning Objectives**:
- Stand up a local Feast repository from Parquet files already on disk, declaring the
  entity, the sources and the feature views in code.
- Ask Feast for a training set by handing it the entity-timestamp pairs it should answer
  for, and ask it for a single moment the way a live system would.
- Check Feast's answers against the hand-written join from `05`, feature by feature, and
  read a disagreement as a configuration defect rather than a rounding difference.

`05` builds the offline join, the as-of retrieval and the skew measurement by hand, which
is what makes the rules visible. This notebook hands the same job to **Feast**, an
open-source feature store, and then checks that the two agree.

The check is the point. A feature store is configured, not written: the timestamp field,
the entity key and the time to live are declarations, and a wrong one produces a store that
runs, answers every query, and answers them wrong. Comparing against an implementation whose
rules you can read is how you find that out.

**Prerequisites**: `pip install 'feast>=0.40'` (included in the `[mlops]` extra).

```python
"""Feature Store with Feast: the same retrievals as notebook 05, checked against it."""
```

## Settings

`TRAINING_START`, `TRAINING_END` and `AS_OF_DATE` bound the two retrievals. Left unset they
are derived from the case study's fold geometry, as in `05` and for the same reason: a date
typed here is a claim about where a fold boundary falls, and rebuilding the case study moves
every boundary while the literal does not.

`FEATURE_TTL_DAYS` is the time to live declared on both Feast views, and it is worth reading
carefully. Two days on a daily feature table means a value from the previous session is
still servable, so Feast answers a query for a date with no row of its own by carrying the
previous row forward. That behaviour is what a store is for and it is also what makes the
parity check below non-trivial: the hand-written join matches on exact keys and returns
nothing for such a date, so the two paths only agree where the event set is restricted to
keys both sources actually hold.

`SOURCE_PAD_DAYS` widens the Parquet copies handed to Feast on both sides of the training
window, so the time-to-live lookback and the as-of query have rows to reach.

`PARITY_TOLERANCE` is how far apart the two implementations may be per feature. It is set
well below anything a correct join could produce, because the difference being looked for is
a wrong row, not a rounding error: a store that picked a different date is off by whatever
the feature moved that day, which is many orders of magnitude larger than this.

`N_SAMPLE_ASSETS`, `LIQUIDITY_WINDOW_DAYS` and `LIQUIDITY_RANK_DAYS` pick the names the as-of
retrieval asks about, by average dollar volume over the tail of the training window.

```python
CASE_STUDY_ID = "us_equities_panel"
PRIMARY_LABEL = "fwd_ret_1d"
TRAINING_START = None
TRAINING_END = None
AS_OF_DATE = None
TRAINING_LOOKBACK_DAYS = 91
N_SAMPLE_ASSETS = 8
FEATURE_TTL_DAYS = 2
SOURCE_PAD_DAYS = 30
LIQUIDITY_WINDOW_DAYS = 21
LIQUIDITY_RANK_DAYS = 30
PARITY_TOLERANCE = 1e-10
```

```python
import shutil
import tempfile
import warnings
from datetime import timedelta
from pathlib import Path

import numpy as np
import pandas as pd
import polars as pl
import yaml

from utils.cv_splits import generate_cv_splits
from utils.paths import get_case_study_dir

# Named, not blanket: a bare ignore would also hide the convergence and numerical
# warnings a reader needs to see.
warnings.filterwarnings("ignore", category=FutureWarning, module="feast")

CASE_DIR = get_case_study_dir(CASE_STUDY_ID)

FINANCIAL_FEATURES = ["past_ret_21d", "vol_21d", "rsi_14", "sharpe_21d"]
MODEL_FEATURES = ["garch_cond_vol", "ffd_log_price", "ffd_log_volume"]
ALL_FEATURES = FINANCIAL_FEATURES + MODEL_FEATURES

print("Feature Store with Feast: Live Integration")
print("=" * 60)
```

## 1. Import Feast

Four classes carry the declaration. `Entity` names the thing rows are about and the column
that identifies it. `FileSource` points at a table and says which of its columns is the
event timestamp. `FeatureView` binds a source to a typed list of features and a time to
live. `FeatureStore` is the repository those declarations are registered into and the object
every query goes through.

```python
from feast import Entity, FeatureStore, FeatureView, Field, FileSource
from feast.data_format import ParquetFormat
from feast.types import Float64
from feast.value_type import ValueType

print("Feast imported successfully")
```

## 2. Prepare Feast-compatible source files

Feast's offline store needs a datetime timestamp for its point-in-time joins, and these
Parquet files carry a `Date`. Temporary copies are written with the column cast, and only
over the padded window, so nothing here writes into the case study's own artifacts. In
production the cast happens in the job that materializes the source, once, rather than
beside every query.

```python
feast_tmp = tempfile.mkdtemp(prefix="feast_ml4t_")
feast_data_dir = Path(feast_tmp) / "data"
feast_data_dir.mkdir()
```

```python
def folds_by_coverage(path, windows: list[tuple]) -> list[object]:
    """Pair each evaluation window with the artifact fold that covers it, by date.

    A fold id is a label the writer chose, not a property of the data, and two tools can
    number the same folds in opposite directions. Joining a stored id to a freshly
    generated one then pairs each date with the wrong fitted state and still succeeds,
    because both ids exist. Ordering both sides by the dates they cover is the pairing
    that holds whatever numbering wrote the file.
    """
    coverage = (
        pl.scan_parquet(path)
        .group_by("fold")
        .agg(pl.max("timestamp").alias("last_covered"))
        .collect()
        .sort("last_covered")
    )
    stored = coverage["fold"].to_list()
    if len(stored) != len(windows):
        raise ValueError(
            f"model_based.parquet carries {len(stored)} folds and this notebook derived "
            f"{len(windows)} evaluation windows. They cannot be paired by date, so the "
            "artifact does not describe the geometry this case study now declares; "
            "regenerate it against the current stage 04."
        )
    order = sorted(range(len(windows)), key=lambda i: windows[i])
    paired: list[object] = [None] * len(windows)
    for position, window_index in enumerate(order):
        paired[window_index] = stored[position]
    return paired


financial_src = CASE_DIR / "features" / "financial.parquet"
model_src = CASE_DIR / "features" / "model_based.parquet"

financial_feast_path = feast_data_dir / "financial.parquet"
model_feast_path = feast_data_dir / "model_based.parquet"

setup = yaml.safe_load((CASE_DIR / "config" / "setup.yaml").read_text())
holdout_start = pd.Timestamp(setup["evaluation"]["holdout_start"])
holdout_end = pd.Timestamp(setup["evaluation"]["holdout_end"])
timeline = pl.scan_parquet(financial_src).select("timestamp").unique().collect()
cv_splits = generate_cv_splits(timeline, case_study_id=CASE_STUDY_ID, label_buffer="1D")
# A feature value is servable for a session the model was evaluated on, so the date range
# handed to Feast is the union of the validation windows and the holdout.
validation_spans = [
    (pd.Timestamp(split["val_start"]).date(), pd.Timestamp(split["val_end"]).date())
    for split in cv_splits
]
model_windows = [*validation_spans, (holdout_start.date(), holdout_end.date())]
# Where the table carries a fold column each stock-date appears once per fold, so the
# window and the fold are applied together; where it does not, the date range selects it.
FOLD_KEYED_ARTIFACT = "fold" in pl.scan_parquet(model_src).collect_schema().names()
model_folds = (
    folds_by_coverage(model_src, model_windows)
    if FOLD_KEYED_ARTIFACT
    else [None] * len(model_windows)
)
# By date rather than by list position: which end of the list holds the latest window
# depends on the numbering convention, and the dates do not.
last_validation_span = max(validation_spans)

training_end = TRAINING_END or str(last_validation_span[1])
training_start = TRAINING_START or str(
    max(
        last_validation_span[0],
        pd.Timestamp(training_end).date() - pd.Timedelta(days=TRAINING_LOOKBACK_DAYS),
    )
)
as_of_date = AS_OF_DATE or str(
    pl.scan_parquet(financial_src)
    .filter(pl.col("timestamp") >= holdout_start.date())
    .select(pl.min("timestamp"))
    .collect()
    .item()
)
assert pd.Timestamp(training_end) < holdout_start, (
    f"training window ends {training_end}, on or after the holdout opens "
    f"{holdout_start.date()}; the offline join would mix data the model may not see"
)
print(f"Offline training window {training_start} to {training_end}; serving as of {as_of_date}")

train_start = pd.Timestamp(training_start).date()
train_end = pd.Timestamp(training_end).date()
# Padded on both sides so the lookback and the as-of query have rows to reach.
filter_start = train_start - timedelta(days=SOURCE_PAD_DAYS)
filter_end = train_end + timedelta(days=SOURCE_PAD_DAYS)

(
    pl.scan_parquet(financial_src)
    .select(["symbol", "timestamp", *FINANCIAL_FEATURES])
    .filter((pl.col("timestamp") >= filter_start) & (pl.col("timestamp") <= filter_end))
    .with_columns(pl.col("timestamp").cast(pl.Datetime("ns")))
    .collect()
    .write_parquet(financial_feast_path)
)
```

```python
def window_terms(clipped: list[tuple]) -> list[pl.Expr]:
    """One predicate per evaluation window, to be OR-ed into a single filter.

    Where the artifact carries no fold column a window is a date range and nothing else.
    Where it does, the same window also names the fold whose fitted state that range is
    evaluated under, and the two are applied together: each fold holds its own value for
    the same stock-date, so a date range alone returns one row per fold with no rule for
    choosing between them.
    """
    if FOLD_KEYED_ARTIFACT:
        return [
            (pl.col("fold") == fold) & pl.col("timestamp").is_between(start, end)
            for fold, start, end in clipped
        ]
    return [pl.col("timestamp").is_between(start, end) for _, start, end in clipped]
```

The model-based copy is built with one filter over the union of the evaluation windows,
rather than one frame per window concatenated: a concat would double-count any date two
windows cover, and would do it silently.

```python
_span_lo = pd.Timestamp(filter_start).date()
_span_hi = pd.Timestamp(filter_end).date()
model_spans = [
    (fold, max(_span_lo, start), min(_span_hi, end))
    for fold, (start, end) in zip(model_folds, model_windows, strict=True)
    if start <= _span_hi and end >= _span_lo
]
if not model_spans:
    raise ValueError(
        f"No validation window and not the holdout covers {_span_lo}..{_span_hi}; "
        "the materialization range lies outside every window this model was evaluated on"
    )
model_features = (
    pl.scan_parquet(model_src)
    .filter(pl.any_horizontal(*window_terms(model_spans)))
    .select(["symbol", "timestamp", *MODEL_FEATURES])
    .collect()
    .sort(["timestamp", "symbol"])
)
assert not model_features.select(pl.struct("symbol", "timestamp").is_duplicated().any()).item(), (
    "a stock-date resolved to more than one model-based row; the evaluation windows "
    "this notebook derived are not disjoint over the materialization range"
)
model_features.with_columns(pl.col("timestamp").cast(pl.Datetime("ns"))).write_parquet(
    model_feast_path
)

fin_rows = pl.scan_parquet(financial_feast_path).select(pl.len()).collect().item()
mod_rows = pl.scan_parquet(model_feast_path).select(pl.len()).collect().item()
print(f"Financial features: {fin_rows:,} rows → {financial_feast_path.name}")
print(f"Model-based features: {mod_rows:,} rows → {model_feast_path.name}")
```

## 3. Create the Feast repository

A Feast repo needs a `feature_store.yaml` that declares the project name,
provider, registry location, and store backends. We use the local file
provider with SQLite for both the registry and online store.

```python
feast_config = {
    "project": "ml4t_feature_store",
    "provider": "local",
    "registry": {"path": str(Path(feast_tmp) / "registry.db")},
    "online_store": {"type": "sqlite", "path": str(Path(feast_tmp) / "online.db")},
    "offline_store": {"type": "file"},
    "entity_key_serialization_version": 3,
}
config_path = Path(feast_tmp) / "feature_store.yaml"
config_path.write_text(yaml.dump(feast_config))
print(f"Feast repo: {feast_tmp}")
print(yaml.dump(feast_config, default_flow_style=False))
```

## 4. Define entity and feature views

The entity is `symbol`, which is the identifier every case-study artifact in this repository
uses. Each view maps to one Parquet source and declares its features with types, because
Feast validates the schema at registration rather than at query time.

The time to live is what bounds how far back a query may reach. At `FEATURE_TTL_DAYS` a
request for a date with no row of its own is answered from the most recent row within that
many days, and a request older than that returns nothing rather than a stale value. Set it
too short and legitimate queries come back empty; too long and the store answers with data
that no longer describes the moment asked about.

```python
symbol_entity = Entity(
    name="symbol",
    join_keys=["symbol"],
    value_type=ValueType.STRING,
    description="Stock ticker symbol",
)

financial_source = FileSource(
    path=str(financial_feast_path.resolve()),
    timestamp_field="timestamp",
    file_format=ParquetFormat(),
)

model_source = FileSource(
    path=str(model_feast_path.resolve()),
    timestamp_field="timestamp",
    file_format=ParquetFormat(),
)

financial_fv = FeatureView(
    name="financial_features",
    entities=[symbol_entity],
    schema=[Field(name=col, dtype=Float64) for col in FINANCIAL_FEATURES],
    source=financial_source,
    ttl=timedelta(days=FEATURE_TTL_DAYS),
)

model_fv = FeatureView(
    name="model_features",
    entities=[symbol_entity],
    schema=[Field(name=col, dtype=Float64) for col in MODEL_FEATURES],
    source=model_source,
    ttl=timedelta(days=FEATURE_TTL_DAYS),
)

print("Defined:")
print(f"  Entity:  {symbol_entity.name} (join_key={symbol_entity.join_key})")
print(f"  View 1:  {financial_fv.name} ({len(FINANCIAL_FEATURES)} features)")
print(f"  View 2:  {model_fv.name} ({len(MODEL_FEATURES)} features)")
```

## 5. Apply the feature store

`store.apply()` registers the entity and feature views in the Feast registry.
After this call, the feature store knows where to find each feature, what
entity key to join on, and what timestamp field governs point-in-time
correctness.

```python
store = FeatureStore(repo_path=feast_tmp)
store.apply([symbol_entity, financial_fv, model_fv])

registered_views = store.list_feature_views()
registered_entities = store.list_entities()
print(f"Registered {len(registered_entities)} entities, {len(registered_views)} feature views")
for fv in registered_views:
    print(f"  {fv.name}: {[f.name for f in fv.features]}")
```

## 6. Offline retrieval: the point-in-time join

The query Feast is built around. It is handed a frame of `(symbol, event_timestamp)` pairs -
the moments a training set needs answered - and returns, for each pair, the most recent
feature values at or before that moment, within the time to live. The features here are
known after the session close, the label is the next session's return, and a position acts
no earlier than that next session, so the pairing is a next-bar decision.

### Restricting the event set, and why the parity check needs it

The two implementations answer differently for a date that has no feature row of its own.
Feast carries the previous row forward, within its time to live; the exact-key join in `05`
returns nothing. Both are right for what they are, so comparing them on such a date would
report a disagreement that is a property of the question rather than of either
implementation.

Dates like that exist wherever the training window spans an embargo gap between consecutive
validation windows, or runs past the last one. Whether this particular window contains any
is a fact about the fold geometry, and the counts printed below say so: the events dropped
are split by which source was missing, because a key absent from the model features is the
gap this restriction exists for, while one absent from the financial features is a
different problem arriving in the same total.

```python
requested = (
    pl.scan_parquet(CASE_DIR / "labels" / f"{PRIMARY_LABEL}.parquet")
    .filter(
        (pl.col("timestamp") >= pl.lit(pd.Timestamp(training_start).date()))
        & (pl.col("timestamp") <= pl.lit(pd.Timestamp(training_end).date()))
    )
    .select("symbol", "timestamp", pl.col(PRIMARY_LABEL).alias("label"))
    .collect()
)
financial_keys = (
    pl.scan_parquet(financial_feast_path)
    .select(pl.col("symbol"), pl.col("timestamp").cast(pl.Date))
    .unique()
    .collect()
)
labels = requested.join(
    model_features.select("symbol", "timestamp"), on=["symbol", "timestamp"]
).join(financial_keys, on=["symbol", "timestamp"])
if labels.is_empty():
    raise ValueError(
        f"No event in {training_start}..{training_end} has a row in both feature sources. "
        "Both join paths would return nothing, and the parity check below reads an "
        "empty difference as a match, so it would report every feature matching."
    )
if labels.height < requested.height:
    dropped = requested.join(
        labels.select("symbol", "timestamp"), on=["symbol", "timestamp"], how="anti"
    )
    no_model = dropped.join(
        model_features.select("symbol", "timestamp"), on=["symbol", "timestamp"], how="anti"
    ).height
    no_financial = dropped.join(financial_keys, on=["symbol", "timestamp"], how="anti").height
    # The two counts overlap where an event is missing from both sources.
    both = no_model + no_financial - dropped.height
    print(
        f"Dropped {dropped.height:,} of {requested.height:,} events: "
        f"{no_model:,} with no fitted model-feature vintage, "
        f"{no_financial:,} with no financial-feature row, "
        f"{both:,} missing from both."
    )

entity_df = labels.select(
    "symbol",
    pl.col("timestamp").cast(pl.Datetime("ns")).alias("event_timestamp"),
).to_pandas()

print(f"Entity DataFrame: {len(entity_df):,} rows")
print(f"Date range: {entity_df['event_timestamp'].min()} → {entity_df['event_timestamp'].max()}")
print(f"Symbols: {entity_df['symbol'].nunique()}")
```

```python
feature_refs = [f"financial_features:{col}" for col in FINANCIAL_FEATURES] + [
    f"model_features:{col}" for col in MODEL_FEATURES
]

feast_training = store.get_historical_features(
    entity_df=entity_df,
    features=feature_refs,
).to_df()

feast_training = feast_training.dropna(subset=ALL_FEATURES)
feast_training = feast_training.sort_values(["event_timestamp", "symbol"]).reset_index(drop=True)

print(f"Feast offline join: {len(feast_training):,} rows × {len(feast_training.columns)} columns")
feast_training.head()
```

## 7. Check Feast against the hand-written join

The same event set, answered twice: once by Feast's point-in-time join and once by the exact
key join `05` builds. Both should return the same rows with the same values.

A disagreement here is a configuration defect, not a numerical one. The two paths read the
same Parquet files, so a difference means one of them selected a different row: a wrong
timestamp field, an entity key that does not identify a row uniquely, or a fold vintage
picked by id rather than by date. Each of those produces a store that runs and answers
wrongly, which is why the comparison is per feature and the tolerance is set far below
anything a correct join could produce.

One misconfiguration this check cannot reach is a time to live set too long. Every event
compared here has a feature row on its own date, so both paths take that row and no lookback
happens; the time to live could be a year without moving a single value. Catching that needs
events with no row of their own, at ages either side of the declared limit, which is a
different test from this one.

```python
def polars_offline_join() -> pl.DataFrame:
    """Reproduce the manual offline join from notebook 05."""
    events = labels.select("symbol", "timestamp")

    financial = (
        pl.scan_parquet(financial_feast_path)
        .with_columns(pl.col("timestamp").cast(pl.Date))
        .select(["symbol", "timestamp", *FINANCIAL_FEATURES])
        .join(events.lazy(), on=["symbol", "timestamp"], how="inner")
    )
    model_based = (
        pl.scan_parquet(model_feast_path)
        .with_columns(pl.col("timestamp").cast(pl.Date))
        .select(["symbol", "timestamp", *MODEL_FEATURES])
        .join(events.lazy(), on=["symbol", "timestamp"], how="inner")
    )
    return (
        financial.join(model_based, on=["symbol", "timestamp"], how="inner")
        .drop_nulls(ALL_FEATURES)
        .collect()
        .sort(["timestamp", "symbol"])
    )


polars_result = polars_offline_join()
print(f"Polars offline join: {polars_result.height:,} rows")
print(f"Feast offline join:  {len(feast_training):,} rows")
print(f"Row count match:     {polars_result.height == len(feast_training)}")
```

```python
# Compare the complete one-row-per-event results.
polars_pd = polars_result.to_pandas().rename(columns={"timestamp": "event_timestamp"})
# Align timestamp dtype/tz with Feast (ns + UTC) for the join key.
polars_pd["event_timestamp"] = pd.to_datetime(polars_pd["event_timestamp"], utc=True)
feast_training["event_timestamp"] = pd.to_datetime(feast_training["event_timestamp"], utc=True)

merged = polars_pd.merge(
    feast_training,
    on=["symbol", "event_timestamp"],
    how="inner",
    suffixes=("_polars", "_feast"),
)
assert len(merged) > 0, "Nothing to compare - a parity check over zero rows proves nothing."
assert len(polars_pd) == len(feast_training) == len(merged)

mismatches = []
for col in ALL_FEATURES:
    diff = (merged[f"{col}_polars"] - merged[f"{col}_feast"]).abs()
    max_diff = float(diff.max()) if len(diff) else 0.0
    mismatches.append(
        {"feature": col, "max_abs_diff": max_diff, "match": max_diff < PARITY_TOLERANCE}
    )

match_df = pd.DataFrame(mismatches)
match_df
```

```python
n_match = int(match_df["match"].sum())
n_total = len(match_df)
mismatch_cols = match_df.loc[~match_df["match"], "feature"].tolist()
print(f"Parity: {n_match}/{n_total} features match within {PARITY_TOLERANCE:g}.")
if mismatch_cols:
    print(f"Differing columns: {mismatch_cols}")
assert n_match == n_total, f"Feast parity failed for {mismatch_cols}"
```

The assertion above is what makes this a check rather than a display: the notebook stops if
any feature disagrees, so a reader who sees the cells below knows the two paths agreed.

What the agreement establishes is narrower than it looks. The source holds one fitted
model-feature vintage per walk-forward fold, and both paths select the vintage valid at each
decision date before comparing. That selection is the part that had to be got right in both;
without it, one stock-date resolves to several rows and the join fans out, or resolves to a
vintage fitted on later data. Agreement says the two implementations made the same
selection. It does not say the selection is the correct one - that is what the fold-by-date
pairing above is for.

## 8. Online-style as-of retrieval

A live system calls `get_online_features`, which reads a key-value store that a
materialization job keeps current. Standing that up needs infrastructure this notebook does
not have, so the request shape is reproduced with `get_historical_features` and one
timestamp per entity: the same query, answered from the offline store.

The shape is what matters here - a handful of names, one moment, one row each - because it
is the shape a model server issues per decision. What it does not reproduce is the latency
or the consistency question, which is whether the online store holds what the offline store
would have answered.

```python
def sample_assets(n_assets: int) -> list[str]:
    from data import load_us_equities

    rank_from = pd.Timestamp(training_end).date() - pd.Timedelta(days=LIQUIDITY_RANK_DAYS)
    prices = load_us_equities(start_date=training_start, end_date=training_end)
    return (
        prices.lazy()
        .sort("symbol", "timestamp")
        .with_columns((pl.col("adj_close") * pl.col("adj_volume")).alias("dollar_volume"))
        .with_columns(
            pl.col("dollar_volume")
            .rolling_mean(LIQUIDITY_WINDOW_DAYS)
            .over("symbol")
            .alias("avg_dollar_volume")
        )
        .filter(
            (pl.col("timestamp") >= rank_from)
            & (pl.col("timestamp") <= pd.Timestamp(training_end).date())
        )
        .group_by("symbol")
        .agg(pl.col("avg_dollar_volume").mean().alias("mean_dollar_volume"))
        .sort("mean_dollar_volume", descending=True)
        .head(n_assets)
        .collect()
        .get_column("symbol")
        .to_list()
    )


sampled = sample_assets(N_SAMPLE_ASSETS)
as_of_entity_df = pd.DataFrame(
    {
        "symbol": sampled,
        "event_timestamp": pd.Timestamp(as_of_date),
    }
)

online_style = store.get_historical_features(
    entity_df=as_of_entity_df,
    features=feature_refs,
).to_df()

print(f"Online-style snapshot for {len(sampled)} assets as of {as_of_date}:")
online_style.sort_values("symbol")
```

## 9. Feature registry introspection

The registry is Feast's own record of what has been declared, and reading it back is how an
operator answers "where does this served value come from" without opening the code that
declared it. It is the library's version of the lineage table `05` assembles by hand.

```python
registry_rows = []
for fv in store.list_feature_views():
    registry_rows.append(
        {
            "feature_view": fv.name,
            "entity": ", ".join(field.name for field in fv.entity_columns),
            "features": len(fv.features),
            "ttl": str(fv.ttl),
            "source_type": type(fv.batch_source).__name__,
        }
    )

registry_df = pd.DataFrame(registry_rows)
print("Feast Feature Registry:")
registry_df
```

## 10. Clean up

The repository was written under a temporary directory, so removing it leaves nothing
behind. A production registry and online store persist across sessions and across the
services that read them, which is most of what makes them shared infrastructure.

```python
shutil.rmtree(feast_tmp, ignore_errors=True)
print(f"Cleaned up temporary Feast repo: {feast_tmp}")
```

## Key Takeaways

1. A feature store is configured rather than written. The entity, the timestamp field and
   the time to live are declarations, and a wrong one produces a store that runs, answers
   every query and answers them wrongly. That is why the check below the retrieval exists.
2. Check a store against an implementation whose rules you can read, feature by feature, and
   set the tolerance far below anything a correct join could produce. The failure being
   looked for is a wrong row, which is off by whatever the feature moved that day, not a
   rounding difference.
3. Compare the two on questions they answer the same way. A store carries the last value
   forward within its time to live and an exact-key join does not, so a date with no row of
   its own makes them disagree by construction; restrict the event set first, and print what
   was dropped and why.
4. Reading agreement correctly is as important as getting it. Agreement says both paths
   selected the same fitted vintage for each decision date. It does not say the selection
   was the right one, which is a separate argument the fold-by-date pairing makes.
5. The library buys shared infrastructure, not correctness. Feast supplies a registry, an
   online store and multi-team isolation, and needs them stood up and their timestamps typed
   for it; `05` needs none of that and does the same retrievals on the Parquet files
   directly. Choose on which of those you need.

**Known limitations**

- The online retrieval here is `get_historical_features` with one timestamp per entity. It
  reproduces the request shape and not the online store, so nothing here exercises the
  materialization job or the consistency between the two tiers, which is where a live
  feature store's own failures are.
- The registry, the online store and the source copies all live in a temporary directory
  that this notebook deletes. A production registry persists and is shared, and most of what
  makes it governance is that other services read the same one.
- The parity check covers one training window on one case study. It establishes that the two
  implementations agree on those events, not that the Feast declaration is correct for a
  window with a different fold geometry.

**Next**: See `06_mlflow_experiments` for experiment tracking with MLflow.