DQNでトレードの売買判断を行うAIの試作 〜性能測定編〜

本記事で紹介している内容は、DQN（ディープQネットワーク）を用いた日経平均トレードの技術的な解説およびシミュレーション事例であり、特定の投資行動や金融商品の購入・売却を勧誘するものではありません。
また、記載された運用成績や利回りは過去のバックテストまたはシミュレーション結果に基づいており、将来の成果を保証するものではありません。
投資には元本割れや損失が発生するリスクがあり、最終的な投資判断はご自身の責任でお願いいたします。
本記事の内容を参考にして生じたいかなる損失についても、一切の責任を負いかねます。
投資にあたっては、必ずご自身で十分な調査・ご判断のうえ、必要に応じて専門家等にご相談ください。

関連記事一覧

1. DQNでトレードの売買判断を行うAIの試作 〜データ準備編〜
2. DQNでトレードの売買判断を行うAIの試作 〜環境・トレーニング編〜
3. （本記事）DQNでトレードの売買判断を行うAIの試作 〜性能測定編〜

はじめに

前回までの記事では、AIトレーダーのデータ準備から学習プロセスまでを詳しく解説しました。27年分の株価データを使って35万ステップの学習を行い、ResNetを使った特徴抽出器でAIが複雑な市場パターンを学習できるようになりました。

しかし、「学習したAIは本当に使えるのか？」という疑問が残ります。機械学習では、学習データでの性能が良くても、未知のデータで同じ性能を発揮できるとは限りません。これを「汎化性能」と呼びます。

今回は、学習したDQNモデルの性能を客観的に評価する方法について解説します。特に、性能測定の仕組みを、初心者の方にも分かりやすく説明していきます。

なぜ性能測定が重要なのか？

投資の世界では「過去の成績は将来の成果を保証しない」という注意書きがよくついています。これは機械学習でも同じです。学習期間（1997-2024年）で優秀な成績を収めても、未来の相場で同じように利益を出せるかは別問題です。

性能測定では以下の点を確認します：

1. 汎化性能：未知のデータでも機能するか？
2. 安定性：一時的な好成績ではなく、継続的に利益を出せるか？
3. リスク管理：大きな損失を避けられるか？
4. 実用性：現実的な取引コストを考慮しても利益が出るか？

これらを客観的に評価するために、様々な指標を計算する必要があります。

性能測定の全体像

後述するeval.pyは、学習したモデルを未知のデータでテストするバックテストシステムです。実際の投資では「タイムマシン」は使えませんが、過去のデータを使って「もしその時点でAIを使っていたら」をシミュレートできます。

テスト期間の設定

start = "2023-01-01"
end = "2025-06-11"
test_data = generate_env_data(start, end, ticker="^N225")

学習期間（1997-2024年）とは異なる期間（2023-2025年）でテストすることで、AIが「見たことのない相場」でどの程度機能するかを確認します。これが汎化性能の評価です。
区間が一部学習区間とかぶっているのは、ウィンドウサイズ(130日)と移動平均線のNaNデータでドロップされる日数(75日)を考慮して設定しています。

段階的な性能評価

for i in range(10000, 350001, 10000):
    model = DQN.load(f"nikkei_cp_1997-01-01_2024-01-01_{i}_steps.zip")

1万ステップごとに保存されたモデルをすべてテストします。

• 学習の進捗：どの段階で性能が向上したか
• 過学習の検出：学習が進みすぎて性能が悪化していないか
• 最適な学習期間：どのタイミングで学習を止めるべきか

を確認できます。AIは「学習しすぎ」で性能が悪化することがあります。

決定論的な行動選択

action, _ = model.predict(obs, deterministic=True)

テスト時はdeterministic=Trueを指定して、ランダム性を排除します。学習時は探索のためにランダムな行動を取りますが、実際の運用では一貫した判断が必要だからです。

性能指標の計算

後述するcalc_performance.pyは、トレーディング戦略の性能を多角的に評価するコードを書いています。単純な「利益が出たか」だけでなく、投資の質を総合的に判断します。

