Pythonを用いてFXのバックテストを行う3ステップ｜データ取得・実行・改善まで解説【サンプルコードあり】

import bisect
import csv
from collections import deque
from datetime import datetime, timedelta

import backtest_config as config

# 日足・4時間足・1時間足が上昇トレンドのとき、15分足のゴールデンクロスで買い、
# 30pipsの損切りまたは40pipsの利益確定で決済するロング専用EAを検証します。

# 使い方:
# 1. backtest_config.py の設定値を必要に応じて変更します。
# 2. GMO-API フォルダで次のコマンドを実行します。
#    python backtest_fx.py
# 3. 結果は data/backtests/USD_JPY/YYYYMMDD_HHMMSS フォルダへ保存されます。
#
# 上の import backtest_config as config で設定ファイルを読み込んでいます。
# このサンプルはロング（買い）だけをバックテストします。
# ASK価格だけを使う基礎サンプルのため、スプレッドなどの取引コストは考慮しません。

# -----------------------------------------------------------------------------
# バックテストで使用する基本情報
# -----------------------------------------------------------------------------

# 各時間足のローソク足1本が、何分または何時間なのかを設定します。
# CSVのtimestamp_jstはローソク足の開始時刻なので、この時間を足すことで
# ローソク足が確定した時刻を求めます。
INTERVAL_DURATIONS = {
    "15min": timedelta(minutes=15),
    "1hour": timedelta(hours=1),
    "4hour": timedelta(hours=4),
    "1day": timedelta(days=1),
}

# trades.csvへ出力する日本語の列名と、その並び順です。
TRADE_COLUMNS = [
    "売買方向", "エントリー日時", "エントリー価格", "決済日時", "決済価格",
    "決済理由", "保有本数", "損益（pips）", "損益（円）",
]


def load_candles(interval, csv_path):
    """fetch_gmo_candles.py形式のCSVを読み込む。"""

    # 指定したCSVがない場合は、対象の時間足を示して処理を止めます。
    if not csv_path.exists():
        raise FileNotFoundError(
            f"CSVが見つかりません: {csv_path}\n"
            f"fetch_gmo_candles.pyで {interval} のデータを取得してください。"
        )

    # CSVの各行を、バックテストで扱いやすい辞書へ変換して格納します。
    candles = []
    with csv_path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)

        # 必要な5列がCSVにそろっているか確認します。
        required_columns = {"timestamp_jst", "open", "high", "low", "close"}
        if not required_columns.issubset(reader.fieldnames or []):
            raise ValueError(f"CSVの列が不足しています: {csv_path}")

        for row in reader:
            # 日時はdatetime、価格はfloatへ変換します。
            # 文字列のままでは日時の比較や価格の計算がしにくいためです。
            candles.append(
                {
                    "open_time": datetime.fromisoformat(row["timestamp_jst"]),
                    "open": float(row["open"]),
                    "high": float(row["high"]),
                    "low": float(row["low"]),
                    "close": float(row["close"]),
                }
            )

    # CSVの並び順に関係なく、古いローソク足から順番に並べます。
    candles.sort(key=lambda candle: candle["open_time"])
    for candle in candles:
        # 未完成の上位足を参照しないよう、利用可能になる時刻を記録します。
        candle["close_time"] = candle["open_time"] + INTERVAL_DURATIONS[interval]
    return candles


def add_sma(candles, period, column_name):
    """各ローソク足へ単純移動平均線を追加する。"""

    # windowには、SMAの計算に必要な直近の終値だけを残します。
    window = deque()
    window_sum = 0.0
    for candle in candles:
        # 新しい終値を計算対象へ加えます。
        window.append(candle["close"])
        window_sum += candle["close"]

        # 指定期間より古い終値は計算対象から外します。
        if len(window) > period:
            window_sum -= window.popleft()

        # 必要な本数がそろうまではSMAを計算できないためNoneを入れます。
        candle[column_name] = window_sum / period if len(window) == period else None


