FXの一目均衡表でおすすめの設定は？5通貨3,080ケースで検証

"""5通貨の15分足で一目均衡表616設定を一括検証する。

三役好転・三役逆転を確認しながら価格が雲を抜けた確定足をシグナルとし、
次の15分足の始値で約定する。出口は全設定でTP50・SL30pipsに固定する。
現在位置から見える雲には、基準線期間前に計算された先行スパンを使用する。
標準ライブラリだけで実行できる。
"""

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/ichimoku_cloud/run_ichimoku_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")

# 8 × 7 × 11 = 616設定。標準設定9・26・52を必ず含みます。
# TP/SLやエントリー条件は固定し、一目均衡表の期間だけを比較します。
TENKAN_PERIODS = (5, 7, 9, 11, 13, 15, 17, 19)
KIJUN_PERIODS = (20, 23, 26, 29, 32, 35, 38)
SENKOU_B_PERIODS = (40, 44, 48, 52, 56, 60, 64, 68, 72, 76, 80)
TAKE_PROFIT_PIPS = 50
STOP_LOSS_PIPS = 30

# 先行スパンのシフト幅と遅行スパンの比較幅は、各設定の基準線期間とします。
# 最大の基準線期間と先行スパン2期間を足し、全設定で計算開始位置を揃えます。
MAX_WARMUP_BARS = max(KIJUN_PERIODS) + max(SENKOU_B_PERIODS)

# __file__は、このPythonファイル自身の場所を表します。
# どのフォルダから実行しても、同じ入力・出力先を参照できます。
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 = (
    "通貨ペア", "売買方向", "転換線期間", "基準線期間", "先行スパン2期間",
    "シフト期間", "エントリー方式", "TP（pips）", "SL（pips）", "取引回数",
    "勝ち取引数", "負け取引数", "勝率（%）", "合計損益（pips）",
    "平均損益（pips）", "プロフィットファクター",
    "最大ドローダウン（pips）", "平均保有時間（分）", "最終足決済数",
    "全期間プラス月数", "全期間マイナス月数", "直近6か月取引回数",
    "直近6か月合計損益（pips）", "直近6か月平均損益（pips）",
    "直近6か月プロフィットファクター", "直近6か月プラス月数",
    "直近6か月マイナス月数", "記事候補条件合格",
)

# 記事候補として採用するための最低条件です。
# 全期間の成績だけでなく、直近6か月にも利益が残る設定へ絞ります。
MIN_CANDIDATE_TRADES = 100
MIN_CANDIDATE_PROFIT_FACTOR = 1.05
MIN_RECENT_POSITIVE_MONTHS = 3


def iter_months(start_date, end_date):
    """開始月から終了月までをYYYY-MM形式で返す。"""
    result = []
    year, month = start_date.year, start_date.month

    # 12月の次を翌年1月へ進めながら、検証対象月を順番に作ります。
    while (year, month) <= (end_date.year, end_date.month):
        result.append(f"{year:04d}-{month:02d}")
        month += 1
        if month == 13:
            year, month = year + 1, 1
    return result


# 全期間の対象月と、検証終了月を含む直近6か月です。
BACKTEST_MONTHS = tuple(iter_months(BACKTEST_START_DATE, BACKTEST_END_DATE))
RECENT_SIX_MONTHS = frozenset(BACKTEST_MONTHS[-6:])


def load_candles(path):
    """検証前の計算用足と検証期間のOHLCを読み込む。"""
    # 検証開始前の足は指標計算だけに使い、売買には使用しません。
    # dequeは上限を超えると古い足を捨てるため、必要本数だけを保持できます。
    warmup = deque(maxlen=MAX_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"]),
            }

            # 開始日より前なら計算用、それ以降なら検証対象として分けます。
            (warmup if timestamp.date() < BACKTEST_START_DATE else backtest).append(candle)
    return list(warmup) + backtest, len(warmup), len(backtest)


def rolling_midpoint(candles, period):
    """直近period本の最高値と最安値の中間値を返す。"""
    # period本が揃う前は計算できないため、先頭をNoneで埋めます。
    values = [None] * len(candles)
    for index in range(period - 1, len(candles)):
        # 現在足を含む直近period本だけを計算窓として取り出します。
        window = candles[index - period + 1:index + 1]

        # 転換線・基準線・先行スパン2は、期間内最高値と最安値の中間です。
        values[index] = (
            max(candle["high"] for candle in window)
            + min(candle["low"] for candle in window)
        ) / 2.0
    return values


