알고리즘 트레이딩 백테스팅 실전: Python Backtrader·전략 구현·성과 평가·리스크 관리

import pandas as pd
import numpy as np
from datetime import datetime


class WalkForwardAnalyzer:
    """Walk-Forward 분석기: 과적합 방지를 위한 순차적 검증"""

    def __init__(self, data: pd.DataFrame, train_ratio: float = 0.7,
                 n_splits: int = 5):
        self.data = data
        self.train_ratio = train_ratio
        self.n_splits = n_splits

    def generate_splits(self) -> list[dict]:
        """학습/테스트 구간을 순차적으로 생성"""
        total_len = len(self.data)
        split_size = total_len // self.n_splits
        splits = []

        for i in range(self.n_splits):
            # 학습 구간: 시작 ~ 현재 분할점
            train_end = split_size * (i + 1)
            train_start = max(0, train_end - int(split_size * self.train_ratio * (i + 1)))

            # 테스트 구간: 학습 종료 ~ 다음 분할점
            test_start = train_end
            test_end = min(total_len, test_start + split_size)

            if test_start >= total_len:
                break

            splits.append({
                "fold": i + 1,
                "train": self.data.iloc[train_start:train_end],
                "test": self.data.iloc[test_start:test_end],
                "train_period": f"{self.data.index[train_start]} ~ {self.data.index[train_end-1]}",
                "test_period": f"{self.data.index[test_start]} ~ {self.data.index[test_end-1]}"
            })

        return splits

    def run_walk_forward(self, strategy_fn, optimize_fn) -> pd.DataFrame:
        """Walk-Forward 최적화 실행"""
        splits = self.generate_splits()
        results = []

        for split in splits:
            # 1. 학습 데이터로 파라미터 최적화
            best_params = optimize_fn(split["train"])

            # 2. 테스트 데이터로 성과 검증
            test_result = strategy_fn(split["test"], best_params)

            results.append({
                "fold": split["fold"],
                "train_period": split["train_period"],
                "test_period": split["test_period"],
                "params": best_params,
                "return": test_result["total_return"],
                "sharpe": test_result["sharpe_ratio"],
                "max_drawdown": test_result["max_drawdown"]
            })

        return pd.DataFrame(results)

import backtrader as bt
import yfinance as yf


# 기본 구조 이해: Cerebro 엔진 설정
cerebro = bt.Cerebro()

# 1. 데이터 피드 추가
data = bt.feeds.PandasData(
    dataname=yf.download("AAPL", start="2020-01-01", end="2025-12-31"),
    datetime=None,  # 인덱스를 날짜로 사용
    open="Open",
    high="High",
    low="Low",
    close="Close",
    volume="Volume"
)
cerebro.adddata(data)

# 2. 전략 추가
# cerebro.addstrategy(MyStrategy)

# 3. 브로커 설정
cerebro.broker.setcash(100000)           # 초기 자본
cerebro.broker.setcommission(commission=0.001)  # 수수료 0.1%

# 4. 사이저 설정 (포지션 크기)
cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

# 5. 분석기 추가
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")

# 실행
print(f"초기 자산: {cerebro.broker.getvalue():,.0f}")
results = cerebro.run()
print(f"최종 자산: {cerebro.broker.getvalue():,.0f}")

# CSV 파일에서 데이터 로드
class CustomCSVData(bt.feeds.GenericCSVData):
    """커스텀 CSV 데이터 피드"""
    params = (
        ("dtformat", "%Y-%m-%d"),
        ("datetime", 0),
        ("open", 1),
        ("high", 2),
        ("low", 3),
        ("close", 4),
        ("volume", 5),
        ("openinterest", -1),
    )


# 여러 종목 동시 백테스트
tickers = ["AAPL", "MSFT", "GOOGL"]
cerebro = bt.Cerebro()

for ticker in tickers:
    df = yf.download(ticker, start="2020-01-01", end="2025-12-31")
    data = bt.feeds.PandasData(dataname=df, name=ticker)
    cerebro.adddata(data)