基本的な収益性指標

年利（Annual Return）

annual_return = (1 + cumulative_return) ** (250 / total_days) - 1

年利は最も基本的な指標です。例えば、2年間で20%の利益が出た場合、年利は約9.5%となります。これにより、異なる期間の投資成果を公平に比較できます。

250という数字は、年間の営業日数（土日祝日を除いた取引日数）を表しています。

累積リターン

cumulative_return = (equity_curve[-1] / equity_curve[0]) - 1

投資期間全体での総リターンです。100万円が120万円になれば、累積リターンは20%となります。

リスク指標：最大ドローダウン

投資で最も恐ろしいのは「大きな損失」です。最大ドローダウンは、過去の最高値からどれだけ資産が減少したかを示します。

peak = equity_curve[0]
max_drawdown = 0
for i, value in enumerate(equity_curve):
    if value > peak:
        peak = value
    drawdown = (peak - value) / peak
    if drawdown > max_drawdown:
        max_drawdown = drawdown

例えば、資産が150万円まで増えた後に120万円まで減った場合、ドローダウンは20%です。この指標が小さいほど、安定した投資戦略と言えます。

トレード分析：勝率と損益比

AIの取引を個別に分析することで、戦略の特性を理解できます。

トレードの抽出

trades = []
position = None
for i, action in enumerate(action_history):
    if position is None:
        if action in [0, 2]:  # ロング or ショート開始
            position = action
            entry_step = i
            entry_value = equity_curve[i]
    else:
        if action != position or i == len(action_history) - 1:
            exit_step = i
            exit_value = equity_curve[min(i, len(equity_curve) - 1)]
            pnl = (exit_value / entry_value) - 1 if position == 0 else (entry_value / exit_value) - 1
            trades.append({"pnl": pnl, "holding_period": exit_step - entry_step})

この処理では、AIの行動履歴から個別のトレードを抽出します。例えば：

• 10日目に「買い」→ 15日目に「様子見」に変更 → 1つのロングトレード（5日間保有）
• 20日目に「売り」→ 25日目に「買い」に変更 → 1つのショートトレード（5日間保有）

勝率（Win Rate）

wins = [t for t in trades if t["pnl"] > 0]
losses = [t for t in trades if t["pnl"] <= 0]
win_rate = len(wins) / len(trades) if trades else 0

勝率は利益を出したトレードの割合です。ただし、勝率が高ければ良いというわけではありません。勝率30%でも、勝ちトレードの利益が負けトレードの損失を大きく上回れば、全体では利益になります。

損益比（Win/Loss Ratio）

avg_win = np.mean([t["pnl"] for t in wins]) if wins else 0
avg_loss = np.mean([t["pnl"] for t in losses]) if losses else 0
wl_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else float("inf")

損益比は、平均的な勝ちトレードの利益と平均的な負けトレードの損失の比率です。例えば、平均的な勝ちで3%の利益、平均的な負けで1%の損失なら、損益比は3.0となります。

期待値（Expectancy）

expectancy = win_rate * avg_win + (1 - win_rate) * avg_loss

期待値は、1回のトレードで期待できる平均的な利益です。これが正の値であれば、長期的には利益が期待できる戦略と言えます。

例：勝率40%、平均勝ち3%、平均負け1%の場合
期待値 = 0.4 × 3% + 0.6 × (-1%) = 1.2% - 0.6% = 0.6%

プロフィットファクター

profit_factor = sum(t["pnl"] for t in wins) / abs(sum(t["pnl"] for t in losses)) if losses else float("inf")

プロフィットファクターは、総利益と総損失の比率です。1.0を超えれば利益となります。プロの投資家は通常、1.5以上を目指します。

リスク調整後リターン：シャープレシオ

利益だけでなく、「どれだけのリスクを取って利益を得たか」も重要です。シャープレシオは、リスク1単位あたりの超過リターンを示します。

