Pythonアルゴリズムトレーディング実践ガイド：バックテストフレームワーク・戦略開発・リスク管理

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

def fetch_market_data(
    tickers: list,
    start_date: str = "2020-01-01",
    end_date: str = None,
    interval: str = "1d",
) -> dict:
    """市場データ収集"""
    if end_date is None:
        end_date = datetime.now().strftime("%Y-%m-%d")

    data = {}
    for ticker in tickers:
        try:
            df = yf.download(
                ticker,
                start=start_date,
                end=end_date,
                interval=interval,
                progress=False,
            )
            if not df.empty:
                data[ticker] = df
                print(f"{ticker}: {len(df)} rows loaded ({df.index[0]} ~ {df.index[-1]})")
            else:
                print(f"{ticker}: No data available")
        except Exception as e:
            print(f"{ticker}: Error - {e}")

    return data

# データ収集
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "SPY"]
market_data = fetch_market_data(tickers, start_date="2020-01-01")

# データ確認
aapl = market_data["AAPL"]
print(f"\nAAPL データ概要:")
print(f"  期間: {aapl.index[0]} ~ {aapl.index[-1]}")
print(f"  データポイント: {len(aapl)}")
print(f"  カラム: {list(aapl.columns)}")

import requests
import pandas as pd

class AlphaVantageClient:
    """Alpha Vantage APIクライアント"""

    BASE_URL = "https://www.alphavantage.co/query"

    def __init__(self, api_key: str):
        self.api_key = api_key

    def get_daily(self, symbol: str, outputsize: str = "full") -> pd.DataFrame:
        """日足データの取得"""
        params = {
            "function": "TIME_SERIES_DAILY_ADJUSTED",
            "symbol": symbol,
            "outputsize": outputsize,
            "apikey": self.api_key,
        }
        response = requests.get(self.BASE_URL, params=params)
        data = response.json()

        if "Time Series (Daily)" not in data:
            raise ValueError(f"API error: {data.get('Note', data.get('Error Message', 'Unknown'))}")

        df = pd.DataFrame.from_dict(data["Time Series (Daily)"], orient="index")
        df.columns = ["Open", "High", "Low", "Close", "Adj Close", "Volume", "Dividend", "Split"]
        df = df.astype(float)
        df.index = pd.to_datetime(df.index)
        df = df.sort_index()

        return df

    def get_intraday(self, symbol: str, interval: str = "5min") -> pd.DataFrame:
        """分足データの取得"""
        params = {
            "function": "TIME_SERIES_INTRADAY",
            "symbol": symbol,
            "interval": interval,
            "outputsize": "full",
            "apikey": self.api_key,
        }
        response = requests.get(self.BASE_URL, params=params)
        data = response.json()

        time_series_key = f"Time Series ({interval})"
        if time_series_key not in data:
            raise ValueError(f"API error: {data}")

        df = pd.DataFrame.from_dict(data[time_series_key], orient="index")
        df.columns = ["Open", "High", "Low", "Close", "Volume"]
        df = df.astype(float)
        df.index = pd.to_datetime(df.index)
        df = df.sort_index()

        return df

# 使用例
# client = AlphaVantageClient(api_key="YOUR_API_KEY")
# daily_data = client.get_daily("AAPL")

import pandas as pd
import numpy as np
from backtesting import Backtest, Strategy
from backtesting.lib import crossover

class MovingAverageCrossover(Strategy):
    """移動平均クロスオーバー戦略
    - 短期移動平均が長期移動平均を上回ったら買い
    - 短期移動平均が長期移動平均を下回ったら売り
    """
    fast_period = 10  # 短期移動平均期間
    slow_period = 30  # 長期移動平均期間

    def init(self):
        close = self.data.Close
        self.fast_ma = self.I(lambda x: pd.Series(x).rolling(self.fast_period).mean(), close)
        self.slow_ma = self.I(lambda x: pd.Series(x).rolling(self.slow_period).mean(), close)

    def next(self):
        # ゴールデンクロス：買い
        if crossover(self.fast_ma, self.slow_ma):
            if not self.position:
                self.buy()

        # デッドクロス：売り
        elif crossover(self.slow_ma, self.fast_ma):
            if self.position:
                self.position.close()

