Vibe-Trading ML 策略技能实战：基于 scikit-learn 的 Walk-Forward 机器学习预测策略

Vibe-Trading ML 策略技能实战：基于 scikit-learn 的 Walk-Forward 机器学习预测策略

【免费下载链接】Vibe-Trading"Vibe-Trading: Your Personal Trading Agent"项目地址: https://gitcode.com/GitHub_Trending/vi/Vibe-Trading

Vibe-Trading 将机器学习策略封装为ml-strategy技能（SKILL.md），为 Agent 提供一套基于 scikit-learn 的可直接复制运行的完整策略管线：从 OHLCV 数据校验、多因子特征工程，到未来 N 日收益方向标签构建、walk-forward 滚动训练与信号生成。读完本文，你将掌握这套防数据泄漏（data leakage）、带输出契约（no NaN / 数值裁剪）的SignalEngine的每一行实现，理解特征、模型、参数与信号约定的设计意图，并能在任何 OHLCV 数据集上落地自己的机器学习预测策略。

技能定位：Vibe-Trading 策略类技能之一

Vibe-Trading 的内置技能库按类别组织，ml-strategy属于Strategy（策略）类别，与strategy-generate、cross-market-strategy、technical-basic、candlestick、ichimoku、elliott-wave、smc、multi-factor等同列（见 README_zh.md）。该类别共 19 个技能，ml-strategy的职责非常聚焦：用机器学习模型预测未来收益方向并生成交易信号，且明确声明"适用于任何 OHLCV 数据"（Suitable for any OHLCV data）。

技能文件采用统一的 frontmatter 元数据格式（name/description/category），由技能加载器读取。从 技能加载器实现 可以看到，Skill数据类解析 SKILL.md 的 frontmatter 得到名称、描述与类别，正文则通过load_skill工具按需注入 Agent 上下文。这种"渐进式披露"设计意味着：ml-strategy的完整管线代码平时不会全部塞进系统提示，而是在 Agent 需要时被精确加载。

ml-strategy还深度接入 Vibe-Trading 的 Swarm 多智能体编排：ml_quant_lab预设（ml_quant_lab.yaml）中，Feature Engineer 与 Data Scientist 两个角色的skills字段都声明了ml-strategy，并通过load_skill("ml-strategy")获取特征工程最佳实践与金融 ML 设计标准，随后由 Backtest Engineer 对产物做严格样本外（OOS）验证。这说明该技能不仅面向单个 Agent，也是多智能体量化实验室的标准参考手册。

信号逻辑：五步流水线

ml-strategy的信号生成遵循严格有序的五步流程，每一步都在为"避免未来函数泄漏、保证输出可被下游直接消费"服务：

1. 校验输入（Validate input）：检查 OHLCV 列是否齐全、最小行数是否达标、NaN 占比是否过高——不合格的标的直接跳过，绝不进入训练流程；
2. 特征工程（Feature engineering）：从原始 OHLCV 构建动量、波动率、RSI、均线比、量比等多维因子，所有特征统一做消毒处理（inf替换、除零防护）；
3. 标签构建（Label construction）：未来 N 日收益 > 0 记为正类（1），< 0 记为负类（0）；
4. Walk-forward 训练：采用扩展窗口（expanding）或滑动窗口（sliding），只用历史数据训练，逐日向前滚动预测；
5. 信号生成：将predict_proba[:, 1]映射到[-1.0, 1.0]，或使用predict得到{-1, 0, 1}离散信号；输出保证干净（无 NaN、数值已裁剪）。

其中第 3 步与第 4 步的组合是整套管线防泄漏的关键：标签基于未来收益构造，而训练样本只允许取"标签在预测时刻已可观测"的历史切片——这一点在测试中被专门验证（见下文"防泄漏的正确性验证"）。

完整 SignalEngine 示例：推荐的全管线实现

这是该技能推荐的标准全流程实现，复制即可运行，安全性已内建。它是后续特征工程、模型选型、参数调优讨论的基准代码。

