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

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

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}")