Backtrader 2026: Python 백테스팅 엔진으로 전략을 100배 더 빠르게 검증 — 완벽 가이드

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Backtrader 설치 (2026년 5월 기준 v1.9.81.127)
pip install backtrader

# 선택사항: 확장된 지표를 위한 TA-Lib 래퍼 설치
pip install TA-Lib

# 선택사항: 플로팅을 위한 matplotlib 설치
pip install matplotlib

import backtrader as bt
print(bt.__version__)

# 예상 출력: 1.9.81.127 또는 이상

import backtrader as bt
import datetime

class SmaCross(bt.Strategy):
    params = dict(fast=10, slow=30)

    def __init__(self):
        self.fast_sma = bt.indicators.SMA(period=self.p.fast)
        self.slow_sma = bt.indicators.SMA(period=self.p.slow)
        self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)

    def next(self):
        if not self.position:  # 시장에 없음
            if self.crossover > 0:  # 빠른 SMA가 느린 SMA를 상향 돌파
                self.buy()
        elif self.crossover < 0:  # 빠른 SMA가 느린 SMA를 하향 돌파
            self.sell()

# Cerebro 엔진 생성
cerebro = bt.Cerebro()
cerebro.addstrategy(SmaCross)

# 데이터 로드
import yfinance as yf
df = yf.download("AAPL", start="2020-01-01", end="2025-01-01")
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

# 초기 자금 설정
cerebro.broker.setcash(10000.0)

# 백테스트 실행
print(f"시작 포트폴리오 가치: {cerebro.broker.getvalue():.2f}")
cerebro.run()
print(f"최종 포트폴리오 가치: {cerebro.broker.getvalue():.2f}")

# 결과 플롯
cerebro.plot()

class RSIMeanReversion(bt.Strategy):
    params = dict(rsi_period=14, oversold=30, overbought=70)

    def __init__(self):
        self.rsi = bt.indicators.RSI(period=self.p.rsi_period)

    def next(self):
        if not self.position:
            if self.rsi < self.p.oversold:
                self.buy()
        else:
            if self.rsi > self.p.overbought:
                self.sell()

    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy():
                print(f"매수 체결가 {order.executed.price:.2f}")
            else:
                print(f"매도 체결가 {order.executed.price:.2f}")

class BollingerBreakout(bt.Strategy):
    params = dict(period=20, devfactor=2.0)

    def __init__(self):
        self.bbands = bt.indicators.BollingerBands(
            period=self.p.period, devfactor=self.p.devfactor
        )
        self.atr = bt.indicators.ATR(period=14)

    def next(self):
        if not self.position:
            if self.data.close > self.bbands.lines.top:
                # ATR 기반 사이징으로 돌파 매수
                size = int(self.broker.getvalue() * 0.02 / self.atr[0])
                self.buy(size=size)
        else:
            if self.data.close < self.bbands.lines.mid:
                self.sell()

    def notify_trade(self, trade):
        if trade.isclosed:
            print(f"트레이드 손익: {trade.pnlcomm:.2f}")

class MultiTimeframeMomentum(bt.Strategy):
    params = dict(daily_period=20, weekly_period=10)

    def __init__(self):
        # 일간 SMA
        self.daily_sma = bt.indicators.SMA(self.data0, period=self.p.daily_period)
        # 주간 SMA (data1을 주간 리샘플링 데이터로 사용)
        self.weekly_sma = bt.indicators.SMA(self.data1, period=self.p.weekly_period)

    def next(self):
        # 일간 및 주간 추세가 일치할 때만 거래
        if (self.data0.close > self.daily_sma[0] and
            self.data1.close > self.weekly_sma[0] and
            not self.position):
            self.buy()
        elif (self.data0.close < self.daily_sma[0] and
              self.data1.close < self.weekly_sma[0] and
              self.position):
            self.sell()

import backtrader.feeds as btfeeds
import yfinance as yf

# 다운로드 및 피드
df = yf.download("SPY", start="2020-01-01", end="2026-01-01")
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

data = btfeeds.GenericCSVData(
    dataname='btc_usd.csv',
    dtformat='%Y-%m-%d',
    datetime=0, open=1, high=2, low=3, close=4, volume=5,
    fromdate=datetime.datetime(2020, 1, 1),
    todate=datetime.datetime(2026, 1, 1)
)
cerebro.adddata(data)

# 변동성 필터 트레이딩을 위한 SPY 및 VIX 추가
spy_data = bt.feeds.PandasData(dataname=spy_df, name="SPY")
vix_data = bt.feeds.PandasData(dataname=vix_df, name="VIX")

cerebro.adddata(spy_data)
cerebro.adddata(vix_data)

# 전략 내: self.datas[0] = SPY, self.datas[1] = VIX

import ccxt

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1d", since=1577836800000)

# pandas DataFrame으로 변환
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)

data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

import backtrader as bt

class SmaCross(bt.Strategy):
    params = dict(fast=10, slow=30)

    def __init__(self):
        self.fast_sma = bt.indicators.SMA(period=self.p.fast)
        self.slow_sma = bt.indicators.SMA(period=self.p.slow)
        self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)

    def next(self):
        if not self.position and self.crossover > 0:
            self.buy()
        elif self.position and self.crossover < 0:
            self.sell()

