FXのMACDで最適な期間・設定値は？5通貨・3,100ケースで検証

"""5通貨・LONG/SHORTでMACD 31設定とTP/SL 10組を一括検証する。"""

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


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

# MACDの計算、シグナル判定、売買には15分足だけを使います。
ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)
# 入力CSVに合わせてASK価格を読み込みます。
# BID価格、スプレッド、手数料、スリッページはこの検証に含みません。
PRICE_TYPE = "ASK"
# 開始日と終了日のローソク足も検証対象に含めます。
BACKTEST_START_DATE = date(2024, 1, 1)
BACKTEST_END_DATE = date(2026, 7, 31)

# 全通貨・全方向を同じ候補群で比較し、ブログのTOP5を公平に選べるようにする。
SYMBOLS = ("USD_JPY", "EUR_JPY", "GBP_JPY", "EUR_USD", "AUD_USD")
SIDES = ("LONG", "SHORT")

# 各要素は「（短期EMA期間, 長期EMA期間, シグナルEMA期間）」です。
# 例：定番の(12, 26, 9)は、短期EMA12、長期EMA26、シグナルEMA9を表します。
# MACD線 = 短期EMA - 長期EMA、シグナル線 = MACD線のEMAです。
MACD_SETTINGS = (
    (3, 15, 3),
    (3, 20, 3),
    (3, 40, 3),
    (3, 60, 3),
    (5, 15, 5),
    (5, 20, 5),
    (5, 26, 9),
    (6, 15, 9),
    (6, 18, 5),
    (6, 22, 4),
    (7, 12, 9),
    (7, 15, 7),
    (7, 18, 7),
    (8, 20, 9),
    (8, 22, 16),
    (8, 26, 14),
    (8, 35, 12),
    (10, 22, 14),
    (10, 26, 12),
    (12, 20, 9),
    (12, 26, 9),
    (12, 50, 12),
    (14, 22, 14),
    (16, 20, 12),
    (16, 21, 11),
    (16, 30, 7),
    (16, 38, 6),
    (20, 30, 6),
    (20, 35, 5),
    (22, 50, 7),
    (24, 50, 6),
)

# 各要素は「（利益確定幅, 損切り幅）」で、単位はpipsです。
# 例：(100, 20)なら、100pipsの利益または20pipsの損失で決済します。
TP_SL_PAIRS = (
    (80, 10),
    (100, 10),
    (120, 10),
    (150, 10),
    (180, 10),
    (100, 20),
    (120, 20),
    (120, 30),
    (150, 30),
    (150, 50),
)

# 5通貨 × LONG/SHORTの10実験へ、同じMACD・TP/SL候補を割り当てます。
# 1実験310ケース、全体では31 × 10 × 5 × 2 = 3,100ケースです。
EXPERIMENTS = {
    (symbol, side): {"settings": MACD_SETTINGS, "tp_sl": TP_SL_PAIRS}
    for symbol in SYMBOLS
    for side in SIDES
}

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


def load_candles(path):
    """fetch_gmo_candles.py形式のOHLC CSVを読み込む。"""
    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"])
            # datetime同士ではなくdate同士で比較することで、終了日の夜の足も含めます。
            if BACKTEST_START_DATE <= timestamp.date() <= BACKTEST_END_DATE:
                candles.append({
                    # timestamp_jstは15分足が始まった時刻です。
                    "open_time": timestamp,
                    "close_time": timestamp + ENTRY_TIMEFRAME_DURATION,
                    # CSVの文字列を計算可能なfloatへ変換します。
                    "open": float(row["open"]),
                    "high": float(row["high"]),
                    "low": float(row["low"]),
                    "close": float(row["close"]),
                })
    return candles


def calculate_ema(values, period):
    """最初のperiod本の単純平均を初期値としてEMAを計算する。"""
    if period <= 0:
        raise ValueError("EMA期間は1以上である必要があります")
    # EMAをまだ計算できない位置にはNoneを入れます。
    # EMA3なら、最初の2本はNone、3本目が最初のEMAになります。
    result = [None] * len(values)
    if len(values) < period:
        return result
    # 最初のEMAだけは、最初のperiod本の単純平均を使います。
    result[period - 1] = sum(values[:period]) / period
    # EMAは新しい価格を強く反映します。期間が短いほどalphaは大きくなります。
    # EMA = 今回価格 × alpha + 前回EMA × (1 - alpha)
    alpha = 2.0 / (period + 1)
    for index in range(period, len(values)):
        result[index] = alpha * values[index] + (1.0 - alpha) * result[index - 1]
    return result