def compute_sharpe_ratio(equity_curve, yearly_risk_free_rate=0.0065, periods_per_year=252):
    daily_log_returns = np.diff(np.log(equity_curve))
    risk_free_daily = (1 + yearly_risk_free_rate) ** (1 / periods_per_year) - 1
    excess_returns = daily_log_returns - risk_free_daily
    sharpe_daily = np.mean(excess_returns) / np.std(excess_returns, ddof=1)
    sharpe_annualized = sharpe_daily * np.sqrt(periods_per_year)
    return sharpe_annualized

risk_free_dailyは年率0.65%の無リスク利率（日本の国債利回りを想定）を日次に変換します。シャープレシオは「無リスク資産を上回る超過リターン」を評価するため、この調整が必要です。

シャープレシオの解釈

• 2.0以上：非常に優秀
• 1.0-2.0：リスクに見合ったリターン
• 0.5-1.0：普通
• 0.5未満：改善が必要
• 負の値：無リスク資産より劣る

視覚化による分析：エクイティカーブとシグナル

数値だけでなく、グラフによる視覚化も重要です。eval.pyでは、資産推移と買い状態を同時に表示します。

fig, ax = plt.subplots(figsize=(12, 6))
ax.plot(steps, equity_curve, label="Equity Curve", color="blue")
ax.scatter(buy_steps, buy_values, color="green", marker="^", s=100, label="BUY Signal")

# 株価データを追加
ax2 = ax.twinx()
ax2.plot(steps, stock_prices, label="Stock Price (N225)", color="red", linestyle="dashed", alpha=0.7)

このグラフからエクイティカーブの形状が右肩上がりであれば、継続的な利益成長をできていると確認できます。
AIの資産推移が青線で、株価が赤線です。

売買タイミングの妥当性

緑の三角マーカー（買い状態）が、株価の底値付近に集中していれば、AIが適切なタイミングで買い判断を行っていることが分かります。
株価が下落しているのにAIの資産が増加していれば、ショート戦略が機能していることになります。

最終行動の分析

final_action_counts = Counter(final_actions)
count_0 = final_action_counts.get(0, 0)  # 買い
count_1 = final_action_counts.get(1, 0)  # 様子見
count_2 = final_action_counts.get(2, 0)  # 売り

各学習段階でのモデルが、テスト期間の最終日にどの行動を選択したかを集計します。最終的に成績の良かったモデルだけを集めて、多数決をとるのに役立ちます。

現実的な制約の考慮

バックテストでは、現実の取引で発生する様々な制約を考慮する必要があります。

取引コスト

transaction_cost=0.001  # 0.1%の手数料

現実の取引では手数料がかかります。頻繁な取引を行うAIの場合、手数料が利益を大きく削る可能性があります。

スリッページ

実際の取引では、注文価格と約定価格に差が生じることがあります。現在のシステムでは考慮していません。

流動性制約

大きな資金を運用する場合、市場への影響を考慮する必要があります。個人投資家レベルでは問題になりませんが、機関投資家レベルでは重要です。

性能評価の実例

実際の性能評価結果の例を見てみましょう。良かったモデルを一つピックアップしてみました。赤線が株価で青線が資産推移です。緑のマークは買いで入っている期間を表しています。

このモデルのパフォーマンス指標は以下のようになります。

• アクション[0:買,1:待,2:売]:0, ステップ:373, 累積リワード:1.1794, 資産:1866872, リターン:0.0, トレード回数:31, 明日: 38278,株価:38278
• 年利: 51.95%
• 年間シャープレシオ: 2.1699
• 最大ドローダウン: 9.17%
• 最大ドローダウン期間: 194日目から198日目
• 勝率: 80.00%
• 平均勝ち%: 3.6138%
• 平均負け%: -7.1540%
• W/Lレシオ: 0.51
• 期待値: 1.4602%
• プロフィットファクター: 2.02
• 取引日数: 373
• 平均勝ち期間: 15.0
• 平均負け期間: 12.0

良い点

