FXのEMA（指数平滑移動平均線）の最適な期間設定は？5通貨で100通りを検証

"""15分足のEMAクロスを100パターン一括検証する。

シグナルが確定した次の15分足の始値で約定し、反対クロスで決済する。
EMAの初期値には各期間と同じ本数の終値のSMAを使用する。
標準ライブラリだけで実行できる。
"""

import csv
import time
from datetime import date, datetime, timedelta
from pathlib import Path


# 使い方:
# 1. fetch_gmo_candles.pyで、5通貨の15分足を取得します。
# 2. GMO-APIフォルダで次のコマンドを実行します。
#    python backtest_research/exponential_moving_average/run_ema_backtest.py
# 3. 結果は、このファイルと同じフォルダにあるresultsへ保存されます。

# 検証する5つの通貨ペアです。
SYMBOLS = ("USD_JPY", "EUR_JPY", "GBP_JPY", "EUR_USD", "AUD_USD")

# EMAの計算、クロス判定、売買には15分足だけを使用します。
ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)

# 既存のローソク足CSVに合わせ、ASK価格を読み込みます。
# スプレッドなどの取引コストは、このスクリプトでは考慮しません。
PRICE_TYPE = "ASK"

# バックテストで使用する期間です。
# CSV内のtimestamp_jstの日付が、この開始日から終了日までの足だけを使います。
# 開始日と終了日のローソク足も検証対象に含みます。
BACKTEST_START_DATE = date(2024, 1, 1)
BACKTEST_END_DATE = date(2026, 7, 31)

# 短期EMA10種類と長期EMA10種類の組み合わせを検証します。
# すべて「短期 < 長期」なので、10 × 10 = 100パターンです。
DEFAULT_FAST_PERIODS = (5, 10, 15, 20, 25, 30, 35, 40, 45, 50)
DEFAULT_SLOW_PERIODS = (60, 75, 100, 120, 150, 200, 250, 300, 400, 500)

# __file__は、このPythonファイル自身の場所を表します。
# どのフォルダから実行しても、同じ入力・出力先を参照できます。
SCRIPT_DIR = Path(__file__).resolve().parent
GMO_API_DIR = SCRIPT_DIR.parents[1]
DEFAULT_DATA_DIR = GMO_API_DIR / "data" / "candles"
DEFAULT_OUTPUT_DIR = SCRIPT_DIR / "results"

# EMA設定ごとの集計結果に保存する列です。
RESULT_COLUMNS = (
    "通貨ペア",
    "短期EMA期間",
    "長期EMA期間",
    "取引回数",
    "勝ち取引数",
    "負け取引数",
    "勝率（%）",
    "合計損益（pips）",
    "平均損益（pips）",
    "プロフィットファクター",
    "最大ドローダウン（pips）",
)

# 1回ごとの取引履歴に保存する列です。
TRADE_COLUMNS = (
    "取引番号",
    "エントリー日時",
    "エントリー価格",
    "決済日時",
    "決済価格",
    "決済理由",
    "保有時間（分）",
    "損益（pips）",
)


def load_candles(path):
    """fetch_gmo_candles.pyが作るOHLC CSVを読み込む。"""
    candles = []

    # CSVの日時、始値、終値だけを読み込みます。
    # 今回は高値・安値による利確や損切りを行わないため、highとlowは不要です。
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        for row in csv.DictReader(file):
            timestamp = datetime.fromisoformat(row["timestamp_jst"])

            # 指定期間より前または後のローソク足は、検証に使用しません。
            # datetimeではなくdate同士で比較するため、終了日の23:45足なども含まれます。
            if not BACKTEST_START_DATE <= timestamp.date() <= BACKTEST_END_DATE:
                continue

            candles.append(
                {
                    "open_time": timestamp,
                    "close_time": timestamp + ENTRY_TIMEFRAME_DURATION,
                    "open": float(row["open"]),
                    "close": float(row["close"]),
                }
            )
    return candles


