【収益評価】自作ロジックの「期待値」を算出し戦略の優位性を測る

自動化・運用

※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。

バックテストを回して勝率やプロフィットファクターを確認する段階まで進んでいるなら、次に把握すべき指標は「期待値(Expectancy)」です。

期待値とは、1トレードあたりの平均的な損益見込みを示す数値です。この値がプラスでなければ、どれだけトレードを繰り返しても資産は増えません。

しかし、勝率が高いだけで「勝てる戦略」と判断してしまう落とし穴があります。勝率80%でも1回の負けが大きければ、トータルで資金は減り続けます。

原因は、勝率と損益比率(リスクリワード)を統合して評価する指標を持っていないことです。期待値はこの2つを1つの数値に集約し、戦略の優位性を一発で判定できます。

本記事では、トレード履歴から期待値を算出するPythonコードと、モンテカルロシミュレーション(Monte Carlo Simulation)で将来の資産推移を確率的に可視化するコードを提示します。

コードはすべてコピペで動く構成です。自分の売買ロジックの出力をCSVで渡すだけで期待値分析が完了するので、ぜひ実戦に組み込んでください。

期待値の計算式と戦略評価における位置づけ

期待値の数学的定義

トレードにおける期待値(Expectancy)は以下の式で算出します。

期待値 = 勝率 × 平均利益 + 敗率 × 平均損失

ここで平均損失はマイナス値なので、実質的には「平均利益の貢献 − 平均損失の貢献」という引き算です。この値が0より大きければ、トレードを繰り返すほど資産が増える正の優位性(Positive Edge)を持つと判断できます。

勝率だけでは判断できない理由

勝率(Win Rate)90%でも、1回の負けが利益9回分を吹き飛ばす戦略なら期待値はゼロ以下です。逆に勝率30%でも、1回の勝ちが負け3回分を上回れば期待値はプラスになります。

期待値は勝率と損益比率(Risk Reward Ratio)を統合する唯一の指標です。戦略を「続ける価値があるか」を判断する最重要の数値として位置づけてください。

期待値がプラスでも安心できない場合

期待値がプラスであっても、分散(バラつき)が大きい戦略は途中の最大ドローダウン(Maximum Drawdown)で資金が尽きるリスクがあります。

そこでモンテカルロシミュレーションを使い、「期待値プラスの戦略を1,000回繰り返したらどうなるか」を確率分布で確認します。これが本記事の後半で扱う発展パートです。

【コピペOK】期待値算出バックテストコード

まず必要なライブラリをインストールしてください。


pip install pandas numpy yfinance matplotlib

以下がメインコードです。


import numpy as np
import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt
from typing import Tuple, Dict

# ==============================
# 設定エリア
# ==============================
TICKER: str = "7203.T"                # 銘柄コード(トヨタ自動車)
START_DATE: str = "2019-01-01"        # データ取得開始日
END_DATE: str = "2024-12-31"          # データ取得終了日
SHORT_WINDOW: int = 5                 # 短期移動平均の期間
LONG_WINDOW: int = 25                # 長期移動平均の期間
INITIAL_CAPITAL: float = 1_000_000.0  # 初期資金(円)
TRADE_UNIT: int = 100                 # 売買単位(株)
COMMISSION_PCT: float = 0.05          # 手数料率(%):片道

# ==============================
# データ取得
# ==============================
def fetch_ohlcv(ticker: str, start: str, end: str) -> pd.DataFrame:
    '指定銘柄のOHLCVデータを取得する'
    df: pd.DataFrame = yf.download(ticker, start=start, end=end, auto_adjust=True)
    df = df.droplevel("Ticker", axis=1) if isinstance(df.columns, pd.MultiIndex) else df
    df.columns = [c.lower() for c in df.columns]
    return df