# データ準備
data = yf.download("AAPL", start="2020-01-01", end="2025-12-31", progress=False)
data.columns = data.columns.droplevel(1) if isinstance(data.columns, pd.MultiIndex) else data.columns

# バックテスト実行
bt = Backtest(
    data,
    MovingAverageCrossover,
    cash=100000,
    commission=0.001,  # 0.1%手数料
    exclusive_orders=True,
)

results = bt.run()
print("=== Moving Average Crossover Results ===")
print(f"トータルリターン: {results['Return [%]']:.2f}%")
print(f"年率リターン: {results['Return (Ann.) [%]']:.2f}%")
print(f"シャープレシオ: {results['Sharpe Ratio']:.2f}")
print(f"最大ドローダウン: {results['Max. Drawdown [%]']:.2f}%")
print(f"勝率: {results['Win Rate [%]']:.2f}%")
print(f"総取引回数: {results['# Trades']}")

# パラメータ最適化
optimization_results = bt.optimize(
    fast_period=range(5, 25, 5),
    slow_period=range(20, 60, 10),
    maximize="Sharpe Ratio",
    constraint=lambda p: p.fast_period < p.slow_period,
)
print(f"\n最適パラメータ: fast={optimization_results._strategy.fast_period}, slow={optimization_results._strategy.slow_period}")

class RSIMeanReversion(Strategy):
    """RSI平均回帰戦略
    - RSIが売られすぎゾーン（30以下）に入ったら買い
    - RSIが買われすぎゾーン（70以上）に入ったら売り
    """
    rsi_period = 14
    rsi_oversold = 30
    rsi_overbought = 70

    def init(self):
        close = pd.Series(self.data.Close)
        delta = close.diff()
        gain = delta.where(delta > 0, 0.0)
        loss = (-delta).where(delta < 0, 0.0)

        avg_gain = gain.rolling(window=self.rsi_period).mean()
        avg_loss = loss.rolling(window=self.rsi_period).mean()

        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))

        self.rsi = self.I(lambda: rsi, name="RSI")

    def next(self):
        if self.rsi[-1] < self.rsi_oversold:
            if not self.position:
                self.buy()
        elif self.rsi[-1] > self.rsi_overbought:
            if self.position:
                self.position.close()

# バックテスト実行
bt_rsi = Backtest(
    data,
    RSIMeanReversion,
    cash=100000,
    commission=0.001,
    exclusive_orders=True,
)

results_rsi = bt_rsi.run()
print("=== RSI Mean Reversion Results ===")
print(f"トータルリターン: {results_rsi['Return [%]']:.2f}%")
print(f"シャープレシオ: {results_rsi['Sharpe Ratio']:.2f}")
print(f"最大ドローダウン: {results_rsi['Max. Drawdown [%]']:.2f}%")
print(f"勝率: {results_rsi['Win Rate [%]']:.2f}%")

class BollingerBandsBreakout(Strategy):
    """ボリンジャーバンドブレイクアウト戦略
    - 価格が下側バンドにタッチしたら買い（平均回帰を期待）
    - 価格が上側バンドにタッチしたら売り
    - ストップロス：エントリー価格から2%下落時に損切り
    """
    bb_period = 20
    bb_std = 2.0
    stop_loss_pct = 0.02

    def init(self):
        close = pd.Series(self.data.Close)
        self.sma = self.I(lambda: close.rolling(self.bb_period).mean(), name="SMA")
        std = close.rolling(self.bb_period).std()
        self.upper = self.I(lambda: close.rolling(self.bb_period).mean() + self.bb_std * std, name="Upper")
        self.lower = self.I(lambda: close.rolling(self.bb_period).mean() - self.bb_std * std, name="Lower")

    def next(self):
        price = self.data.Close[-1]

        # 下側バンドタッチ：買い
        if price <= self.lower[-1]:
            if not self.position:
                self.buy(sl=price * (1 - self.stop_loss_pct))

        # 上側バンドタッチ：売り
        elif price >= self.upper[-1]:
            if self.position:
                self.position.close()