def build_ichimoku(candles, tenkan_period, kijun_period, senkou_b_period, cache=None):
    """現在足で利用可能な転換線・基準線・雲・遅行比較値を作る。"""
    # 一目均衡表の役割が逆転しない期間の組み合わせだけを許可します。
    if not (0 < tenkan_period < kijun_period < senkou_b_period):
        raise ValueError("期間は 0 < 転換線 < 基準線 < 先行スパン2 にしてください")
    cache = {} if cache is None else cache

    def midpoint(period):
        # 同じ期間の中間値は複数設定で共有し、繰り返し計算を避けます。
        if period not in cache:
            cache[period] = rolling_midpoint(candles, period)
        return cache[period]

    tenkan = midpoint(tenkan_period)
    kijun = midpoint(kijun_period)
    span_b_source = midpoint(senkou_b_period)
    cloud_a = [None] * len(candles)
    cloud_b = [None] * len(candles)
    lagging_reference = [None] * len(candles)
    for index in range(kijun_period, len(candles)):
        # 現在位置の雲には、基準線期間前に計算された値が表示されます。
        # 現在足の値をそのまま雲に使わないことで、表示位置を正しく再現します。
        source_index = index - kijun_period

        # 先行スパン1 =（転換線 + 基準線）÷ 2 です。
        if tenkan[source_index] is not None and kijun[source_index] is not None:
            cloud_a[index] = (tenkan[source_index] + kijun[source_index]) / 2.0

        # 先行スパン2は長期期間の中間値です。
        # 遅行スパン相当の判定では、現在終値と基準線期間前の終値を比較します。
        cloud_b[index] = span_b_source[source_index]
        lagging_reference[index] = candles[source_index]["close"]
    return tenkan, kijun, cloud_a, cloud_b, lagging_reference


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