# ==============================
# シグナル生成
# ==============================
def generate_signals(df: pd.DataFrame, short_w: int, long_w: int) -> pd.DataFrame:
    'ゴールデンクロス・デッドクロスでシグナルを生成する'
    df = df.copy()
    df["sma_short"] = df["close"].rolling(window=short_w).mean()
    df["sma_long"] = df["close"].rolling(window=long_w).mean()
    df["signal"] = 0
    df.loc[df["sma_short"] > df["sma_long"], "signal"] = 1
    df.loc[df["sma_short"] <= df["sma_long"], "signal"] = -1
    df["position"] = df["signal"].diff()
    return df.dropna()

# ==============================
# バックテスト実行(トレードログ生成)
# ==============================
def run_backtest(
    df: pd.DataFrame,
    initial_capital: float,
    commission_pct: float,
    trade_unit: int,
) -> pd.DataFrame:
    'シグナルに基づきトレードを実行し、損益ログを返す'
    cash: float = initial_capital
    shares: int = 0
    entry_price: float = 0.0
    trades: list = []

    for i, row in df.iterrows():
        price: float = row["close"]

        # --- 買いエントリー ---
        if row["position"] == 2 and shares == 0:
            buy_shares: int = (int(cash / (price * trade_unit))) * trade_unit
            if buy_shares >= trade_unit:
                cost: float = price * buy_shares
                fee: float = cost * (commission_pct / 100.0)
                cash -= (cost + fee)
                shares = buy_shares
                entry_price = price

        # --- 売りエグジット ---
        elif row["position"] == -2 and shares > 0:
            revenue: float = price * shares
            fee: float = revenue * (commission_pct / 100.0)
            cash += (revenue - fee)
            pnl: float = (price - entry_price) * shares
            pnl_pct: float = (price - entry_price) / entry_price * 100.0
            total_fee: float = (entry_price * shares + revenue) * (commission_pct / 100.0)
            net_pnl: float = pnl - total_fee

            trades.append({
                "entry_date": ',
                "exit_date": i,
                "entry_price": entry_price,
                "exit_price": price,
                "shares": shares,
                "gross_pnl": pnl,
                "commission": total_fee,
                "net_pnl": net_pnl,
                "net_pnl_pct": pnl_pct - (commission_pct * 2 / 100.0 * 100),
            })
            shares = 0

    return pd.DataFrame(trades)

# ==============================
# 期待値の算出
# ==============================
def calc_expectancy(df_trades: pd.DataFrame) -> Dict[str, float]:
    'トレードログから期待値と関連指標を算出する'
    if len(df_trades) == 0:
        return {}

    wins: pd.Series = df_trades.loc[df_trades["net_pnl"] > 0, "net_pnl"]
    losses: pd.Series = df_trades.loc[df_trades["net_pnl"] <= 0, "net_pnl"]

    total_trades: int = len(df_trades)
    win_count: int = len(wins)
    loss_count: int = len(losses)

    win_rate: float = win_count / total_trades
    loss_rate: float = loss_count / total_trades
    avg_win: float = wins.mean() if win_count > 0 else 0.0
    avg_loss: float = losses.mean() if loss_count > 0 else 0.0

    expectancy: float = win_rate * avg_win + loss_rate * avg_loss
    risk_reward: float = abs(avg_win / avg_loss) if avg_loss != 0 else float("inf")
    profit_factor: float = wins.sum() / abs(losses.sum()) if losses.sum() != 0 else float("inf")

    return {
        "total_trades": total_trades,
        "win_count": win_count,
        "loss_count": loss_count,
        "win_rate_pct": win_rate * 100,
        "avg_win": avg_win,
        "avg_loss": avg_loss,
        "risk_reward_ratio": risk_reward,
        "expectancy_per_trade": expectancy,
        "profit_factor": profit_factor,
        "total_net_pnl": df_trades["net_pnl"].sum(),
    }

