アルゴリズムトレーディングバックテスト実践: 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, 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       # 範囲外は適用しない
)