# バックテスト実行
bt_bb = Backtest(
    data,
    BollingerBandsBreakout,
    cash=100000,
    commission=0.001,
    exclusive_orders=True,
)

results_bb = bt_bb.run()
print("=== Bollinger Bands Breakout Results ===")
print(f"トータルリターン: {results_bb['Return [%]']:.2f}%")
print(f"シャープレシオ: {results_bb['Sharpe Ratio']:.2f}")
print(f"最大ドローダウン: {results_bb['Max. Drawdown [%]']:.2f}%")
print(f"勝率: {results_bb['Win Rate [%]']:.2f}%")

import numpy as np
import pandas as pd
from scipy import stats

class RiskMetrics:
    """リスク管理メトリクス計算機"""

    def __init__(self, returns: pd.Series, risk_free_rate: float = 0.04):
        """
        Args:
            returns: 日次リターンシリーズ
            risk_free_rate: 無リスク金利（年率、デフォルト4%）
        """
        self.returns = returns.dropna()
        self.risk_free_rate = risk_free_rate
        self.daily_rf = (1 + risk_free_rate) ** (1/252) - 1

    def sharpe_ratio(self) -> float:
        """シャープレシオの計算"""
        excess_returns = self.returns - self.daily_rf
        if excess_returns.std() == 0:
            return 0.0
        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]
        if len(downside_returns) == 0 or downside_returns.std() == 0:
            return 0.0
        downside_std = downside_returns.std()
        return np.sqrt(252) * excess_returns.mean() / downside_std

    def maximum_drawdown(self) -> float:
        """最大ドローダウン（MDD）の計算"""
        cumulative = (1 + self.returns).cumprod()
        peak = cumulative.expanding().max()
        drawdown = (cumulative - peak) / peak
        return drawdown.min()

    def value_at_risk(self, confidence: float = 0.95) -> float:
        """VaR（Value at Risk）の計算 - ヒストリカル法"""
        return np.percentile(self.returns, (1 - confidence) * 100)

    def conditional_var(self, confidence: float = 0.95) -> float:
        """CVaR（条件付きVaR）の計算"""
        var = self.value_at_risk(confidence)
        return self.returns[self.returns <= var].mean()

    def calmar_ratio(self) -> float:
        """カルマーレシオの計算（年率リターン / MDD）"""
        annual_return = (1 + self.returns.mean()) ** 252 - 1
        mdd = abs(self.maximum_drawdown())
        if mdd == 0:
            return 0.0
        return annual_return / mdd

    def summary(self) -> dict:
        """全リスクメトリクスの要約"""
        annual_return = (1 + self.returns.mean()) ** 252 - 1
        annual_volatility = self.returns.std() * np.sqrt(252)

        return {
            "年率リターン": f"{annual_return:.2%}",
            "年率ボラティリティ": f"{annual_volatility:.2%}",
            "シャープレシオ": f"{self.sharpe_ratio():.2f}",
            "ソルティノレシオ": f"{self.sortino_ratio():.2f}",
            "最大ドローダウン（MDD）": f"{self.maximum_drawdown():.2%}",
            "VaR（95%）": f"{self.value_at_risk():.2%}",
            "CVaR（95%）": f"{self.conditional_var():.2%}",
            "カルマーレシオ": f"{self.calmar_ratio():.2f}",
            "総取引日数": len(self.returns),
            "プラスリターン日": f"{(self.returns > 0).sum()} ({(self.returns > 0).mean():.1%})",
        }

# 使用例
# SPYの日次リターン計算
spy = yf.download("SPY", start="2020-01-01", end="2025-12-31", progress=False)
daily_returns = spy["Close"].pct_change().dropna()