def make_completed_trend_lookup(candles, sma_column):
    """指定時刻までに確定した上位足のトレンドを検索できる形にする。"""

    # SMAを計算できたローソク足だけをトレンド判定に使います。
    usable = [candle for candle in candles if candle[sma_column] is not None]
    close_times = [candle["close_time"] for candle in usable]

    # 終値がSMAより上なら1、下なら-1、同じなら0です。
    trends = [
        1 if candle["close"] > candle[sma_column]
        else -1 if candle["close"] < candle[sma_column]
        else 0
        for candle in usable
    ]
    return close_times, trends


def trend_at(lookup, signal_time):
    """signal_timeまでに確定済みの最新トレンドを返す。"""

    close_times, trends = lookup

    # bisectを使い、シグナル時刻以前に確定した最新の上位足を探します。
    # 形成途中の上位足を使わないことで、未来の情報を参照するのを防ぎます。
    index = bisect.bisect_right(close_times, signal_time) - 1
    return trends[index] if index >= 0 else None


def has_long_entry_signal(index, candles, trend_lookups):
    """確定した15分足と上位足から、買いシグナルの有無を返す。"""

    # 最初の1本は直前のSMAと比較できないため、エントリーしません。
    if index == 0:
        return False

    # 直前の15分足と、現在の15分足を取り出します。
    previous = candles[index - 1]
    current = candles[index]

    # 短期SMAと長期SMAの計算に必要な本数がそろうまで待ちます。
    required = ("fast_sma", "slow_sma")
    if any(previous[name] is None or current[name] is None for name in required):
        return False

    # 現在の15分足が確定した時点で、各上位足のトレンドを調べます。
    signal_time = current["close_time"]
    higher_trends = [trend_at(lookup, signal_time) for lookup in trend_lookups.values()]
    if any(trend is None for trend in higher_trends):
        return False

    # 短期SMAが長期SMAを下から上へ抜けたか確認します。
    crossed_up = (
        previous["fast_sma"] <= previous["slow_sma"]
        and current["fast_sma"] > current["slow_sma"]
    )
    # ゴールデンクロスが発生し、すべての上位足が上昇方向なら買いです。
    if crossed_up and all(trend == 1 for trend in higher_trends):
        return True
    return False


def exit_from_candle(position, candle):
    """15分足の高値・安値から損切りまたは利益確定を判定する。"""

    # 設定したpipsを価格差へ変換し、損切り価格と利益確定価格を作ります。
    entry_price = position["entry_price"]
    stop_price = entry_price - config.STOP_LOSS_PIPS * config.PIP_SIZE
    take_price = entry_price + config.TAKE_PROFIT_PIPS * config.PIP_SIZE
    # 安値が損切り価格以下なら損切り、高値が利益確定価格以上なら利益確定です。
    stop_hit = candle["low"] <= stop_price
    take_hit = candle["high"] >= take_price

    # 1本の足で両方に触れた場合は、設定ファイルで優先順位を決めます。
    if stop_hit and take_hit:
        if config.STOP_FIRST_WHEN_BOTH_HIT:
            return stop_price, "損切り"
        return take_price, "利益確定"
    if stop_hit:
        return stop_price, "損切り"
    if take_hit:
        return take_price, "利益確定"
    return None


def close_position(position, exit_time, exit_price, exit_reason):
    """決済価格からpips損益と金額の損益を計算する。"""

    # ロングなので「決済価格 - エントリー価格」が値動きによる損益です。
    profit_pips = (exit_price - position["entry_price"]) / config.PIP_SIZE

    # pipsを価格差へ戻し、取引数量を掛けて概算損益を求めます。
    profit = profit_pips * config.PIP_SIZE * config.POSITION_SIZE_UNITS

    # trades.csvへ1行として保存できる辞書を返します。
    return {
        "売買方向": "ロング",
        "エントリー日時": position["entry_time"].isoformat(),
        "エントリー価格": f'{position["entry_price"]:.6f}',
        "決済日時": exit_time.isoformat(),
        "決済価格": f"{exit_price:.6f}",
        "決済理由": exit_reason,
        "保有本数": position["holding_bars"],
        "損益（pips）": f"{profit_pips:.2f}",
        "損益（円）": f"{profit:.0f}",
    }


