如何把 machine-learning-for-trading 的特征表接入 Feast 特征存储

如何把 machine-learning-for-trading 的特征表接入 Feast 特征存储

【免费下载链接】machine-learning-for-tradingCode for Machine Learning for Trading, 3rd edition — from data sourcing to live execution.项目地址: https://gitcode.com/GitHub_Trending/ma/machine-learning-for-trading

在 machine-learning-for-trading 仓库中，第 26 章的配套 notebook 05b_feast_live 完成的任务是：把us_equities_panel案例研究产出的两张特征 Parquet 表（features/financial.parquet与features/model_based.parquet）声明成 Feast 的 feature view，跑一次训练用的 point-in-time 离线 join 和一次 live 系统的 as-of 检索，再逐特征核对 Feast 的答案与 05_feast_feature_store 中手写 Polars join 是否一致。核对是这套流程的重心：feature store 是配置出来的而不是写出来的，timestamp 字段、entity key 或 TTL 配错会产生一个"能运行、能回答、但答错"的 store。适用环境是仓库文档支持的ml4tDocker 镜像或本地uv环境；notebook 声明的前置依赖是feast>=0.40，而仓库 pyproject.toml 实际把feast>=0.61同时列在主依赖和[mlops]extra 里，uv sync后直接可用。

准备条件：环境与案例研究产物

环境

环境搭建（Docker 或本地 uv）完全按 docs/installation.md 走，两条路径都能覆盖本章全部 notebook。装好后用同一个脚本确认依赖可用：

# 本地 uv 路径 uv run python scripts/verify_installation.py # Docker 路径 docker compose run --rm ml4t python scripts/verify_installation.py

输出逐组件 PASS/FAIL，全为 PASS 再继续。

特征表与标签

05b 读取的三张表都不随仓库自带，需要先用 us_equities_panel 案例研究的前四个 stage 生成，从仓库根目录按顺序执行：

uv run python case_studies/us_equities_panel/02_labels.py # 产出 labels/fwd_ret_1d.parquet 等 uv run python case_studies/us_equities_panel/03_financial_features.py # 产出 features/financial.parquet uv run python case_studies/us_equities_panel/04_model_based_features.py # 产出 features/model_based.parquet

各 stage 会检查自己需要的上游产物，缺什么会明确提示先跑哪个 stage。两点注意：

• 这批 stage 读取 NASDAQ Data Link 的美国股票日线数据，数据集缺失时会以DataNotFoundError停下并给出该数据集的下载命令；US equities 数据集需要在.env里配置 NASDAQ Data Link 的 API key，细节见 docs/running-notebooks.md 的 Data Requirements 一节。
• docs/running-notebooks.md 中 1.6 GB 的us_equities_panel产物下载只覆盖run_log/（registry、预测、回测），不含features/和labels/，所以下载产物不能替代上面三个 stage。

配置 Feast 仓库与 Feast 兼容的源文件

05b 把整个 Feast 仓库（registry、online store、源文件副本）都放在tempfile.mkdtemp(prefix="feast_ml4t_")创建的临时目录里，保证不写入案例研究的任何产物：

import tempfile from pathlib import Path feast_tmp = tempfile.mkdtemp(prefix="feast_ml4t_") feast_data_dir = Path(feast_tmp) / "data" feast_data_dir.mkdir()

feature_store.yaml

Feast 仓库需要一个声明项目名、provider、registry 位置和 store 后端的配置文件，notebook 用本地 file provider，registry 和 online store 都指向 SQLite 文件：

import yaml 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))

写出带 datetime 时间戳的源副本

案例研究的 Parquet 文件里timestamp是Date类型，而 Feast 的 offline store 做 point-in-time join 需要 datetime，所以 notebook 在临时目录写两份副本，把该列转成pl.Datetime("ns")，并且只在训练窗口两侧各放宽SOURCE_PAD_DAYS = 30天的范围内取数，让 TTL 回看和 as-of 查询都有行可够到：

import polars as pl financial_src = CASE_DIR / "features" / "financial.parquet" financial_feast_path = feast_data_dir / "financial.parquet" ( 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) )

model_based.parquet的副本走同样的转换。filter_start/filter_end由派生出的训练窗口 ± 30 天得到。训练窗口本身不手写日期：notebook 从case_studies/us_equities_panel/config/setup.yaml的 holdout 边界和 walk-forward split 派生——训练窗口取最后一个 validation 窗口的尾部、回看 91 天（TRAINING_LOOKBACK_DAYS = 91），as-of 日期取 holdout 内第一个 session；也可以在 parameters cell 里把TRAINING_START/TRAINING_END/AS_OF_DATE设成日期字符串手动固定。派生之后有一条断言守住治理边界：训练窗口结束日期必须早于 holdout 开始日期，否则抛错拒绝构造训练集。

声明 entity、source 与 feature view

四个类承载全部声明：Entity命名行描述的对象及识别列，FileSource指向表并指明哪一列是 event timestamp，FeatureView把 source 绑定到带类型的特征列表和 TTL，FeatureStore是注册和查询的入口：

from datetime import timedelta 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 FINANCIAL_FEATURES = ["past_ret_21d", "vol_21d", "rsi_14", "sharpe_21d"] MODEL_FEATURES = ["garch_cond_vol", "ffd_log_price", "ffd_log_volume"] FEATURE_TTL_DAYS = 2 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(), ) 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_features view 同形：name="model_features"，source 指向 model_feast_path， # schema 由 MODEL_FEATURES 三个 Float64 字段构成