metrics = RiskMetrics(daily_returns.squeeze(), risk_free_rate=0.04)
summary = metrics.summary()

print("=== SPY Risk Metrics ===")
for key, value in summary.items():
    print(f"  {key}: {value}")

class PositionSizer:
    """ポジションサイジングアルゴリズム"""

    @staticmethod
    def kelly_criterion(win_rate: float, avg_win: float, avg_loss: float) -> float:
        """ケリー基準による最適ポジションサイズの計算

        Args:
            win_rate: 勝率（0〜1）
            avg_win: 平均利益率（正の値）
            avg_loss: 平均損失率（正の値）

        Returns:
            最適ベット比率（0〜1）
        """
        if avg_loss == 0:
            return 0.0

        # Kelly Formula: f = (bp - q) / b
        # b = avg_win / avg_loss（オッズ比）
        # p = win_rate, q = 1 - win_rate
        b = avg_win / avg_loss
        p = win_rate
        q = 1 - p

        kelly = (b * p - q) / b

        # 負の値ならベットしない
        return max(0.0, kelly)

    @staticmethod
    def half_kelly(win_rate: float, avg_win: float, avg_loss: float) -> float:
        """ハーフケリー：ケリー基準の半分で保守的アプローチ"""
        full_kelly = PositionSizer.kelly_criterion(win_rate, avg_win, avg_loss)
        return full_kelly / 2

    @staticmethod
    def fixed_fractional(equity: float, risk_per_trade: float,
                          entry_price: float, stop_loss_price: float) -> int:
        """固定比率（Fixed Fractional）ポジションサイジング

        Args:
            equity: 現在の資本金
            risk_per_trade: 取引あたりリスク比率（例：0.02 = 2%）
            entry_price: エントリー価格
            stop_loss_price: ストップロス価格

        Returns:
            購入する株数
        """
        risk_amount = equity * risk_per_trade
        risk_per_share = abs(entry_price - stop_loss_price)

        if risk_per_share == 0:
            return 0

        shares = int(risk_amount / risk_per_share)
        return max(0, shares)

    @staticmethod
    def volatility_based(equity: float, target_volatility: float,
                          asset_volatility: float) -> float:
        """ボラティリティベースのポジションサイジング

        Args:
            equity: 現在の資本金
            target_volatility: 目標ポートフォリオボラティリティ（年率）
            asset_volatility: 資産ボラティリティ（年率）

        Returns:
            ポジション比重（0〜1）
        """
        if asset_volatility == 0:
            return 0.0

        weight = target_volatility / asset_volatility
        return min(weight, 1.0)  # 最大100%

# 使用例
sizer = PositionSizer()

# ケリー基準の計算
win_rate = 0.55
avg_win = 0.03   # 平均3%利益
avg_loss = 0.02  # 平均2%損失

kelly = sizer.kelly_criterion(win_rate, avg_win, avg_loss)
half = sizer.half_kelly(win_rate, avg_win, avg_loss)
print(f"ケリー基準: {kelly:.2%}")
print(f"ハーフケリー: {half:.2%}")

# 固定比率ポジションサイジング
equity = 100000
entry = 150.0
stop_loss = 147.0
shares = sizer.fixed_fractional(equity, 0.02, entry, stop_loss)
print(f"購入株数: {shares}株（エントリー: {entry}、ストップロス: {stop_loss}）")

import pandas as pd
import numpy as np
from backtesting import Backtest