import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier from sklearn.linear_model import LogisticRegression from sklearn.preprocessing import StandardScaler def validate_data(df: pd.DataFrame, min_rows: int = 300) -> bool: """Check that OHLCV data meets minimum quality for ML training. Args: df: DataFrame with DatetimeIndex. min_rows: Minimum number of rows required. Returns: True if data is usable. """ required = {"open", "high", "low", "close", "volume"} if not required.issubset(df.columns): return False if len(df) < min_rows: return False if df["close"].isnull().mean() > 0.2: return False return True def build_features(df: pd.DataFrame) -> pd.DataFrame: """Build a machine-learning feature matrix from OHLCV data. All features are guarded against division-by-zero and sanitized (inf replaced with NaN) so downstream code never sees inf values. Args: df: DataFrame containing open, high, low, close, and volume columns. Returns: DataFrame with feature columns prefixed by 'f_'. """ c = df["close"] v = df["volume"] ret = c.pct_change(fill_method=None) features = pd.DataFrame(index=df.index) features["f_ret_5d"] = c.pct_change(5, fill_method=None) features["f_ret_20d"] = c.pct_change(20, fill_method=None) features["f_vol_20d"] = ret.rolling(20).std() features["f_ma_ratio"] = c / c.rolling(20).mean() features["f_volume_ratio"] = v / v.rolling(20).mean() # RSI(14) — guard: loss=0 in zero-volatility periods produces inf delta = c.diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rs = gain / loss.replace(0, np.nan) features["f_rsi_14"] = 100 - (100 / (1 + rs)) # Bollinger Band position — guard: bb_upper == bb_lower when std=0 ma20 = c.rolling(20).mean() std20 = c.rolling(20).std() bb_upper = ma20 + 2 * std20 bb_lower = ma20 - 2 * std20 bb_range = (bb_upper - bb_lower).replace(0, np.nan) features["f_bb_position"] = (c - bb_lower) / bb_range # Intraday features features["f_high_low_ratio"] = (df["high"] - df["low"]) / c features["f_close_open_ratio"] = (c - df["open"]) / df["open"] features["f_skew_20d"] = ret.rolling(20).skew() # Sanitize: replace all inf with NaN (NaN handled by walk-forward) features = features.replace([np.inf, -np.inf], np.nan) return features def walk_forward_predict( features: pd.DataFrame, labels: pd.Series, min_train_size: int = 252, retrain_freq: int = 20, model_type: str = "random_forest", window_type: str = "expanding", sliding_size: int = 504, prediction_horizon: int = 5, ) -> pd.Series: """Walk-forward training and prediction to avoid future data leakage. Args: features: Feature matrix aligned with labels by row index. labels: Binary labels (0/1), representing the direction of future N-day returns. min_train_size: Minimum training-set size in trading days. retrain_freq: Retrain the model every N days. model_type: One of "random_forest" / "gradient_boosting" / "ridge". window_type: "expanding" uses all history; "sliding" uses a fixed lookback. sliding_size: Lookback window size when window_type is "sliding". prediction_horizon: Number of bars each target label looks ahead. Returns: Predicted signal series with range [-1.0, 1.0], no NaN values. """ predictions = pd.Series(0.0, index=features.index) model = None scaler = None if prediction_horizon < 1: raise ValueError("prediction_horizon must be >= 1") for i in range(min_train_size, len(features)): # Retrain every retrain_freq days if model is None or (i - min_train_size) % retrain_freq == 0: # A label at row t is observable only once t + horizon <= i. train_stop = max(0, i - prediction_horizon + 1) start = ( max(0, train_stop - sliding_size) if window_type == "sliding" else 0 ) X_train = features.iloc[start:train_stop].values y_train = labels.iloc[start:train_stop].values # Drop rows with NaN valid = ~(np.isnan(X_train).any(axis=1) | np.isnan(y_train)) X_train = X_train[valid] y_train = y_train[valid] if len(X_train) < 50: continue # Standardization: fit only on training set scaler = StandardScaler() X_train = scaler.fit_transform(X_train) # Build the model if model_type == "random_forest": model = RandomForestClassifier( n_estimators=100, max_depth=5, random_state=42, ) elif model_type == "gradient_boosting": model = GradientBoostingClassifier( n_estimators=100, max_depth=3, learning_rate=0.05, random_state=42, ) elif model_type == "ridge": model = LogisticRegression(penalty="l2", C=1.0, random_state=42) else: raise ValueError(f"Unsupported model_type: {model_type}") model.fit(X_train, y_train) # Predict today X_today = features.iloc[i : i + 1].values if np.isnan(X_today).any(): predictions.iloc[i] = 0.0 continue X_today = scaler.transform(X_today) if hasattr(model, "predict_proba"): prob = model.predict_proba(X_today)[0, 1] predictions.iloc[i] = prob * 2 - 1 # [0,1] -> [-1,1] else: predictions.iloc[i] = float(model.predict(X_today)[0]) # Output contract: no NaN, clipped to [-1, 1] predictions = predictions.fillna(0.0).clip(-1.0, 1.0) return predictions class SignalEngine: """Complete ML strategy with built-in data validation and safety.""" def generate(self, data_map: dict) -> dict: """Generate signals for each symbol. Args: data_map: code -> OHLCV DataFrame. Returns: code -> signal Series in [-1.0, 1.0]. """ signals = {} for code, df in data_map.items(): if not validate_data(df): print(f"[WARN] {code}: data quality insufficient, skipping") continue features = build_features(df) prediction_horizon = 5 future_returns = ( df["close"].shift(-prediction_horizon) / df["close"] - 1 ) labels = (future_returns > 0).astype(float).where(future_returns.notna()) signal = walk_forward_predict( features, labels, prediction_horizon=prediction_horizon, ) signals[code] = signal return signals

