VectorBT：闪电般快速的 Python 回测库处理 100 万笔以上交易 — 2026 量化指南

import vectorbt as vbt
import numpy as np
import pandas as pd

# 获取数据 — VectorBT 封装了 yfinance 方便调用
price = vbt.YFData.download(
    "BTC-USD",
    start="2020-01-01",
    end="2026-01-01",
    interval="1d"
).get("Close")

print(f"Data shape: {price.shape}")  # (2,210,) — 日收盘价
print(f"Data type: {type(price)}")   # <class 'pandas.core.series.Series'>

# 基础安装
pip install vectorbt

# 全部可选依赖（推荐）
pip install "vectorbt[all]"

import vectorbt as vbt
print(vbt.__version__)  # 0.27.2 或更高

# requirements.txt
vectorbt==0.27.2
numba==0.60.0
numpy==1.26.4
pandas==2.2.3
yfinance==0.2.54
plotly==5.24.1

xcode-select --install  # 如果 Numba 安装失败，先运行这个

import vectorbt as vbt
import pandas as pd

# 下载历史数据
price = vbt.YFData.download(
    ["AAPL", "MSFT", "GOOGL"],
    start="2020-01-01",
    end="2026-01-01"
).get("Close")

# 生成快慢均线
fast_ma = vbt.MA.run(price, window=20)
slow_ma = vbt.MA.run(price, window=50)

# 创建入场和出场信号
entries = fast_ma.ma_crossed_above(slow_ma)
exits = fast_ma.ma_crossed_below(slow_ma)

# 运行组合模拟
portfolio = vbt.Portfolio.from_signals(
    price,
    entries=entries,
    exits=exits,
    init_cash=100_000,
    fees=0.001,        # 每笔 0.1% 佣金
    slippage=0.0005    # 5 个基点滑点
)

# 结果
print(portfolio.total_return())
print(portfolio.sharpe_ratio())

import vectorbt as vbt

price = vbt.YFData.download("BTC-USD", start="2020-01-01", end="2026-01-01").get("Close")

# 定义参数范围
fast_windows = np.arange(5, 51, 5)    # [5, 10, 15, ..., 50]
slow_windows = np.arange(20, 201, 10)  # [20, 30, 40, ..., 200]

# 运行向量化网格搜索
fast_ma, slow_ma = vbt.MA.run_combs(
    price,
    window=fast_windows,
    r=2,
    short_names=["fast", "slow"]
)

entries = fast_ma.ma_crossed_above(slow_ma)
exits = fast_ma.ma_crossed_below(slow_ma)

portfolio = vbt.Portfolio.from_signals(
    price, entries=exits, exits=entries,
    init_cash=10_000, fees=0.001
)

# 找到最佳组合
best_idx = portfolio.sharpe_ratio().idxmax()
print(f"Best params: {best_idx}")
print(f"Sharpe: {portfolio.sharpe_ratio().loc[best_idx]:.2f}")

import vectorbt as vbt
from datetime import datetime
import pandas as pd

price = vbt.YFData.download("ETH-USD", start="2021-01-01", end="2026-01-01").get("Close")

# 滚动前进配置
n_splits = 10
split_size = len(price) // n_splits
results = []

for i in range(n_splits):
    # 定义训练/测试窗口
    train_start = i * split_size
    train_end = train_start + split_size - 60
    test_end = train_start + split_size
    
    train_price = price.iloc[train_start:train_end]
    test_price = price.iloc[train_end:test_end]
    
    # 在训练集上优化
    fast_ma = vbt.MA.run(train_price, np.arange(5, 31, 5))
    slow_ma = vbt.MA.run(train_price, np.arange(20, 101, 10))
    
    entries = fast_ma.ma_crossed_above(slow_ma)
    exits = fast_ma.ma_crossed_below(slow_ma)
    
    train_pf = vbt.Portfolio.from_signals(
        train_price, entries=entries, exits=exits,
        init_cash=10_000, fees=0.001
    )
    
    best = train_pf.sharpe_ratio().idxmax()
    best_fast, best_slow = best
    
    # 在样本外测试
    test_fast = vbt.MA.run(test_price, window=best_fast)
    test_slow = vbt.MA.run(test_price, window=best_slow)
    test_entries = test_fast.ma_crossed_above(test_slow)
    test_exits = test_fast.ma_crossed_below(test_slow)
    
    test_pf = vbt.Portfolio.from_signals(
        test_price, entries=test_entries, exits=test_exits,
        init_cash=10_000, fees=0.001
    )
    
    results.append({
        "split": i,
        "fast": best_fast,
        "slow": best_slow,
        "train_sharpe": train_pf.sharpe_ratio().loc[best],
        "test_sharpe": test_pf.sharpe_ratio(),
        "test_return": test_pf.total_return()
    })

results_df = pd.DataFrame(results)
print(results_df[["test_sharpe", "test_return"]].mean())

import vectorbt as vbt
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import pandas as pd
import numpy as np

