Hummingbot 第11章：Quants Lab 研究环境

Quants Lab 是 Hummingbot 的研究环境，提供 Jupyter Notebook 集成和回测引擎，帮助用户分析和优化交易策略。

Jupyter Notebook 集成

安装 Quants Lab

# 通过 pip 安装 Quants Lab
pip install hummingbot-quantslab

# 安装 Jupyter
pip install jupyterlab notebook

# 启动 Jupyter
jupyter lab

导入 Hummingbot 数据模块

# quants_lab_setup.ipynb
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

from hummingbot.client.hummingbot_application import HummingbotApplication
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory
from hummingbot.connector.exchange.binance import BinanceExchange

# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

print("Quants Lab 环境初始化完成")

回测引擎

回测引擎概述

Hummingbot 的回测引擎支持在历史数据上模拟策略运行：

• 事件驱动模拟：基于真实市场数据的 tick 级模拟
• 支持任意交易对：只要有历史 K 线数据即可回测
• 盈亏计算：自动计算交易费用、滑点等成本
• 详细报告：生成包含多种统计指标的 HTML 报告

获取历史数据

# download_data.ipynb
from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory

# 创建 K 线数据源
candles = CandlesFactory.get_candles(
    connector="binance",
    trading_pair="BTC-USDT",
    interval="1h",
    max_records=5000  # 最多获取 5000 根 K 线
)

# 启动数据获取
candles.start()

# 等待数据准备完成
import time
while not candles.ready:
    time.sleep(1)
    print("等待数据下载...")

# 获取数据
df = candles.candles_df
print(f"数据范围: {df['timestamp'].min()} 至 {df['timestamp'].max()}")
print(f"数据量: {len(df)} 条 K 线")

# 查看数据前几行
print(df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].head())

# 保存到本地 CSV
df.to_csv("btc_usdt_1h.csv", index=False)
print("数据已保存到 btc_usdt_1h.csv")

编写回测策略

# backtest_strategy.py
from hummingbot.strategy_v2.backtesting import BacktestingEngine
from hummingbot.strategy_v2.controllers import ControllerBase
from pydantic import BaseModel


class SimpleMMConfig(BaseModel):
    """简单做市策略配置"""
    bid_spread: float = 0.005
    ask_spread: float = 0.005
    order_amount: float = 0.01


class SimpleMMController(ControllerBase):
    """回测用的做市控制器"""

    def __init__(self, config: SimpleMMConfig):
        super().__init__(config)
        self.config = config

    async def create_orders(self, strategy, current_price: float):
        """根据当前价格创建做市订单"""
        bid_price = current_price * (1 - self.config.bid_spread)
        ask_price = current_price * (1 + self.config.ask_spread)

        # 记录回测信号
        strategy.add_signal(
            price=current_price,
            bid_price=bid_price,
            ask_price=ask_price,
            bid_spread=self.config.bid_spread,
            ask_spread=self.config.ask_spread
        )

        return {
            "bids": [{"price": bid_price, "amount": self.config.order_amount}],
            "asks": [{"price": ask_price, "amount": self.config.order_amount}]
        }

运行回测

# run_backtest.ipynb
from hummingbot.strategy_v2.backtesting import BacktestingEngine
import pandas as pd

# 加载历史数据
df = pd.read_csv("btc_usdt_1h.csv")

# 配置回测引擎
engine = BacktestingEngine(
    data=df,
    initial_balance=10000.0,  # 初始资金（USDT）
    trading_fee=0.001,        # 交易费率 (0.1%)
    slippage=0.0005,          # 滑点 (0.05%)
    start_date="2024-01-01",
    end_date="2024-12-31"
)

# 运行回测
results = engine.run(
    controller_class=SimpleMMController,
    controller_config=SimpleMMConfig(
        bid_spread=0.005,
        ask_spread=0.005,
        order_amount=0.01
    )
)