各函数职责拆解

• validate_data(df, min_rows=300)：数据质量的守门员。要求列集合必须包含open/high/low/close/volume全部五列；样本量低于min_rows（默认 300 个交易日）直接拒绝；收盘价缺失率超过 20% 同样拒绝。这一层保证进入训练的数据具备最低统计意义，避免用几根 K 线训练出的模型自欺欺人。
• build_features(df)：特征工厂。产出 10 个以f_前缀命名的默认因子（详见下节特征表），并在返回前用replace([np.inf, -np.inf], np.nan)统一消毒——文档注释明确指出，这一设计保证"下游代码永远看不到 inf 值"，而 NaN 交由 walk-forward 的样本过滤处理。
• walk_forward_predict(...)：防泄漏训练与预测的核心。逐日滚动，每retrain_freq天重训一次；每次重训时用train_stop = max(0, i - prediction_horizon + 1)把训练集截止点前移一个预测视界，确保被训练的标签在"预测时刻 i"确实已经可观测；标准化器StandardScaler只在训练集上fit，杜绝标准化泄漏；单日特征含 NaN 时该日输出 0.0 中性信号；最终统一fillna(0.0).clip(-1.0, 1.0)。
• SignalEngine.generate(data_map)：面向多标的的入口。接收code -> OHLCV DataFrame的字典，逐标的执行校验→特征→标签→walk-forward 预测，返回code -> signal Series。标签构建使用了shift(-prediction_horizon)的前移技巧，where(future_returns.notna())把"视界尚未走完"的尾部标签保持为 NaN——这些行不会进入训练。

防泄漏的正确性验证：测试如何锁定行为

技能文档不是空谈，Vibe-Trading 用真实测试固定了 walk-forward 的防泄漏语义。仓库中的 test_ml_strategy_skill.py 通过 AST 解析直接从 SKILL.md 提取 Python 代码块执行，这意味着文档代码块本身就是被测试的产物。三个测试分别验证：

• test_walk_forward_purges_labels_not_observable_at_prediction_time：构造 70 行数据、min_train_size=60、retrain_freq=100（只重训一次）、prediction_horizon=5，断言第一次fit收到的训练集最后一个样本行号是 55——即60 - 5 + 1 = 56个样本中的最后一行（索引 55），证明"标签在预测时刻不可观测"的尾部样本被精确剔除；
• test_one_bar_horizon_preserves_existing_training_window：同样的数据在prediction_horizon=1时训练集末行是 59，即完整 60 行都保留——视界为 1 时无需剔除，语义自洽；
• test_signal_engine_preserves_unavailable_future_labels_as_nan：用 10 行 close 数据驱动SignalEngine.generate，断言前 5 个标签为1.0、后 5 个标签为NaN，同时确认prediction_horizon被正确传为 5——锁定"未来标签不可用时保持 NaN"的行为。

这组测试是理解该技能正确性契约的最佳入口：防泄漏不是注释里的愿望，而是被 CI 固定下来的行为。

特征工程参考：默认因子表

build_features()是自定义扩展点，下表列出全部默认特征，可按需增删。

这 10 个因子覆盖了四类信息维度：动量（ret_5d、ret_20d）、波动（vol_20d、bb_position、skew_20d）、量能（volume_ratio）与日内结构（high_low_ratio、close_open_ratio）。其中rsi_14与bb_position是教科书式的"易除零陷阱"：零波动时段loss=0会让 RSI 产生 inf、std=0会让布林带宽为 0，代码分别用.replace(0, np.nan)规避——这正体现了该技能"安全性内建"的工程取向。

值得注意的是，pct_change与rolling均未指定fill_method之外的参数，而是显式传入fill_method=None，避免默认前向填充把停牌/缺失日期"伪装"成真实行情——这是金融时间序列工程中容易忽视、却直接影响特征质量的细节。若你的数据包含复权除权跳空，建议先对齐数据口径（参考仓库对价格口径的既有约束，见 README_zh.md 关于复权/未复权价混用告警的描述），再做特征计算。

模型选型指南