class SMACrossoverStrategy(bt.Strategy):
    """이동평균 교차 전략: 골든크로스 매수, 데드크로스 매도"""

    params = (
        ("fast_period", 20),   # 단기 이동평균
        ("slow_period", 50),   # 장기 이동평균
        ("printlog", True),
    )

    def __init__(self):
        self.sma_fast = bt.indicators.SMA(
            self.data.close, period=self.params.fast_period
        )
        self.sma_slow = bt.indicators.SMA(
            self.data.close, period=self.params.slow_period
        )
        self.crossover = bt.indicators.CrossOver(self.sma_fast, self.sma_slow)

        # 주문 추적
        self.order = None
        self.buy_price = None
        self.buy_comm = None

    def log(self, txt, dt=None):
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            print(f"[{dt}] {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"BUY  | Price: {order.executed.price:.2f}, "
                    f"Cost: {order.executed.value:.2f}, "
                    f"Comm: {order.executed.comm:.2f}"
                )
                self.buy_price = order.executed.price
                self.buy_comm = order.executed.comm
            else:
                self.log(
                    f"SELL | Price: {order.executed.price:.2f}, "
                    f"Cost: {order.executed.value:.2f}, "
                    f"Comm: {order.executed.comm:.2f}"
                )

        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log("Order Canceled/Margin/Rejected")

        self.order = None

    def next(self):
        if self.order:
            return  # 대기 중인 주문이 있으면 스킵

        if not self.position:
            # 포지션 없음 -> 매수 신호 확인
            if self.crossover > 0:  # 골든크로스
                self.log(f"BUY SIGNAL | Close: {self.data.close[0]:.2f}")
                self.order = self.buy()
        else:
            # 포지션 보유 -> 매도 신호 확인
            if self.crossover < 0:  # 데드크로스
                self.log(f"SELL SIGNAL | Close: {self.data.close[0]:.2f}")
                self.order = self.sell()


# 전략 실행
cerebro = bt.Cerebro()
data = bt.feeds.PandasData(
    dataname=yf.download("AAPL", start="2020-01-01", end="2025-12-31")
)
cerebro.adddata(data)
cerebro.addstrategy(SMACrossoverStrategy, fast_period=20, slow_period=50)
cerebro.broker.setcash(100000)
cerebro.broker.setcommission(commission=0.001)
results = cerebro.run()
cerebro.plot()

class RSIMeanReversionStrategy(bt.Strategy):
    """RSI 기반 평균회귀 전략: 과매도 매수, 과매수 매도"""

    params = (
        ("rsi_period", 14),
        ("oversold", 30),     # 과매도 기준
        ("overbought", 70),   # 과매수 기준
        ("stake", 100),       # 거래 수량
    )

    def __init__(self):
        self.rsi = bt.indicators.RSI(
            self.data.close, period=self.params.rsi_period
        )
        self.order = None

    def next(self):
        if self.order:
            return

        if not self.position:
            # 과매도 구간 진입 -> 매수
            if self.rsi[0] < self.params.oversold:
                self.order = self.buy(size=self.params.stake)
        else:
            # 과매수 구간 진입 -> 매도
            if self.rsi[0] > self.params.overbought:
                self.order = self.sell(size=self.params.stake)

class BollingerBandStrategy(bt.Strategy):
    """볼린저 밴드 전략: 하단 밴드 터치 매수, 상단 밴드 터치 매도"""

    params = (
        ("bb_period", 20),
        ("bb_dev", 2.0),      # 표준편차 배수
        ("stop_loss", 0.03),  # 3% 손절
    )

    def __init__(self):
        self.bb = bt.indicators.BollingerBands(
            self.data.close,
            period=self.params.bb_period,
            devfactor=self.params.bb_dev
        )
        self.order = None
        self.entry_price = None

    def next(self):
        if self.order:
            return

        if not self.position:
            # 가격이 하단 밴드 아래로 하락 -> 매수
            if self.data.close[0] < self.bb.lines.bot[0]:
                self.order = self.buy()
                self.entry_price = self.data.close[0]
        else:
            # 매도 조건: 상단 밴드 돌파 또는 손절
            if self.data.close[0] > self.bb.lines.top[0]:
                self.order = self.sell()
            elif self.entry_price and \
                 self.data.close[0] < self.entry_price * (1 - self.params.stop_loss):
                self.order = self.sell()  # 손절

