FXのCCIでおすすめの設定は？5通貨1,340ケースで検証

"""5通貨の15分足でCCIの期間・基準値を比較する。

CCIが極端な水準の外側から内側へ戻った確定足をシグナルとし、次の
15分足始値でエントリーする。出口は全ケースでTP50/SL30 pipsに固定する。
"""

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

# CCIの計算、シグナル判定、売買には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")

# CCI期間は、直近何本の値動きを基準に現在価格の離れ具合を測るかを表します。
# 一般的な14・20を含め、短期から長期までを比較します。
CCI_PERIODS = (5, 7, 9, 10, 14, 18, 20, 21, 25, 28, 30, 35, 40)

# 上下対称の基準値です。例えば100なら、LONGは-100、SHORTは+100を使います。
CCI_LEVELS = (50, 75, 100, 125, 150, 175, 200, 225, 250, 275, 300)
# CCIで一般的に使われる補正定数です。値の多くが±100付近に収まりやすくなります。
CCI_CONSTANT = 0.015

# 期間と基準値だけを公平に比較するため、出口は全設定で固定します。
TAKE_PROFIT_PIPS = 50
STOP_LOSS_PIPS = 30

# 期間の短い設定だけ先に取引を開始しないよう、全設定で開始位置を揃える。
WARMUP_BARS = max(CCI_PERIODS)

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

# 全結果と通貨・方向別result.csvへ保存する列と、その並び順です。
RESULT_COLUMNS = (
    "通貨ペア", "売買方向", "CCI期間", "CCI基準値", "エントリー方式",
    "CCI理論最大絶対値", "理論上シグナル発生可能",
    "TP（pips）", "SL（pips）", "取引回数", "勝ち取引数", "負け取引数",
    "勝率（%）", "合計損益（pips）", "平均損益（pips）",
    "プロフィットファクター", "最大ドローダウン（pips）",
    "平均保有時間（分）", "最終足決済数", "全期間プラス月数",
    "全期間マイナス月数", "直近6か月取引回数",
    "直近6か月合計損益（pips）", "直近6か月平均損益（pips）",
    "直近6か月プロフィットファクター", "直近6か月プラス月数",
    "直近6か月マイナス月数", "記事候補条件合格",
)

# 月別CSV自体は保存しませんが、月ごとの内部集計を作る際の列順として使います。
MONTHLY_RESULT_COLUMNS = (
    "通貨ペア", "売買方向", "CCI期間", "CCI基準値", "TP（pips）",
    "SL（pips）", "年月", "取引回数", "勝ち取引数", "負け取引数",
    "勝率（%）", "合計損益（pips）", "平均損益（pips）",
    "プロフィットファクター",
)

# 「記事候補条件合格」をTrueにする最低条件です。
# 全期間だけでなく直近6か月も確認し、一時的な好成績を除きやすくします。
MIN_CANDIDATE_TRADES = 100
MIN_CANDIDATE_PROFIT_FACTOR = 1.05
MIN_RECENT_POSITIVE_MONTHS = 3


def load_candles(path):
    """検証開始前のウォームアップ足と検証期間内のOHLCを読む。"""
    # deque(maxlen=40)は上限を超えると古い足を自動的に捨てるため、
    # 検証開始前の全データではなく、CCI計算に必要な直近40本だけを残せます。
    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の文字列を、計算に使えるdatetimeと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:
                # 検証開始前の足はCCIを準備するためだけに使い、売買には使いません。
                warmup.append(candle)
            else:
                backtest.append(candle)
    return list(warmup) + backtest, len(warmup), len(backtest)