• 年利51.95%：日経平均のリターンを上回る
• シャープレシオ2.169：リスク調整後でも優秀
• 最大ドローダウン9.17%：許容範囲内のリスク
• 勝率80.00%：勝率も高い
• 期待値1.4602%：1回のトレードで平均1.46%の利益期待

注意点

• W/Lレシオ0.51：負けトレードで失う%が勝ちトレードの利益の約2倍

総合評価

プロフィットファクター2.02は、プロの投資家が目指す1.5を上回り、実用的な戦略と評価できます。

まとめ：性能測定の重要性と限界

今回は、DQNを用いた株価テクニカル分析AIの性能測定について詳しく解説しました。重要なポイントをまとめると：

主要な評価指標

1. 収益性：年利、累積リターン
2. リスク：最大ドローダウン、シャープレシオ
3. 安定性：勝率、期待値、プロフィットファクター
4. 効率性：取引頻度、保有期間

限界と注意点

1. 過去データの限界：未来の相場は過去と異なる可能性
2. ブラックスワン：予期しない大きな市場変動への対応
3. 心理的要因：実際の運用では感情的な判断が介入
4. 技術的制約：スリッページ、流動性などの現実的制約

AIトレーダーの性能測定は、単なる数値の計算ではありません。投資戦略の本質を理解し、リスクを適切に管理するための重要なプロセスです。どんなに優秀なAIでも、適切な評価と継続的な監視なしには実用化できません。

機械学習による投資は魅力的ですが、「AIが全てを解決してくれる」という幻想は禁物です。適切な性能測定により、AIの能力と限界を正しく理解し、人間の判断と組み合わせることで、より良い投資成果を目指すことができるでしょう。

コードの全体

calc_performance.py

import numpy as np

def calculate_performance_metrics(equity_curve, action_history):
    """
    トレーディングモデルのパフォーマンス指標を計算

    Parameters:
    - equity_curve (list): 資産残高の推移
    - action_history (list): 各ステップでのアクション（0:ロング, 1:フラット, 2:ショート）

    Returns:
    dict: 各種パフォーマンス指標
    """
    daily_returns = []
    for i in range(1, len(equity_curve)):
        r = (equity_curve[i] / equity_curve[i - 1]) - 1
        daily_returns.append(r)

    # ───────────────────────────────
    # トレード単位に分割するためのロジック
    trades = []
    position = None
    entry_step = None
    entry_value = None

    for i, action in enumerate(action_history):
        if position is None:
            if action in [0, 2]:  # ロング or ショート開始
                position = action
                entry_step = i
                entry_value = equity_curve[i]
        else:
            # ポジション変更 or 最後まで保有
            if action != position or i == len(action_history) - 1:
                exit_step = i
                exit_value = equity_curve[min(i, len(equity_curve) - 1)]
                pnl = (
                    (exit_value / entry_value) - 1
                    if position == 0
                    else (entry_value / exit_value) - 1
                )
                trades.append({"pnl": pnl, "holding_period": exit_step - entry_step})
                # 次のポジションを開始
                if action in [0, 2]:
                    position = action
                    entry_step = i
                    entry_value = equity_curve[i]
                else:
                    position = None
                    entry_step = None
                    entry_value = None

    # 勝ちトレードと負けトレードを分ける
    wins = [t for t in trades if t["pnl"] > 0]
    losses = [t for t in trades if t["pnl"] <= 0]

    # ───────────────────────────────
    # 各種指標の計算
    total_days = len(daily_returns)
    cumulative_return = (equity_curve[-1] / equity_curve[0]) - 1
    annual_return = (
        (1 + cumulative_return) ** (250 / total_days) - 1 if total_days > 0 else 0
    )

    # 最大ドローダウン
    peak = equity_curve[0]
    max_drawdown = 0
    max_dd_start = 0
    max_dd_end = 0
    current_dd_start = 0
    for i, value in enumerate(equity_curve):
        if value > peak:
            peak = value
            current_dd_start = i
        drawdown = (peak - value) / peak
        if drawdown > max_drawdown:
            max_drawdown = drawdown
            max_dd_start = current_dd_start
            max_dd_end = i

    # 勝率、平均勝ち・負け%
    win_rate = len(wins) / len(trades) if trades else 0
    avg_win = np.mean([t["pnl"] for t in wins]) if wins else 0
    avg_loss = np.mean([t["pnl"] for t in losses]) if losses else 0
    wl_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else float("inf")
    expectancy = win_rate * avg_win + (1 - win_rate) * avg_loss
    profit_factor = (
        sum(t["pnl"] for t in wins) / abs(sum(t["pnl"] for t in losses))
        if losses
        else float("inf")
    )

    # 追加: 勝ち/負けトレードの平均保持期間
    avg_win_holding = np.mean([t["holding_period"] for t in wins]) if wins else 0
    avg_loss_holding = np.mean([t["holding_period"] for t in losses]) if losses else 0

    return {
        "annual_return": annual_return * 100,
        "max_drawdown": max_drawdown * 100,
        "max_drawdown_period": f"{max_dd_start}日目から{max_dd_end}日目",
        "win_rate": win_rate * 100,
        "avg_win": avg_win * 100,
        "avg_loss": avg_loss * 100,
        "wl_ratio": wl_ratio,
        "expectancy": expectancy * 100,
        "profit_factor": profit_factor,
        "total_days": total_days,
        "avg_win_holding_period": avg_win_holding,
        "avg_loss_holding_period": avg_loss_holding,
        "total_trades": len(trades),
    }