def run_backtest(candles, trend_lookups):
    """15分足を時刻順に処理し、取引履歴を作る。"""

    # tradesは完了した取引、positionは現在保有している買いポジションです。
    trades = []
    position = None

    # シグナルが出た足の終値ではなく、次の足の始値でエントリーするための印です。
    pending_long_entry = False

    # 15分足を古いものから1本ずつ処理します。
    for index, candle in enumerate(candles):
        # 直前の足で確定したシグナルを、現在足の始値で約定します。
        if position is None and pending_long_entry:
            position = {
                "entry_time": candle["open_time"],
                "entry_price": candle["open"],
                "holding_bars": 0,
            }
        # 一度確認したシグナルはここでリセットします。
        pending_long_entry = False

        # ポジションを持っている間は、現在足で決済条件に触れたか確認します。
        if position is not None:
            position["holding_bars"] += 1
            exit_result = exit_from_candle(position, candle)
            if exit_result is not None:
                exit_price, reason = exit_result
                # 決済した取引を履歴へ追加し、ポジションを空に戻します。
                trades.append(close_position(position, candle["close_time"], exit_price, reason))
                position = None
        # この足の終値でシグナルを判定し、次の足まで保留します。
        if position is None:
            pending_long_entry = has_long_entry_signal(index, candles, trend_lookups)

    # データ末尾で保有中なら、最後の終値で決済します。
    if position is not None:
        last = candles[-1]
        trades.append(close_position(position, last["close_time"], last["close"], "データ終了"))
    return trades


def calculate_summary(trades):
    """取引履歴から主な評価指標を作る。"""

    # 取引ごとの損益を順番に足し、最大ドローダウンを計算します。
    cumulative_profit = 0.0
    peak_profit = 0.0
    max_drawdown = 0.0
    for trade in trades:
        cumulative_profit += float(trade["損益（円）"])
        peak_profit = max(peak_profit, cumulative_profit)
        drawdown = peak_profit - cumulative_profit
        max_drawdown = max(max_drawdown, drawdown)

    # 利益がプラスの取引を勝ち、マイナスの取引を負けとして分けます。
    profits = [float(trade["損益（円）"]) 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))
    # プロフィットファクターは「総利益 ÷ 総損失」です。
    profit_factor = gross_profit / gross_loss if gross_loss else None

    # summary.csvへ1行で出力する集計結果を作ります。
    summary = {
        "通貨ペア": config.SYMBOL,
        "取引回数": len(trades),
        "勝ち取引数": len(wins),
        "負け取引数": len(losses),
        "勝率（%）": f"{(len(wins) / len(trades) * 100) if trades else 0:.2f}",
        "合計損益（円）": f"{sum(profits):.0f}",
        "合計損益（pips）": f'{sum(float(trade["損益（pips）"]) for trade in trades):.2f}',
        "プロフィットファクター": f"{profit_factor:.2f}" if profit_factor is not None else "計算不可",
        "最大ドローダウン（円）": f"{max_drawdown:.0f}",
    }
    return summary


def make_output_dir():
    """実行日時を使って結果の保存先を作る。"""

    # 実行日時を「20260803_223000」のような文字列にします。
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_dir = config.OUTPUT_BASE_DIR / timestamp
    output_dir.mkdir(parents=True)
    return output_dir