# 加载数据
price = vbt.YFData.download("BTC-USD", start="2020-01-01", end="2026-01-01").get("Close")
returns = price.pct_change()

# 构建特征矩阵：滞后收益
lags = 5
features = pd.concat([returns.shift(i) for i in range(1, lags + 1)], axis=1)
features.columns = [f"lag_{i}" for i in range(1, lags + 1)]

# 目标：次日收益为正为 1，否则为 0
target = (returns.shift(-1) > 0).astype(int)

# 清洗并切分
data = pd.concat([features, target], axis=1).dropna()
X = data.iloc[:, :-1]
y = data.iloc[:, -1]
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, shuffle=False
)

# 训练分类器
clf = RandomForestClassifier(n_estimators=200, max_depth=5, random_state=42)
clf.fit(X_train, y_train)

# 在测试集上生成预测
predictions = pd.Series(
    clf.predict(X_test),
    index=X_test.index,
    name="prediction"
)

# 创建信号：预测上涨日入场，预测下跌日出场
test_price = price.loc[X_test.index]
entries = predictions == 1
exits = predictions == 0

# 用 ML 信号运行回测
ml_portfolio = vbt.Portfolio.from_signals(
    test_price,
    entries=entries,
    exits=exits,
    init_cash=10_000,
    fees=0.001,
    freq="1D"
)

print(f"ML Strategy Return: {ml_portfolio.total_return():.2%}")
print(f"ML Strategy Sharpe: {ml_portfolio.sharpe_ratio():.2f}")
print(f"Buy & Hold Return: {(test_price.iloc[-1] / test_price.iloc[0] - 1):.2%}")

import vectorbt as vbt
import numpy as np

# 多资产价格数据
data = vbt.YFData.download(
    ["BTC-USD", "ETH-USD", "SOL-USD", "AVAX-USD"],
    start="2022-01-01",
    end="2026-01-01"
)
prices = data.get("Close")

# 简单的反波动率加权
returns = prices.pct_change().dropna()
volatility = returns.rolling(30).std().iloc[-1]
weights = 1 / volatility
weights = weights / weights.sum()

print("Portfolio weights:")
for symbol, w in weights.items():
    print(f"  {symbol}: {w:.2%}")

# 回测该配置
portfolio = vbt.Portfolio.from_holding(
    prices,
    weights=weights,
    init_cash=100_000,
    fees=0.001
)

print(f"\nCAGR: {portfolio.total_return() ** (1/4) - 1:.2%}")
print(f"Sharpe: {portfolio.sharpe_ratio():.2f}")
print(f"Max Drawdown: {portfolio.max_drawdown():.2%}")

import vectorbt as vbt
import numpy as np
from numba import njit

@njit
def custom_momentum_nb(price, period):
    """Numba 加速的动量指标。"""
    momentum = np.empty_like(price)
    momentum[:period] = np.nan
    for i in range(period, len(price)):
        momentum[i] = (price[i] / price[i - period] - 1) * 100
    return momentum

# 用 IndicatorFactory 包装
CustomMomentum = vbt.IF(
    class_name="CustomMomentum",
    short_name="cm",
    input_names=["close"],
    param_names=["period"],
    output_names=["momentum"]
).with_custom_func(custom_momentum_nb)

# 使用
price = vbt.YFData.download("BTC-USD", start="2023-01-01").get("Close")
cm = CustomMomentum.run(price, period=[7, 14, 30])
print(cm.momentum)

import vectorbt as vbt

price = vbt.YFData.download("BTC-USD", start="2023-01-01").get("Close")

entries = vbt.MA.run(price, 10).ma_crossed_above(vbt.MA.run(price, 30))

portfolio = vbt.Portfolio.from_signals(
    price,
    entries=entries,
    exits=None,  # 让止损/止盈处理出场
    sl_stop=0.05,     # 5% 止损
    tp_stop=0.15,     # 15% 止盈
    tsl_stop=0.08,    # 8% 移动止损
    init_cash=10_000,
    fees=0.001
)

print(f"Return: {portfolio.total_return():.2%}")
print(f"Win rate: {portfolio.trades.win_rate():.2%}")
print(f"Avg trade: {portfolio.trades.returns.mean():.2%}")

from multiprocessing import Pool
import vectorbt as vbt
import numpy as np

def run_chunk(param_chunk):
    price = vbt.YFData.download("BTC-USD").get("Close")
    fast_ma = vbt.MA.run(price, param_chunk[:, 0])
    slow_ma = vbt.MA.run(price, param_chunk[:, 1])
    entries = fast_ma.ma_crossed_above(slow_ma)
    exits = fast_ma.ma_crossed_below(slow_ma)
    pf = vbt.Portfolio.from_signals(price, entries, exits, init_cash=10_000)
    return pf.sharpe_ratio()

# 把 360 个参数拆分到 4 个进程
params = np.array(np.meshgrid(np.arange(5, 41, 5), np.arange(20, 121, 10))).T.reshape(-1, 2)
chunks = np.array_split(params, 4)

with Pool(4) as p:
    results = p.map(run_chunk, chunks)