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

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


class WalkForwardAnalyzer:
    """Walk-Forward Analyzer: sequential validation to prevent overfitting"""

    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]:
        """Generate sequential train/test splits"""
        total_len = len(self.data)
        split_size = total_len // self.n_splits
        splits = []

        for i in range(self.n_splits):
            # Training period: start ~ current split point
            train_end = split_size * (i + 1)
            train_start = max(0, train_end - int(split_size * self.train_ratio * (i + 1)))

            # Testing period: training end ~ next split point
            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:
        """Execute Walk-Forward optimization"""
        splits = self.generate_splits()
        results = []

        for split in splits:
            # 1. Optimize parameters on training data
            best_params = optimize_fn(split["train"])

            # 2. Validate performance on test data
            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


# Understanding the basic structure: Cerebro engine setup
cerebro = bt.Cerebro()

# 1. Add data feed
data = bt.feeds.PandasData(
    dataname=yf.download("AAPL", start="2020-01-01", end="2025-12-31"),
    datetime=None,  # Use index as datetime
    open="Open",
    high="High",
    low="Low",
    close="Close",
    volume="Volume"
)
cerebro.adddata(data)

# 2. Add strategy
# cerebro.addstrategy(MyStrategy)

# 3. Broker settings
cerebro.broker.setcash(100000)           # Initial capital
cerebro.broker.setcommission(commission=0.001)  # 0.1% commission

# 4. Sizer settings (position size)
cerebro.addsizer(bt.sizers.PercentSizer, percents=95)

# 5. Add analyzers
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")

# Execute
print(f"Initial portfolio value: {cerebro.broker.getvalue():,.0f}")
results = cerebro.run()
print(f"Final portfolio value: {cerebro.broker.getvalue():,.0f}")

# Loading data from CSV files
class CustomCSVData(bt.feeds.GenericCSVData):
    """Custom CSV data feed"""
    params = (
        ("dtformat", "%Y-%m-%d"),
        ("datetime", 0),
        ("open", 1),
        ("high", 2),
        ("low", 3),
        ("close", 4),
        ("volume", 5),
        ("openinterest", -1),
    )


