Vibe-Trading 技能库完全使用指南：83 个技能 + 27 个多智能体团队

最小 Demo：用 MCP 完成一次回归预测 + 回测

这个 Demo 演示如何用 Vibe-Trading 的 MCP 工具链，零代码写管线、零配置拉数据，完成一次完整的"预测→验证"闭环。

目标

用滚动线性回归预测次日涨跌，回测验证效果。

思路

收盘价 → 计算 5 日动量因子 → 线性回归预测次日方向 → 生成交易信号 → 回测

Step 1：拉数据（1 个 MCP 调用）

工具: get_market_data
参数:
  codes: ["AAPL.US"]
  start_date: "2023-01-01"
  end_date: "2024-12-31"
  source: "auto"

返回 AAPL 日线 OHLCV，无需写任何 yfinance 代码。

Step 2：写策略（回归预测核心）

创建 config.json：

{
  "source": "auto",
  "codes": ["AAPL.US"],
  "start_date": "2023-01-01",
  "end_date": "2024-12-31",
  "interval": "1D",
  "initial_cash": 100000,
  "commission": 0.001,
  "engine": "daily"
}

创建 code/signal_engine.py（回归预测的最小实现）：

import pandas as pd
import numpy as np
from typing import Dict
from sklearn.linear_model import LinearRegression

class SignalEngine:
    def __init__(self, lookback=60, forecast_horizon=1):
        self.lookback = lookback
        self.forecast_horizon = forecast_horizon

    def _build_features(self, df):
        close = df["close"]
        features = pd.DataFrame(index=df.index)
        features["returns"] = close.pct_change()
        features["ma5"] = close.rolling(5).mean() / close - 1
        features["ma20"] = close.rolling(20).mean() / close - 1
        features["vol5"] = features["returns"].rolling(5).std()
        features["rsi"] = self._rsi(close, 14)
        features["volume_ratio"] = df["volume"] / df["volume"].rolling(20).mean()
        return features, close

    def _rsi(self, close, period=14):
        delta = close.diff()
        gain = delta.clip(lower=0)
        loss = -delta.clip(upper=0)
        avg_gain = gain.ewm(alpha=1/period, adjust=False).mean()
        avg_loss = loss.ewm(alpha=1/period, adjust=False).mean()
        rs = avg_gain / avg_loss.replace(0, np.nan)
        return 100 - 100 / (1 + rs)

    def generate(self, data_map: Dict[str, pd.DataFrame]) -> Dict[str, pd.Series]:
        signals = {}
        for code, df in data_map.items():
            features, close = self._build_features(df)
            target = close.pct_change(self.forecast_horizon).shift(-self.forecast_horizon)

            signal = pd.Series(0.0, index=df.index)

            for t in range(self.lookback, len(df) - self.forecast_horizon):
                train_features = features.iloc[t-self.lookback:t].dropna()
                train_target = target.iloc[t-self.lookback:t].dropna()
                common_idx = train_features.index.intersection(train_target.index)
                X_train = train_features.loc[common_idx].values
                y_train = train_target.loc[common_idx].values

                X_pred = features.iloc[t:t+1].values
                if len(X_train) < 10 or np.isnan(X_pred).any():
                    continue

                model = LinearRegression()
                model.fit(X_train, y_train)
                pred = model.predict(X_pred)[0]

                signal.iloc[t] = np.clip(pred * 20, -1.0, 1.0)

            signals[code] = signal.fillna(0)
        return signals

核心逻辑：

1. 构造 6 个特征（收益率、均线偏离、波动率、RSI、量比）
2. 用过去 60 天数据滚动训练 LinearRegression
3. 预测次日涨跌幅 → 乘以 20 放大信号 → clip 到 [-1, 1]

Step 3：回测验证（1 个 MCP 调用）

工具: backtest
参数: run_dir=<包含上面两个文件的目录>

返回 metrics.csv：

annual_return, sharpe_ratio, max_drawdown, win_rate, trade_count
0.xxx,         x.xxx,        -0.xxx,       0.xxx,    xxx

Step 4（可选）：替换模型

把 LinearRegression 换成 Ridge 或 LogisticRegression：

from sklearn.linear_model import Ridge
model = Ridge(alpha=1.0)

或者做方向分类而非回归：

from sklearn.linear_model import LogisticRegression
model = LogisticRegression()
# target 改为 0/1：次日涨 → 1，跌 → 0
y_train = (train_target > 0).astype(int).values
pred_prob = model.predict_proba(X_pred)[0][1]
signal.iloc[t] = (pred_prob - 0.5) * 2  # 映射到 [-1, 1]

关键要点

• MCP 工具帮你做的事：数据获取、回测引擎、绩效统计 — 你只写信号逻辑
• 前视偏差防护：shift(-1) 确保预测目标在未来，iloc[t] 只用 t 时刻及之前的数据训练
• 最小依赖：只需 pandas + numpy + sklearn，无需引入任何回测框架

这个 Demo 的完整链路：get_market_data → write_file(config.json) → write_file(signal_engine.py) → backtest → read_file(metrics.csv)，5 个 MCP 调用完成一次完整的回归预测+验证闭环。