# ==============================
# 結果表示
# ==============================
def show_expectancy_report(stats: Dict[str, float]) -> None:
    '期待値レポートをコンソールに表示する'
    if not stats:
        print("トレードが0件です。シグナル設定を見直してください。")
        return

    print("=" * 55)
    print("        期 待 値 レ ポ ー ト")
    print("=" * 55)
    print(f"  総トレード数       : {stats['total_trades']:>10} 回")
    print(f"  勝ちトレード数     : {stats['win_count']:>10} 回")
    print(f"  負けトレード数     : {stats['loss_count']:>10} 回")
    print(f"  勝率               : {stats['win_rate_pct']:>10.2f} %")
    print(f"  平均利益           : {stats['avg_win']:>10,.0f} 円")
    print(f"  平均損失           : {stats['avg_loss']:>10,.0f} 円")
    print(f"  損益比率(RR)       : {stats['risk_reward_ratio']:>10.2f}")
    print(f"  プロフィットファクター: {stats['profit_factor']:>10.2f}")
    print(f"  ─────────────────────────────────")
    print(f"  ★ 1トレード期待値 : {stats['expectancy_per_trade']:>10,.0f} 円")
    print(f"  ─────────────────────────────────")
    print(f"  累計純損益         : {stats['total_net_pnl']:>10,.0f} 円")
    print("=" * 55)

    if stats["expectancy_per_trade"] > 0:
        print("  判定: 正の優位性あり → 継続検討に値する")
    else:
        print("  判定: 優位性なし → ロジック改善が必要")

# ==============================
# 損益分布の可視化
# ==============================
def plot_pnl_distribution(df_trades: pd.DataFrame) -> None:
    'トレードごとの損益をヒストグラムで表示する'
    fig, axes = plt.subplots(1, 2, figsize=(14, 5))

    axes[0].bar(range(len(df_trades)), df_trades["net_pnl"], color=[
        "steelblue" if v > 0 else "tomato" for v in df_trades["net_pnl"]
    ])
    axes[0].set_title("Net P&L per Trade")
    axes[0].set_xlabel("Trade #")
    axes[0].set_ylabel("JPY")
    axes[0].axhline(y=0, color="black", linewidth=0.8)
    axes[0].grid(axis="y", alpha=0.3)

    axes[1].hist(df_trades["net_pnl"], bins=20, color="slategray", edgecolor="white")
    axes[1].axvline(x=df_trades["net_pnl"].mean(), color="red", linestyle="--", label="Mean")
    axes[1].set_title("P&L Distribution")
    axes[1].set_xlabel("JPY")
    axes[1].set_ylabel("Frequency")
    axes[1].legend()
    axes[1].grid(axis="y", alpha=0.3)

    plt.tight_layout()
    plt.show()

# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    data: pd.DataFrame = fetch_ohlcv(TICKER, START_DATE, END_DATE)
    signals: pd.DataFrame = generate_signals(data, SHORT_WINDOW, LONG_WINDOW)
    df_trades: pd.DataFrame = run_backtest(
        signals, INITIAL_CAPITAL, COMMISSION_PCT, TRADE_UNIT,
    )
    stats: Dict[str, float] = calc_expectancy(df_trades)
    show_expectancy_report(stats)
    if len(df_trades) > 0:
        plot_pnl_distribution(df_trades)

コードの処理フロー解説

上記のコードは、以下の6ステップで構成されています。

* ステップ1(データ取得)yfinanceで指定銘柄のOHLCVデータをダウンロードし、カラム名を小文字に統一する

* ステップ2(シグナル生成):短期・長期SMAのクロスで買い・売りシグナルを生成する。position列の2が買い、-2が売りを表す

* ステップ3(バックテスト実行):シグナルに従い仮想売買を行い、1トレードごとの粗利益・手数料・純損益を記録する

* ステップ4(期待値算出):勝ちトレードと負けトレードを分離し、勝率・平均利益・平均損失から期待値を計算する

* ステップ5(レポート表示):期待値・勝率・損益比率・プロフィットファクター(Profit Factor)をコンソールに出力する

* ステップ6(可視化):トレードごとの損益棒グラフと、損益分布のヒストグラムを描画する

generate_signals関数を自分のロジックに差し替えれば、任意の戦略の期待値を即座に測定できます。

期待値レポートの読み解き方

各指標の判断基準

レポートに出力される指標の合格ラインを以下にまとめます。