def write_csv(path, rows, fieldnames):
    """指定した列の順番で、結果をCSVへ保存する。"""

    # 保存先フォルダがなければ作成します。
    path.parent.mkdir(parents=True, exist_ok=True)

    # "w"は新規書き込みです。日時別フォルダなので以前の結果は上書きしません。
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main():
    """CSVの読み込みから結果の保存までを順番に実行する。"""

    # 設定ファイルに書かれた4種類のローソク足CSVを読み込みます。
    print(f"{config.SYMBOL}のローソク足CSVを読み込んでいます。")
    all_candles = {
        interval: load_candles(interval, path)
        for interval, path in config.CANDLE_CSV_PATHS.items()
    }

    # エントリー判断に使う15分足へ、短期SMAと長期SMAを追加します。
    entry_candles = all_candles["15min"]
    add_sma(entry_candles, config.ENTRY_FAST_SMA_PERIOD, "fast_sma")
    add_sma(entry_candles, config.ENTRY_SLOW_SMA_PERIOD, "slow_sma")

    # 1時間足、4時間足、日足へSMAを追加し、トレンドを検索できる形にします。
    trend_lookups = {}
    for interval, period in config.TREND_SMA_PERIODS.items():
        sma_column = "trend_sma"
        add_sma(all_candles[interval], period, sma_column)
        trend_lookups[interval] = make_completed_trend_lookup(
            all_candles[interval], sma_column
        )

    # 時刻順にバックテストし、取引履歴と集計結果を作ります。
    trades = run_backtest(entry_candles, trend_lookups)
    summary = calculate_summary(trades)

    # 実行日時のフォルダを作り、取引履歴と集計結果をCSVへ保存します。
    output_dir = make_output_dir()
    write_csv(output_dir / "trades.csv", trades, TRADE_COLUMNS)
    write_csv(output_dir / "summary.csv", [summary], list(summary.keys()))

    print(f'取引回数: {summary["取引回数"]}回')
    print(f'勝率: {summary["勝率（%）"]}%')
    print(f'合計損益: {summary["合計損益（円）"]}円')
    print(f'最大ドローダウン: {summary["最大ドローダウン（円）"]}円')
    print(f"結果の保存先: {output_dir}")


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

from pathlib import Path

# 検証する通貨ペアと価格種別です。
SYMBOL = "USD_JPY"
PRICE_TYPE = "ASK"

# fetch_gmo_candles.py が出力したCSVを読み込む場所です。
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data" / "candles" / SYMBOL
CANDLE_CSV_PATHS = {
    "1day": DATA_DIR / f"1day_{PRICE_TYPE}.csv",
    "4hour": DATA_DIR / f"4hour_{PRICE_TYPE}.csv",
    "1hour": DATA_DIR / f"1hour_{PRICE_TYPE}.csv",
    "15min": DATA_DIR / f"15min_{PRICE_TYPE}.csv",
}

# 結果を保存する親フォルダです。
# 実行時に、この下へ YYYYMMDD_HHMMSS 形式のフォルダが作られます。
OUTPUT_BASE_DIR = BASE_DIR / "data" / "backtests" / SYMBOL

# 売買ルールです。
# 15分足の短期・長期移動平均線が交差したときにエントリーを検討します。
ENTRY_FAST_SMA_PERIOD = 10
ENTRY_SLOW_SMA_PERIOD = 30

# 上位足は、終値が移動平均線より上なら買い方向、下なら売り方向と判定します。
TREND_SMA_PERIODS = {
    "1hour": 20,
    "4hour": 20,
    "1day": 20,
}

# このサンプルではロング（買い）のみを検証します。
# 損切りと利益確定の幅をpips単位で設定します。
STOP_LOSS_PIPS = 20.0
TAKE_PROFIT_PIPS = 40.0

# USD/JPYなど、円を含む通貨ペアは通常 1pips = 0.01円です。
PIP_SIZE = 0.01

# 1回の取引数量です。USD/JPYの場合、損益の単位は概算で円になります。
POSITION_SIZE_UNITS = 10000

# 同じ15分足の中で損切りと利益確定の両方に触れた場合、保守的に損切りを優先します。
STOP_FIRST_WHEN_BOTH_HIT = True