def collect_trades(candles, indicator, side, pip_size, entry_start_index):
    """三役が同方向に揃った雲抜けを、次足始値で売買する。"""
    tenkan, kijun, cloud_a, cloud_b, lagging_reference = indicator

    # positionがNoneなら未保有です。
    # pending_entryは確定足のシグナルを次足始値で実行する予約です。
    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

        # ポジション保有中は、現在足の高値・安値でTP/SL到達を確認します。
        if position is not None:
            tp = TAKE_PROFIT_PIPS * pip_size
            sl = STOP_LOSS_PIPS * pip_size
            if side == "LONG":
                tp_price, sl_price = position["price"] + tp, position["price"] - sl
                tp_hit = candles[index]["high"] >= tp_price
                sl_hit = candles[index]["low"] <= sl_price
            else:
                tp_price, sl_price = position["price"] - tp, position["price"] + sl
                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

        # 雲抜け判定に必要な値がすべて計算済みか確認します。
        required = (
            tenkan[index], kijun[index], cloud_a[index], cloud_b[index],
            lagging_reference[index], cloud_a[index - 1], cloud_b[index - 1],
        )
        if position is not None or any(value is None for value in required):
            continue
        close = candles[index]["close"]
        previous_close = candles[index - 1]["close"]
        cloud_top = max(cloud_a[index], cloud_b[index])
        cloud_bottom = min(cloud_a[index], cloud_b[index])
        previous_top = max(cloud_a[index - 1], cloud_b[index - 1])
        previous_bottom = min(cloud_a[index - 1], cloud_b[index - 1])
        if side == "LONG":
            # LONGは価格の雲上抜けをトリガーに、転換線と遅行条件も確認します。
            # 転換線のクロスそのものを要求せず、基準線より上なら条件成立です。
            signal = (
                previous_close <= previous_top and close > cloud_top
                and tenkan[index] > kijun[index]
                and close > lagging_reference[index]
            )
        else:
            # SHORTはLONGと反対に、価格の雲下抜けと三役逆転を確認します。
            signal = (
                previous_close >= previous_bottom and close < cloud_bottom
                and tenkan[index] < kijun[index]
                and close < lagging_reference[index]
            )
        if signal 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 = [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 {
        "取引回数": 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 summarize(symbol, side, setting, trades):
    """1設定の全期間・月別・直近6か月成績をまとめる。"""
    tenkan_period, kijun_period, senkou_b_period = setting
    metrics = calculate_metrics(trades)

    # 決済時刻を基準に、取引0回の月も含めて月別損益を作ります。
    monthly_profits = []
    for month in BACKTEST_MONTHS:
        selected = [t for t in trades if t["決済時刻"].strftime("%Y-%m") == month]
        monthly_profits.append(calculate_metrics(selected)["合計損益（pips）"])
    recent_trades = [
        trade for trade in trades
        if trade["決済時刻"].strftime("%Y-%m") in RECENT_SIX_MONTHS
    ]
    recent = calculate_metrics(recent_trades)
    recent_profits = monthly_profits[-6:]
    pf = metrics["プロフィットファクター"]

    # 記事候補は取引数、全期間PF、直近成績をすべて満たす設定です。
    candidate = (
        metrics["取引回数"] >= MIN_CANDIDATE_TRADES
        and metrics["合計損益（pips）"] > 0
        and pf != "" and pf >= MIN_CANDIDATE_PROFIT_FACTOR
        and recent["合計損益（pips）"] > 0
        and sum(value > 0 for value in recent_profits) >= MIN_RECENT_POSITIVE_MONTHS
    )
    return {
        "通貨ペア": symbol, "売買方向": side,
        "転換線期間": tenkan_period, "基準線期間": kijun_period,
        "先行スパン2期間": senkou_b_period, "シフト期間": kijun_period,
        "エントリー方式": "三役好転・三役逆転の雲抜け",
        "TP（pips）": TAKE_PROFIT_PIPS, "SL（pips）": STOP_LOSS_PIPS,
        **metrics,
        "最終足決済数": sum(t["決済理由"] == "データ終了" for t in trades),
        "全期間プラス月数": sum(value > 0 for value in monthly_profits),
        "全期間マイナス月数": sum(value < 0 for value in monthly_profits),
        "直近6か月取引回数": recent["取引回数"],
        "直近6か月合計損益（pips）": recent["合計損益（pips）"],
        "直近6か月平均損益（pips）": recent["平均損益（pips）"],
        "直近6か月プロフィットファクター": recent["プロフィットファクター"],
        "直近6か月プラス月数": sum(value > 0 for value in recent_profits),
        "直近6か月マイナス月数": sum(value < 0 for value in recent_profits),
        "記事候補条件合格": candidate,
    }


def write_csv(path, rows):
    """結果をExcelでも開きやすいUTF-8 BOM付きCSVへ保存する。"""
    # フォルダが存在しない初回実行時は、自動的に作成します。
    path.parent.mkdir(parents=True, exist_ok=True)

    # utf-8-sigはUTF-8にBOMを付け、Excelの文字化けを起きにくくします。
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=RESULT_COLUMNS)
        writer.writeheader()
        writer.writerows(rows)


def number(row, column):
    """CSV出力用の空文字を考慮して集計列を数値化する。"""
    return float(row[column]) if row[column] != "" else 0.0


def format_setting(row):
    """記事で使う「転換線・基準線・先行スパン2」表記を返す。"""
    return f"{row['転換線期間']}・{row['基準線期間']}・{row['先行スパン2期間']}"


def print_article_summary(rows):
    """全結果から記事掲載用のMarkdown表を標準出力する。"""
    # 設定の欠落や重複を見逃さないよう、出力前に全件数を確認します。
    expected_rows = len(SYMBOLS) * len(SIDES) * len(TENKAN_PERIODS) * len(
        KIJUN_PERIODS
    ) * len(SENKOU_B_PERIODS)
    if len(rows) != expected_rows:
        raise ValueError(
            f"結果行数が想定と異なります: {len(rows)}（想定: {expected_rows}）"
        )

    print("\n## 通貨・方向別TOP5（候補条件合格後、合計pips順）")
    print("| 通貨 | 方向 | 順位 | 設定 | 取引 | 合計pips | PF | 最大DD |")
    print("|---|---|---:|---:|---:|---:|---:|---:|")
    for symbol in SYMBOLS:
        for side in SIDES:
            selected = [
                row for row in rows
                if row["通貨ペア"] == symbol
                and row["売買方向"] == side
                and row["記事候補条件合格"] is True
            ]
            if not selected:
                print(
                    f"| {symbol.replace('_', '')} | {side} | - | "
                    "候補なし | - | - | - | - |"
                )
                continue

            # 候補条件を通過した設定を合計pips順、同値ならPF順に並べます。
            ranked = sorted(
                selected,
                key=lambda row: (
                    number(row, "合計損益（pips）"),
                    number(row, "プロフィットファクター"),
                ),
                reverse=True,
            )
            for rank, row in enumerate(ranked[:5], start=1):
                print(
                    f"| {symbol.replace('_', '')} | {side} | {rank} | "
                    f"{format_setting(row)} | {row['取引回数']} | "
                    f"{row['合計損益（pips）']} | "
                    f"{row['プロフィットファクター']} | "
                    f"{row['最大ドローダウン（pips）']} |"
                )

    print("\n## 標準設定9・26・52")
    print("| 通貨 | 方向 | 取引 | 合計pips | PF | 最大DD |")
    print("|---|---|---:|---:|---:|---:|")
    standard = [
        row for row in rows
        if row["転換線期間"] == 9
        and row["基準線期間"] == 26
        and row["先行スパン2期間"] == 52
    ]
    for row in standard:
        print(
            f"| {row['通貨ペア'].replace('_', '')} | {row['売買方向']} | "
            f"{row['取引回数']} | {row['合計損益（pips）']} | "
            f"{row['プロフィットファクター']} | "
            f"{row['最大ドローダウン（pips）']} |"
        )

    print("\n## 方向別集計（重複設定の単純合算）")
    print("| 方向 | 取引 | 合計pips | PF中央値 | プラス設定 | 候補合格 |")
    print("|---|---:|---:|---:|---:|---:|")
    settings_per_side = len(rows) // len(SIDES)
    for side in SIDES:
        selected = [row for row in rows if row["売買方向"] == side]
        pfs = sorted(number(row, "プロフィットファクター") for row in selected)
        midpoint = len(pfs) // 2
        median = (pfs[midpoint - 1] + pfs[midpoint]) / 2
        print(
            f"| {side} | {sum(int(row['取引回数']) for row in selected)} | "
            f"{sum(number(row, '合計損益（pips）') for row in selected):.2f} | "
            f"{median:.4f} | "
            f"{sum(number(row, '合計損益（pips）') > 0 for row in selected)}"
            f"/{settings_per_side} | "
            f"{sum(row['記事候補条件合格'] is True for row in selected)} |"
        )


def run_symbol(candles, symbol, entry_start_index, output_dir):
    """1通貨について616設定をLONG・SHORT別に検証する。"""
    # 円を決済通貨にする通貨ペアと、それ以外で1pipsの値が異なります。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001

    # 同じ期間の中間値を設定間で共有するキャッシュです。
    cache = {}
    rows_by_side = {side: [] for side in SIDES}
    for tenkan_period in TENKAN_PERIODS:
        for kijun_period in KIJUN_PERIODS:
            for senkou_b_period in SENKOU_B_PERIODS:
                setting = (tenkan_period, kijun_period, senkou_b_period)
                indicator = build_ichimoku(candles, *setting, cache=cache)
                for side in SIDES:
                    # LONGとSHORTは同じ指標値を使い、売買条件だけを反転します。
                    trades = collect_trades(candles, indicator, side, pip_size, entry_start_index)
                    rows_by_side[side].append(summarize(symbol, side, setting, trades))
    rows = []
    for side in SIDES:
        # 1通貨・1方向の616設定をresult.csvへ保存します。
        write_csv(output_dir / symbol / side / "result.csv", rows_by_side[side])
        rows.extend(rows_by_side[side])
    return rows


def main():
    """5通貨・6,160結果の一目均衡表検証を実行する。"""
    # スクリプト全体の実行時間を測定します。
    started = time.perf_counter()
    settings_count = len(TENKAN_PERIODS) * len(KIJUN_PERIODS) * len(SENKOU_B_PERIODS)
    print(
        f"{len(SYMBOLS)}通貨 × {settings_count}設定 = "
        f"{len(SYMBOLS) * settings_count}設定"
    )
    print(f"LONG・SHORT別の結果行: {len(SYMBOLS) * len(SIDES) * settings_count}ケース")
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    print(f"固定出口: TP{TAKE_PROFIT_PIPS} / SL{STOP_LOSS_PIPS} pips")

    # 通貨・方向別CSVに加え、全結果をまとめるall_rowsも作ります。
    all_rows = []
    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 >= MAX_WARMUP_BARS:
            # 検証開始前の足が十分なら、指標計算だけに使用します。
            entry_start_index = warmup_count
        elif len(candles) > MAX_WARMUP_BARS:
            # 検証前データがなければ、CSV先頭を全設定共通で除外します。
            entry_start_index = MAX_WARMUP_BARS
            print(f"[WARN] {symbol}: 検証前データ不足のため先頭{MAX_WARMUP_BARS}本を除外")
        else:
            raise SystemExit(f"{symbol}のデータが不足しています: {len(candles)}本")
        print(f"[RUN] {symbol}: {settings_count * 2}ケース / 検証{backtest_count}本")
        all_rows.extend(run_symbol(candles, symbol, entry_start_index, DEFAULT_OUTPUT_DIR))

    # 個別CSVに加え、全6,160結果を一括比較できる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}秒")
    # 記事へ貼り付けられるTOP5・標準設定・方向別集計を表示します。
    print_article_summary(all_rows)
    return 0


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