class WalkForwardOptimizer:
    """ウォークフォワード最適化"""

    def __init__(self, data: pd.DataFrame, strategy_class,
                 train_period: int = 252, test_period: int = 63):
        """
        Args:
            data: OHLCVデータ
            strategy_class: 戦略クラス
            train_period: トレーニング期間（取引日、デフォルト1年）
            test_period: テスト期間（取引日、デフォルト3ヶ月）
        """
        self.data = data
        self.strategy_class = strategy_class
        self.train_period = train_period
        self.test_period = test_period

    def run(self, optimization_params: dict, maximize: str = "Sharpe Ratio") -> list:
        """ウォークフォワード分析の実行"""
        results = []
        total_days = len(self.data)
        start_idx = 0

        fold = 1
        while start_idx + self.train_period + self.test_period <= total_days:
            train_end = start_idx + self.train_period
            test_end = train_end + self.test_period

            train_data = self.data.iloc[start_idx:train_end]
            test_data = self.data.iloc[train_end:test_end]

            # トレーニング期間でパラメータ最適化
            bt_train = Backtest(
                train_data, self.strategy_class,
                cash=100000, commission=0.001,
            )
            opt_result = bt_train.optimize(
                **optimization_params,
                maximize=maximize,
            )

            # 最適化パラメータの抽出
            best_params = {}
            for param_name in optimization_params:
                best_params[param_name] = getattr(opt_result._strategy, param_name)

            # テスト期間での検証
            bt_test = Backtest(
                test_data, self.strategy_class,
                cash=100000, commission=0.001,
            )
            # 最適化パラメータでテスト実行
            test_result = bt_test.run(**best_params)

            fold_result = {
                "fold": fold,
                "train_start": train_data.index[0],
                "train_end": train_data.index[-1],
                "test_start": test_data.index[0],
                "test_end": test_data.index[-1],
                "best_params": best_params,
                "train_return": opt_result["Return [%]"],
                "test_return": test_result["Return [%]"],
                "test_sharpe": test_result["Sharpe Ratio"],
                "test_mdd": test_result["Max. Drawdown [%]"],
            }
            results.append(fold_result)

            print(f"Fold {fold}: Train Return={fold_result['train_return']:.2f}%, "
                  f"Test Return={fold_result['test_return']:.2f}%, "
                  f"Params={best_params}")

            start_idx += self.test_period
            fold += 1

        return results

    def summary(self, results: list) -> dict:
        """ウォークフォワード結果の要約"""
        test_returns = [r["test_return"] for r in results]
        test_sharpes = [r["test_sharpe"] for r in results]

        return {
            "総Fold数": len(results),
            "平均テストリターン": f"{np.mean(test_returns):.2f}%",
            "テストリターン標準偏差": f"{np.std(test_returns):.2f}%",
            "プラスリターンFold比率": f"{sum(1 for r in test_returns if r > 0) / len(test_returns):.1%}",
            "平均テストシャープ": f"{np.mean(test_sharpes):.2f}",
        }

# 使用例
# wfo = WalkForwardOptimizer(data, MovingAverageCrossover)
# results = wfo.run(
#     optimization_params={
#         "fast_period": range(5, 25, 5),
#         "slow_period": range(20, 60, 10),
#     },
# )
# print(wfo.summary(results))

class OverfitDetector:
    """オーバーフィット検出器"""

    @staticmethod
    def check_overfit(train_sharpe: float, test_sharpe: float,
                       threshold: float = 0.5) -> dict:
        """オーバーフィット判定

        Args:
            train_sharpe: トレーニング期間のシャープレシオ
            test_sharpe: テスト期間のシャープレシオ
            threshold: 許容パフォーマンス低下率

        Returns:
            オーバーフィット診断結果
        """
        if train_sharpe <= 0:
            return {"is_overfit": True, "reason": "トレーニング期間のパフォーマンス自体がマイナス"}

        degradation = 1 - (test_sharpe / train_sharpe)
        is_overfit = degradation > threshold

        return {
            "is_overfit": is_overfit,
            "train_sharpe": train_sharpe,
            "test_sharpe": test_sharpe,
            "performance_degradation": f"{degradation:.1%}",
            "recommendation": (
                "オーバーフィットの疑い：パラメータ数を減らすかトレーニング期間を延長してください"
                if is_overfit
                else "許容範囲内のパフォーマンス差異"
            ),
        }

    @staticmethod
    def parameter_sensitivity(results_grid: dict) -> dict:
        """パラメータ感度分析
        最適パラメータ周辺でパフォーマンスが急激に低下する場合、オーバーフィットの可能性が高い
        """
        sharpe_values = list(results_grid.values())
        mean_sharpe = np.mean(sharpe_values)
        std_sharpe = np.std(sharpe_values)
        max_sharpe = max(sharpe_values)

        # 最適値が平均より2標準偏差以上高い場合、オーバーフィットの疑い
        is_sensitive = (max_sharpe - mean_sharpe) > 2 * std_sharpe

        return {
            "is_sensitive": is_sensitive,
            "max_sharpe": max_sharpe,
            "mean_sharpe": mean_sharpe,
            "std_sharpe": std_sharpe,
            "recommendation": (
                "パラメータ感度が高い：オーバーフィットリスク"
                if is_sensitive
                else "パラメータに対して安定したパフォーマンス"
            ),
        }

