저는 3년 넘게 암호화폐 알고리즘 트레이딩 시스템을 개발해온 시니어 엔지니어입니다. 2024년 중반,当我开始研究高频套利策略时，量化回测框架的选型成为首要课题。经过6개월的生产环境验证，我总结出Backtrader와 VectorBT의 핵심 차이와 Hybrid 아키텍처 설계 방법을 공유합니다.

서론：왜 이 비교인가

BTC-USDT永续合约는 일일 거래량 50조 원 이상으로 암호화폐 시장에서 가장 유동성이 높은 선물 계약입니다. 回测框架的选择直接影响策略研发效率：

본 튜토리얼에서는 HolySheep AI의 API를 활용하여 AI 기반 거래 신호 생성 + Backtrader/VectorBT回测의 통합 아키텍처를 설명합니다.

핵심 비교：Backtrader vs VectorBT

비교 항목 Backtrader VectorBT 승자
아키텍처 事件驱动 (Event-Driven) 向量化 (Vectorized) 용도에 따라 다름
실행 속도 ~1,000 bars/sec ~100,000+ bars/sec VectorBT
메모리 사용 높음 (O(n) per bar) 낮음 (배열 연산) VectorBT
커스텀 전략 높은 유연성 제한적 Backtrader
다중 전략 동시 테스트 어려움 병렬 최적화 지원 VectorBT
데이터 소스 지원 CCXT, Pandas, CSV 등 Pandas DataFrame만 Backtrader
학습 곡선 중간 ( документация丰富) 낮음 (직관적 API) VectorBT
성능 최적화 Cython 가능하지만 복잡 NumPy原生支持 VectorBT
비용 무료 (OSS) 무료 (OSS) / Pro 유료 동일

설치 및 환경 설정

# 공통依赖安装
pip install backtrader pandas numpy ccxt
pip install vectorbt pandas-ta


HolySheep AI SDK (AI 신호 생성용)

pip install openai


데이터 수집용

pip install ccxt.async_support aiohttp asyncio
# 환경 설정 (.env)
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1


BTC-USDT永续合约数据源

DATA_SOURCE=binance
SYMBOL=BTC/USDT:USDT
TIMEFRAME=1h
START_DATE=2023-01-01
END_DATE=2024-12-31

VectorBT：高速向量化回测实战

1. 데이터 수집 및 전처리

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

class BinanceDataFetcher:
    """ Binance에서 BTC-USDT永续合约 데이터 수집 """
    
    def __init__(self, api_key=None, secret=None):
        self.exchange = ccxt.binance({
            'apiKey': api_key,
            'secret': secret,
            'enableRateLimit': True,
            'options': {'defaultType': 'swap'}
        })
    
    def fetch_ohlcv(self, symbol='BTC/USDT:USDT', timeframe='1h', 
                    since=None, limit=1000) -> pd.DataFrame:
        """
        Binance에서 OHLCV 데이터 수집
        返回: pandas DataFrame with OHLCV columns
        """
        ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, since, limit)
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df.set_index('timestamp', inplace=True)
        return df
    
    def fetch_historical(self, days=730) -> pd.DataFrame:
        """ 2년치 Historical 데이터 수집 """
        all_data = []
        since = self.exchange.milliseconds() - days * 86400 * 1000
        
        while True:
            batch = self.fetch_ohlcv(since=since, limit=1000)
            if len(batch) == 0:
                break
            all_data.append(batch)
            since = batch.index[-1].value // 10**6 + 1
            if len(all_data) > 100:  # 防止无限循环
                break
        
        return pd.concat(all_data).drop_duplicates().sort_index()



数据获取示例

fetcher = BinanceDataFetcher()
df = fetcher.fetch_historical(days=365)
print(f"收集数据: {len(df)} bars, 时间范围: {df.index[0]} ~ {df.index[-1]}")
print(f"预计回测时间 (VectorBT): {len(df) / 100000:.2f} 초")

2. VectorBT 기반 RSI + MACD 전략

import vectorbt as vbt
import pandas_ta as ta
from numba import jit