cerebro = bt.Cerebro()

# 파라미터 그리드 검색: fast=[5,10,15], slow=[20,30,40]
cerebro.optstrategy(
    SmaCross,
    fast=range(5, 20, 5),
    slow=range(20, 50, 10)
)

# 데이터 및 브로커 추가
data = bt.feeds.PandasData(dataname=yf.download("AAPL", start="2020-01-01", end="2025-01-01"))
cerebro.adddata(data)
cerebro.broker.setcash(10000.0)
cerebro.broker.setcommission(commission=0.001)  # 거래당 0.1%

# 분석기 추가
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

# 최적화 실행 (모든 CPU 코어 사용)
results = cerebro.run(maxcpus=4)

# 샤프 비율로 최상의 결과 추출
best = max(results, key=lambda r: r[0].analyzers.sharpe.get_analysis()['sharperatio'] or 0)
print(f"최고 샤프 비율: {best[0].analyzers.sharpe.get_analysis()['sharperatio']:.2f}")
print(f"최고 파라미터: fast={best[0].params.fast}, slow={best[0].params.slow}")

import backtrader as bt
import ccxt

class LiveStrategy(bt.Strategy):
    def __init__(self):
        self.rsi = bt.indicators.RSI(period=14)

    def next(self):
        if not self.position and self.rsi < 30:
            self.buy(size=0.001)  # 0.001 BTC
        elif self.position and self.rsi > 70:
            self.sell(size=0.001)

# 실시간 트레이딩 설정
cerebro = bt.Cerebro()
cerebro.addstrategy(LiveStrategy)

# CCXT 브로커 래퍼 사용
from ccxtbt import CCXTStore
store = CCXTStore(exchange='binance', currency='USDT',
                  config={'apiKey': 'YOUR_KEY', 'secret': 'YOUR_SECRET'})
broker = store.getbroker()
cerebro.setbroker(broker)

# 실시간 데이터 피드 가져오기
data = store.getdata(dataname='BTC/USDT', timeframe=bt.TimeFrame.Minutes, compression=60)
cerebro.adddata(data)

cerebro.run()

FROM python:3.12-slim

WORKDIR /app
RUN pip install --no-cache-dir backtrader pandas numpy matplotlib yfinance

COPY strategy.py .
COPY data/ ./data/

CMD ["python", "strategy.py"]

# docker-compose.yml
version: '3.8'
services:
  backtrader:
    build: .
    volumes:
      - ./results:/app/results
    environment:
      - INITIAL_CASH=100000
    restart: unless-stopped

# 매일 오전 6시 백테스트를 위해 crontab에 추가
0 6 * * * cd /path/to/strategy && /path/to/venv/bin/python run_backtest.py >> logs/backtest.log 2>&1

# 현실적인 커미션: 주당 $0.01, 최소 $1
commission_info = bt.CommissionInfo(
    commission=0.01,
    mult=1.0,
    margin=None,
    commtype=bt.CommissionInfo.COMM_FIXED,
    mincom=1.0
)
cerebro.broker.addcommissioninfo(commission_info)

# 슬리피지: 0.1% 체결 가격 영향
cerebro.broker.set_slippage_perc(perc=0.001)

def walk_forward_analysis(data, train_days=252, test_days=63):
    """견고성을 검증하기 위해 롤링 훈련/테스트 분할을 실행."""
    results = []
    total_bars = len(data)
    start = 0

    while start + train_days + test_days < total_bars:
        train_data = data[start:start + train_days]
        test_data = data[start + train_days:start + train_days + test_days]

        # 훈련 세트에서 최적화, 보지 못한 데이터에서 테스트
        cerebro = bt.Cerebro()
        cerebro.optstrategy(MyStrategy, param1=range(5, 20))
        cerebro.adddata(train_data)
        best_params = cerebro.run()

        cerebro2 = bt.Cerebro()
        cerebro2.addstrategy(MyStrategy, **best_params)
        cerebro2.adddata(test_data)
        result = cerebro2.run()
        results.append(result)

        start += test_days

    return results

class EquityCurve(bt.observer.Observer):
    lines = ('equity',)
    plotinfo = dict(plot=True, subplot=True)

    def next(self):
        self.lines.equity[0] = self._owner.broker.getvalue()

# Cerebro에 추가
cerebro.addobserver(EquityCurve)

import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RiskManagedStrategy(bt.Strategy):
    params = dict(max_risk_per_trade=0.02, max_drawdown=0.15)

    def __init__(self):
        self.peak_value = self.broker.getvalue()

    def next(self):
        current_value = self.broker.getvalue()
        self.peak_value = max(self.peak_value, current_value)
        drawdown = (self.peak_value - current_value) / self.peak_value

        if drawdown > self.p.max_drawdown:
            logger.warning(f"최대 낙폭 도달: {drawdown:.2%}. 모든 포지션 청산.")
            self.close()
            return

        # ... 전략 로직의 나머지

class CustomIndicator(bt.Indicator):
    lines = ('myline',)
    params = dict(period=20)

    def __init__(self):
        self.addminperiod(self.p.period)

    def next(self):
        # 커스텀 계산
        self.lines.myline[0] = sum(self.data.get(size=self.p.period)) / self.p.period