# 输出结果
print("===== 回测结果 =====")
print(f"总收益率: {results['total_return_pct']:.2f}%")
print(f"年化收益率: {results['annual_return_pct']:.2f}%")
print(f"最大回撤: {results['max_drawdown_pct']:.2f}%")
print(f"夏普比率: {results['sharpe_ratio']:.2f}")
print(f"总交易次数: {results['total_trades']}")
print(f"胜率: {results['win_rate']:.2f}%")
print(f"平均盈利: {results['avg_win']:.4f}%")
print(f"平均亏损: {results['avg_loss']:.4f}%")

数据获取

支持的 K 线周期

多交易所数据获取

def get_historical_data(
    exchange: str,
    trading_pair: str,
    interval: str,
    days: int
) -> pd.DataFrame:
    """获取历史 K 线数据"""
    from hummingbot.data_feed.candles_feed.candles_factory import CandlesFactory

    candles = CandlesFactory.get_candles(
        connector=exchange,
        trading_pair=trading_pair,
        interval=interval,
        max_records=days * 24 * 60  # 根据间隔换算
    )

    candles.start()

    # 等待数据下载
    import time
    timeout = 30
    while not candles.ready and timeout > 0:
        time.sleep(1)
        timeout -= 1

    if not candles.ready:
        raise TimeoutError("数据下载超时")

    return candles.candles_df


# 获取多个交易所同一交易对的数据
binance_data = get_historical_data("binance", "BTC-USDT", "1h", 30)
okx_data = get_historical_data("okx", "BTC-USDT", "1h", 30)

# 比较价格差异
merged = pd.merge(
    binance_data[['timestamp', 'close']],
    okx_data[['timestamp', 'close']],
    on='timestamp',
    suffixes=('_binance', '_okx')
)

merged['price_diff_pct'] = (
    (merged['close_okx'] - merged['close_binance'])
    / merged['close_binance'] * 100
)

print(f"最大价差: {merged['price_diff_pct'].max():.3f}%")
print(f"平均价差: {merged['price_diff_pct'].mean():.3f}%")

结果分析

可视化回测结果

# analyze_results.ipynb
import matplotlib.pyplot as plt

# 绘制资金曲线
equity_curve = results['equity_curve']
plt.figure(figsize=(12, 6))
plt.plot(equity_curve.index, equity_curve['equity'], label='资金曲线', linewidth=2)
plt.plot(equity_curve.index, equity_curve['drawdown'], label='回撤', alpha=0.7)
plt.title('策略表现分析')
plt.xlabel('时间')
plt.ylabel('资金 (USDT)')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

# 月度收益热力图
monthly_returns = results['monthly_returns']
plt.figure(figsize=(10, 6))
plt.imshow(monthly_returns.values, cmap='RdYlGn', aspect='auto')
plt.colorbar(label='收益率 (%)')
plt.title('月度收益率热力图')
plt.yticks(range(len(monthly_returns.index)), monthly_returns.index)
plt.xticks(range(len(monthly_returns.columns)), monthly_returns.columns)
plt.show()

策略对比分析

# 对比不同参数的回测结果
configs = [
    {"bid_spread": 0.003, "ask_spread": 0.003, "order_amount": 0.01},
    {"bid_spread": 0.005, "ask_spread": 0.005, "order_amount": 0.01},
    {"bid_spread": 0.010, "ask_spread": 0.010, "order_amount": 0.01},
]

results_list = []
for config in configs:
    result = engine.run(
        controller_class=SimpleMMController,
        controller_config=SimpleMMConfig(**config)
    )
    results_list.append({
        "config": f"Spread {config['bid_spread']*100:.1f}%",
        "return": result['total_return_pct'],
        "sharpe": result['sharpe_ratio'],
        "max_dd": result['max_drawdown_pct'],
        "trades": result['total_trades']
    })

comparison_df = pd.DataFrame(results_list)
print(comparison_df.to_string(index=False))

关键回测指标参考