def calculate_ema(closes, period):
    """期間本数のSMAを初期値としてEMAを未来参照なしで計算する。"""
    if period <= 0:
        raise ValueError("EMA期間は1以上である必要があります")

    # EMAをまだ計算できない場所にはNoneを入れます。
    # 例えばEMA3なら、最初の2本はNoneになります。
    values = [None] * len(closes)
    if len(closes) < period:
        return values

    # EMAの最初の値には、同じ期間の単純移動平均（SMA）を使います。
    # 終値が[1, 2, 3]でEMA3なら、最初のEMAは(1 + 2 + 3) / 3 = 2です。
    seed_index = period - 1
    seed = sum(closes[:period]) / period
    values[seed_index] = seed

    # EMAは直近価格へ大きな重みを置く移動平均です。
    # EMA3ならalphaは2 / (3 + 1) = 0.5になります。
    alpha = 2.0 / (period + 1.0)
    previous_ema = seed
    for index in range(period, len(closes)):
        # 現在EMA = alpha × 現在終値 + (1 - alpha) × 前回EMA
        # 現在までに確定した終値だけを使うため、未来の価格は参照しません。
        previous_ema = alpha * closes[index] + (1.0 - alpha) * previous_ema
        values[index] = previous_ema
    return values


def crossover(previous_fast, previous_slow, fast, slow):
    """ゴールデンクロスなら1、デッドクロスなら-1、それ以外は0。"""
    # 短期EMAが長期EMAを下から上へ抜けた状態です。
    if previous_fast <= previous_slow and fast > slow:
        return 1

    # 短期EMAが長期EMAを上から下へ抜けた状態です。
    if previous_fast >= previous_slow and fast < slow:
        return -1
    return 0


def collect_trades(
    candles,
    fast_ema,
    slow_ema,
    side,
    pip_size,
    evaluation_start_index,
):
    """片側戦略を検証し、1回ごとの取引履歴を返す。"""
    if side not in {"LONG", "SHORT"}:
        raise ValueError("sideはLONGまたはSHORTである必要があります")

    # LONGはゴールデンクロスで入り、デッドクロスで出ます。
    # SHORTはデッドクロスで入り、ゴールデンクロスで出ます。
    entry_signal = 1 if side == "LONG" else -1
    exit_signal = -entry_signal

    # entry_priceがNoneなら、ポジションを持っていない状態です。
    # pendingは、前の足で確定した注文を次の足で実行するための印です。
    entry_price = None
    entry_time = None
    pending_entry = False
    pending_exit = False
    trades = []

    # 全EMA設定を公平に比べるため、EMA500を計算できる共通位置から始めます。
    for index in range(max(1, evaluation_start_index), len(candles)):
        # 前足で確定した注文だけを現在足の始値で約定する。
        if entry_price is not None and pending_exit:
            direction = 1.0 if side == "LONG" else -1.0
            exit_price = candles[index]["open"]
            trades.append(
                make_trade(
                    len(trades) + 1,
                    entry_time,
                    entry_price,
                    candles[index]["open_time"],
                    exit_price,
                    "反対クロス",
                    direction,
                    pip_size,
                )
            )
            entry_price = None
            entry_time = None
        pending_exit = False

        # 前の15分足でエントリー条件が確定していれば、現在足の始値で入ります。
        # クロスした足の終値で約定させないことで、未来参照を防ぎます。
        if entry_price is None and pending_entry:
            entry_price = candles[index]["open"]
            entry_time = candles[index]["open_time"]
        pending_entry = False

        # 前の足と現在足の短期・長期EMAを比較してクロスを判定します。
        previous_fast = fast_ema[index - 1]
        previous_slow = slow_ema[index - 1]
        current_fast = fast_ema[index]
        current_slow = slow_ema[index]
        if None in (previous_fast, previous_slow, current_fast, current_slow):
            continue

        signal = crossover(
            float(previous_fast),
            float(previous_slow),
            float(current_fast),
            float(current_slow),
        )

        # ここでは注文を即時約定せず、次のループで次足始値を使います。
        if entry_price is None and signal == entry_signal:
            pending_entry = True
        elif entry_price is not None and signal == exit_signal:
            pending_exit = True

    # データ終了時にポジションが残っていれば、最終足の終値で決済します。
    if entry_price is not None:
        direction = 1.0 if side == "LONG" else -1.0
        trades.append(
            make_trade(
                len(trades) + 1,
                entry_time,
                entry_price,
                candles[-1]["close_time"],
                candles[-1]["close"],
                "データ終了",
                direction,
                pip_size,
            )
        )
    return trades