import bisect
import csv
from collections import deque
from datetime import datetime, timedelta

import backtest_config as config

# 日足・4時間足・1時間足が上昇トレンドのとき、15分足のゴールデンクロスで買い、
# 30pipsの損切りまたは40pipsの利益確定で決済するロング専用EAを検証します。

# 使い方:
# 1. backtest_config.py の設定値を必要に応じて変更します。
# 2. GMO-API フォルダで次のコマンドを実行します。
#    python backtest_fx.py
# 3. 結果は data/backtests/USD_JPY/YYYYMMDD_HHMMSS フォルダへ保存されます。
#
# 上の import backtest_config as config で設定ファイルを読み込んでいます。
# このサンプルはロング（買い）だけをバックテストします。
# ASK価格だけを使う基礎サンプルのため、スプレッドなどの取引コストは考慮しません。

# -----------------------------------------------------------------------------
# バックテストで使用する基本情報
# -----------------------------------------------------------------------------

# 各時間足のローソク足1本が、何分または何時間なのかを設定します。
# CSVのtimestamp_jstはローソク足の開始時刻なので、この時間を足すことで
# ローソク足が確定した時刻を求めます。
INTERVAL_DURATIONS = {
    "15min": timedelta(minutes=15),
    "1hour": timedelta(hours=1),
    "4hour": timedelta(hours=4),
    "1day": timedelta(days=1),
}

# trades.csvへ出力する日本語の列名と、その並び順です。
TRADE_COLUMNS = [
    "売買方向", "エントリー日時", "エントリー価格", "決済日時", "決済価格",
    "決済理由", "保有本数", "損益（pips）", "損益（円）",
]


def load_candles(interval, csv_path):
    """fetch_gmo_candles.py形式のCSVを読み込む。"""

    # 指定したCSVがない場合は、対象の時間足を示して処理を止めます。
    if not csv_path.exists():
        raise FileNotFoundError(
            f"CSVが見つかりません: {csv_path}\n"
            f"fetch_gmo_candles.pyで {interval} のデータを取得してください。"
        )

    # CSVの各行を、バックテストで扱いやすい辞書へ変換して格納します。
    candles = []
    with csv_path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)

        # 必要な5列がCSVにそろっているか確認します。
        required_columns = {"timestamp_jst", "open", "high", "low", "close"}
        if not required_columns.issubset(reader.fieldnames or []):
            raise ValueError(f"CSVの列が不足しています: {csv_path}")

        for row in reader:
            # 日時はdatetime、価格はfloatへ変換します。
            # 文字列のままでは日時の比較や価格の計算がしにくいためです。
            candles.append(
                {
                    "open_time": datetime.fromisoformat(row["timestamp_jst"]),
                    "open": float(row["open"]),
                    "high": float(row["high"]),
                    "low": float(row["low"]),
                    "close": float(row["close"]),
                }
            )

    # CSVの並び順に関係なく、古いローソク足から順番に並べます。
    candles.sort(key=lambda candle: candle["open_time"])
    for candle in candles:
        # 未完成の上位足を参照しないよう、利用可能になる時刻を記録します。
        candle["close_time"] = candle["open_time"] + INTERVAL_DURATIONS[interval]
    return candles


def add_sma(candles, period, column_name):
    """各ローソク足へ単純移動平均線を追加する。"""

    # windowには、SMAの計算に必要な直近の終値だけを残します。
    window = deque()
    window_sum = 0.0
    for candle in candles:
        # 新しい終値を計算対象へ加えます。
        window.append(candle["close"])
        window_sum += candle["close"]

        # 指定期間より古い終値は計算対象から外します。
        if len(window) > period:
            window_sum -= window.popleft()

        # 必要な本数がそろうまではSMAを計算できないためNoneを入れます。
        candle[column_name] = window_sum / period if len(window) == period else None