def calculate_macd(closes, fast_period, slow_period, signal_period):
    """MACD線、シグナル線、ヒストグラムを返す。"""
    if fast_period >= slow_period:
        raise ValueError("短期EMAは長期EMAより短くしてください")
    fast = calculate_ema(closes, fast_period)
    slow = calculate_ema(closes, slow_period)
    # MACD線は短期EMAから長期EMAを引いた値です。
    # プラスなら短期の価格傾向が長期より上、マイナスなら下にあります。
    macd = [
        fast_value - slow_value
        if fast_value is not None and slow_value is not None else None
        for fast_value, slow_value in zip(fast, slow)
    ]

    # MACD線を計算できる最初の位置から、そのEMAであるシグナル線を作ります。
    first = next((i for i, value in enumerate(macd) if value is not None), None)
    signal = [None] * len(closes)
    if first is not None:
        compact_signal = calculate_ema(macd[first:], signal_period)
        signal[first:] = compact_signal
    # ヒストグラム = MACD線 - シグナル線です。
    # 今回の売買判定では直接使いませんが、一般的なMACD値として返します。
    histogram = [
        macd_value - signal_value
        if macd_value is not None and signal_value is not None else None
        for macd_value, signal_value in zip(macd, signal)
    ]
    return macd, signal, histogram


def collect_trades(candles, macd, signal, side, tp_pips, sl_pips, pip_size):
    """MACDクロスの次足始値で入り、固定TP/SLで決済する。"""
    # positionがNoneなら、ポジションを持っていない状態です。
    # pendingには、現在足で確定した注文を次足始値で実行するための
    # MACD線とシグナル線を一時保存します。
    position = None
    pending = None
    trades = []
    for index in range(1, len(candles)):
        # 前の15分足でクロスが確定していれば、現在足の始値で入ります。
        # クロスした足の終値で入らないため、確定前の情報を使う未来参照を防げます。
        if position is None and pending is not None:
            position = {
                "time": candles[index]["open_time"],
                "price": candles[index]["open"],
                "macd": pending[0],
                "signal": pending[1],
            }
        pending = None

        # ポジション保有中は、現在足の高値・安値がTP/SLへ届いたか確認します。
        if position is not None:
            # pipsを実際の価格差へ変換します。
            # USDJPYなら100pips × 0.01円 = 1円です。
            tp_distance = tp_pips * pip_size
            sl_distance = sl_pips * pip_size
            if side == "LONG":
                # 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:
                # SHORTは価格下落で利益、価格上昇で損失になります。
                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分足だけでは、足の中でTPとSLのどちらへ先に届いたか分かりません。
            # 同一足で両方に届いた場合は、成績を楽観視しないようSLを優先します。
            if sl_hit:
                exit_price, reason = sl_price, f"SL {sl_pips}pips"
            elif tp_hit:
                exit_price, reason = tp_price, f"TP {tp_pips}pips"
            else:
                exit_price = reason = None
            if exit_price is not None:
                # 決済した1取引を記録し、再びポジションなしの状態へ戻します。
                trades.append(make_trade(
                    position, candles[index]["close_time"], exit_price,
                    reason, side, pip_size,
                ))
                position = None

        # 前足と現在足のMACD線・シグナル線を比べ、交差した瞬間を探します。
        current_macd, previous_macd = macd[index], macd[index - 1]
        current_signal, previous_signal = signal[index], signal[index - 1]
        if None in (current_macd, previous_macd, current_signal, previous_signal):
            continue
        if position is None:
            # LONGはMACD線がシグナル線を下から上へ抜けるゴールデンクロスです。
            if (
                side == "LONG"
                and previous_macd <= previous_signal
                and current_macd > current_signal
            ):
                pending = (current_macd, current_signal)
            # SHORTはMACD線がシグナル線を上から下へ抜けるデッドクロスです。
            elif (
                side == "SHORT"
                and previous_macd >= previous_signal
                and current_macd < current_signal
            ):
                pending = (current_macd, current_signal)

    # データの最後までポジションが残った場合は、最終足の終値で決済します。
    if position is not None:
        trades.append(make_trade(
            position, candles[-1]["close_time"], candles[-1]["close"],
            "データ終了", side, pip_size,
        ))
    return trades


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,
    }


