FXのストキャスティクスでおすすめの設定は？5通貨10,500ケースで検証

"""5通貨の15分足でストキャスティクス10,500設定を比較する。

LONGは売られすぎ圏からの上抜け、SHORTは買われすぎ圏からの下抜けを
%Kと%Dのクロスで確認する。シグナル確定の次足始値で入り、固定TP/SLで
決済する。%K期間、基準値、%K平滑化期間、%D期間を比較する。
"""

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/stochastic_oscillator/run_stochastic_backtest.py
# 3. 結果は backtest_research/stochastic_oscillator/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")

# Slow %K = SMA(Raw %K, %K平滑化期間)、%D = SMA(Slow %K, %D期間)
# %K期間は、現在の終値が直近何本の高値・安値のどこにあるかを調べる期間です。
STOCHASTIC_PERIODS = (5, 7, 14, 18, 21, 25, 28)

# Raw %Kを何本平均してSlow %Kにするかを比較します。
# 1の場合はRaw %Kを平均せず、そのままSlow %Kとして使います。
SLOW_K_SMOOTHINGS = (1, 2, 3, 4, 5)

# Slow %Kを何本平均して%Dにするかを比較します。
# 1では%Kと%Dが同じ値になりクロスが発生しないため、2以上にしています。
D_PERIODS = (2, 3, 4, 5, 6)

# 各要素は（買われすぎ, 売られすぎ）。上下は100から見て対称にする。
# 例として80/20なら、%Dが20以下でLONG、80以上でSHORTのクロスを探します。
STOCHASTIC_LEVELS = (
    (55, 45), (60, 40), (70, 30),
    (75, 25), (80, 20), (90, 10),
)

# ストキャスティクス設定を公平に比べるため、出口は全ケースで固定します。
TAKE_PROFIT_PIPS = 50
STOP_LOSS_PIPS = 30

# 最も計算に時間がかかる設定でも、検証開始時に%Kと%Dが揃う本数です。
# 最大%K期間28 + 最大%K平滑化5 + 最大%D期間6 - 2 = 37本です。
WARMUP_BARS = (
    max(STOCHASTIC_PERIODS) + max(SLOW_K_SMOOTHINGS) + max(D_PERIODS) - 2
)

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


def load_candles(path):
    """検証開始前のウォームアップ足と検証期間内のOHLCを読む。"""
    # deque(maxlen=37)は上限を超えると古い足を自動的に捨てます。
    # 検証開始前のデータは指標を準備するためだけに使い、売買には使いません。
    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:
                warmup.append(candle)
            else:
                backtest.append(candle)
    return list(warmup) + backtest, len(warmup), len(backtest)


def simple_moving_average(values, period):
    """Noneを含む先頭区間を考慮して単純移動平均を返す。"""
    if period <= 0:
        raise ValueError("移動平均期間は1以上にしてください")
    # SMA(1)は入力値そのもの。加算と減算を繰り返す実装を通すと、
    # 浮動小数点誤差によって同一であるはずの%Kと%Dに偽クロスが生じる。
    if period == 1:
        return list(values)

    # まだ平均を計算できない先頭部分にはNoneを残します。
    # windowには直近period個の値、totalにはその合計を保持します。
    result = [None] * len(values)
    window = deque()
    total = 0.0
    for index, value in enumerate(values):
        if value is None:
            # Raw %Kなどをまだ計算できない区間では平均窓をリセットします。
            window.clear()
            total = 0.0
            continue
        window.append(value)
        total += value
        if len(window) > period:
            # 必要期間より古くなった値を窓と合計から取り除きます。
            total -= window.popleft()
        if len(window) == period:
            # 必要な本数が揃った位置から平均値を記録します。
            result[index] = total / period
    return result


def calculate_raw_percent_k(candles, period):
    """指定期間のRaw %Kを0～100で計算する。"""
    if period <= 0:
        raise ValueError("%K期間は1以上にしてください")
    raw_k = [None] * len(candles)
    for index in range(period - 1, len(candles)):
        # 現在足を含む直近period本から、最高値と最安値を探します。
        window = candles[index - period + 1:index + 1]
        highest = max(candle["high"] for candle in window)
        lowest = min(candle["low"] for candle in window)
        price_range = highest - lowest
        # 例：最安値100、最高値110、終値108なら、
        # (108 - 100) ÷ (110 - 100) × 100 = 80です。
        # 全足が同値なら値幅が0になるため、中立値の50としてゼロ除算を避けます。
        raw_k[index] = (
            50.0 if price_range == 0
            else (candles[index]["close"] - lowest) / price_range * 100.0
        )
    return raw_k


def calculate_stochastic(candles, period, k_smoothing=3, d_period=3):
    """Slow %Kと%Dを0～100で計算する。"""
    if k_smoothing <= 0 or d_period <= 0:
        raise ValueError("ストキャスティクスの各期間は1以上にしてください")
    # Raw %Kをk_smoothing本平均したものがSlow %Kです。
    # さらにSlow %Kをd_period本平均したものが%Dです。
    raw_k = calculate_raw_percent_k(candles, period)
    slow_k = simple_moving_average(raw_k, k_smoothing)
    percent_d = simple_moving_average(slow_k, d_period)
    return slow_k, percent_d


