FXのADXでおすすめの設定は？期間・基準値を5通貨1,000ケースで検証

"""5通貨・LONG/SHORTでADX期間と基準値の100設定を比較する。

+DIと-DIのクロスで方向を決め、ADXが基準値以上のときだけエントリーする。
出口条件は全設定共通のTP50・SL30に固定する。
"""

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


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

# ADXの計算、シグナル判定、売買には15分足だけを使用します。
ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)

# 既存のローソク足CSVに合わせ、ASK価格を読み込みます。
# BID価格、スプレッド、手数料、スリッページは考慮しません。
PRICE_TYPE = "ASK"

# CSV内のtimestamp_jstの日付が、この開始日から終了日までの足を検証します。
# 開始日と終了日のローソク足も検証対象に含みます。
BACKTEST_START_DATE = date(2024, 1, 1)
BACKTEST_END_DATE = date(2026, 7, 31)

# 検証する5つの通貨ペアと、買い・売りの2方向です。
SYMBOLS = ("USD_JPY", "EUR_JPY", "GBP_JPY", "EUR_USD", "AUD_USD")
SIDES = ("LONG", "SHORT")

# 「ADX期間」は値動きを何本分使って平滑化するかを表します。
# 「ADX基準値」はトレンドが十分に強いと判断する最低値です。
# ADXは方向を表さないため、売買方向は+DIと-DIのクロスで決めます。
ADX_PERIODS = (5, 7, 9, 12, 14, 18, 21, 25, 28, 35)
ADX_THRESHOLDS = (10, 15, 20, 25, 30, 35, 40, 45, 50, 55)
# 最長期間の初回ADXには「2 × period - 1」本目までの値動きが必要です。
WARMUP_BARS = max(ADX_PERIODS) * 2

# ADX設定を公平に比較するため、出口条件はTP50・SL30の1組に固定します。
TP_SL_PAIRS = ((50, 30),)

# __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"

# 期間・基準値・TP/SLごとの集計結果へ保存する列と、その並び順です。
RESULT_COLUMNS = (
    "通貨ペア", "売買方向", "ADX期間", "ADX基準値", "TP（pips）", "SL（pips）",
    "取引回数", "勝ち取引数", "負け取引数", "勝率（%）", "合計損益（pips）",
    "平均損益（pips）", "プロフィットファクター", "最大ドローダウン（pips）",
    "平均保有時間（分）", "最終足決済数",
)


def load_candles(path):
    """検証前のウォームアップ足と検証期間内のOHLCを読み込む。"""
    # deque(maxlen=WARMUP_BARS)は、上限を超えると古い足を自動的に捨てます。
    # そのため、検証開始前の全履歴ではなくADX計算に必要な本数だけを保持します。
    warmup = deque(maxlen=WARMUP_BARS)
    backtest = []
    # utf-8-sigを使うと、Excel向けのBOM付きCSVも読み込めます。
    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
            # CSVの文字列を、日時と計算可能なfloatへ変換します。
            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:
                # 検証開始前の足はADXを準備するためだけに使用します。
                warmup.append(candle)
            else:
                backtest.append(candle)
    return list(warmup) + backtest, len(warmup), len(backtest)