def summarize(symbol, side, fast, slow, signal_period, 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下落したかを調べます。
    # 例：累積損益が100、300、150なら、300から150への150pipsがDDです。
    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（プロフィットファクター）= 勝ち取引の総利益 ÷ 負け取引の総損失です。
    # PFが1より大きければ、取引コストを考えない総利益は総損失を上回ります。
    gross_loss = abs(sum(losses))
    durations = [trade["保有時間（分）"] for trade in trades]
    return {
        "通貨ペア": symbol, "売買方向": side, "短期EMA": fast,
        "長期EMA": slow, "シグナルEMA": signal_period, "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,
    }


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 macd_settings(config):
    """実験設定から有効なMACD期間を重複なしで作る。"""
    # 現行設定のようにsettingsへ候補を直接列挙した場合はこちらを使います。
    if "settings" in config:
        settings = []
        for fast, slow, signal in config["settings"]:
            if fast >= slow:
                raise ValueError(f"短期EMA {fast}は長期EMA {slow}より短くしてください")
            setting = (fast, slow, signal)
            if setting not in settings:
                settings.append(setting)
        return settings

    # fast・slow・signalを別々に指定する過去形式にも対応しています。
    # fast < slowを満たす組み合わせだけを残します。
    settings = [
        (fast, slow, signal)
        for fast in config["fast"]
        for slow in config["slow"]
        for signal in config["signal"]
        if fast < slow
    ]

    for fast, slow in config.get("extra_fast_slow", ()):
        if fast >= slow:
            raise ValueError(f"短期EMA {fast}は長期EMA {slow}より短くしてください")
        for signal in config["signal"]:
            setting = (fast, slow, signal)
            if setting not in settings:
                settings.append(setting)
    return settings


def run_experiment(candles, symbol, side, config, output_dir):
    """1通貨・1売買方向のMACD・TP/SL候補を検証する。"""
    # 同じMACD設定をTP/SLごとに再計算しないよう、最初にまとめて計算します。
    closes = [candle["close"] for candle in candles]
    indicators = {
        setting: calculate_macd(closes, *setting)[:2]
        for setting in macd_settings(config)
    }
    # USDJPY・EURJPY・GBPJPYは1pips=0.01円、それ以外は0.0001です。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    rows = []
    for (fast, slow, signal_period), (macd, signal) in indicators.items():
        for tp, sl in config["tp_sl"]:
            trades = collect_trades(
                candles, macd, signal, side, tp, sl, pip_size,
            )
            rows.append(summarize(
                symbol, side, fast, slow, signal_period, tp, sl, trades,
            ))
    # 1通貨・1方向の310ケースを、1つのresult.csvへ保存します。
    write_csv(output_dir / symbol / side / "result.csv", rows)
    return rows


def main():
    """3,100ケースを実行し、通貨別結果と全件結果を保存する。"""
    # スクリプト全体の実行時間を測ります。
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    # 設定から実際のケース数を計算するため、候補数を変えても表示が追従します。
    total_cases = sum(
        len(macd_settings(config)) * len(config["tp_sl"])
        for config in EXPERIMENTS.values()
    )
    print(
        f"MACD設定{len(MACD_SETTINGS)}種類 × TP/SL {len(TP_SL_PAIRS)}組 "
        f"× 5通貨 × 2方向 = {total_cases}ケース"
    )
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    all_rows = []
    candles_by_symbol = {}
    # 同じ通貨のCSVをLONGとSHORTで2回読まないよう、最初に1回だけ読み込みます。
    for symbol in dict.fromkeys(symbol for symbol, _side in EXPERIMENTS):
        path = DEFAULT_DATA_DIR / symbol / f"{ENTRY_TIMEFRAME}_{PRICE_TYPE}.csv"
        if not path.is_file():
            raise SystemExit(f"入力CSVが見つかりません: {path}")
        candles = load_candles(path)
        symbol_configs = [
            config
            for (experiment_symbol, _side), config in EXPERIMENTS.items()
            if experiment_symbol == symbol
        ]
        # 最長EMAとシグナルEMAを計算できるだけのローソク足があるか確認します。
        required = max(
            slow + signal
            for config in symbol_configs
            for _fast, slow, signal in macd_settings(config)
        )
        if len(candles) <= required:
            raise SystemExit(f"{symbol}のデータが不足しています: {len(candles)}本")
        candles_by_symbol[symbol] = candles
        print(f"[LOAD] {symbol}: {len(candles)}本")

    # 通貨・売買方向を1組ずつ検証し、全結果もall_rowsへ集めます。
    for (symbol, side), config in EXPERIMENTS.items():
        cases = len(macd_settings(config)) * len(config["tp_sl"])
        print(f"[RUN] {symbol} {side}: {cases}ケース")
        all_rows.extend(run_experiment(
            candles_by_symbol[symbol], symbol, side, config, DEFAULT_OUTPUT_DIR,
        ))
    # 通貨別CSVに加え、3,100ケースを一括比較できる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())