代码中的默认实现细节与之对应：随机森林默认n_estimators=100, max_depth=5（浅树抗过拟合）；梯度提升默认max_depth=3, learning_rate=0.05（小步长配合浅树，进一步抑制过拟合）；ridge实际使用带 L2 惩罚的LogisticRegression(penalty="l2", C=1.0)，即"岭"式线性分类器，适合作为快速基线。三者统一random_state=42保证结果可复现。

参数说明

几个关键参数的设计意图：

• min_train_size=252对应约一年的交易日，是扩展窗口的起点长度——太短则统计意义不足，太长则冷启动成本高；
• retrain_freq=20约等于每月重训一次，在"模型时效性"与"训练成本"之间取平衡；
• prediction_horizon=5定义标签视界，即预测未来 5 日收益方向；代码在入口处校验prediction_horizon >= 1，非法值直接抛ValueError；
• window_type决定训练窗口形态：expanding用全部历史（样本更多，但可能混入过时 regime），sliding只用最近sliding_size根 K 线（504≈ 两年交易日），更贴近 regime 变化频繁的市场；
• threshold=0.0是信号过滤阈值，abs(signal) < threshold的信号被置 0，可用于在回测/实盘前剔除弱信号（代码示例中该参数未显式使用，属于文档声明的可调契约）。

常见陷阱：代码已解决 vs 仍需人工判断

文档明确区分了两类问题。代码已内置防护的有：数据泄漏（标签可观测性剔除）、标准化泄漏（scaler 只在训练集拟合）、inf/NaN 传播（消毒 + 样本过滤）、重训频率（retrain_freq）。仍需你判断的陷阱有三类：

1. 过拟合（Overfitting）：树过深（max_depth > 10）、特征过多、训练集过小都会导致。建议保持max_depth=3~5，特征数< 15；
2. 类别不平衡（Class imbalance）：牛市环境涨跌比可能达 7:3，模型会偏向预测多数类。必要时使用class_weight="balanced"或 SMOTE 重采样；
3. 前视偏差的非泄漏形态（Look-ahead bias）：用今日收盘计算特征、又预测今日信号，这不算"未来泄漏"，但实践中应确保特征只用 T-1 及更早数据。对应到仓库的 swarm 预设，Feature Engineer 的角色提示也强调"所有特征必须严格 point-in-time 对齐，用 t−1 信息预测 t 期收益"（见 ml_quant_lab.yaml），并把相关性 > 0.85 的特征剔除、按 1%/99% 分位数去极值作为特征工程标准。

依赖与运行环境

pip install scikit-learn pandas numpy

技能依赖仅三个库：scikit-learn（模型与标准化）、pandas（数据与特征）、numpy（数值操作）。在 Vibe-Trading 项目中，你可以在 Agent 会话中通过load_skill("ml-strategy")加载本文档，或直接运行 CLI：vibe-trading run -p "Backtest a BTC-USDT 20/50 moving-average strategy for 2024 and summarize return and drawdown"（示例见 README_zh.md），将本管线的信号接入回测。数据获取可借助仓库内置的数据加载器（get_market_data工具或 MCP 同源 loader 注册表），它们输出归一化 OHLCV，与本技能的输入约定天然兼容。

信号约定

• 连续强度信号：predict_proba[:, 1]（正类概率）经prob * 2 - 1映射到[-1.0, 1.0]；
• 离散信号：使用predict()得到{-1, 0, 1}（空头、中性、多头）；
• 语义：正值 = 看涨方向，负值 = 看跌方向，绝对值 = 置信强度；
• 输出契约：保证无 NaN、无 inf，数值裁剪到[-1.0, 1.0]——任何下游（回测引擎、信号过滤、仓位构建）都可以无条件消费该输出。

这一契约与 Vibe-Trading 整体的严格 JSON/非有限值处理风格一致（仓库在多处强调 NaN 泄漏进输出的危害，参见 README_zh.md 中关于 worker 输出 NaN 泄漏进非严格 JSON 的修复记录），也是该技能能安全嵌入 Agent 工具链的基石。

参考路径速览

• 技能文档本体：agent/src/skills/ml-strategy/SKILL.md
• 正确性测试（防泄漏语义锁定）：agent/tests/test_ml_strategy_skill.py
• Swarm 多智能体应用预设：agent/src/swarm/presets/ml_quant_lab.yaml
• 技能加载机制：agent/src/agent/skills.py
• 技能库分类说明：README_zh.md

【免费下载链接】Vibe-Trading"Vibe-Trading: Your Personal Trading Agent"项目地址: https://gitcode.com/GitHub_Trending/vi/Vibe-Trading