def calculate_adx(candles, period):
    """Wilder方式でADX、+DI、-DIを計算する。"""
    if period <= 0:
        raise ValueError("ADX期間は1以上にしてください")
    size = len(candles)
    # 必要本数が揃う前は計算できないため、結果をいったんNoneで埋めます。
    adx = [None] * size
    plus_di = [None] * size
    minus_di = [None] * size
    if size <= period:
        return adx, plus_di, minus_di

    # TRは窓開けを含む実質的な値幅です。
    # +DMは上方向の値動き、-DMは下方向の値動きを表します。
    true_ranges = [0.0] * size
    plus_dm = [0.0] * size
    minus_dm = [0.0] * size
    for index in range(1, size):
        current = candles[index]
        previous = candles[index - 1]
        # TRは次の3つのうち最大の値です。
        # 1. 現在足の高値－安値
        # 2. 現在足の高値－前足終値
        # 3. 現在足の安値－前足終値
        true_ranges[index] = max(
            current["high"] - current["low"],
            abs(current["high"] - previous["close"]),
            abs(current["low"] - previous["close"]),
        )
        upward = current["high"] - previous["high"]
        downward = previous["low"] - current["low"]
        # 上昇幅と下落幅を比べ、大きい方向だけをDMとして採用します。
        # 両方を同時に採用しないことがWilder方式のポイントです。
        plus_dm[index] = upward if upward > downward and upward > 0 else 0.0
        minus_dm[index] = downward if downward > upward and downward > 0 else 0.0

    # 最初の平滑値は、直近period本の単純な合計から開始します。
    smoothed_tr = sum(true_ranges[1:period + 1])
    smoothed_plus = sum(plus_dm[1:period + 1])
    smoothed_minus = sum(minus_dm[1:period + 1])
    dx = [None] * size
    for index in range(period, size):
        if index > period:
            # Wilderの平滑化:
            # 前回値－前回値÷期間＋今回値
            smoothed_tr = smoothed_tr - smoothed_tr / period + true_ranges[index]
            smoothed_plus = smoothed_plus - smoothed_plus / period + plus_dm[index]
            smoothed_minus = smoothed_minus - smoothed_minus / period + minus_dm[index]
        if smoothed_tr == 0:
            plus_di[index] = minus_di[index] = 0.0
        else:
            # +DIと-DIは、それぞれの方向の値動きが全体の値幅に占める割合です。
            plus_di[index] = 100.0 * smoothed_plus / smoothed_tr
            minus_di[index] = 100.0 * smoothed_minus / smoothed_tr
        denominator = plus_di[index] + minus_di[index]
        # DXは+DIと-DIの離れ具合を0～100で表します。
        dx[index] = (
            100.0 * abs(plus_di[index] - minus_di[index]) / denominator
            if denominator else 0.0
        )

    # 最初のADXは、計算済みのDXをperiod本平均して作ります。
    # そのため、ADX14の最初の値はindex=27に入ります。
    first_adx_index = period * 2 - 1
    if first_adx_index >= size:
        return adx, plus_di, minus_di
    adx[first_adx_index] = sum(dx[period:first_adx_index + 1]) / period
    for index in range(first_adx_index + 1, size):
        # 2つ目以降のADXもWilder方式でなめらかに更新します。
        adx[index] = (adx[index - 1] * (period - 1) + dx[index]) / period
    return adx, plus_di, minus_di


def make_trade(position, candle, exit_price, reason, side, pip_size):
    """決済した1取引の保有時間、pips損益、決済理由を作る。"""
    # LONGは値上がり、SHORTは値下がりしたときに損益をプラスにします。
    direction = 1.0 if side == "LONG" else -1.0
    return {
        # 日時差なので、土日やデータ欠損をまたぐ時間も含みます。
        "保有時間（分）": (
            candle["close_time"] - position["time"]
        ).total_seconds() / 60,
        # 円通貨は1pips=0.01円、それ以外は1pips=0.0001で換算します。
        "損益（pips）": direction * (exit_price - position["price"]) / pip_size,
        "決済理由": reason,
    }