def calculate_cci(candles, period, constant=CCI_CONSTANT):
    """典型価格と平均偏差を使った標準的なCCIを返す。"""
    if period <= 0:
        raise ValueError("CCI期間は1以上にしてください")
    if constant <= 0:
        raise ValueError("CCI定数は0より大きくしてください")

    # 典型価格 =（高値 + 安値 + 終値）÷ 3 です。
    # 終値だけでなく、その足の高値・安値も含めた代表価格として使います。
    typical_prices = [
        (candle["high"] + candle["low"] + candle["close"]) / 3.0
        for candle in candles
    ]
    # period本が揃う前はCCIを計算できないため、先頭をNoneで埋めます。
    cci = [None] * len(candles)
    for index in range(period - 1, len(candles)):
        # 現在足を含む直近period本だけを計算窓として取り出します。
        window = typical_prices[index - period + 1:index + 1]
        # averageは期間内の典型価格の単純平均です。
        average = sum(window) / period
        # 各典型価格が平均からどれだけ離れたかを絶対値で平均します。
        mean_deviation = sum(abs(value - average) for value in window) / period
        # CCI =（現在の典型価格 - 平均）÷（0.015 × 平均偏差）です。
        # 全足が同値で平均偏差が0なら、中立値の0としてゼロ除算を避けます。
        cci[index] = (
            0.0
            if mean_deviation == 0.0
            else (typical_prices[index] - average) / (constant * mean_deviation)
        )
    return cci


def theoretical_cci_max(period, constant=CCI_CONSTANT):
    """指定期間のCCIが取り得る最大絶対値を返す。"""
    # 短いCCIでは、計算上どうしても届かない大きな基準値があります。
    # 例: 期間5の最大絶対値は約166.67なので、基準値200には届きません。
    return period / (2.0 * constant)


def is_level_reachable(period, level):
    """CCIが理論上その基準値へ到達できるかを返す。"""
    return level <= theoretical_cci_max(period)


def make_trade(position, candle, exit_price, reason, side, pip_size):
    """決済済み取引を作る。"""
    # LONGは値上がり、SHORTは値下がりしたときに損益がプラスになります。
    direction = 1.0 if side == "LONG" else -1.0
    return {
        # 円通貨は1pips=0.01円、それ以外は1pips=0.0001で換算します。
        "損益（pips）": direction * (exit_price - position["price"]) / pip_size,
        # 日時差なので、土日やデータ欠損をまたぐ時間も含みます。
        "保有時間（分）": (
            candle["close_time"] - position["time"]
        ).total_seconds() / 60,
        "決済理由": reason,
        # 月別集計は入力CSVのtimestamp_jst（足の開始時刻）へ帰属させる。
        # 最終日の23:45足を翌月扱いにせず、検証対象月内へ確実に含める。
        "決済時刻": candle["open_time"],
    }


def collect_trades(candles, cci, side, level, pip_size, entry_start_index):
    """CCIが圏外から基準値の内側へ戻った後、次足始値で入る。"""
    if side not in SIDES:
        raise ValueError("sideはLONGまたはSHORTにしてください")
    if level <= 0:
        raise ValueError("CCI基準値は0より大きくしてください")

    # 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 = TAKE_PROFIT_PIPS * pip_size
            sl_distance = STOP_LOSS_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

            # OHLCだけでは足内の到達順が分からないため保守的に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

        # 前足と現在足のCCIを比較し、基準値の内側へ戻った瞬間を探します。
        previous_cci = cci[index - 1]
        current_cci = cci[index]
        if position is not None or previous_cci is None or current_cci is None:
            # 保有中は新規シグナルを無視します。NoneはCCI未計算区間です。
            continue

        if side == "LONG":
            # -level以下から-levelを上抜けたときに売られすぎ圏からの復帰とする。
            crossed_back = previous_cci <= -level and current_cci > -level
        else:
            # +level以上から+levelを下抜けたときに買われすぎ圏からの復帰とする。
            crossed_back = previous_cci >= level and current_cci < level

        if crossed_back and index >= entry_start_index:
            # この足ではまだ約定せず、次のループの足始値で入ります。
            pending_entry = True

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