def make_trade(
    number,
    entry_time,
    entry_price,
    exit_time,
    exit_price,
    reason,
    direction,
    pip_size,
):
    """取引履歴の1行を作成する。"""
    # LONGでは値上がり、SHORTでは値下がりしたときにpipsがプラスになります。
    # USDJPYなどの円通貨は1pips=0.01円、それ以外は1pips=0.0001です。
    return {
        "取引番号": number,
        "エントリー日時": entry_time.isoformat(),
        "エントリー価格": round(entry_price, 6),
        "決済日時": exit_time.isoformat(),
        "決済価格": round(exit_price, 6),
        "決済理由": reason,
        # 足の本数 × 15分ではなく、日時の差から実際の経過時間を求めます。
        # これにより、土日やデータ欠損を挟んだ時間も保有時間に含まれます。
        "保有時間（分）": round(
            (exit_time - entry_time).total_seconds() / 60, 2
        ),
        "損益（pips）": round(
            direction * (exit_price - entry_price) / pip_size, 2
        ),
    }


def summarize(symbol, fast_period, slow_period, trades):
    """取引ごとのpips損益から成績をまとめる。"""
    profits = [trade["損益（pips）"] for trade in trades]

    # 0pipsの取引は勝ちにも負けにも数えませんが、取引回数には含めます。
    wins = [profit for profit in profits if profit > 0]
    losses = [profit for profit in profits if profit < 0]
    gross_profit = sum(wins)
    gross_loss = abs(sum(losses))
    # プロフィットファクター（PF）= 勝ち取引の総利益 / 負け取引の総損失です。
    # PFが1より大きければ、コストを考慮しない損益合計はプラスです。
    profit_factor = gross_profit / gross_loss if gross_loss else None

    # 累積損益の過去最高値から、どこまで下落したかを順番に調べます。
    # その最大値が最大ドローダウンです。
    equity = peak = max_drawdown = 0.0
    for profit in profits:
        equity += profit
        peak = max(peak, equity)
        max_drawdown = max(max_drawdown, peak - equity)

    trade_count = len(profits)
    return {
        "通貨ペア": symbol,
        "短期EMA期間": fast_period,
        "長期EMA期間": slow_period,
        "取引回数": trade_count,
        "勝ち取引数": len(wins),
        "負け取引数": len(losses),
        "勝率（%）": round(len(wins) / trade_count * 100, 2) if trade_count else 0.0,
        "合計損益（pips）": round(sum(profits), 2),
        "平均損益（pips）": round(sum(profits) / trade_count, 2) if trade_count else 0.0,
        "プロフィットファクター": round(profit_factor, 4) if profit_factor is not None else "",
        "最大ドローダウン（pips）": round(max_drawdown, 2),
    }