class AdvancedOrderStrategy(bt.Strategy):
    """다양한 주문 유형 활용 예시"""

    params = (
        ("trail_percent", 0.05),  # 5% 트레일링 스톱
        ("limit_offset", 0.02),   # 2% 리밋 오프셋
    )

    def next(self):
        if not self.position:
            # 리밋 주문: 현재가 대비 2% 낮은 가격에 매수
            limit_price = self.data.close[0] * (1 - self.params.limit_offset)
            self.buy(exectype=bt.Order.Limit, price=limit_price)

        else:
            # 트레일링 스톱: 고점 대비 5% 하락 시 매도
            self.sell(
                exectype=bt.Order.StopTrail,
                trailpercent=self.params.trail_percent
            )

import numpy as np
import pandas as pd


class PerformanceMetrics:
    """트레이딩 전략 성과 지표 계산"""

    def __init__(self, returns: pd.Series, risk_free_rate: float = 0.04):
        self.returns = returns
        self.risk_free_rate = risk_free_rate
        self.daily_rf = (1 + risk_free_rate) ** (1/252) - 1

    def total_return(self) -> float:
        """총 수익률"""
        return (1 + self.returns).prod() - 1

    def annualized_return(self) -> float:
        """연간화 수익률"""
        total = self.total_return()
        n_years = len(self.returns) / 252
        return (1 + total) ** (1 / n_years) - 1

    def sharpe_ratio(self) -> float:
        """샤프 비율: (연간 수익률 - 무위험 수익률) / 연간 변동성"""
        excess_returns = self.returns - self.daily_rf
        return np.sqrt(252) * excess_returns.mean() / excess_returns.std()

    def sortino_ratio(self) -> float:
        """소르티노 비율: 하방 변동성만 고려"""
        excess_returns = self.returns - self.daily_rf
        downside_returns = excess_returns[excess_returns < 0]
        downside_std = np.sqrt((downside_returns ** 2).mean())
        return np.sqrt(252) * excess_returns.mean() / downside_std

    def max_drawdown(self) -> float:
        """최대 낙폭"""
        cumulative = (1 + self.returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        return drawdown.min()

    def calmar_ratio(self) -> float:
        """칼마 비율: 연간 수익률 / 최대 낙폭"""
        mdd = abs(self.max_drawdown())
        if mdd == 0:
            return 0
        return self.annualized_return() / mdd

    def win_rate(self, trades: list[float]) -> float:
        """승률: 수익 거래 비율"""
        if not trades:
            return 0
        wins = sum(1 for t in trades if t > 0)
        return wins / len(trades)

    def profit_factor(self, trades: list[float]) -> float:
        """수익 팩터: 총 이익 / 총 손실"""
        gross_profit = sum(t for t in trades if t > 0)
        gross_loss = abs(sum(t for t in trades if t < 0))
        if gross_loss == 0:
            return float("inf")
        return gross_profit / gross_loss

    def summary(self) -> dict:
        """전체 성과 요약"""
        return {
            "Total Return": f"{self.total_return():.2%}",
            "Annualized Return": f"{self.annualized_return():.2%}",
            "Sharpe Ratio": f"{self.sharpe_ratio():.2f}",
            "Sortino Ratio": f"{self.sortino_ratio():.2f}",
            "Max Drawdown": f"{self.max_drawdown():.2%}",
            "Calmar Ratio": f"{self.calmar_ratio():.2f}",
        }


# 사용 예시
returns = pd.Series(np.random.normal(0.0005, 0.02, 252 * 3))  # 3년 일일 수익률
metrics = PerformanceMetrics(returns)
for key, value in metrics.summary().items():
    print(f"{key}: {value}")

class FullAnalysisStrategy(bt.Strategy):
    """Backtrader 내장 Analyzer를 활용한 성과 분석"""

    params = (("fast", 10), ("slow", 30))

    def __init__(self):
        self.sma_fast = bt.indicators.SMA(period=self.params.fast)
        self.sma_slow = bt.indicators.SMA(period=self.params.slow)
        self.crossover = bt.indicators.CrossOver(self.sma_fast, self.sma_slow)

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


# Analyzer 설정 및 결과 분석
cerebro = bt.Cerebro()
cerebro.addstrategy(FullAnalysisStrategy)
cerebro.adddata(data)
cerebro.broker.setcash(100000)

# 다양한 Analyzer 추가
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe",
                    timeframe=bt.TimeFrame.Days, compression=1)
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
cerebro.addanalyzer(bt.analyzers.SQN, _name="sqn")

