FXのATRでおすすめの設定は？5通貨で期間・基準値を徹底検証

"""5通貨・LONG/SHORTでATR期間と相対ボラティリティ基準を比較する。

EMA20/EMA50クロスで売買方向を決め、ATRが自身の過去分布に対して
指定された水準以上のときだけエントリーする。10組のTP/SLも比較する。
"""

import csv
import time
from bisect import bisect_left, insort
from collections import deque
from datetime import date, datetime, timedelta
from pathlib import Path


# GMO-APIフォルダで次のコマンドを実行します。
# python backtest_research/average_true_range/run_atr_backtest.py

ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)
PRICE_TYPE = "ASK"

BACKTEST_START_DATE = date(2024, 1, 1)
BACKTEST_END_DATE = date(2026, 7, 31)

SYMBOLS = ("USD_JPY", "EUR_JPY", "GBP_JPY", "EUR_USD", "AUD_USD")
SIDES = ("LONG", "SHORT")

# EMAクロスは全ATR設定で共通です。ATR自体は売買方向を決めません。
FAST_EMA_PERIOD = 20
SLOW_EMA_PERIOD = 50

# ATRはWilder方式で計算します。
ATR_PERIODS = (7, 10, 14, 20, 21, 28)

# 現在のATRを、直前100本のATRだけから作った分布と比較します。
# 0.0: 制限なし
# 0.3: 下位30%を除外（上位70%を採用）
# 0.5: 下位50%を除外（上位50%を採用）
# 0.7: 下位70%を除外（高ボラ上位30%だけを採用）
ATR_PERCENTILE_LOOKBACK = 100
ATR_FILTERS = (
    ("制限なし", None),
    ("低ボラ30%除外", 0.3),
    ("低ボラ50%除外", 0.5),
    ("高ボラ30%のみ", 0.7),
)

# 利幅・損失幅とリスクリワードの偏りを避けるため、狭い出口から広い出口まで
# 既存のボリンジャーバンド検証と同じ10組を比較します。
TP_SL_PAIRS = (
    (20, 10),
    (30, 10),
    (30, 20),
    (50, 10),
    (50, 20),
    (50, 30),
    (80, 20),
    (80, 30),
    (100, 30),
    (100, 50),
)

# 最長ATR、EMA、過去ATR分布を検証開始前に準備できる本数です。
WARMUP_BARS = max(
    SLOW_EMA_PERIOD + 1,
    max(ATR_PERIODS) + ATR_PERCENTILE_LOOKBACK + 1,
)

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"

RESULT_COLUMNS = (
    "通貨ペア", "売買方向", "短期EMA期間", "長期EMA期間", "ATR期間",
    "ATRフィルター", "ATR下限パーセンタイル", "ATR分布参照本数",
    "TP（pips）", "SL（pips）", "取引回数", "勝ち取引数", "負け取引数",
    "勝率（%）", "合計損益（pips）", "平均損益（pips）",
    "プロフィットファクター", "最大ドローダウン（pips）",
    "平均保有時間（分）", "最終足決済数",
)


def load_candles(path):
    """検証前のウォームアップ足と検証期間内のOHLCを読み込む。"""
    warmup = deque(maxlen=WARMUP_BARS)
    backtest = []
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        for row in csv.DictReader(file):
            timestamp = datetime.fromisoformat(row["timestamp_jst"])
            if timestamp.date() > BACKTEST_END_DATE:
                continue
            candle = {
                "open_time": timestamp,
                "close_time": timestamp + ENTRY_TIMEFRAME_DURATION,
                "open": float(row["open"]),
                "high": float(row["high"]),
                "low": float(row["low"]),
                "close": float(row["close"]),
            }
            if timestamp.date() < BACKTEST_START_DATE:
                warmup.append(candle)
            else:
                backtest.append(candle)
    return list(warmup) + backtest, len(warmup), len(backtest)


def calculate_ema(candles, period):
    """終値のEMAを計算する。初期値には最初のperiod本の単純平均を使う。"""
    if period <= 0:
        raise ValueError("EMA期間は1以上にしてください")
    values = [None] * len(candles)
    if len(candles) < period:
        return values
    initial_index = period - 1
    values[initial_index] = sum(
        candle["close"] for candle in candles[:period]
    ) / period
    multiplier = 2.0 / (period + 1)
    for index in range(initial_index + 1, len(candles)):
        values[index] = (
            candles[index]["close"] - values[index - 1]
        ) * multiplier + values[index - 1]
    return values


def calculate_atr(candles, period):
    """True RangeをWilder方式で平滑化してATRを計算する。"""
    if period <= 0:
        raise ValueError("ATR期間は1以上にしてください")
    atr = [None] * len(candles)
    if len(candles) <= period:
        return atr
    true_ranges = [None] * len(candles)
    for index in range(1, len(candles)):
        current = candles[index]
        previous_close = candles[index - 1]["close"]
        true_ranges[index] = max(
            current["high"] - current["low"],
            abs(current["high"] - previous_close),
            abs(current["low"] - previous_close),
        )
    atr[period] = sum(true_ranges[1:period + 1]) / period
    for index in range(period + 1, len(candles)):
        atr[index] = (
            atr[index - 1] * (period - 1) + true_ranges[index]
        ) / period
    return atr