def make_completed_trend_lookup(candles, sma_column):
    """指定時刻までに確定した上位足のトレンドを検索できる形にする。"""

    # SMAを計算できたローソク足だけをトレンド判定に使います。
    usable = [candle for candle in candles if candle[sma_column] is not None]
    close_times = [candle["close_time"] for candle in usable]

    # 終値がSMAより上なら1、下なら-1、同じなら0です。
    trends = [
        1 if candle["close"] > candle[sma_column]
        else -1 if candle["close"] < candle[sma_column]
        else 0
        for candle in usable
    ]
    return close_times, trends


def trend_at(lookup, signal_time):
    """signal_timeまでに確定済みの最新トレンドを返す。"""

    close_times, trends = lookup

    # bisectを使い、シグナル時刻以前に確定した最新の上位足を探します。
    # 形成途中の上位足を使わないことで、未来の情報を参照するのを防ぎます。
    index = bisect.bisect_right(close_times, signal_time) - 1
    return trends[index] if index >= 0 else None


def has_long_entry_signal(index, candles, trend_lookups):
    """確定した15分足と上位足から、買いシグナルの有無を返す。"""

    # 最初の1本は直前のSMAと比較できないため、エントリーしません。
    if index == 0:
        return False

    # 直前の15分足と、現在の15分足を取り出します。
    previous = candles[index - 1]
    current = candles[index]

    # 短期SMAと長期SMAの計算に必要な本数がそろうまで待ちます。
    required = ("fast_sma", "slow_sma")
    if any(previous[name] is None or current[name] is None for name in required):
        return False

    # 現在の15分足が確定した時点で、各上位足のトレンドを調べます。
    signal_time = current["close_time"]
    higher_trends = [trend_at(lookup, signal_time) for lookup in trend_lookups.values()]
    if any(trend is None for trend in higher_trends):
        return False

    # 短期SMAが長期SMAを下から上へ抜けたか確認します。
    crossed_up = (
        previous["fast_sma"] <= previous["slow_sma"]
        and current["fast_sma"] > current["slow_sma"]
    )
    # ゴールデンクロスが発生し、すべての上位足が上昇方向なら買いです。
    if crossed_up and all(trend == 1 for trend in higher_trends):
        return True
    return False


def exit_from_candle(position, candle):
    """15分足の高値・安値から損切りまたは利益確定を判定する。"""

    # 設定したpipsを価格差へ変換し、損切り価格と利益確定価格を作ります。
    entry_price = position["entry_price"]
    stop_price = entry_price - config.STOP_LOSS_PIPS * config.PIP_SIZE
    take_price = entry_price + config.TAKE_PROFIT_PIPS * config.PIP_SIZE
    # 安値が損切り価格以下なら損切り、高値が利益確定価格以上なら利益確定です。
    stop_hit = candle["low"] <= stop_price
    take_hit = candle["high"] >= take_price

    # 1本の足で両方に触れた場合は、設定ファイルで優先順位を決めます。
    if stop_hit and take_hit:
        if config.STOP_FIRST_WHEN_BOTH_HIT:
            return stop_price, "損切り"
        return take_price, "利益確定"
    if stop_hit:
        return stop_price, "損切り"
    if take_hit:
        return take_price, "利益確定"
    return None


def close_position(position, exit_time, exit_price, exit_reason):
    """決済価格からpips損益と金額の損益を計算する。"""

    # ロングなので「決済価格 - エントリー価格」が値動きによる損益です。
    profit_pips = (exit_price - position["entry_price"]) / config.PIP_SIZE

    # pipsを価格差へ戻し、取引数量を掛けて概算損益を求めます。
    profit = profit_pips * config.PIP_SIZE * config.POSITION_SIZE_UNITS

    # trades.csvへ1行として保存できる辞書を返します。
    return {
        "売買方向": "ロング",
        "エントリー日時": position["entry_time"].isoformat(),
        "エントリー価格": f'{position["entry_price"]:.6f}',
        "決済日時": exit_time.isoformat(),
        "決済価格": f"{exit_price:.6f}",
        "決済理由": exit_reason,
        "保有本数": position["holding_bars"],
        "損益（pips）": f"{profit_pips:.2f}",
        "損益（円）": f"{profit:.0f}",
    }