指標 合格ライン 解説
期待値(1トレードあたり) 0円超 マイナスなら即改善が必要
勝率(Win Rate) 戦略による 単体では意味がない。損益比率とセットで評価する
損益比率(Risk Reward Ratio) 1.0以上 1.0未満なら勝率60%以上が必要になる
プロフィットファクター 1.5以上 1.0〜1.2は手数料負けのリスクが高い

期待値がプラスでもプロフィットファクターが1.2未満の場合、スリッページや手数料の増加で簡単に優位性が消えます。「ギリギリプラス」の戦略は過信してはいけません。

期待値を改善する2つのアプローチ

期待値を上げる方法は大きく2つに分かれます。

1つ目は損切りラインの最適化です。平均損失を小さくすれば、敗率×平均損失の項が縮小して期待値が上がります。ただし損切りが浅すぎると勝率が下がるため、トレードオフの検証が必要です。

2つ目は利確ルールの改善です。トレーリングストップ(Trailing Stop)を導入すると、勝ちトレードの利益を伸ばしやすくなり平均利益が上がります。

【コピペOK】モンテカルロシミュレーションで将来の資産推移を確認

期待値がプラスでも、ドローダウンの深さは運に左右されます。以下のコードをメインコードの末尾に追加してください。


# ==============================
# モンテカルロシミュレーション
# ==============================
NUM_SIMULATIONS: int = 1000           # シミュレーション回数
NUM_TRADES_FUTURE: int = 200          # 将来想定トレード数

def run_monte_carlo(
    df_trades: pd.DataFrame,
    initial_capital: float,
    n_simulations: int,
    n_trades: int,
) -> np.ndarray:
    'トレード損益をランダム抽出して資産推移をシミュレーションする'
    pnl_array: np.ndarray = df_trades["net_pnl"].values
    all_curves: np.ndarray = np.zeros((n_simulations, n_trades + 1))
    all_curves[:, 0] = initial_capital

    for sim in range(n_simulations):
        sampled: np.ndarray = np.random.choice(pnl_array, size=n_trades, replace=True)
        all_curves[sim, 1:] = initial_capital + np.cumsum(sampled)

    return all_curves

def plot_monte_carlo(all_curves: np.ndarray, initial_capital: float) -> None:
    'モンテカルロシミュレーションの結果を可視化する'
    fig, ax = plt.subplots(figsize=(14, 6))

    percentile_5: np.ndarray = np.percentile(all_curves, 5, axis=0)
    percentile_50: np.ndarray = np.percentile(all_curves, 50, axis=0)
    percentile_95: np.ndarray = np.percentile(all_curves, 95, axis=0)

    for curve in all_curves[:100]:
        ax.plot(curve, color="lightgray", alpha=0.3, linewidth=0.5)

    ax.plot(percentile_5, color="tomato", linewidth=1.5, label="5th Percentile (Worst)")
    ax.plot(percentile_50, color="steelblue", linewidth=2.0, label="50th Percentile (Median)")
    ax.plot(percentile_95, color="seagreen", linewidth=1.5, label="95th Percentile (Best)")
    ax.axhline(y=initial_capital, color="black", linestyle="--", linewidth=0.8)

    ax.set_title(f"Monte Carlo Simulation ({len(all_curves)} runs)")
    ax.set_xlabel("Trade #")
    ax.set_ylabel("Equity (JPY)")
    ax.legend()
    ax.grid(alpha=0.3)
    plt.tight_layout()
    plt.show()

    final_values: np.ndarray = all_curves[:, -1]
    ruin_rate: float = np.mean(final_values < initial_capital * 0.5) * 100
    print(f"n--- モンテカルロ結果 ({len(all_curves)}回) ---")
    print(f"  最終資産 中央値 : {np.median(final_values):>12,.0f} 円")
    print(f"  最終資産 5%ile  : {np.percentile(final_values, 5):>12,.0f} 円")
    print(f"  最終資産 95%ile : {np.percentile(final_values, 95):>12,.0f} 円")
    print(f"  破産確率(50%減) : {ruin_rate:>12.2f} %")

