FXの移動平均線の最適な期間設定は？5通貨で100通りのSMAを検証

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

15分足のシグナルが確定した次の足の始値で約定し、反対クロスで決済する。
標準ライブラリだけで実行できる。
"""

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

# 使い方:
# 1. fetch_gmo_candles.pyで、5通貨の15分足を取得します。
# 2. GMO-APIフォルダで次のコマンドを実行します。
#    python backtest_research/moving_average/run_ma_backtest.py
# 3. 結果は backtest_research/moving_average/results に通貨別で保存されます。

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

# 売買の判断には15分足だけを使います。
ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)
PRICE_TYPE = "ASK"

# 15分足で検証する短期SMAと長期SMAです。
# すべての短期期間がすべての長期期間より短いため、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)

# 入力CSVと検証結果の保存場所です。
# __file__を基準にするため、どのフォルダからコマンドを実行しても同じ場所を参照します。
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"

# 結果CSVへ出力する列と、その並び順です。
RESULT_COLUMNS = (
    "通貨ペア",
    "短期SMA期間",
    "長期SMA期間",
    "取引回数",
    "勝ち取引数",
    "負け取引数",
    "勝率（%）",
    "合計損益（pips）",
    "平均損益（pips）",
    "プロフィットファクター",
    "最大ドローダウン（pips）",
)

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


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

    # 始値と終値をバックテストで使いやすい数値に変換します。
    # 高値と安値は今回のクロス決済では使わないため読み込みません。
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        for row in csv.DictReader(file):
            timestamp = datetime.fromisoformat(row["timestamp_jst"])
            candles.append(
                {
                    "open_time": timestamp,
                    "close_time": timestamp + ENTRY_TIMEFRAME_DURATION,
                    "open": float(row["open"]),
                    "close": float(row["close"]),
                }
            )

    return candles


def calculate_sma(closes, period):
    """終値の単純移動平均を未来参照なしで計算する。"""
    # 例：終値が [100, 101, 102, 103]、期間が3の場合、
    # SMAは [None, None, 101, 102] になります。
    # 101 = (100 + 101 + 102) ÷ 3、102 = (101 + 102 + 103) ÷ 3です。
    # SMAをまだ計算できない最初の期間にはNoneを入れます。
    values = [None] * len(closes)
    window_sum = 0.0

    for index, close in enumerate(closes):
        # 新しい終値を合計へ追加し、期間より古い終値を合計から外します。
        # 上のSMA3の4本目なら、103を加えた後、期間外になった100を引きます。
        window_sum += close
        if index >= period:
            window_sum -= closes[index - period]

        # 必要な本数が揃った位置からSMAを記録します。
        # SMA3なら、3本目にあたるindex=2から計算できます。
        if index >= period - 1:
            values[index] = window_sum / period
    return values


def crossover(previous_fast, previous_slow, fast, slow):
    """ゴールデンクロスなら1、デッドクロスなら-1、それ以外は0。"""
    # 短期SMAが長期SMAを下から上へ抜けた場合です。
    # 例：前の足が「短期99、長期100」、現在足が「短期101、長期100」なら、
    # 短期SMAが下側から上側へ移ったため、ゴールデンクロスとして1を返します。
    if previous_fast <= previous_slow and fast > slow:
        return 1

    # 短期SMAが長期SMAを上から下へ抜けた場合です。
    # 例：前の足が「短期101、長期100」、現在足が「短期99、長期100」なら、
    # 短期SMAが上側から下側へ移ったため、デッドクロスとして-1を返します。
    if previous_fast >= previous_slow and fast < slow:
        return -1
    return 0


def collect_trades(
    candles,
    fast_sma,
    slow_sma,
    side,
    pip_size,
    evaluation_start_index,
):
    """片側戦略を検証し、1回ごとの取引履歴を返す。"""
    # LONGはゴールデンクロスで入り、デッドクロスで出ます。
    # SHORTはその反対です。
    entry_signal = 1 if side == "LONG" else -1
    exit_signal = -entry_signal

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

    # 全設定を公平に比べるため、最長のSMA500が計算できる位置から始めます。
    for index in range(max(1, evaluation_start_index), len(candles)):
        # 前の15分足で決済が決まっていれば、現在足の始値で決済します。
        if entry_price is not None and pending_exit:
            direction = 1.0 if side == "LONG" else -1.0
            exit_price = candles[index]["open"]

            # USDJPYのLONGを150.00円で買い、150.20円で決済した例では、
            # (150.20 - 150.00) ÷ 0.01 = 20pipsです。
            # SHORTはdirectionが-1になるため、値下がりしたときに利益がプラスになります。
            trades.append(
                {
                    "取引番号": len(trades) + 1,
                    "エントリー日時": entry_time.isoformat(),
                    "エントリー価格": round(entry_price, 6),
                    "決済日時": candles[index]["open_time"].isoformat(),
                    "決済価格": round(exit_price, 6),
                    "決済理由": "反対クロス",
                    "保有時間（分）": (index - entry_index) * 15,
                    "損益（pips）": round(
                        direction * (exit_price - entry_price) / pip_size, 2
                    ),
                }
            )
            entry_price = None
            entry_time = None
            entry_index = None
        pending_exit = False

        # 前の15分足でエントリーが決まっていれば、現在足の始値で約定します。
        # 例：10:00開始の足が10:15に確定してクロスした場合、
        # クロスした足の終値ではなく、10:15開始足の始値で約定します。
        if entry_price is None and pending_entry:
            entry_price = candles[index]["open"]
            entry_time = candles[index]["open_time"]
            entry_index = index
        pending_entry = False

        # 直前と現在の短期SMA・長期SMAを比べてクロスを判定します。
        previous_fast = fast_sma[index - 1]
        previous_slow = slow_sma[index - 1]
        current_fast = fast_sma[index]
        current_slow = slow_sma[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)
        )

        # ポジションがない場合は、15分足でクロスした次の足でエントリーします。
        # ここではまだ約定せず、pending_entryをTrueにして次のループまで待ちます。
        if entry_price is None and signal == entry_signal:
            pending_entry = True

        # ポジション保有中に反対クロスが出たら、次の15分足で決済します。
        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
        exit_price = candles[-1]["close"]
        trades.append(
            {
                "取引番号": len(trades) + 1,
                "エントリー日時": entry_time.isoformat(),
                "エントリー価格": round(entry_price, 6),
                "決済日時": candles[-1]["close_time"].isoformat(),
                "決済価格": round(exit_price, 6),
                "決済理由": "データ終了",
                "保有時間（分）": (len(candles) - entry_index) * 15,
                "損益（pips）": round(
                    direction * (exit_price - entry_price) / pip_size, 2
                ),
            }
        )
    return trades


def summarize(symbol, fast_period, slow_period, trades):
    """取引ごとのpips損益から、成績をまとめる。"""
    profits = [trade["損益（pips）"] for trade in trades]
    # プラスなら勝ち、マイナスなら負けとして分けます。
    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))

    # プロフィットファクターは「総利益 ÷ 総損失」です。
    # 例：勝ち取引の合計が300pips、負け取引の合計が-200pipsなら、
    # プロフィットファクターは 300 ÷ 200 = 1.5 です。
    profit_factor = gross_profit / gross_loss if gross_loss else None

    # 取引順に累積損益を作り、過去の最高値からの最大下落幅を求めます。
    # 例：累積損益が [100, 250, 180, 50] と推移した場合、
    # 過去最高の250から50まで200pips下落するため、最大DDは200pipsです。
    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,
        "短期SMA期間": fast_period,
        "長期SMA期間": 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通貨について、15分足のMA100パターンを検証して順次保存する。"""
    # 15分足の終値から、検証で必要なすべてのSMAを先に計算します。
    all_periods = set(fast_periods) | set(slow_periods)
    closes = [candle["close"] for candle in candles]
    entry_sma_by_period = {
        period: calculate_sma(closes, period) for period in all_periods
    }
    # 全組み合わせを最長MAが計算可能になった同じ時点から比較します。
    evaluation_start_index = max(all_periods)

    # 円を含む通貨ペアは1pips=0.01円、それ以外は1pips=0.0001です。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001

    # LONGとSHORTを分け、result.csvを検証中に1回だけ開きます。
    for side in ("LONG", "SHORT"):
        result_path = output_dir / symbol / side / "result.csv"
        result_path.parent.mkdir(parents=True, exist_ok=True)

        # "w"で開くため、再実行時は以前のresult.csvを上書きします。
        with result_path.open("w", encoding="utf-8-sig", newline="") as file:
            writer = csv.DictWriter(file, fieldnames=RESULT_COLUMNS)
            writer.writeheader()

            # 短期10種類 × 長期10種類を順番に検証します。
            for fast_period in fast_periods:
                for slow_period in slow_periods:
                    if fast_period >= slow_period:
                        continue

                    trades = collect_trades(
                        candles,
                        entry_sma_by_period[fast_period],
                        entry_sma_by_period[slow_period],
                        side,
                        pip_size,
                        evaluation_start_index,
                    )

                    # 1設定の検証が終わった時点で取引履歴を保存します。
                    # 次の設定へ進む前にtradesが置き換わるため、全設定分を保持しません。
                    trade_path = (
                        output_dir
                        / symbol
                        / side
                        / "trades"
                        / f"SMA_{fast_period}_{slow_period}.csv"
                    )
                    write_csv(trade_path, trades, TRADE_COLUMNS)

                    # 取引履歴を保存した直後に、この設定の検証結果を1行追加します。
                    writer.writerow(
                        summarize(symbol, fast_period, slow_period, trades)
                    )


def write_csv(path, rows, columns):
    """検証結果をCSVへ保存する。"""
    # 保存先フォルダがなければ作成します。
    path.parent.mkdir(parents=True, exist_ok=True)

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


def main():
    """CSVの読み込みから検証結果の保存までを実行する。"""
    # スクリプト全体の実行時間を測り始めます。
    start_time = time.perf_counter()

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

    print(f"有効なMA組み合わせ: {combinations}パターン")

    # 5通貨を1つずつ検証します。
    for symbol in SYMBOLS:
        # この通貨の15分足CSVを読み込みます。
        path = DEFAULT_DATA_DIR / symbol / f"{ENTRY_TIMEFRAME}_{PRICE_TYPE}.csv"
        candles = load_candles(path)

        # 100パターンを実行し、全結果を保存します。
        print(f"[RUN] {symbol}: 15min")
        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


# このファイルを直接実行したときだけmain()を動かします。
if __name__ == "__main__":
    raise SystemExit(main())