def run_backtest(candles, trend_lookups):
    """15分足を時刻順に処理し、取引履歴を作る。"""

    # tradesは完了した取引、positionは現在保有している買いポジションです。
    trades = []
    position = None

    # シグナルが出た足の終値ではなく、次の足の始値でエントリーするための印です。
    pending_long_entry = False

    # 15分足を古いものから1本ずつ処理します。
    for index, candle in enumerate(candles):
        # 直前の足で確定したシグナルを、現在足の始値で約定します。
        if position is None and pending_long_entry:
            position = {
                "entry_time": candle["open_time"],
                "entry_price": candle["open"],
                "holding_bars": 0,
            }
        # 一度確認したシグナルはここでリセットします。
        pending_long_entry = False

        # ポジションを持っている間は、現在足で決済条件に触れたか確認します。
        if position is not None:
            position["holding_bars"] += 1
            exit_result = exit_from_candle(position, candle)
            if exit_result is not None:
                exit_price, reason = exit_result
                # 決済した取引を履歴へ追加し、ポジションを空に戻します。
                trades.append(close_position(position, candle["close_time"], exit_price, reason))
                position = None
        # この足の終値でシグナルを判定し、次の足まで保留します。
        if position is None:
            pending_long_entry = has_long_entry_signal(index, candles, trend_lookups)

    # データ末尾で保有中なら、最後の終値で決済します。
    if position is not None:
        last = candles[-1]
        trades.append(close_position(position, last["close_time"], last["close"], "データ終了"))
    return trades


def calculate_summary(trades):
    """取引履歴から主な評価指標を作る。"""

    # 取引ごとの損益を順番に足し、最大ドローダウンを計算します。
    cumulative_profit = 0.0
    peak_profit = 0.0
    max_drawdown = 0.0
    for trade in trades:
        cumulative_profit += float(trade["損益（円）"])
        peak_profit = max(peak_profit, cumulative_profit)
        drawdown = peak_profit - cumulative_profit
        max_drawdown = max(max_drawdown, drawdown)

    # 利益がプラスの取引を勝ち、マイナスの取引を負けとして分けます。
    profits = [float(trade["損益（円）"]) 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))
    # プロフィットファクターは「総利益 ÷ 総損失」です。
    profit_factor = gross_profit / gross_loss if gross_loss else None

    # summary.csvへ1行で出力する集計結果を作ります。
    summary = {
        "通貨ペア": config.SYMBOL,
        "取引回数": len(trades),
        "勝ち取引数": len(wins),
        "負け取引数": len(losses),
        "勝率（%）": f"{(len(wins) / len(trades) * 100) if trades else 0:.2f}",
        "合計損益（円）": f"{sum(profits):.0f}",
        "合計損益（pips）": f'{sum(float(trade["損益（pips）"]) for trade in trades):.2f}',
        "プロフィットファクター": f"{profit_factor:.2f}" if profit_factor is not None else "計算不可",
        "最大ドローダウン（円）": f"{max_drawdown:.0f}",
    }
    return summary


def make_output_dir():
    """実行日時を使って結果の保存先を作る。"""

    # 実行日時を「20260803_223000」のような文字列にします。
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_dir = config.OUTPUT_BASE_DIR / timestamp
    output_dir.mkdir(parents=True)
    return output_dir


