FXのボリンジャーバンドでおすすめの設定は？期間・σを5通貨で徹底検証

"""5通貨・LONG/SHORTでボリンジャーバンド逆張りを一括検証する。

終値がいったんバンドの外へ出た後、バンド内へ戻った足でシグナルを確定する。
次の15分足の始値で約定し、固定pipsのTPまたはSLで決済する。
"""

import csv
import math
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/bollinger_band/run_bollinger_band_backtest.py
# 3. 結果は、このファイルと同じフォルダにあるresultsへ保存されます。

# ボリンジャーバンドの計算、シグナル判定、売買には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")

# 「期間」は移動平均と標準偏差を計算するローソク足の本数です。
# 「σ」は移動平均から上下バンドまでの標準偏差の倍率です。
BB_PERIODS = (5, 10, 15, 20, 21, 25, 30, 40, 50, 75, 100)
BB_SIGMAS = (1.0, 1.5, 2.0, 2.5, 3.0)
# 検証開始時点から最長期間のバンドを計算できるよう、開始前の足を使います。
WARMUP_BARS = max(BB_PERIODS)
# 各要素は「（利益確定幅, 損切り幅）」で、単位はpipsです。
# 例：(30, 10)なら、30pipsの利益または10pipsの損失で決済します。
TP_SL_PAIRS = (
    (20, 10),
    (30, 10),
    (30, 20),
    (50, 10),
    (50, 20),
    (50, 30),
    (80, 20),
    (80, 30),
    (100, 30),
    (100, 50),
)

# __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 = (
    "通貨ペア", "売買方向", "期間", "σ", "TP（pips）", "SL（pips）",
    "取引回数", "勝ち取引数", "負け取引数", "勝率（%）",
    "合計損益（pips）", "平均損益（pips）", "プロフィットファクター",
    "最大ドローダウン（pips）", "平均保有時間（分）", "最終足決済数",
)


def load_candles(path):
    """検証開始前のウォームアップ足と検証期間内のOHLCを読み込む。"""
    # deque(maxlen=100)は、新しい足を追加すると古い足を自動的に捨てます。
    # そのため、検証開始前の全履歴ではなく直近100本だけを保持できます。
    warmup = deque(maxlen=WARMUP_BARS)
    backtest_candles = []
    # 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:
                # 全履歴ではなく、指標計算に必要な直近本数だけを保持します。
                warmup.append(candle)
            else:
                backtest_candles.append(candle)
    return list(warmup) + backtest_candles, len(warmup), len(backtest_candles)


def calculate_bollinger_bands(closes, period, sigma):
    """終値のSMAと母標準偏差（ddof=0）から上下バンドを計算する。"""
    if period < 2:
        raise ValueError("ボリンジャーバンド期間は2以上にしてください")
    if sigma <= 0:
        raise ValueError("σは0より大きくしてください")

    # 必要本数が揃う前は計算できないため、結果をいったんNoneで埋めます。
    middle = [None] * len(closes)
    upper = [None] * len(closes)
    lower = [None] * len(closes)
    window = deque()
    window_sum = 0.0
    window_sum_squares = 0.0

    # 終値の合計と二乗和を更新しながら計算すると、各位置で期間分を
    # 毎回足し直すより高速に全設定を検証できます。
    for index, value in enumerate(closes):
        window.append(value)
        window_sum += value
        window_sum_squares += value * value
        if len(window) > period:
            removed = window.popleft()
            window_sum -= removed
            window_sum_squares -= removed * removed
        if len(window) < period:
            continue

        # middleは単純移動平均線（SMA）です。
        mean = window_sum / period
        # 丸め誤差でごく小さい負数になる場合があるため0を下限にします。
        variance = max(window_sum_squares / period - mean * mean, 0.0)
        standard_deviation = math.sqrt(variance)
        # 上側バンド = SMA + σ × 標準偏差
        # 下側バンド = SMA - σ × 標準偏差
        middle[index] = mean
        upper[index] = mean + sigma * standard_deviation
        lower[index] = mean - sigma * standard_deviation

    return middle, upper, lower