def calculate_metrics(trades):
    """指定された取引群の基本成績を計算する。"""
    profits = [trade["損益（pips）"] for trade in trades]
    # 0pipsは勝ちにも負けにも含めませんが、取引回数には含めます。
    wins = [profit for profit in profits if profit > 0]
    losses = [profit for profit in profits if profit < 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 {
        "取引回数": 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 iter_months(start_date, end_date):
    """開始月から終了月までをYYYY-MM形式で返す。"""
    # 2024年12月の次を2025年1月に進めながら、対象月を順番に作ります。
    year, month = start_date.year, start_date.month
    result = []
    while (year, month) <= (end_date.year, end_date.month):
        result.append(f"{year:04d}-{month:02d}")
        month += 1
        if month == 13:
            year += 1
            month = 1
    return result


# 全期間は2024-01～2026-07の31か月、直近6か月は最後の6要素です。
BACKTEST_MONTHS = tuple(iter_months(BACKTEST_START_DATE, BACKTEST_END_DATE))
RECENT_SIX_MONTHS = frozenset(BACKTEST_MONTHS[-6:])


def build_monthly_rows(symbol, side, period, level, trades):
    """取引を決済月別に集計し、取引0回の月も含めて返す。"""
    rows = []
    for month in BACKTEST_MONTHS:
        # 決済がこの年月に属する取引だけを抜き出します。
        selected = [
            trade for trade in trades
            if trade["決済時刻"].strftime("%Y-%m") == month
        ]
        metrics = calculate_metrics(selected)
        rows.append({
            "通貨ペア": symbol,
            "売買方向": side,
            "CCI期間": period,
            "CCI基準値": level,
            "TP（pips）": TAKE_PROFIT_PIPS,
            "SL（pips）": STOP_LOSS_PIPS,
            "年月": month,
            **{column: metrics[column] for column in MONTHLY_RESULT_COLUMNS[7:]},
        })
    return rows


def summarize(symbol, side, period, level, trades, monthly_rows):
    """1設定の全期間・月別・直近6か月成績を集計する。"""
    # まず全期間の成績を作り、その後に同じ方法で直近6か月を集計します。
    metrics = calculate_metrics(trades)
    recent_trades = [
        trade for trade in trades
        if trade["決済時刻"].strftime("%Y-%m") in RECENT_SIX_MONTHS
    ]
    recent_metrics = calculate_metrics(recent_trades)
    monthly_profits = [row["合計損益（pips）"] for row in monthly_rows]
    recent_monthly_profits = [
        row["合計損益（pips）"]
        for row in monthly_rows
        if row["年月"] in RECENT_SIX_MONTHS
    ]
    theoretical_max = theoretical_cci_max(period)
    signal_possible = is_level_reachable(period, level)
    profit_factor = metrics["プロフィットファクター"]
    # 記事候補は、取引数・全期間PF・直近成績をすべて満たした設定です。
    candidate_passed = (
        signal_possible
        and metrics["取引回数"] >= MIN_CANDIDATE_TRADES
        and metrics["合計損益（pips）"] > 0
        and profit_factor != ""
        and profit_factor >= MIN_CANDIDATE_PROFIT_FACTOR
        and recent_metrics["合計損益（pips）"] > 0
        and sum(profit > 0 for profit in recent_monthly_profits)
        >= MIN_RECENT_POSITIVE_MONTHS
    )
    return {
        "通貨ペア": symbol,
        "売買方向": side,
        "CCI期間": period,
        "CCI基準値": level,
        "エントリー方式": "圏外からの復帰",
        "CCI理論最大絶対値": round(theoretical_max, 2),
        "理論上シグナル発生可能": signal_possible,
        "TP（pips）": TAKE_PROFIT_PIPS,
        "SL（pips）": STOP_LOSS_PIPS,
        **metrics,
        "最終足決済数": sum(
            trade["決済理由"] == "データ終了" for trade in trades
        ),
        "全期間プラス月数": sum(profit > 0 for profit in monthly_profits),
        "全期間マイナス月数": sum(profit < 0 for profit in monthly_profits),
        "直近6か月取引回数": recent_metrics["取引回数"],
        "直近6か月合計損益（pips）": recent_metrics["合計損益（pips）"],
        "直近6か月平均損益（pips）": recent_metrics["平均損益（pips）"],
        "直近6か月プロフィットファクター": recent_metrics["プロフィットファクター"],
        "直近6か月プラス月数": sum(
            profit > 0 for profit in recent_monthly_profits
        ),
        "直近6か月マイナス月数": sum(
            profit < 0 for profit in recent_monthly_profits
        ),
        "記事候補条件合格": candidate_passed,
    }


def write_csv(path, rows, columns=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=columns)
        writer.writeheader()
        writer.writerows(rows)


def run_symbol(candles, symbol, entry_start_index, output_dir):
    """1通貨について理論上比較可能な全CCI設定を検証する。"""
    # USDJPY・EURJPY・GBPJPYは1pips=0.01円、それ以外は0.0001です。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    # LONGとSHORTを別々のCSVへ保存するため、方向別に結果を貯めます。
    rows_by_side = {side: [] for side in SIDES}

    # CCIは期間ごとに1回だけ計算し、方向・基準値間で共有する。
    for period in CCI_PERIODS:
        cci = calculate_cci(candles, period)
        for side in SIDES:
            for level in CCI_LEVELS:
                # 期間5・基準値200など、計算上届かない組み合わせは除外します。
                if not is_level_reachable(period, level):
                    continue
                trades = collect_trades(
                    candles, cci, side, level, pip_size, entry_start_index,
                )
                # 月別CSVは保存しませんが、月数と直近6か月の列を作るため集計します。
                setting_monthly_rows = build_monthly_rows(
                    symbol, side, period, level, trades,
                )
                rows_by_side[side].append(
                    summarize(
                        symbol, side, period, level, trades,
                        setting_monthly_rows,
                    )
                )

    # 1通貨・1方向の134ケースをresult.csvへ保存し、全件用にも返します。
    rows = []
    for side in SIDES:
        side_rows = rows_by_side[side]
        write_csv(output_dir / symbol / side / "result.csv", side_rows)
        rows.extend(side_rows)
    return rows


def main():
    """5通貨・2方向の理論上比較可能なCCI設定を検証する。"""
    # スクリプト全体の実行時間を測り始めます。
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    if TAKE_PROFIT_PIPS <= 0 or STOP_LOSS_PIPS <= 0:
        raise SystemExit("TPとSLは1pips以上にしてください")

    # 候補の直積から、CCIの理論上限へ届かない組み合わせを取り除きます。
    valid_parameter_pairs = tuple(
        (period, level)
        for period in CCI_PERIODS
        for level in CCI_LEVELS
        if is_level_reachable(period, level)
    )
    excluded_parameter_pairs = len(CCI_PERIODS) * len(CCI_LEVELS) - len(
        valid_parameter_pairs
    )
    total_cases = len(SYMBOLS) * len(SIDES) * len(valid_parameter_pairs)
    print(
        f"{len(SYMBOLS)}通貨 × {len(SIDES)}方向 × "
        f"CCI期間{len(CCI_PERIODS)}種類 × 基準値{len(CCI_LEVELS)}種類 / "
        f"有効な期間・基準値{len(valid_parameter_pairs)}組 = {total_cases}ケース"
    )
    print(f"理論上到達不能のため除外: {excluded_parameter_pairs}組")
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    print(f"固定出口: TP{TAKE_PROFIT_PIPS} / SL{STOP_LOSS_PIPS} pips")

    # 通貨・方向別result.csvに加え、全結果をまとめるall_rowsも作ります。
    all_rows = []
    cases_per_symbol = len(SIDES) * len(valid_parameter_pairs)
    for symbol in SYMBOLS:
        # この通貨の15分足ASK CSVを読み込みます。
        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:
            # 検証開始前の足が十分なら、その足をCCI計算だけに使います。
            entry_start_index = warmup_count
            warmup_description = f"検証開始前{warmup_count}本"
        elif len(candles) > WARMUP_BARS:
            # 検証前データがなければ、CSV先頭40本を全設定共通で除外します。
            # 短い期間だけ先に取引を始めないためです。
            entry_start_index = WARMUP_BARS
            warmup_description = f"CSV先頭{WARMUP_BARS}本"
            print(f"[WARN] {symbol}: 検証前データ不足のため{warmup_description}を使用")
        else:
            raise SystemExit(f"{symbol}のデータが不足しています: {len(candles)}本")

        print(
            f"[RUN] {symbol}: {cases_per_symbol}ケース / 検証{backtest_count}本 / "
            f"ウォームアップ: {warmup_description}"
        )
        all_rows.extend(run_symbol(
            candles, symbol, entry_start_index, DEFAULT_OUTPUT_DIR,
        ))

    # 個別CSVに加え、全1,340ケースを一括比較できるCSVも保存します。
    write_csv(DEFAULT_OUTPUT_DIR / "all_results.csv", all_rows)
    print(f"完了: {DEFAULT_OUTPUT_DIR}")
    print(f"出力行数: {len(all_rows)} / 実行時間: {time.perf_counter() - started:.2f}秒")
    return 0


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