def calculate_atr_percentile_ranks(atr, lookback):
    """現在値を直前lookback個のATRと比較したパーセンタイル順位を返す。

    現在足のATRを比較分布へ含めないため、フィルターに未来情報は入らない。
    例えば0.7以上なら、現在ATRが過去分布の上位30%相当である。
    """
    if lookback <= 0:
        raise ValueError("ATR分布参照本数は1以上にしてください")
    ranks = [None] * len(atr)
    window = deque()
    sorted_window = []
    for index, value in enumerate(atr):
        if value is None:
            continue
        if len(sorted_window) == lookback:
            # bisect_leftにより「現在値未満」の過去値の割合を求めます。
            ranks[index] = bisect_left(sorted_window, value) / lookback
        window.append(value)
        insort(sorted_window, value)
        if len(window) > lookback:
            # 現在値を追加した後、次回に備えて最古の値を1個だけ削除します。
            old = window.popleft()
            sorted_window.pop(bisect_left(sorted_window, old))
    return ranks


def make_trade(position, candle, exit_price, reason, side, pip_size):
    """決済済み取引の保有時間とpips損益を返す。"""
    direction = 1.0 if side == "LONG" else -1.0
    return {
        "保有時間（分）": (
            candle["close_time"] - position["time"]
        ).total_seconds() / 60,
        "損益（pips）": direction * (exit_price - position["price"]) / pip_size,
        "決済理由": reason,
    }


def collect_trades(candles, fast_ema, slow_ema, atr_ranks, side,
                   minimum_percentile, tp, sl, pip_size, entry_start_index):
    """EMAクロスとATR相対基準で入り、固定TP/SLで決済する。"""
    position = None
    pending_entry = False
    trades = []
    for index in range(1, len(candles)):
        if position is None and pending_entry and index >= entry_start_index:
            position = {
                "time": candles[index]["open_time"],
                "price": candles[index]["open"],
            }
        pending_entry = False

        if position is not None:
            tp_distance = tp * pip_size
            sl_distance = sl * pip_size
            if side == "LONG":
                tp_price = position["price"] + tp_distance
                sl_price = position["price"] - sl_distance
                tp_hit = candles[index]["high"] >= tp_price
                sl_hit = candles[index]["low"] <= sl_price
            else:
                tp_price = position["price"] - tp_distance
                sl_price = position["price"] + sl_distance
                tp_hit = candles[index]["low"] <= tp_price
                sl_hit = candles[index]["high"] >= sl_price
            # 足内の到達順が不明な場合は保守的にSLを優先します。
            if sl_hit:
                trades.append(make_trade(
                    position, candles[index], sl_price, "SL", side, pip_size,
                ))
                position = None
            elif tp_hit:
                trades.append(make_trade(
                    position, candles[index], tp_price, "TP", side, pip_size,
                ))
                position = None

        ema_values = (
            fast_ema[index - 1], slow_ema[index - 1],
            fast_ema[index], slow_ema[index],
        )
        if position is not None or any(value is None for value in ema_values):
            continue
        if side == "LONG":
            crossed = (
                fast_ema[index - 1] <= slow_ema[index - 1]
                and fast_ema[index] > slow_ema[index]
            )
        else:
            crossed = (
                fast_ema[index - 1] >= slow_ema[index - 1]
                and fast_ema[index] < slow_ema[index]
            )
        # 制限なしではATR順位が未計算でもEMAクロスを採用します。
        atr_allowed = minimum_percentile is None or (
            atr_ranks[index] is not None
            and atr_ranks[index] >= minimum_percentile
        )
        if crossed and atr_allowed:
            pending_entry = True

    if position is not None:
        trades.append(make_trade(
            position, candles[-1], candles[-1]["close"], "データ終了",
            side, pip_size,
        ))
    return trades


def summarize(symbol, side, atr_period, filter_name, minimum_percentile,
              tp, sl, trades):
    """1つのATR設定の全期間成績をまとめる。"""
    profits = [trade["損益（pips）"] for trade in trades]
    wins = [value for value in profits if value > 0]
    losses = [value for value in profits if value < 0]
    equity = peak = max_drawdown = 0.0
    for profit in profits:
        equity += profit
        peak = max(peak, equity)
        max_drawdown = max(max_drawdown, peak - equity)
    count = len(profits)
    gross_loss = abs(sum(losses))
    durations = [trade["保有時間（分）"] for trade in trades]
    return {
        "通貨ペア": symbol, "売買方向": side,
        "短期EMA期間": FAST_EMA_PERIOD, "長期EMA期間": SLOW_EMA_PERIOD,
        "ATR期間": atr_period, "ATRフィルター": filter_name,
        "ATR下限パーセンタイル": (
            "" if minimum_percentile is None else int(minimum_percentile * 100)
        ),
        "ATR分布参照本数": ATR_PERCENTILE_LOOKBACK,
        "TP（pips）": tp, "SL（pips）": sl,
        "取引回数": count, "勝ち取引数": len(wins), "負け取引数": len(losses),
        "勝率（%）": round(len(wins) / count * 100, 2) if count else 0.0,
        "合計損益（pips）": round(sum(profits), 2),
        "平均損益（pips）": round(sum(profits) / count, 2) if count else 0.0,
        "プロフィットファクター": (
            round(sum(wins) / gross_loss, 4) if gross_loss else ""
        ),
        "最大ドローダウン（pips）": round(max_drawdown, 2),
        "平均保有時間（分）": (
            round(sum(durations) / count, 2) if count else 0.0
        ),
        "最終足決済数": sum(
            trade["決済理由"] == "データ終了" for trade in trades
        ),
    }