def check_survivorship_bias(tickers: list, start_date: str) -> dict:
    """サバイバーシップバイアスのチェック
    現在存在する銘柄のみでバックテストすると生存者バイアスが発生

    推奨：上場廃止/合併銘柄も含まれたデータセットを使用
    """
    warnings = []

    # 現時点のインデックス構成銘柄のみでテストする場合の警告
    if all(yf.Ticker(t).info.get("marketCap", 0) > 0 for t in tickers[:5]):
        warnings.append(
            "現在上場中の銘柄のみが含まれています。"
            "過去に上場廃止や合併された銘柄が欠落しており、"
            "リターンが過大評価される可能性があります"
        )

    return {
        "ticker_count": len(tickers),
        "warnings": warnings,
        "recommendation": "ポイントインタイムデータセットの使用を推奨（例：CRSP、Sharadar）",
    }

def validate_no_lookahead(strategy_code: str) -> list:
    """ルックアヘッドバイアスチェック（静的コード分析）"""
    warnings = []

    # 未来データを参照するパターンの検査
    dangerous_patterns = [
        ("shift(-", "shift(-N)による未来データ参照を検出"),
        (".iloc[-1]", "最終行参照 - コンテキストにより未来参照の可能性あり"),
        ("resample", "リサンプリングで未来データが含まれる可能性あり"),
    ]

    for pattern, description in dangerous_patterns:
        if pattern in strategy_code:
            warnings.append(f"警告: {description} - '{pattern}' 発見")

    if not warnings:
        warnings.append("明示的なルックアヘッドバイアスパターンは検出されませんでした")

    return warnings

class RealisticBacktestConfig:
    """現実的なバックテスト設定"""

    @staticmethod
    def get_config(asset_type: str = "us_equity") -> dict:
        """資産タイプ別の現実的な取引コスト設定"""
        configs = {
            "us_equity": {
                "commission": 0.001,     # 0.1%手数料
                "slippage": 0.0005,      # 0.05%スリッページ
                "spread": 0.0001,        # 0.01%スプレッド（大型株）
                "market_impact": 0.0002, # 0.02%マーケットインパクト
            },
            "jp_equity": {
                "commission": 0.001,     # 0.1%（証券会社手数料）
                "slippage": 0.001,       # 0.1%スリッページ
                "spread": 0.0005,        # 0.05%スプレッド
                "tax": 0.0,              # 株式取引の売却益課税は別途
            },
            "crypto": {
                "commission": 0.001,     # 0.1%（メイカー手数料）
                "slippage": 0.002,       # 0.2%スリッページ
                "spread": 0.001,         # 0.1%スプレッド
            },
        }
        return configs.get(asset_type, configs["us_equity"])

    @staticmethod
    def total_cost_per_trade(config: dict) -> float:
        """取引あたりの総コスト計算"""
        return sum(config.values())

# コスト確認
for asset_type in ["us_equity", "jp_equity", "crypto"]:
    config = RealisticBacktestConfig.get_config(asset_type)
    total = RealisticBacktestConfig.total_cost_per_trade(config)
    print(f"{asset_type}: 取引あたり総コスト 約{total:.3%}")