def make_trade(position, exit_time, exit_price, reason, side, pip_size):
    """決済した1取引の保有時間、pips損益、決済理由を作る。"""
    # LONGは値上がり、SHORTは値下がりしたときに損益をプラスにします。
    direction = 1.0 if side == "LONG" else -1.0
    return {
        # 日時差なので、土日やデータ欠損をまたぐ時間も含みます。
        "保有時間（分）": (exit_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, upper, lower, side, tp_pips, sl_pips, pip_size,
    entry_start_index,
):
    """バンド外から内側へ復帰した次足で入り、固定TP/SLで決済する。"""
    # positionがNoneならポジションを持っていません。
    # pending_entryは、現在足で確定した注文を次足始値で実行する予約です。
    position = None
    pending_entry = False
    trades = []

    # 前足と比較するため、2本目（index=1）から順番に処理します。
    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

        # ポジション保有中は、現在足の高値・安値でTP/SL到達を確認します。
        if position is not None:
            tp_distance = tp_pips * pip_size
            sl_distance = sl_pips * 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:
                exit_price, reason = sl_price, "SL"
            elif tp_hit:
                exit_price, reason = tp_price, "TP"
            else:
                exit_price = reason = None
            if exit_price is not None:
                trades.append(make_trade(
                    position, candles[index]["close_time"], exit_price,
                    reason, side, pip_size,
                ))
                position = None

        # 前足と現在足の終値・バンドを比較し、バンド内へ戻った瞬間を探します。
        current_close = candles[index]["close"]
        previous_close = candles[index - 1]["close"]
        if side == "LONG":
            current_band = lower[index]
            previous_band = lower[index - 1]
            signal = (
                previous_band is not None
                and current_band is not None
                and previous_close < previous_band
                and current_close >= current_band
            )
        else:
            current_band = upper[index]
            previous_band = upper[index - 1]
            signal = (
                previous_band is not None
                and current_band is not None
                and previous_close > previous_band
                and current_close <= current_band
            )
        # ポジションがないときだけ、次足の注文を予約します。
        if position is None and signal:
            pending_entry = True

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


def summarize(symbol, side, period, sigma, 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,
        "期間": period,
        "σ": sigma,
        "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):
    """集計結果を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=RESULT_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)


def run_experiment(candles, symbol, side, entry_start_index, output_dir):
    """1通貨・1売買方向の期間・σ・TP/SL候補を検証する。"""
    # 同じ期間・σのバンドをTP/SLごとに再計算しないよう先に作ります。
    closes = [candle["close"] for candle in candles]
    indicators = {
        (period, sigma): calculate_bollinger_bands(closes, period, sigma)[1:]
        for period in BB_PERIODS
        for sigma in BB_SIGMAS
    }
    # USDJPY・EURJPY・GBPJPYは1pips=0.01円、それ以外は0.0001です。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    rows = []
    for (period, sigma), (upper, lower) in indicators.items():
        for tp, sl in TP_SL_PAIRS:
            trades = collect_trades(
                candles, upper, lower, side, tp, sl, pip_size,
                entry_start_index,
            )
            rows.append(summarize(
                symbol, side, period, sigma, tp, sl, trades,
            ))
    # 1通貨・1方向の550ケースを1つのresult.csvへ保存します。
    write_csv(output_dir / symbol / side / "result.csv", rows)
    return rows


def main():
    """全5,500ケースを実行し、通貨別結果と全件結果を保存する。"""
    # スクリプト全体の実行時間を測定します。
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    # 設定数から実際のケース数を計算するため、候補を変えても表示が追従します。
    total_cases = (
        len(BB_PERIODS) * len(BB_SIGMAS) * len(TP_SL_PAIRS)
        * len(SYMBOLS) * len(SIDES)
    )
    print(
        f"期間{len(BB_PERIODS)}種類 × σ {len(BB_SIGMAS)}種類 "
        f"× TP/SL {len(TP_SL_PAIRS)}組 × 5通貨 × 2方向 "
        f"= {total_cases}ケース"
    )
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")

    # 同じ通貨のCSVをLONGとSHORTで2回読まないよう、先に1回だけ読み込みます。
    candles_by_symbol = {}
    entry_start_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_start_index = warmup_count
            warmup_description = f"検証開始前{warmup_count}本"
        else:
            # CSVに検証開始前の足がない場合は、検証期間内の先頭100本を全設定で
            # 共通のウォームアップにします。短い期間だけ先に取引しないためです。
            if len(candles) <= WARMUP_BARS:
                raise SystemExit(
                    f"{symbol}のデータが不足しています: "
                    f"{len(candles)}本（ウォームアップに必要{WARMUP_BARS}本）"
                )
            entry_start_index = WARMUP_BARS
            warmup_description = f"CSV先頭{WARMUP_BARS}本"
            first_entry_time = candles[entry_start_index]["open_time"]
            print(
                f"[WARN] {symbol}: 検証開始前の足が{warmup_count}本のため、"
                f"{warmup_description}をウォームアップに使用します。"
                f"エントリー開始: {first_entry_time.isoformat()}"
            )
        candles_by_symbol[symbol] = candles
        entry_start_indexes[symbol] = entry_start_index
        print(
            f"[LOAD] {symbol}: 検証{backtest_count}本 "
            f"/ ウォームアップ: {warmup_description}"
        )

    # 通貨・方向別に検証しながら、全結果もall_rowsへ集めます。
    all_rows = []
    cases_per_experiment = len(BB_PERIODS) * len(BB_SIGMAS) * len(TP_SL_PAIRS)
    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_start_indexes[symbol], DEFAULT_OUTPUT_DIR,
            ))

    # 個別CSVに加え、5,500ケースを一括比較できる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())