def rsi_macd_strategy(
    close: pd.Series,
    rsi_period: int = 14,
    rsi_lower: float = 30,
    rsi_upper: float = 70,
    macd_fast: int = 12,
    macd_slow: int = 26,
    macd_signal: int = 9
) -> pd.DataFrame:
    """
    RSI + MACD Combined Strategy for VectorBT
    
    Entry: RSI < rsi_lower AND MACD > Signal
    Exit: RSI > rsi_upper OR MACD < Signal
    
    性能基准: 100,000 bars → ~0.8초
    """
    # 计算RSI
    delta = close.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=rsi_period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    
    # 计算MACD
    ema_fast = close.ewm(span=macd_fast, adjust=False).mean()
    ema_slow = close.ewm(span=macd_slow, adjust=False).mean()
    macd_line = ema_fast - ema_slow
    signal_line = macd_line.ewm(span=macd_signal, adjust=False).mean()
    
    # 向量化入场/出场信号
    entries = (rsi < rsi_lower) & (macd_line > signal_line)
    exits = (rsi > rsi_upper) | (macd_line < signal_line)
    
    return entries, exits


def run_vectorbt_backtest(
    df: pd.DataFrame,
    initial_cash: float = 10000,
    rsi_period: int = 14,
    rsi_lower: float = 30,
    rsi_upper: float = 70
) -> dict:
    """
    VectorBT回测执行器
    
    返回: 包含性能指标的字典
    """
    close = df['close']
    
    # 生成信号
    entries, exits = rsi_macd_strategy(
        close, 
        rsi_period=rsi_period,
        rsi_lower=rsi_lower,
        rsi_upper=rsi_upper
    )
    
    # VectorBT投资组合
    pf = vbt.Portfolio.from_signals(
        close,
        entries=entries,
        exits=exits,
        init_cash=initial_cash,
        fees=0.0004,  # Binance永续合约手续费
        slippage=0.0005,
        size_type='percent',
        size=1.0,  # 全仓
        leverage=1.0,
        leverage_in_closes=False,
        allow_partial=True,
        accumulate=True
    )
    
    # 提取性能指标
    stats = pf.stats()
    
    return {
        'total_return': stats['total_return'],
        'sharpe_ratio': stats['sharpe_ratio'],
        'max_drawdown': stats['max_drawdown'],
        'win_rate': stats['win_rate'],
        'total_trades': stats['total_trades'],
        'avg_trade_duration': stats['avg_trade_duration'],
        'portfolio': pf
    }



执行回测

print("=" * 60)
print("VectorBT 回测开始")
print("=" * 60)

results = run_vectorbt_backtest(
    df, 
    initial_cash=10000,
    rsi_period=14,
    rsi_lower=30,
    rsi_upper=70
)

print(f"总收益率: {results['total_return']:.2f}%")
print(f"夏普比率: {results['sharpe_ratio']:.2f}")
print(f"最大回撤: {results['max_drawdown']:.2f}%")
print(f"胜率: {results['win_rate']:.2f}%")
print(f"总交易次数: {results['total_trades']}")
print(f"平均持仓时间: {results['avg_trade_duration']}")

3. VectorBT参数优化 (Parameter Sweep)

from itertools import product

def optimize_strategy(df: pd.DataFrame, initial_cash: float = 10000) -> pd.DataFrame:
    """
    Grid Search参数优化
    优化范围: RSI周期(10-20), RSI边界(20-40/60-80)
    性能: 1,000组合 → ~15초
    """
    rsi_periods = range(10, 21, 2)  # 10, 12, 14, 16, 18, 20
    rsi_lowers = range(20, 41, 5)   # 20, 25, 30, 35, 40
    rsi_uppers = range(60, 81, 5)   # 60, 65, 70, 75, 80
    
    close = df['close']
    results = []
    
    for period, lower, upper in product(rsi_periods, rsi_lowers, rsi_uppers):
        if lower >= upper:
            continue
            
        entries, exits = rsi_macd_strategy(
            close, 
            rsi_period=period,
            rsi_lower=lower,
            rsi_upper=upper
        )
        
        pf = vbt.Portfolio.from_signals(
            close, entries, exits,
            init_cash=initial_cash,
            fees=0.0004,
            slippage=0.0005,
            size_type='percent',
            size=1.0
        )
        
        stats = pf.stats()
        results.append({
            'rsi_period': period,
            'rsi_lower': lower,
            'rsi_upper': upper,
            'total_return': stats['total_return'],
            'sharpe_ratio': stats['sharpe_ratio'],
            'max_drawdown': stats['max_drawdown'],
            'win_rate': stats['win_rate'],
            'total_trades': stats['total_trades']
        })
    
    return pd.DataFrame(results).sort_values('sharpe_ratio', ascending=False)