# Multi-asset simultaneous backtest
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):
    """Moving Average Crossover: buy on golden cross, sell on death cross"""

    params = (
        ("fast_period", 20),   # Short-term moving average
        ("slow_period", 50),   # Long-term moving average
        ("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)

        # Order tracking
        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  # Skip if pending order exists

        if not self.position:
            # No position -> check buy signal
            if self.crossover > 0:  # Golden cross
                self.log(f"BUY SIGNAL | Close: {self.data.close[0]:.2f}")
                self.order = self.buy()
        else:
            # Holding position -> check sell signal
            if self.crossover < 0:  # Death cross
                self.log(f"SELL SIGNAL | Close: {self.data.close[0]:.2f}")
                self.order = self.sell()


# Strategy execution
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-based mean reversion: buy on oversold, sell on overbought"""

    params = (
        ("rsi_period", 14),
        ("oversold", 30),     # Oversold threshold
        ("overbought", 70),   # Overbought threshold
        ("stake", 100),       # Trade quantity
    )

    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:
            # Entering oversold zone -> buy
            if self.rsi[0] < self.params.oversold:
                self.order = self.buy(size=self.params.stake)
        else:
            # Entering overbought zone -> sell
            if self.rsi[0] > self.params.overbought:
                self.order = self.sell(size=self.params.stake)

class BollingerBandStrategy(bt.Strategy):
    """Bollinger Bands: buy at lower band touch, sell at upper band touch"""

    params = (
        ("bb_period", 20),
        ("bb_dev", 2.0),      # Standard deviation multiplier
        ("stop_loss", 0.03),  # 3% stop-loss
    )

    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:
            # Price drops below lower band -> buy
            if self.data.close[0] < self.bb.lines.bot[0]:
                self.order = self.buy()
                self.entry_price = self.data.close[0]
        else:
            # Sell conditions: upper band breakout or stop-loss
            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()  # Stop-loss

class AdvancedOrderStrategy(bt.Strategy):
    """Advanced order type usage example"""

    params = (
        ("trail_percent", 0.05),  # 5% trailing stop
        ("limit_offset", 0.02),   # 2% limit offset
    )

    def next(self):
        if not self.position:
            # Limit order: buy at 2% below current price
            limit_price = self.data.close[0] * (1 - self.params.limit_offset)
            self.buy(exectype=bt.Order.Limit, price=limit_price)

        else:
            # Trailing stop: sell when 5% below peak
            self.sell(
                exectype=bt.Order.StopTrail,
                trailpercent=self.params.trail_percent
            )

import numpy as np
import pandas as pd


class PerformanceMetrics:
    """Trading strategy performance metric calculator"""

    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:
        """Total return"""
        return (1 + self.returns).prod() - 1

    def annualized_return(self) -> float:
        """Annualized return"""
        total = self.total_return()
        n_years = len(self.returns) / 252
        return (1 + total) ** (1 / n_years) - 1

    def sharpe_ratio(self) -> float:
        """Sharpe Ratio: (Annualized Return - Risk-Free Rate) / Annualized Volatility"""
        excess_returns = self.returns - self.daily_rf
        return np.sqrt(252) * excess_returns.mean() / excess_returns.std()

    def sortino_ratio(self) -> float:
        """Sortino Ratio: considers only downside volatility"""
        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:
        """Maximum Drawdown"""
        cumulative = (1 + self.returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        return drawdown.min()

    def calmar_ratio(self) -> float:
        """Calmar Ratio: Annualized Return / Maximum Drawdown"""
        mdd = abs(self.max_drawdown())
        if mdd == 0:
            return 0
        return self.annualized_return() / mdd

    def win_rate(self, trades: list[float]) -> float:
        """Win Rate: proportion of profitable trades"""
        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:
        """Profit Factor: Gross Profit / Gross Loss"""
        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:
        """Full performance summary"""
        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}",
        }


# Usage example
returns = pd.Series(np.random.normal(0.0005, 0.02, 252 * 3))  # 3 years of daily returns
metrics = PerformanceMetrics(returns)
for key, value in metrics.summary().items():
    print(f"{key}: {value}")

class FullAnalysisStrategy(bt.Strategy):
    """Performance analysis using Backtrader built-in Analyzers"""

    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 setup and result analysis
cerebro = bt.Cerebro()
cerebro.addstrategy(FullAnalysisStrategy)
cerebro.adddata(data)
cerebro.broker.setcash(100000)

# Add various Analyzers
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 results
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):
    """Kelly Criterion position sizing"""

    params = (
        ("fraction", 0.5),  # Half-Kelly (conservative)
    )

    def _getsizing(self, comminfo, cash, data, isbuy):
        # Calculate win rate and payoff ratio from recent trade history
        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=win rate, R=payoff ratio
            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))  # Max 25%
        else:
            kelly = 0.02  # Initial 2%

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


class FixedRiskSizer(bt.Sizer):
    """Fixed risk percentage sizing: prevent losing more than N% of capital per trade"""

    params = (
        ("risk_percent", 0.02),  # 2% risk
        ("stop_distance", 0.05), # 5% stop distance
    )

    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):
    """Strategy with integrated stop-loss, take-profit, and trailing stop"""

    params = (
        ("sma_period", 20),
        ("stop_loss", 0.03),      # 3% stop-loss
        ("take_profit", 0.06),    # 6% take-profit (2:1 risk-reward)
        ("trail_percent", 0.04),  # 4% trailing stop
    )

    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():
                # On buy fill, set stop-loss and take-profit simultaneously
                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():
                # On sell fill, cancel the opposite order
                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 basic usage example (for comparison)
import vectorbt as vbt

# Download data
price = vbt.YFData.download("AAPL", start="2020-01-01", end="2025-12-31").get("Close")

# SMA crossover backtest (fast execution via vectorized operations)
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:
    """Parameter perturbation test: verify strategy stability against parameter changes"""

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

    def generate_perturbations(self) -> list[dict]:
        """Generate perturbation combinations around base parameters"""
        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:
        """Run backtests with perturbed parameters and compare results"""
        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: coefficient of variation of Sharpe ratio (lower = more stable)
        stability_score = df["sharpe"].std() / abs(df["sharpe"].mean())
        print(f"Stability Score (CV): {stability_score:.4f}")
        print(f"  - Below 0.1: Very stable")
        print(f"  - 0.1~0.3: Stable")
        print(f"  - Above 0.3: Unstable (overfitting suspected)")

        return df

class TradingSystemRecovery:
    """Trading system failure recovery handler"""

    def handle_data_feed_failure(self):
        """On data feed failure"""
        # 1. Switch to alternative data source (e.g., Yahoo -> Alpha Vantage)
        # 2. Decide to hold/close positions based on last valid price
        # 3. Verify missing intervals after data recovery
        pass

    def handle_order_rejection(self):
        """On order rejection"""
        # 1. Check rejection reason (insufficient funds, price limits, etc.)
        # 2. Adjust parameters and resubmit order
        # 3. Pause strategy on consecutive rejections
        pass

    def handle_position_mismatch(self):
        """On position mismatch"""
        # 1. Synchronize broker API with internal state
        # 2. Analyze mismatch cause (partial fills, network errors, etc.)
        # 3. Adjust positions after manual verification
        pass

    def handle_extreme_drawdown(self, current_drawdown: float):
        """On extreme drawdown"""
        # 1. Stop new entries when drawdown exceeds threshold (e.g., -15%)
        # 2. Gradually close existing positions
        # 3. Send admin alert notification
        if current_drawdown < -0.15:
            print("ALERT: Emergency stop triggered")
            # self.close_all_positions()
            # self.notify_admin()

# Backtrader slippage settings
cerebro = bt.Cerebro()

# Fixed slippage: add fixed points per trade
cerebro.broker.set_slippage_fixed(fixed=0.05)

# Percentage slippage: add N% of price
cerebro.broker.set_slippage_perc(perc=0.001)  # 0.1%

# Commission structure
cerebro.broker.setcommission(
    commission=0.001,    # 0.1% commission
    margin=None,
    mult=1.0
)

# Production-realistic settings
cerebro.broker.set_slippage_perc(
    perc=0.001,          # 0.1% slippage
    slip_open=True,      # Apply slippage to open price
    slip_limit=True,     # Apply to limit orders
    slip_match=True,     # Apply at matching price
    slip_out=False       # Do not apply outside range
)