if __name__ == "__main__":
    if len(df_trades) >= 10:
        all_curves: np.ndarray = run_monte_carlo(
            df_trades, INITIAL_CAPITAL, NUM_SIMULATIONS, NUM_TRADES_FUTURE,
        )
        plot_monte_carlo(all_curves, INITIAL_CAPITAL)
    else:
        print("トレード数が10未満のためシミュレーションをスキップしました。")

コードの処理フロー解説

上記のコードは、以下の4ステップで構成されています。

* ステップ1(損益配列の準備):バックテストで得られた各トレードの純損益をnumpy配列に変換する

* ステップ2(ランダム抽出):損益配列から復元抽出(Bootstrap)で将来200トレード分を生成し、累積和で資産推移を計算する。これを1,000回繰り返す

* ステップ3(パーセンタイル算出):1,000本の資産曲線から5パーセンタイル(最悪ケース)、50パーセンタイル(中央値)、95パーセンタイル(好調ケース)を抽出する

* ステップ4(可視化と破産確率):3本のパーセンタイル曲線をグラフに重ね、最終資産が初期資金の50%を割る確率を「破産確率」として出力する

NUM_TRADES_FUTUREを変えれば、半年先・1年先など任意の期間をシミュレーションできます。破産確率が5%を超える場合はロット(TRADE_UNIT)を下げるか戦略自体の見直しを検討してください。

よくあるエラーと対処法

ValueError: No trades to analyze(トレード数が0件)

シグナルが一度も発火していない状態です。移動平均の期間がデータ期間に対して長すぎることが主な原因です。

以下を試してください。

* SHORT_WINDOWを3〜5、LONG_WINDOWを15〜25の範囲に設定する

* START_DATEをさらに過去に広げ、最低でも3年分のデータを確保する

* signals["position"].value_counts()を出力してシグナル発生状況を確認する

期待値が極端に大きい(または小さい)値になる

トレード数が5件未満など、サンプルが少なすぎると統計的に信頼できない値が出ます。期待値の信頼性はトレード数に依存します。

最低でも30トレード以上のサンプルで評価してください。トレード数が少ない場合は、対象銘柄を複数に増やすか、バックテスト期間を延長してサンプルを確保することが必要です。

モンテカルロシミュレーションの結果が毎回変わる

乱数を使っているため結果が変動するのは正常な動作です。再現性が必要な場合は、run_monte_carlo関数の先頭にnp.random.seed(42)を追加してください。

以下を試してください。

* np.random.seed(42) を関数冒頭に1行追加する

* NUM_SIMULATIONSを1,000以上に設定し、統計的なブレを抑える

* 複数回実行して中央値が大きく変動しないことを確認する

まとめ

この記事では、Pythonでトレード戦略の期待値を算出し、モンテカルロシミュレーションで将来の資産推移を確率的に評価する方法を解説しました。

要点を整理します。

* 期待値は「勝率 × 平均利益 + 敗率 × 平均損失」で算出し、プラスでなければその戦略を続ける意味がない

* 勝率だけで戦略を評価するのは危険であり、損益比率(Risk Reward Ratio)と組み合わせて期待値で判断する

* プロフィットファクターは1.5以上を目安とし、1.0〜1.2の戦略はコスト変動で優位性が消える可能性がある

* モンテカルロシミュレーションで5パーセンタイルの資産曲線を確認し、破産確率5%以下を安全基準とする

* 期待値の信頼性はサンプル数に依存するため、最低30トレード以上で評価する

次のステップとして、期待値算出の対象を複数銘柄・複数タイムフレームに拡大し、「どの市場条件で優位性が崩れるか」を分析することを推奨します。設定エリアのTICKERをリストに変更してループ処理にすれば、ポートフォリオ全体の期待値マップを作成できます。

さらに、ケリー基準(Kelly Criterion)と組み合わせれば、期待値に基づく最適なポジションサイズを算出でき、資金管理の精度が一段上がります。

タイトルとURLをコピーしました