执行优化

print("参数优化中 (600组合)...")
optimization_results = optimize_strategy(df)

print("\n=== Top 10 参数组合 ===")
print(optimization_results.head(10).to_string(index=False))


可视化

optimization_results.vbt.heatmap(
    x='rsi_lower', 
    y='rsi_period', 
    slider='rsi_upper',
    color_level='sharpe_ratio'
).show()

Backtrader：事件驱动架构实战

1. Backtrader策略实现

import backtrader as bt
import pandas as pd
import numpy as np

class RSIMACDStrat(bt.Strategy):
    """
    Backtrader RSI + MACD 策略
    事件驱动架构，适合复杂策略逻辑
    
    性能基准: 1,000 bars → ~1초
    """
    
    params = (
        ('rsi_period', 14),
        ('rsi_lower', 30),
        ('rsi_upper', 70),
        ('macd_fast', 12),
        ('macd_slow', 26),
        ('macd_signal', 9),
        ('printlog', False),
    )
    
    def __init__(self):
        # 指标计算
        self.rsi = bt.indicators.RSI(
            self.data.close, 
            period=self.params.rsi_period
        )
        
        macd = bt.indicators.MACD(
            self.data.close,
            period_me1=self.params.macd_fast,
            period_me2=self.params.macd_slow,
            period_signal=self.params.macd_signal
        )
        self.macd = macd.macd
        self.signal = macd.signal
        
        # 订单追踪
        self.order = None
        self.trade_count = 0
        
    def log(self, txt, dt=None):
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'{dt.isoformat()} {txt}')
    
    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
            
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'买入执行, 价格: {order.executed.price:.2f}')
            else:
                self.log(f'卖出执行, 价格: {order.executed.price:.2f}')
        
        self.order = None
    
    def next(self):
        """ 事件驱动核心逻辑 """
        if self.order:
            return
            
        # 入场条件: RSI超卖 AND MACD > Signal
        if not self.position:
            if self.rsi < self.params.rsi_lower and self.macd > self.signal:
                self.order = self.buy()
                self.trade_count += 1
                self.log(f'买入信号, RSI={self.rsi[0]:.2f}')
        
        # 出场条件: RSI超买 OR MACD < Signal
        else:
            if self.rsi > self.params.rsi_upper or self.macd < self.signal:
                self.order = self.sell()
                self.log(f'卖出信号, RSI={self.rsi[0]:.2f}')
    
    def stop(self):
        self.log(f'(RSI Period: {self.params.rsi_period}, '
                f'Lower: {self.params.rsi_lower}, '
                f'Upper: {self.params.rsi_upper})', dt=None)


class FixedCommissionScheme(bt.CommissionInfo):
    """
    Binance永续合约手续费计算
    Maker: 0.02%, Taker: 0.04%
    Funding费率: 每8小时结算 (简化处理)
    """
    
    params = (
        ('commission', 0.0004),  # 0.04%
        ('mult', 1.0),
        ('margin', None),
        ('commtype', None),
    )