def collect_trades(candles, adx, plus_di, minus_di, side, threshold, tp, sl,
                   pip_size, entry_start_index):
    """DIクロスかつADX基準値以上で入り、固定TP/SLで決済する。"""
    # positionがNoneならポジションを持っていません。
    # pending_entryは、現在足で確定した注文を次足始値で実行する予約です。
    position = None
    pending_entry = False
    trades = []
    # 前足と比較するため、2本目（index=1）から順番に処理します。
    for index in range(1, len(candles)):
        # 前の15分足でシグナルが確定していれば、現在足の始値で入ります。
        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

        # ポジション保有中は、現在足の高値・安値でTP/SL到達を確認します。
        if position is not None:
            tp_distance = tp * pip_size
            sl_distance = sl * pip_size
            if side == "LONG":
                # LONGは上昇でTP、下落でSLへ到達します。
                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:
                # SHORTは下落でTP、上昇でSLへ到達します。
                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
            # 15分足内の到達順を判別できない場合は、保守的に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

        # シグナル判定には現在足と前足のDI、および現在足のADXが必要です。
        values = (
            adx[index], plus_di[index], minus_di[index],
            plus_di[index - 1], minus_di[index - 1],
        )
        # ポジション保有中や、指標をまだ計算できない位置では判定しません。
        if position is not None or any(value is None for value in values):
            continue
        if side == "LONG":
            # LONGは+DIが-DIを下から上へ抜けた瞬間を探します。
            crossed = (
                plus_di[index - 1] <= minus_di[index - 1]
                and plus_di[index] > minus_di[index]
            )
        else:
            # SHORTは-DIが+DIを下から上へ抜けた瞬間を探します。
            crossed = (
                minus_di[index - 1] <= plus_di[index - 1]
                and minus_di[index] > plus_di[index]
            )
        # ADXが基準値以上なら、十分なトレンド強度があるものと判断します。
        if crossed and adx[index] >= threshold:
            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, period, threshold, tp, sl, trades):
    """取引ごとのpips損益から、1設定の全期間成績をまとめる。"""
    profits = [trade["損益（pips）"] for trade in trades]
    # 0pipsは勝ちにも負けにも含めませんが、取引回数には含めます。
    wins = [value for value in profits if value > 0]
    losses = [value for value in profits if value < 0]
    # 累積損益の過去最高値から、最大で何pips下落したかを計算します。
    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)
    # PF（プロフィットファクター）= 総利益 ÷ 総損失です。
    gross_loss = abs(sum(losses))
    durations = [trade["保有時間（分）"] for trade in trades]
    return {
        "通貨ペア": symbol, "売買方向": side, "ADX期間": period,
        "ADX基準値": threshold, "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へ保存する。"""
    # resultsや通貨・方向別フォルダがなければ自動的に作成します。
    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):
    """1通貨・1方向でADX期間×基準値の100設定を検証する。"""
    indicators = {period: calculate_adx(candles, period) for period in ADX_PERIODS}
    # USDJPY・EURJPY・GBPJPYは1pips=0.01円、それ以外は0.0001です。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    rows = []
    tp, sl = TP_SL_PAIRS[0]
    for period, (adx, plus_di, minus_di) in indicators.items():
        for threshold in ADX_THRESHOLDS:
            trades = collect_trades(
                candles, adx, plus_di, minus_di, side, threshold, tp, sl,
                pip_size, entry_start_index,
            )
            rows.append(summarize(
                symbol, side, period, threshold, tp, sl, trades,
            ))
    # 全件CSVとは別に、通貨・売買方向ごとの100設定も保存します。
    write_csv(output_dir / symbol / side / "result.csv", rows)
    return rows


def main():
    """5通貨・2方向・ADX100設定の全1,000ケースを実行する。"""
    # スクリプト全体の実行時間を測定します。
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    # 設定数から実際のケース数を計算するため、候補を変えても表示が追従します。
    total_cases = (
        len(SYMBOLS) * len(SIDES) * len(ADX_PERIODS) * len(ADX_THRESHOLDS)
    )
    print(
        f"5通貨 × 2方向 × ADX期間{len(ADX_PERIODS)}種類 × "
        f"基準値{len(ADX_THRESHOLDS)}種類 = {total_cases}ケース"
    )
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")

    # 同じ通貨のCSVをLONGとSHORTで2回読まないよう、先に1回だけ読み込みます。
    candles_by_symbol = {}
    entry_indexes = {}
    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:
            # CSVに検証開始前の足がない場合は、検証期間内の先頭70本を全設定で
            # 共通のウォームアップにします。短い期間だけ先に取引しないためです。
            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
        print(f"[LOAD] {symbol}: 検証{backtest_count}本 / ウォームアップ: {warmup_description}")

    # 通貨・方向別に検証しながら、全結果もall_rowsへ集めます。
    all_rows = []
    cases_per_experiment = len(ADX_PERIODS) * len(ADX_THRESHOLDS)
    for symbol in SYMBOLS:
        for side in SIDES:
            print(f"[RUN] {symbol} {side}: {cases_per_experiment}ケース")
            all_rows.extend(run_experiment(
                candles_by_symbol[symbol], symbol, side, entry_indexes[symbol],
                DEFAULT_OUTPUT_DIR,
            ))
    # 通貨・方向別CSVに加え、全1,000ケースをまとめたCSVも保存します。
    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())