def make_trade(position, candle, exit_price, reason, side, pip_size):
    """決済済み取引を作る。"""
    # LONGは値上がり、SHORTは値下がりしたときに損益がプラスになります。
    direction = 1.0 if side == "LONG" else -1.0
    return {
        # USDJPYを150.00円で買って150.50円で決済した場合は、
        # (150.50 - 150.00) ÷ 0.01 = 50pipsです。
        "損益（pips）": direction * (exit_price - position["price"]) / pip_size,
        "保有時間（分）": (
            candle["close_time"] - position["time"]
        ).total_seconds() / 60,
        "決済理由": reason,
    }


def collect_trades(candles, slow_k, percent_d, side, upper, lower, pip_size,
                   entry_start_index):
    """現在%Dがゾーン内にあるK/Dクロスを次足始値で約定させる。"""
    if side not in SIDES:
        raise ValueError("sideはLONGまたはSHORTにしてください")
    # 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 = TAKE_PROFIT_PIPS * pip_size
            sl_distance = STOP_LOSS_PIPS * pip_size
            if side == "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:
                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

        values = slow_k[index - 1], percent_d[index - 1], slow_k[index], percent_d[index]
        # ポジション保有中や、指標をまだ計算できない位置では判定しません。
        if position is not None or any(value is None for value in values):
            continue
        previous_k, previous_d, current_k, current_d = values
        if side == "LONG":
            # LONGは%Kが%Dを下から上へ抜けた瞬間を探します。
            crossed = previous_k <= previous_d and current_k > current_d
            in_extreme_zone = current_d <= lower
        else:
            # SHORTは%Kが%Dを上から下へ抜けた瞬間を探します。
            crossed = previous_k >= previous_d and current_k < current_d
            in_extreme_zone = current_d >= upper

        # ここではまだ約定せず、次のループで次足始値を使います。
        if crossed and in_extreme_zone 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 summarize(symbol, side, period, k_smoothing, d_period, upper, lower,
              trades):
    """1設定の取引成績を集計する。"""
    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]
    # 取引順の累積損益について、過去最高値からの最大下落幅を求めます。
    # 累積損益が[100, 250, 180, 50]なら、最大DDは250 - 50 = 200pipsです。
    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, "%K期間": period,
        "%K平滑化期間": k_smoothing, "%D期間": d_period,
        "買われすぎ": upper, "売られすぎ": lower,
        "TP（pips）": TAKE_PROFIT_PIPS, "SL（pips）": STOP_LOSS_PIPS,
        "取引回数": 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を保存する。"""
    # 保存先フォルダが存在しない初回実行時は、自動的に作成します。
    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 run_symbol(candles, symbol, entry_start_index, output_dir):
    """1通貨について2,100ケースを検証する。"""
    # 円を決済通貨にする通貨ペアは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}

    # Raw %Kは%K期間ごとに1回だけ計算する。指標175組を同時に保持せず、
    # 1組ずつ検証して破棄することでメモリ使用量を抑える。
    for period in STOCHASTIC_PERIODS:
        raw_k = calculate_raw_percent_k(candles, period)
        for k_smoothing in SLOW_K_SMOOTHINGS:
            slow_k = simple_moving_average(raw_k, k_smoothing)
            for d_period in D_PERIODS:
                percent_d = simple_moving_average(slow_k, d_period)
                for side in SIDES:
                    for upper, lower in STOCHASTIC_LEVELS:
                        # 1つの「期間・平滑化・%D・基準値・方向」を検証します。
                        trades = collect_trades(
                            candles, slow_k, percent_d, side, upper, lower,
                            pip_size, entry_start_index,
                        )
                        rows_by_side[side].append(summarize(
                            symbol, side, period, k_smoothing, d_period,
                            upper, lower, trades,
                        ))

    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方向のストキャスティクス10,500ケースを検証する。"""
    # スクリプト全体の実行時間を測り始めます。
    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以上にしてください")
    # 定数を変更しても表示が追従するよう、実際の候補数からケース数を計算します。
    total_cases = (
        len(SYMBOLS) * len(SIDES) * len(STOCHASTIC_PERIODS)
        * len(STOCHASTIC_LEVELS) * len(SLOW_K_SMOOTHINGS) * len(D_PERIODS)
    )
    print(
        f"{len(SYMBOLS)}通貨 × {len(SIDES)}方向 × "
        f"期間{len(STOCHASTIC_PERIODS)}種類 × "
        f"基準値{len(STOCHASTIC_LEVELS)}種類 × "
        f"K平滑化{len(SLOW_K_SMOOTHINGS)}種類 × "
        f"D期間{len(D_PERIODS)}種類 = {total_cases}ケース"
    )
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    print(f"固定出口: TP{TAKE_PROFIT_PIPS} / SL{STOP_LOSS_PIPS} pips")

    # 通貨別result.csvに加え、全結果をまとめたall_results.csvも作ります。
    all_rows = []
    cases_per_symbol = (
        len(SIDES) * len(STOCHASTIC_PERIODS) * len(STOCHASTIC_LEVELS)
        * len(SLOW_K_SMOOTHINGS) * len(D_PERIODS)
    )
    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:
            # 検証開始前の足が十分なら、その足を指標計算だけに使います。
            entry_start_index = warmup_count
            warmup_description = f"検証開始前{warmup_count}本"
        elif len(candles) > WARMUP_BARS:
            # 検証前データがなければ、CSV先頭37本を全設定共通で除外します。
            # 短い期間だけ先に取引しないため、設定間の開始位置が揃います。
            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,
        ))

    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


# このファイルを直接実行したときだけmain()を動かします。
if __name__ == "__main__":
    raise SystemExit(main())