def write_csv(path, rows, fieldnames):
    """指定した列の順番で、結果をCSVへ保存する。"""

    # 保存先フォルダがなければ作成します。
    path.parent.mkdir(parents=True, exist_ok=True)

    # "w"は新規書き込みです。日時別フォルダなので以前の結果は上書きしません。
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main():
    """CSVの読み込みから結果の保存までを順番に実行する。"""

    # 設定ファイルに書かれた4種類のローソク足CSVを読み込みます。
    print(f"{config.SYMBOL}のローソク足CSVを読み込んでいます。")
    all_candles = {
        interval: load_candles(interval, path)
        for interval, path in config.CANDLE_CSV_PATHS.items()
    }

    # エントリー判断に使う15分足へ、短期SMAと長期SMAを追加します。
    entry_candles = all_candles["15min"]
    add_sma(entry_candles, config.ENTRY_FAST_SMA_PERIOD, "fast_sma")
    add_sma(entry_candles, config.ENTRY_SLOW_SMA_PERIOD, "slow_sma")

    # 1時間足、4時間足、日足へSMAを追加し、トレンドを検索できる形にします。
    trend_lookups = {}
    for interval, period in config.TREND_SMA_PERIODS.items():
        sma_column = "trend_sma"
        add_sma(all_candles[interval], period, sma_column)
        trend_lookups[interval] = make_completed_trend_lookup(
            all_candles[interval], sma_column
        )

    # 時刻順にバックテストし、取引履歴と集計結果を作ります。
    trades = run_backtest(entry_candles, trend_lookups)
    summary = calculate_summary(trades)

    # 実行日時のフォルダを作り、取引履歴と集計結果をCSVへ保存します。
    output_dir = make_output_dir()
    write_csv(output_dir / "trades.csv", trades, TRADE_COLUMNS)
    write_csv(output_dir / "summary.csv", [summary], list(summary.keys()))

    print(f'取引回数: {summary["取引回数"]}回')
    print(f'勝率: {summary["勝率（%）"]}%')
    print(f'合計損益: {summary["合計損益（円）"]}円')
    print(f'最大ドローダウン: {summary["最大ドローダウン（円）"]}円')
    print(f"結果の保存先: {output_dir}")


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

from pathlib import Path

# 検証する通貨ペアと価格種別です。
SYMBOL = "USD_JPY"
PRICE_TYPE = "ASK"

# fetch_gmo_candles.py が出力したCSVを読み込む場所です。
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data" / "candles" / SYMBOL
CANDLE_CSV_PATHS = {
    "1day": DATA_DIR / f"1day_{PRICE_TYPE}.csv",
    "4hour": DATA_DIR / f"4hour_{PRICE_TYPE}.csv",
    "1hour": DATA_DIR / f"1hour_{PRICE_TYPE}.csv",
    "15min": DATA_DIR / f"15min_{PRICE_TYPE}.csv",
}

# 結果を保存する親フォルダです。
# 実行時に、この下へ YYYYMMDD_HHMMSS 形式のフォルダが作られます。
OUTPUT_BASE_DIR = BASE_DIR / "data" / "backtests" / SYMBOL

# 売買ルールです。
# 15分足の短期・長期移動平均線が交差したときにエントリーを検討します。
ENTRY_FAST_SMA_PERIOD = 10
ENTRY_SLOW_SMA_PERIOD = 30

# 上位足は、終値が移動平均線より上なら買い方向、下なら売り方向と判定します。
TREND_SMA_PERIODS = {
    "1hour": 20,
    "4hour": 20,
    "1day": 20,
}

# このサンプルではロング（買い）のみを検証します。
# 損切りと利益確定の幅をpips単位で設定します。
STOP_LOSS_PIPS = 20.0
TAKE_PROFIT_PIPS = 40.0

# USD/JPYなど、円を含む通貨ペアは通常 1pips = 0.01円です。
PIP_SIZE = 0.01

# 1回の取引数量です。USD/JPYの場合、損益の単位は概算で円になります。
POSITION_SIZE_UNITS = 10000

# 同じ15分足の中で損切りと利益確定の両方に触れた場合、保守的に損切りを優先します。
STOP_FIRST_WHEN_BOTH_HIT = True