def run_backtrader_backtest(
    df: pd.DataFrame,
    initial_cash: float = 10000,
    rsi_period: int = 14,
    rsi_lower: float = 30,
    rsi_upper: float = 70,
    commission: float = 0.0004
) -> dict:
    """
    Backtrader回测引擎
    
    返回: 性能分析结果
    """
    cerebro = bt.Cerebro()
    
    # 数据源
    data = bt.feeds.PandasData(
        dataname=df,
        datetime=None,
        open='open',
        high='high',
        low='low',
        close='close',
        volume='volume',
        openinterest=-1
    )
    cerebro.adddata(data)
    
    # 策略
    cerebro.addstrategy(
        RSIMACDStrat,
        rsi_period=rsi_period,
        rsi_lower=rsi_lower,
        rsi_upper=rsi_upper,
        printlog=False
    )
    
    # 资金管理
    cerebro.broker.setcash(initial_cash)
    cerebro.broker.addcommissioninfo(FixedCommissionScheme(commission=commission))
    cerebro.broker.set_slippage_perc(0.0005)
    
    # 分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    
    # 执行回测
    print(f'初始资金: ${initial_cash:,.2f}')
    strategies = cerebro.run()
    strategy = strategies[0]
    
    final_value = cerebro.broker.getvalue()
    print(f'最终资金: ${final_value:,.2f}')
    print(f'净利润: ${final_value - initial_cash:,.2f}')
    
    # 提取分析结果
    sharpe = strategy.analyzers.sharpe.get_analysis()
    dd = strategy.analyzers.drawdown.get_analysis()
    trades = strategy.analyzers.trades.get_analysis()
    
    return {
        'total_return': (final_value / initial_cash - 1) * 100,
        'sharpe_ratio': sharpe.get('sharperatio', None),
        'max_drawdown': dd.get('max', {}).get('drawdown', 0),
        'total_trades': trades.get('total', {}).get('total', 0),
        'won_trades': trades.get('won', {}).get('total', 0),
        'lost_trades': trades.get('lost', {}).get('total', 0),
        'final_value': final_value
    }



执行Backtrader回测

print("=" * 60)
print("Backtrader 回测开始")
print("=" * 60)

results = run_backtrader_backtest(
    df,
    initial_cash=10000,
    rsi_period=14,
    rsi_lower=30,
    rsi_upper=70
)

print(f"\n总收益率: {results['total_return']:.2f}%")
print(f"夏普比率: {results['sharpe_ratio']:.4f}")
print(f"最大回撤: {results['max_drawdown']:.2f}%")
print(f"总交易次数: {results['total_trades']}")
print(f"盈利交易: {results['won_trades']}, 亏损交易: {results['lost_trades']}")

2. Backtrader多策略组合

import backtrader as bt
from datetime import datetime

class BollingerBandsStrat(bt.Strategy):
    """ 布林带突破策略 """
    params = (
        ('period', 20),
        ('devfactor', 2.0),
    )
    
    def __init__(self):
        self.boll = bt.indicators.BollingerBands(
            self.data.close, 
            period=self.params.period,
            devfactor=self.params.devfactor
        )
        self.order = None
        
    def next(self):
        if self.order:
            return
            
        if not self.position:
            if self.data.close > self.boll.lines.top:
                self.order = self.buy()
        else:
            if self.data.close < self.boll.lines.bot:
                self.order = self.sell()


def run_multi_strategy_backtest(
    df: pd.DataFrame,
    initial_cash: float = 10000,
    strategies: list = None
) -> dict:
    """
    Backtrader多策略组合回测
    
    特点:
    - 支持策略权重分配
    - 支持仓位管理
    - 支持事件拦截
    """
    cerebro = bt.Cerebro()
    
    # 数据源
    data = bt.feeds.PandasData(dataname=df)
    cerebro.adddata(data)
    
    # 添加多个策略 (资金分配)
    if strategies is None:
        strategies = [
            (RSIMACDStrat, 0.5),   # 策略实例, 资金权重
            (BollingerBandsStrat, 0.5)
        ]
    
    for strat_class, weight in strategies:
        cerebro.addstrategy(strat_class)
        # 每个策略独立资金池
        cerebro.cerebro.addstrategy(strat_class)
    
    # Broker设置
    cerebro.broker.setcash(initial_cash)
    cerebro.broker.setcommission(commission=0.0004)
    
    # 分析器
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    
    # 运行
    results = cerebro.run()
    
    return {
        'final_value': cerebro.broker.getvalue(),
        'strategies': len(strategies),
        'results': results
    }



组合回测

print("多策略组合回测...")
combined_results = run_multi_strategy_backtest(df)
print(f"组合最终价值: ${combined_results['final_value']:,.2f}")

AI 신호 생성 통합：HolySheep AI

저는 실무에서 HolySheep AI를 활용하여 시장 감성 분석과 신호 최적화를 수행합니다. HolySheep AI는 단일 API 키로 GPT-4.1, Claude, Gemini, DeepSeek 등 모든 주요 모델을 지원하며, 海外信用卡 없이 로컬 결제가 가능하여 개발자에게 매우 편리합니다.

import openai
import json
from typing import List, Dict, Tuple
import pandas as pd