def compute_sharpe_ratio(equity_curve, yearly_risk_free_rate=0.0065, periods_per_year=252):
    """
    equity_curve            : 資産残高のリストまたはnumpy配列（各時点の口座残高）
    yearly_risk_free_rate   : 年率の無リスク利率（例: 0.01 → 1%）
    periods_per_year        : 年間取引日数（例: 252）
    """
    # equity_curve をnumpy配列に変換
    equity_curve = np.array(equity_curve)
    # 各日の対数リターンの計算
    daily_log_returns = np.diff(np.log(equity_curve))

    # 年率の無リスク利率を日次に変換
    risk_free_daily = (1 + yearly_risk_free_rate) ** (1 / periods_per_year) - 1

    # 超過リターン
    excess_returns = daily_log_returns - risk_free_daily

    # 日次シャープレシオ ※分母が0にならないよう注意
    sharpe_daily = (
        np.mean(excess_returns) / np.std(excess_returns, ddof=1)
        if np.std(excess_returns, ddof=1) != 0
        else 0
    )

    # 年率シャープレシオに換算
    sharpe_annualized = sharpe_daily * np.sqrt(periods_per_year)

    return sharpe_annualized

eval.py

import matplotlib.pyplot as plt
import pandas as pd
from main import NikkeiEnv
from data import generate_env_data
from collections import Counter
from stable_baselines3 import DQN
from calc_performance import calculate_performance_metrics, compute_sharpe_ratio


# 例: 手動でデータを追加する
manual_data = {
    "Date": ["2025-06-11"],
    "Open": [38420],
    "High": [38420],
    "Low": [38420],
    "Close": [38420],
    "Volume": [100000000],
    "VIX": [16.98],
    "Japan_10Y_Rate": [1.31],
    "US_10Y_Rate": [4.33],
}

# DataFrame に変換
manual_data = pd.DataFrame(manual_data)
manual_data.set_index("Date", inplace=True)

# ──────────────────────────────
# バックテスト（テストデータ上で方策を実行）
start = "2023-01-01"
end = "2025-06-11"
test_data = generate_env_data(start, end, ticker="^N225")  # 日経平均
# test_data = generate_env_data(start, end, ticker="^N225", manual_data=manual_data)
window_size = 130
# バックテスト用（評価用）環境：通常の環境オブジェクトを利用
test_env = NikkeiEnv(
    test_data,
    window_size=window_size,
    transaction_cost=0.001,
    risk_limit=0.5,
    trade_penalty=0.00,
)