特征类型必须在这里声明，因为 Feast 在注册时校验 schema，而不是查询时。TTL 是 2 天：对日线特征表，这意味着上一个 session 的值仍可服务——查询一个本身没有行的日期时，Feast 会把上一行带过来回答。这个行为是 store 的本意，也是下面 parity 检查变得非平凡的原因（手写的精确 key join 对这种日期返回空）。

Apply 并执行 point-in-time 训练 join

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()

apply()把 entity 和两个 view 注册进 registry。之后的查询由store发出。执行成功的直接信号（文档示例输出）：

Registered 1 entities, 2 feature views

训练 join 的输入是一个(symbol, event_timestamp)事件表，取自标签文件在训练窗口内的行。这里有一个必须先做的限制：把事件集收缩到两张特征源都有精确行的 key 上（notebook 用 labels 与 model 特征、financial 特征做三次 join 求交，并打印被丢弃事件的计数，其中区分"缺 model vintage"和"缺 financial 行"两种原因）。原因是 Feast 在 TTL 内会把上一行向前带，而 05 的精确 key join 不会——在只有单侧有行的日期上比较，分歧是问题本身的性质，不是任何一方的实现错误。事件集为空时 notebook 直接抛ValueError，因为后续检查对空集会把一切差异都读成"匹配"。

import pandas as pd 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, # 两列：symbol 和 event_timestamp（ns 精度 datetime） 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)

feature 引用的写法是"<view 名>:<特征名>"。entity_df的event_timestamp列是 Feast 约定的时间戳列名。join 结果的行数由训练窗口与受限后的事件集决定，不是固定值。

验证：与手写 join 逐特征对账

这一步把同一事件集分别用 Feast 的 point-in-time join 和 05 的精确 key join 回答一遍，逐特征比较：

PARITY_TOLERANCE = 1e-10 mismatches = [] for col in ALL_FEATURES: diff = (merged[f"{col}_polars"] - merged[f"{col}_feast"]).abs() mismatches.append( {"feature": col, "max_abs_diff": float(diff.max()), "match": float(diff.max()) < PARITY_TOLERANCE} ) match_df = pd.DataFrame(mismatches) assert int(match_df["match"].sum()) == len(match_df), f"Feast parity failed for {mismatch_cols}"

成功条件（文档示例输出）：

Parity: 7/7 features match within 1e-10.

任何一项不匹配，notebook 在该断言处停下，并打印出分歧的列名。这个检查找的是"选错了行"（错的时间戳字段、不能唯一标识行的 entity key、按 id 而非按日期选 fold vintage），不是舍入误差——tolerance 刻意设得远低于任何正确 join 可能产生的差异。要注意它能证明什么、不能证明什么：两边一致说明两条路径为每个决策日期做了同样的fold vintage 选择，不说明该选择本身正确；另外它抓不到"TTL 设得过长"，因为比较的所有事件在自己日期上都有行，TTL 改成一年也不会移动任何值——抓那种错误需要没有自身行的事件。

As-of 检索与 registry 回读

live 系统本来调get_online_features，读由 materialization job 维护的 key-value store；05b 没有那套基础设施，所以用get_historical_features加"每个 entity 一个时间戳"复现同样的请求形状（8 个按成交额抽样的 symbol、as-of 日期），从 offline store 回答：

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()

复现的是请求形状：少数几个名字、一个时刻、每个名字一行——模型服务每次决策发出的就是这种查询。它不复现延迟和 online/offline 两层的一致性，那是 live feature store 自己的故障面，本 notebook 不覆盖。

最后用 registry 回读回答"这个被服务的值从哪来"，这是 05 手工汇编 lineage 表的 library 版本：

for fv in store.list_feature_views(): print(fv.name, fv.entity_columns, len(fv.features), fv.ttl, type(fv.batch_source).__name__)

清理

notebook 的最后一格删除本次创建的临时 Feast 仓库：

import shutil shutil.rmtree(feast_tmp, ignore_errors=True)

副作用仅限tempfile.mkdtemp创建的那个临时目录（本例的 registry.db、online.db 和源副本），不触碰案例研究产物。生产环境的 registry 和 online store 会跨 session、跨服务持久存在并被多方读取——这正是"共享基础设施"与本文临时仓库的主要区别。

已知限制

• 这里没有真正的 online store：online 路径由get_historical_features模拟，materialization job 与两层一致性都没被验证。
• parity 检查覆盖的是这一个案例研究的一个训练窗口；它证明两条实现在这批事件上一致，不证明 Feast 声明在别的 fold 几何下也正确。
• registry、online store 和源副本都在被删除的临时目录里；需要持久、可共享的 registry 时，要自己把feature_store.yaml的路径换成持久位置。

下一步

• 想看 Feast 自动化了什么、先手工做一遍同样的检索规则，读 05_feast_feature_store（它的FEATURE_TTL_DAYS是 1，并用"多服务一个 session"量化了时间戳规则错一位对特征向量的影响）。
• 实验追踪接着看同章的 06_mlflow_experiments。
• 用 Papermill 以缩减参数跑通整个章节：uv run pytest tests/test_chapter_notebooks.py -v -k "26_mlops_governance"；按 26_mlops_governance/README.md 的说明，执行本章 notebook 时不要设置MPLBACKEND=Agg或PLOTLY_RENDERER=json，否则会压掉执行版 notebook 应携带的图。

【免费下载链接】machine-learning-for-tradingCode for Machine Learning for Trading, 3rd edition — from data sourcing to live execution.项目地址: https://gitcode.com/GitHub_Trending/ma/machine-learning-for-trading