results = cerebro.run()
strat = results[0]

# 결과 출력
print("=== Sharpe Ratio ===")
print(f"  Sharpe: {strat.analyzers.sharpe.get_analysis().get('sharperatio', 'N/A')}")

print("\n=== Drawdown ===")
dd = strat.analyzers.drawdown.get_analysis()
print(f"  Max Drawdown: {dd.max.drawdown:.2f}%")
print(f"  Max Drawdown Period: {dd.max.len} days")

print("\n=== Trade Analysis ===")
ta = strat.analyzers.trades.get_analysis()
print(f"  Total Trades: {ta.total.closed}")
print(f"  Won: {ta.won.total}")
print(f"  Lost: {ta.lost.total}")

class KellyCriterionSizer(bt.Sizer):
    """켈리 기준 포지션 사이징"""

    params = (
        ("fraction", 0.5),  # Half-Kelly (보수적)
    )

    def _getsizing(self, comminfo, cash, data, isbuy):
        # 최근 거래 기록에서 승률과 손익비 계산
        trades = self.strategy.analyzers.trades.get_analysis()

        if hasattr(trades, "won") and trades.total.closed > 10:
            win_rate = trades.won.total / trades.total.closed
            avg_win = trades.won.pnl.average if trades.won.total > 0 else 0
            avg_loss = abs(trades.lost.pnl.average) if trades.lost.total > 0 else 1

            # Kelly: f = W - (1-W)/R, where W=승률, R=손익비
            if avg_loss > 0:
                kelly = win_rate - (1 - win_rate) / (avg_win / avg_loss)
            else:
                kelly = 0

            kelly = max(0, min(kelly * self.params.fraction, 0.25))  # 최대 25%
        else:
            kelly = 0.02  # 초기에는 2%

        target_value = cash * kelly
        size = int(target_value / data.close[0])
        return max(size, 1)


class FixedRiskSizer(bt.Sizer):
    """고정 리스크 비율 사이징: 각 거래에서 자본의 N% 이상 손실 방지"""

    params = (
        ("risk_percent", 0.02),  # 2% 리스크
        ("stop_distance", 0.05), # 5% 손절 거리
    )

    def _getsizing(self, comminfo, cash, data, isbuy):
        risk_amount = cash * self.params.risk_percent
        price = data.close[0]
        stop_distance = price * self.params.stop_distance
        size = int(risk_amount / stop_distance)
        return max(size, 1)

class RiskManagedStrategy(bt.Strategy):
    """손절·이익 실현·트레일링 스톱이 통합된 전략"""

    params = (
        ("sma_period", 20),
        ("stop_loss", 0.03),      # 3% 손절
        ("take_profit", 0.06),    # 6% 이익 실현 (2:1 손익비)
        ("trail_percent", 0.04),  # 4% 트레일링 스톱
    )

    def __init__(self):
        self.sma = bt.indicators.SMA(period=self.params.sma_period)
        self.order = None
        self.stop_order = None
        self.profit_order = None

    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy():
                # 매수 체결 시 손절/이익 실현 주문 동시 설정
                stop_price = order.executed.price * (1 - self.params.stop_loss)
                profit_price = order.executed.price * (1 + self.params.take_profit)

                self.stop_order = self.sell(
                    exectype=bt.Order.Stop,
                    price=stop_price
                )
                self.profit_order = self.sell(
                    exectype=bt.Order.Limit,
                    price=profit_price
                )

            elif order.issell():
                # 매도 체결 시 반대쪽 주문 취소
                if self.stop_order and self.stop_order.status in [
                    order.Submitted, order.Accepted
                ]:
                    self.cancel(self.stop_order)
                if self.profit_order and self.profit_order.status in [
                    order.Submitted, order.Accepted
                ]:
                    self.cancel(self.profit_order)
                self.stop_order = None
                self.profit_order = None

        self.order = None

    def next(self):
        if self.order:
            return

        if not self.position:
            if self.data.close[0] > self.sma[0]:
                self.order = self.buy()

# VectorBT 기본 사용 예시 (비교 참고)
import vectorbt as vbt