# 最後のアクションを保存するリスト
final_actions = []
for i in range(10000, 350001, 10000):
    obs = test_env.reset()
    done = False
    action_history = []

    num_steps = i
    model = DQN.load(
        f"nikkei_cp_1997-01-01_2024-01-01_{num_steps}_steps.zip",
        env=test_env,
    )
    print("## Step", num_steps)
    while not done:
        # 決定論的に行動を選択
        action, _ = model.predict(obs, deterministic=True)
        action_history.append(action)
        obs, reward, done, info = test_env.step(action)
        if done:  # エピソード終了時のアクションを保存
            final_actions.append(int(action))

    # テスト期間中のエクイティカーブを取得
    equity_curve = test_env.get_equity_curve()
    sharpe = compute_sharpe_ratio(equity_curve, yearly_risk_free_rate=0.0065)

    # パフォーマンス指標の計算と表示

    metrics = calculate_performance_metrics(equity_curve, action_history)
    print("=== パフォーマンス指標 ===")
    print(f"年利: {metrics['annual_return']:.2f}%")
    print("年間シャープレシオ:", sharpe)
    print(f"最大ドローダウン: {metrics['max_drawdown']:.2f}%")
    print(f"最大ドローダウン期間: {metrics['max_drawdown_period']}")
    print(f"勝率: {metrics['win_rate']:.2f}%")
    print(f"平均勝ち%: {metrics['avg_win']:.4f}%")
    print(f"平均負け%: {metrics['avg_loss']:.4f}%")
    print(f"W/Lレシオ: {metrics['wl_ratio']:.2f}")
    print(f"期待値: {metrics['expectancy']:.4f}%")
    print(f"プロフィットファクター: {metrics['profit_factor']:.2f}")
    print(f"取引日数: {metrics['total_days']}")
    print(f"平均勝ち期間: {metrics['avg_win_holding_period']}")
    print(f"平均負け期間: {metrics['avg_loss_holding_period']}")
    buy_steps = [i for i, a in enumerate(action_history) if a == 0]
    buy_values = [equity_curve[i] for i in buy_steps]
    steps = range(len(equity_curve))

    fig, ax = plt.subplots(figsize=(12, 6))
    ax.plot(steps, equity_curve, label="Equity Curve", color="blue")
    ax.scatter(
        buy_steps, buy_values, color="green", marker="^", s=100, label="BUY Signal"
    )
    ax.set_xlabel("Step (Day)")
    ax.set_ylabel("Asset Balance")
    ax.set_title("Equity Curve with Buy Signals")
    ax.tick_params(axis="y", labelcolor="blue")

    # 株価データを追加（赤）
    stock_prices = test_data["Open"].values
    offset = len(stock_prices) - len(equity_curve)
    stock_prices = stock_prices[
        offset:
    ]  # offsetにはwindow_size, 75のdrop_na区間、その他欠損データが含まれる
    ax2 = ax.twinx()
    ax2.plot(
        steps,
        stock_prices,
        label="Stock Price (N225)",
        color="red",
        linestyle="dashed",
        alpha=0.7,
    )
    ax2.set_ylabel("Stock Price", color="red")
    ax2.tick_params(axis="y", labelcolor="red")

    fig.suptitle("Equity Curve & Stock Price with Buy Signals")
    ax.legend(loc="upper left")
    ax2.legend(loc="upper right")
    plt.savefig(
        f"equity_curve{num_steps}_{start}_{end}.png", dpi=300, bbox_inches="tight"
    )
    plt.show()

# 最後のアクションの出現回数をカウント
final_action_counts = Counter(final_actions)

# 0, 1, 2 の出現回数を取得（存在しない場合は0）
count_0 = final_action_counts.get(0, 0)
count_1 = final_action_counts.get(1, 0)
count_2 = final_action_counts.get(2, 0)

# 結果を表示
print(f"{end} 買い:{count_0}, 待ち:{count_1}, 売り:{count_2}")