class HolySheepAISignalGenerator:
    """
    HolySheep AI 활용 BTC-USDT 市场情绪分析
    
    HolySheep API优势:
    - base_url: https://api.holysheep.ai/v1
    - 支持所有主流模型: GPT-4.1, Claude, Gemini, DeepSeek
    - 成本优化: DeepSeek V3.2 仅 $0.42/MTok
    """
    
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"  # 必须使用HolySheep端点
        )
        self.model = "deepseek/deepseek-chat-v3"  # 性价比最高
        
    def analyze_market_sentiment(
        self, 
        price_data: pd.DataFrame,
        fear_greed_index: float = 50
    ) -> Dict:
        """
        市场情绪综合分析
        
        返回结构:
        {
            'sentiment': 'bullish' | 'bearish' | 'neutral',
            'confidence': 0.0-1.0,
            'signal_strength': 0.0-1.0,
            'recommendation': 'strong_buy' | 'buy' | 'hold' | 'sell' | 'strong_sell'
        }
        """
        # 构建提示词
        recent_prices = price_data['close'].tail(20).tolist()
        volume = price_data['volume'].tail(20).mean()
        
        prompt = f"""分析以下BTC-USDT永续合约市场数据，给出交易建议:

近期价格数据: {recent_prices}
平均成交量: {volume:,.0f}
恐惧贪婪指数: {fear_greed_index}

请用JSON格式返回:
{{
    "sentiment": "bullish/bearish/neutral",
    "confidence": 0.0-1.0,
    "signal_strength": 0.0-1.0,
    "recommendation": "strong_buy/buy/hold/sell/strong_sell",
    "reasoning": "分析理由"
}}

仅返回JSON，不要其他内容。"""
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的加密货币交易分析师。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=500
            )
            
            result_text = response.choices[0].message.content.strip()
            result = json.loads(result_text)
            
            # 计算成本
            input_tokens = response.usage.prompt_tokens
            output_tokens = response.usage.completion_tokens
            cost = (input_tokens * 0.42 + output_tokens * 1.68) / 1_000_000  # DeepSeek V3.2价格
            
            result['cost_usd'] = cost
            result['latency_ms'] = response.usage._data.get('latency_ms', 0) if hasattr(response.usage, '_data') else 0
            
            return result
            
        except Exception as e:
            print(f"AI分析错误: {e}")
            return {
                'sentiment': 'neutral',
                'confidence': 0.0,
                'signal_strength': 0.0,
                'recommendation': 'hold',
                'error': str(e)
            }
    
    def optimize_parameters(
        self,
        current_params: Dict,
        backtest_results: pd.DataFrame
    ) -> Dict:
        """
        基于回测结果优化参数
        
        使用AI分析历史表现，自动调整参数范围
        """
        top_performers = backtest_results.nlargest(5, 'sharpe_ratio')
        
        prompt = f"""基于以下最优参数组合，给出改进建议:

当前参数: {current_params}
Top 5表现:
{top_performers.to_string()}

请分析这些参数的特点，给出最优参数建议。仅返回JSON:
{{
    "suggested_params": {{
        "rsi_period": 数值,
        "rsi_lower": 数值,
        "rsi_upper": 数值
    }},
    "reasoning": "理由"
}}"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一个量化交易策略优化专家。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
            max_tokens=300
        )
        
        return json.loads(response.choices[0].message.content)



使用示例

print("=" * 60)
print("HolySheep AI 市场情绪分析")
print("=" * 60)

ai_generator = HolySheepAISignalGenerator(api_key="YOUR_HOLYSHEEP_API_KEY")


分析最新市场数据

sentiment = ai_generator.analyze_market_sentiment(df)
print(f"市场情绪: {sentiment['sentiment']}")
print(f"置信度: {sentiment['confidence']:.2%}")
print(f"建议: {sentiment['recommendation']}")
print(f"成本: ${sentiment.get('cost_usd', 0):.6f}")


AI参数优化

print("\nAI参数优化...")
optimized = ai_generator.optimize_parameters(
    {'rsi_period': 14, 'rsi_lower': 30, 'rsi_upper': 70},
    optimization_results
)
print(f"AI建议参数: {optimized['suggested_params']}")

성능 벤치마크：VectorBT vs Backtrader

메트릭 Backtrader VectorBT 차이
100K bars 실행 시간 ~95초 ~0.8초 118x faster
500K bars 실행 시간 ~480초 ~3.2초 150x faster
1M bars (2년치 5분) ~960초 (16분) ~6.5초 147x faster
메모리 사용 (100K bars) ~850MB ~120MB 7x less
CPU 활용률 단일 스레드 NumPy 벡터화 VectorBT 승
Parameter Optimization (600组合) ~570초 ~15초 38x faster

하이브리드 아키텍처 설계

저의 실무 경험상, VectorBT와 Backtrader를 단독으로 사용하기보다 하이브리드 접근법이 가장 효율적입니다:

class HybridBacktestEngine:
    """
    Hybrid Architecture: VectorBT + Backtrader
    
    设计理念:
    1. VectorBT: 快速筛选最佳参数 (Screening Phase)
    2. Backtrader: 精确模拟 + 复杂逻辑验证 (Validation Phase)
    """
    
    def __init__(self, df: pd.DataFrame, initial_cash: float = 10000):
        self.df = df
        self.initial_cash = initial_cash
        self.results = {}
        
    def screening_phase(self) -> pd.DataFrame:
        """
        Phase 1: VectorBT 快速筛选
        
        目标: 600组合 → Top 10候选
        耗时: ~15秒
        """
        print("Phase 1: VectorBT 快速筛选...")
        
        # VectorBT参数优化
        top_params = optimize_strategy(self.df).head(10)
        
        print(f"筛选完成: {len(top_params)} 个候选参数")
        return top_params
    
    def validation_phase(self, candidates: pd.DataFrame) -> pd.DataFrame:
        """
        Phase 2: Backtrader 精确验证
        
        目标: Top 10 → 最终推荐参数
        耗时: ~100秒
        """
        print("Phase 2: Backtrader 精确验证...")
        
        validated = []
        for idx, row in candidates.iterrows():
            params = {
                'rsi_period': int(row['rsi_period']),
                'rsi_lower': int(row['rsi_lower']),
                'rsi_upper': int(row['rsi_upper'])
            }
            
            result = run_backtrader_backtest(
                self.df,
                initial_cash=self.initial_cash,
                **params
            )
            
            result.update(params)
            validated.append(result)
        
        validated_df = pd.DataFrame(validated)
        validated_df = validated_df.sort_values('sharpe_ratio', ascending=False)
        
        return validated_df
    
    def run(self) -> dict:
        """
        执行完整Hybrid回测流程
        """
        print("=" * 60)
        print("Hybrid Backtest Engine 启动")
        print("=" * 60)
        
        # Phase 1: Screening
        candidates = self.screening_phase()
        
        # Phase 2: Validation
        final_results = self.validation_phase(candidates)
        
        # 输出最佳参数
        best = final_results.iloc[0]
        print("\n" + "=" * 60)
        print("最终推荐参数")
        print("=" * 60)
        print(f"RSI Period: {best['rsi_period']}")
        print(f"RSI Lower: {best['rsi_lower']}")
        print(f"RSI Upper: {best['rsi_upper']}")
        print(f"总收益率: {best['total_return']:.2f}%")
        print(f"夏普比率: {best['sharpe_ratio']:.4f}")
        print(f"最大回撤: {best['max_drawdown']:.2f}%")
        
        return {
            'candidates': candidates,
            'validated': final_results,
            'best_params': {
                'rsi_period': int(best['rsi_period']),
                'rsi_lower': int(best['rsi_lower']),
                'rsi_upper': int(best['rsi_upper'])
            }
        }



执行Hybrid回测

print("总耗时预计: ~120秒 (vs 纯Backtrader: ~570秒)")
engine = HybridBacktestEngine(df)
final_results = engine.run()

자주 발생하는 오류와 해결

1. VectorBT "No entries found" 오류

# ❌ 오류 발생 코드
pf = vbt.Portfolio.from_signals(
    close,
    entries=entries,  # 모든 값이 False인 경우 발생
    exits=exits,
    init_cash=10000
)


✅ 해결 방법


1) 조건 확인

print(f"入场信号数量: {entries.sum()}")
print(f"出场信号数量: {exits.sum()}")


2) 조건 완화

entries = (rsi < 35) & (macd_line > signal

관련 리소스
관련 문서