# 데이터 다운로드
price = vbt.YFData.download("AAPL", start="2020-01-01", end="2025-12-31").get("Close")

# SMA 교차 백테스트 (벡터 연산으로 빠르게 실행)
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=100000,
    fees=0.001
)

print(portfolio.stats())

class ParameterStabilityTest:
    """파라미터 섭동 테스트: 파라미터 변경에 대한 전략 안정성 검증"""

    def __init__(self, base_params: dict, perturbation: float = 0.1):
        self.base_params = base_params
        self.perturbation = perturbation

    def generate_perturbations(self) -> list[dict]:
        """기본 파라미터 주변의 섭동 조합 생성"""
        variations = []
        for key, value in self.base_params.items():
            if isinstance(value, (int, float)):
                delta = value * self.perturbation
                for factor in [-1, -0.5, 0, 0.5, 1]:
                    perturbed = self.base_params.copy()
                    new_val = value + delta * factor
                    perturbed[key] = int(new_val) if isinstance(value, int) else new_val
                    variations.append(perturbed)
        return variations

    def run_stability_test(self, backtest_fn) -> pd.DataFrame:
        """섭동된 파라미터로 백테스트를 실행하고 결과 비교"""
        variations = self.generate_perturbations()
        results = []

        for params in variations:
            result = backtest_fn(params)
            results.append({
                **params,
                "sharpe": result["sharpe_ratio"],
                "return": result["total_return"],
                "max_dd": result["max_drawdown"]
            })

        df = pd.DataFrame(results)

        # 안정성 점수: 샤프 비율의 변동 계수 (낮을수록 안정)
        stability_score = df["sharpe"].std() / abs(df["sharpe"].mean())
        print(f"Stability Score (CV): {stability_score:.4f}")
        print(f"  - 0.1 미만: 매우 안정")
        print(f"  - 0.1~0.3: 안정")
        print(f"  - 0.3 초과: 불안정 (오버피팅 의심)")

        return df

class TradingSystemRecovery:
    """트레이딩 시스템 장애 복구 핸들러"""

    def handle_data_feed_failure(self):
        """데이터 피드 장애 시"""
        # 1. 대체 데이터 소스로 전환 (예: Yahoo -> Alpha Vantage)
        # 2. 마지막 유효 가격 기반으로 포지션 유지/청산 판단
        # 3. 데이터 복구 후 누락 구간 검증
        pass

    def handle_order_rejection(self):
        """주문 거부 시"""
        # 1. 거부 사유 확인 (잔고 부족, 가격 제한 등)
        # 2. 파라미터 조정 후 재주문
        # 3. 연속 거부 시 전략 일시 중지
        pass

    def handle_position_mismatch(self):
        """포지션 불일치 시"""
        # 1. 브로커 API와 내부 상태 동기화
        # 2. 불일치 원인 분석 (부분 체결, 네트워크 오류 등)
        # 3. 수동 확인 후 포지션 조정
        pass

    def handle_extreme_drawdown(self, current_drawdown: float):
        """극단적 드로다운 발생 시"""
        # 1. 드로다운이 한계치 초과 시 (예: -15%) 신규 진입 중단
        # 2. 기존 포지션 단계적 청산
        # 3. 관리자 알림 발송
        if current_drawdown < -0.15:
            print("ALERT: Emergency stop triggered")
            # self.close_all_positions()
            # self.notify_admin()

# Backtrader 슬리피지 설정
cerebro = bt.Cerebro()

# 고정 슬리피지: 각 거래에 고정 포인트 추가
cerebro.broker.set_slippage_fixed(fixed=0.05)

# 비율 슬리피지: 가격의 N% 추가
cerebro.broker.set_slippage_perc(perc=0.001)  # 0.1%

# 수수료 체계
cerebro.broker.setcommission(
    commission=0.001,    # 0.1% 수수료
    margin=None,
    mult=1.0
)

# 실전에 가까운 설정
cerebro.broker.set_slippage_perc(
    perc=0.001,          # 0.1% 슬리피지
    slip_open=True,      # 시가에도 슬리피지 적용
    slip_limit=True,     # 리밋 주문에도 적용
    slip_match=True,     # 일치하는 가격에서 적용
    slip_out=False       # 범위 밖은 적용 안 함
)