def run_grid(candles, symbol, fast_periods, slow_periods, output_dir):
    """1通貨についてEMA100パターンをLONG・SHORT別に検証する。"""
    if not candles:
        raise ValueError(f"{symbol}のローソク足データが空です")

    # 同じ期間のEMAを何度も計算しないよう、必要な期間を先にまとめます。
    all_periods = set(fast_periods) | set(slow_periods)
    closes = [candle["close"] for candle in candles]
    ema_by_period = {
        period: calculate_ema(closes, period) for period in all_periods
    }

    # EMA500の初期値はindex=499で確定する。前足と現在足を使うため
    # index=500から全設定を共通条件で評価する。
    evaluation_start_index = max(all_periods)
    # 円を決済通貨にする通貨ペアと、それ以外で1pipsの値が異なります。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001

    # LONGとSHORTは別フォルダ、別result.csvへ保存します。
    for side in ("LONG", "SHORT"):
        result_path = output_dir / symbol / side / "result.csv"
        result_path.parent.mkdir(parents=True, exist_ok=True)
        with result_path.open("w", encoding="utf-8-sig", newline="") as file:
            writer = csv.DictWriter(file, fieldnames=RESULT_COLUMNS)
            writer.writeheader()
            # 1設定ずつ検証・保存するため、全取引履歴を同時に保持しません。
            for fast_period in fast_periods:
                for slow_period in slow_periods:
                    if fast_period >= slow_period:
                        continue
                    trades = collect_trades(
                        candles,
                        ema_by_period[fast_period],
                        ema_by_period[slow_period],
                        side,
                        pip_size,
                        evaluation_start_index,
                    )
                    trade_path = (
                        output_dir
                        / symbol
                        / side
                        / "trades"
                        / f"EMA_{fast_period}_{slow_period}.csv"
                    )
                    write_csv(trade_path, trades, TRADE_COLUMNS)
                    writer.writerow(
                        summarize(symbol, fast_period, slow_period, trades)
                    )


def write_csv(path, rows, columns):
    """UTF-8 BOM付きCSVを保存する。"""
    # フォルダが存在しない初回実行時は、自動的に作成します。
    path.parent.mkdir(parents=True, exist_ok=True)

    # utf-8-sigはUTF-8にBOMを付け、Excelでの文字化けを起きにくくします。
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=columns)
        writer.writeheader()
        writer.writerows(rows)


def main():
    """5通貨のEMAクロス検証を実行する。"""
    # スクリプト全体の実行時間を測定します。
    start_time = time.perf_counter()

    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")

    # 短期EMAが長期EMAより短い、有効な組み合わせ数を数えます。
    combinations = sum(
        fast < slow for fast in DEFAULT_FAST_PERIODS for slow in DEFAULT_SLOW_PERIODS
    )
    if combinations == 0:
        raise SystemExit("短期期間 < 長期期間となる組み合わせがありません")

    print(f"有効なEMA組み合わせ: {combinations}パターン")
    print(
        "検証期間: "
        f"{BACKTEST_START_DATE.isoformat()} ～ {BACKTEST_END_DATE.isoformat()}"
    )
    # 5通貨を1つずつ読み込み、各100パターンを検証します。
    for symbol in SYMBOLS:
        path = DEFAULT_DATA_DIR / symbol / f"{ENTRY_TIMEFRAME}_{PRICE_TYPE}.csv"
        if not path.is_file():
            raise SystemExit(f"入力CSVが見つかりません: {path}")
        candles = load_candles(path)
        if len(candles) <= max(DEFAULT_SLOW_PERIODS):
            raise SystemExit(
                f"{symbol}のデータが不足しています: "
                f"{len(candles)}本（{max(DEFAULT_SLOW_PERIODS) + 1}本以上必要）"
            )
        print(f"[RUN] {symbol}: {ENTRY_TIMEFRAME}")
        run_grid(
            candles,
            symbol,
            DEFAULT_FAST_PERIODS,
            DEFAULT_SLOW_PERIODS,
            DEFAULT_OUTPUT_DIR,
        )

    elapsed_seconds = time.perf_counter() - start_time
    print(f"完了: {DEFAULT_OUTPUT_DIR}")
    print(f"実行時間: {elapsed_seconds:.2f}秒")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())