最小 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，无需引入任何回测框架