def write_csv(path, rows, fieldnames=RESULT_COLUMNS):
    """Excelでも開きやすいUTF-8 BOM付きCSVを保存する。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def run_experiment(candles, symbol, side, entry_start_index, output_dir,
                   fast_ema, slow_ema):
    """1通貨・1方向でATR期間6種類×基準4種類×出口10組を検証する。"""
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    rows = []
    for atr_period in ATR_PERIODS:
        atr = calculate_atr(candles, atr_period)
        atr_ranks = calculate_atr_percentile_ranks(
            atr, ATR_PERCENTILE_LOOKBACK,
        )
        for filter_name, minimum_percentile in ATR_FILTERS:
            for tp, sl in TP_SL_PAIRS:
                trades = collect_trades(
                    candles, fast_ema, slow_ema, atr_ranks, side,
                    minimum_percentile, tp, sl, pip_size, entry_start_index,
                )
                rows.append(summarize(
                    symbol, side, atr_period, filter_name, minimum_percentile,
                    tp, sl, trades,
                ))
    write_csv(output_dir / symbol / side / "result.csv", rows)
    return rows


def main():
    """5通貨・2方向・ATR24設定・出口10組の全2,400ケースを実行する。"""
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    if FAST_EMA_PERIOD >= SLOW_EMA_PERIOD:
        raise SystemExit("短期EMA期間は長期EMA期間より短くしてください")
    total_cases = (
        len(SYMBOLS) * len(SIDES) * len(ATR_PERIODS) * len(ATR_FILTERS)
        * len(TP_SL_PAIRS)
    )
    print(
        f"5通貨 × 2方向 × ATR期間{len(ATR_PERIODS)}種類 × "
        f"フィルター{len(ATR_FILTERS)}種類 × "
        f"TP/SL{len(TP_SL_PAIRS)}組 = {total_cases}ケース"
    )
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    print(
        f"共通シグナル: EMA{FAST_EMA_PERIOD}/EMA{SLOW_EMA_PERIOD}クロス / "
        f"ATR分布: 直前{ATR_PERCENTILE_LOOKBACK}本"
    )

    candles_by_symbol = {}
    entry_indexes = {}
    emas_by_symbol = {}
    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, warmup_count, backtest_count = load_candles(path)
        if backtest_count == 0:
            raise SystemExit(f"{symbol}の検証期間内データがありません")
        if warmup_count >= WARMUP_BARS:
            entry_index = warmup_count
            warmup_description = f"検証開始前{warmup_count}本"
        else:
            if len(candles) <= WARMUP_BARS:
                raise SystemExit(f"{symbol}のデータが不足しています")
            entry_index = WARMUP_BARS
            warmup_description = f"CSV先頭{WARMUP_BARS}本"
            print(f"[WARN] {symbol}: {warmup_description}をウォームアップに使用")
        candles_by_symbol[symbol] = candles
        entry_indexes[symbol] = entry_index
        emas_by_symbol[symbol] = (
            calculate_ema(candles, FAST_EMA_PERIOD),
            calculate_ema(candles, SLOW_EMA_PERIOD),
        )
        print(
            f"[LOAD] {symbol}: 検証{backtest_count}本 / "
            f"ウォームアップ: {warmup_description}"
        )

    all_rows = []
    cases_per_experiment = (
        len(ATR_PERIODS) * len(ATR_FILTERS) * len(TP_SL_PAIRS)
    )
    for symbol in SYMBOLS:
        for side in SIDES:
            print(f"[RUN] {symbol} {side}: {cases_per_experiment}ケース")
            fast_ema, slow_ema = emas_by_symbol[symbol]
            all_rows.extend(run_experiment(
                candles_by_symbol[symbol], symbol, side,
                entry_indexes[symbol], DEFAULT_OUTPUT_DIR, fast_ema, slow_ema,
            ))
    write_csv(DEFAULT_OUTPUT_DIR / "all_results.csv", all_rows)
    print(f"完了: {DEFAULT_OUTPUT_DIR}")
    print(f"実行時間: {time.perf_counter() - started:.2f}秒")
    return 0


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