FXのRSIで最適な期間・設定値は？5通貨・1,800ケースで検証

"""5通貨の15分足でRSI逆張りと固定TP/SLを検証する。

RSIはWilder方式で計算する。売られすぎ・買われすぎ圏から戻り、
LONGは陽線、SHORTは陰線になった足でシグナルを確定する。
次の15分足の始値で約定し、固定pipsのTPまたはSLで決済する。
"""

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

# 検証する5つの通貨ペアです。
SYMBOLS = ("USD_JPY", "EUR_JPY", "GBP_JPY", "EUR_USD", "AUD_USD")

# RSIの計算、シグナル判定、売買には15分足だけを使用します。
ENTRY_TIMEFRAME = "15min"
ENTRY_TIMEFRAME_DURATION = timedelta(minutes=15)

# 既存のローソク足CSVに合わせ、ASK価格を読み込みます。
# スプレッドなどの取引コストは、このスクリプトでは考慮しません。
PRICE_TYPE = "ASK"

# CSV内のtimestamp_jstの日付が、この開始日から終了日までの足を使います。
# 開始日と終了日のローソク足も検証対象に含みます。
BACKTEST_START_DATE = date(2024, 1, 1)
BACKTEST_END_DATE = date(2026, 7, 31)

# RSI期間6種類と、買われすぎ・売られすぎ水準5種類を組み合わせます。
# 6 × 5 = 30パターンを、LONG・SHORTに分けて検証します。
RSI_PERIODS = (5, 7, 9, 14, 21, 28)
# 各要素は「（買われすぎ水準, 売られすぎ水準）」です。
RSI_LEVELS = ((60, 40), (65, 35), (70, 30), (75, 25), (80, 20))

# 検証する「（TP, SL）」の固定pipsペアです。
# 事前検証でLONGの成績と通貨間の安定性が良かった5組に絞ります。
TP_SL_PAIRS = (
    (100, 10),
    (100, 30),
    (100, 50),
    (150, 30),
    (150, 50),
)

# __file__は、このPythonファイル自身の場所を表します。
# どのフォルダから実行しても、同じ入力・出力先を参照できます。
SCRIPT_DIR = Path(__file__).resolve().parent
GMO_API_DIR = SCRIPT_DIR.parents[1]
DEFAULT_DATA_DIR = GMO_API_DIR / "data" / "candles"
# 検証結果は、このスクリプトと同じフォルダのresultsへ保存します。
DEFAULT_OUTPUT_DIR = SCRIPT_DIR / "results"

# RSI設定ごとの集計結果に保存する列です。
RESULT_COLUMNS = (
    "通貨ペア", "売買方向", "RSI期間", "買われすぎ", "売られすぎ",
    "TP（pips）", "SL（pips）",
    "取引回数", "勝ち取引数", "負け取引数", "勝率（%）",
    "合計損益（pips）", "平均損益（pips）", "プロフィットファクター",
    "最大ドローダウン（pips）", "平均保有時間（分）",
)

def load_candles(path):
    """fetch_gmo_candles.pyが作るOHLC CSVを読み込む。"""
    candles = []

    # RSI用の終値、約定用の始値、SL判定用の高値・安値を読み込みます。
    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同士で比較し、終了日の23:45足なども含めます。
            if BACKTEST_START_DATE <= timestamp.date() <= BACKTEST_END_DATE:
                candles.append({
                    "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"]),
                })
    return candles


def calculate_rsi(closes, period):
    """Wilderの平滑化（初期値はperiod本の平均）でRSIを計算する。"""
    if period <= 0:
        raise ValueError("RSI期間は1以上である必要があります")
    # RSIをまだ計算できない位置にはNoneを入れます。
    # RSI14には14本分の価格変化が必要なので、最初の14要素はNoneです。
    values = [None] * len(closes)
    if len(closes) <= period:
        return values

    # 最初のRSIは、直近period本の上昇幅と下落幅の単純平均から作ります。
    losses = []
    gains = []
    for index in range(1, period + 1):
        change = closes[index] - closes[index - 1]
        gains.append(max(change, 0.0))
        losses.append(max(-change, 0.0))
    average_gain = sum(gains) / period
    average_loss = sum(losses) / period
    values[period] = rsi_value(average_gain, average_loss)

    # 2つ目以降はWilderの平滑化を使います。
    # 平均上昇幅 = (前回平均 × (期間 - 1) + 今回上昇幅) / 期間
    for index in range(period + 1, len(closes)):
        change = closes[index] - closes[index - 1]
        gain = max(change, 0.0)
        loss = max(-change, 0.0)
        average_gain = (average_gain * (period - 1) + gain) / period
        average_loss = (average_loss * (period - 1) + loss) / period
        values[index] = rsi_value(average_gain, average_loss)
    return values


def rsi_value(average_gain, average_loss):
    """平均上昇幅と平均下落幅を0～100のRSIへ変換する。"""
    # 下落が一度もなければRSIは100です。値動きが全くなければ中立の50とします。
    if average_loss == 0:
        return 100.0 if average_gain > 0 else 50.0
    # 上昇が一度もなければRSIは0です。
    if average_gain == 0:
        return 0.0

    # RSI = 100 - 100 / (1 + 平均上昇幅 / 平均下落幅)
    relative_strength = average_gain / average_loss
    return 100.0 - 100.0 / (1.0 + relative_strength)


def collect_trades(
    candles, rsi, side, upper, lower, take_profit_pips, stop_loss_pips,
    pip_size, start_index,
):
    """RSIの圏外復帰と反発足を確認し、固定TPまたはSLで決済する。"""
    if side not in {"LONG", "SHORT"}:
        raise ValueError("sideはLONGまたはSHORTである必要があります")

    # positionがNoneなら、ポジションを持っていない状態です。
    # pendingには、現在足で確定した注文を次足始値で実行するためのRSIを保存します。
    position = None
    pending_entry_rsi = None
    trades = []
    for index in range(max(1, start_index), len(candles)):
        # 前の15分足でエントリー条件が確定していれば、現在足の始値で入ります。
        # シグナル足の終値で約定させないことで、未来参照を防ぎます。
        if position is None and pending_entry_rsi is not None:
            position = {
                "time": candles[index]["open_time"],
                "price": candles[index]["open"],
                "rsi": pending_entry_rsi,
            }
        pending_entry_rsi = None

        # ポジション保有中は、現在足の高値・安値でTP/SL到達を確認します。
        if position is not None:
            take_profit_distance = take_profit_pips * pip_size
            stop_distance = stop_loss_pips * pip_size
            if side == "LONG":
                take_profit_price = position["price"] + take_profit_distance
                stop_price = position["price"] - stop_distance
                take_profit_hit = candles[index]["high"] >= take_profit_price
                stop_hit = candles[index]["low"] <= stop_price
            else:
                take_profit_price = position["price"] - take_profit_distance
                stop_price = position["price"] + stop_distance
                take_profit_hit = candles[index]["low"] <= take_profit_price
                stop_hit = candles[index]["high"] >= stop_price

            # 15分足だけでは足の中でTPとSLのどちらへ先に達したか分かりません。
            # 同じ足で両方へ達した場合は、成績を楽観視しないようSLを優先します。
            if stop_hit:
                # SLはエントリー時に指定済みとし、指定SL価格で決済します。
                exit_price = stop_price
                exit_reason = f"SL {stop_loss_pips}pips"
            elif take_profit_hit:
                # TPはエントリー時に指定済みとし、指定TP価格で決済します。
                exit_price = take_profit_price
                exit_reason = f"TP {take_profit_pips}pips"
            else:
                exit_price = None
                exit_reason = None

            if exit_price is not None:
                trades.append(make_trade(
                    len(trades) + 1, position, candles[index]["close_time"],
                    exit_price, rsi[index], exit_reason, side, pip_size,
                ))
                position = None

        # 前足と現在足のRSIを比較し、水準を跨いだ瞬間だけシグナルにします。
        current = rsi[index]
        previous = rsi[index - 1]
        if previous is None or current is None:
            continue

        if position is None:
            # LONGはRSIが売られすぎ圏から戻り、同じ足が陽線なら
            # 上方向への反発を確認できたものとして次足の注文を予約します。
            if (
                side == "LONG"
                and previous <= lower < current
                and candles[index]["close"] > candles[index]["open"]
            ):
                pending_entry_rsi = current
            # SHORTはRSIが買われすぎ圏から戻り、同じ足が陰線なら
            # 下方向への反落を確認できたものとして次足の注文を予約します。
            elif (
                side == "SHORT"
                and previous >= upper > current
                and candles[index]["close"] < candles[index]["open"]
            ):
                pending_entry_rsi = current

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


def make_trade(number, position, exit_time, exit_price, exit_rsi, reason, side, pip_size):
    """取引履歴の1行を作成する。"""
    # LONGでは値上がり、SHORTでは値下がりしたときにpipsがプラスになります。
    direction = 1.0 if side == "LONG" else -1.0
    return {
        "取引番号": number,
        "エントリー日時": position["time"].isoformat(),
        "エントリー価格": round(position["price"], 6),
        "エントリー時RSI": round(position["rsi"], 4),
        "決済日時": exit_time.isoformat(),
        "決済価格": round(exit_price, 6),
        "決済時RSI": round(exit_rsi, 4) if exit_rsi is not None else "",
        "決済理由": reason,
        # 日時の差を使うため、土日やデータ欠損を挟んだ時間も含みます。
        "保有時間（分）": round((exit_time - position["time"]).total_seconds() / 60, 2),
        # 円通貨は1pips=0.01円、それ以外は1pips=0.0001として換算します。
        "損益（pips）": round(direction * (exit_price - position["price"]) / pip_size, 2),
    }


def summarize(
    symbol, side, period, upper, lower, take_profit_pips, stop_loss_pips, trades
):
    """取引ごとのpips損益からRSI設定の成績をまとめる。"""
    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]
    # PF = 勝ち取引の総利益 / 負け取引の総損失です。
    gross_loss = abs(sum(losses))

    # 累積損益の過去最高値からの最大下落幅を求めます。
    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)
    durations = [trade["保有時間（分）"] for trade in trades]
    return {
        "通貨ペア": symbol, "売買方向": side, "RSI期間": period,
        "買われすぎ": upper, "売られすぎ": lower, "取引回数": count,
        "TP（pips）": take_profit_pips,
        "SL（pips）": stop_loss_pips,
        "勝ち取引数": 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, columns):
    """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=columns)
        writer.writeheader()
        writer.writerows(rows)


def run_symbol(candles, symbol, output_dir):
    """1通貨についてRSIとTP/SLの150パターンを売買方向別に検証する。"""
    # 同じ期間のRSIを設定ごとに再計算しないよう、最初にまとめて作ります。
    closes = [candle["close"] for candle in candles]
    rsi_by_period = {period: calculate_rsi(closes, period) for period in RSI_PERIODS}
    # 円を決済通貨にする通貨ペアと、それ以外で1pipsの値が異なります。
    pip_size = 0.01 if symbol.endswith("_JPY") else 0.0001
    rows = []
    # LONGとSHORTは別フォルダ、別result.csvへ保存します。
    for side in ("LONG", "SHORT"):
        for period in RSI_PERIODS:
            for upper, lower in RSI_LEVELS:
                for take_profit_pips, stop_loss_pips in TP_SL_PAIRS:
                    trades = collect_trades(
                        candles, rsi_by_period[period], side, upper, lower,
                        take_profit_pips, stop_loss_pips, pip_size,
                        max(RSI_PERIODS) + 1,
                    )
                    # ファイル数を抑えるため、設定ごとの個別取引CSVは保存しません。
                    # 取引履歴は集計が終わると破棄し、成績だけresult.csvへ残します。
                    rows.append(summarize(
                        symbol, side, period, upper, lower,
                        take_profit_pips, stop_loss_pips, trades,
                    ))
        write_csv(
            output_dir / symbol / side / "result.csv",
            [row for row in rows if row["売買方向"] == side], RESULT_COLUMNS,
        )
    return rows


def main():
    """5通貨のRSI逆張り設定を一括検証する。"""
    # スクリプト全体の実行時間を測定します。
    started = time.perf_counter()
    if BACKTEST_START_DATE > BACKTEST_END_DATE:
        raise SystemExit("検証開始日は検証終了日以前に設定してください")
    # 1通貨30設定 × TP/SL 5組 × LONG・SHORT × 5通貨 = 1,500行です。
    # 出力は通貨・方向別result.csv 10個とall_results.csv 1個の計11個です。
    all_rows = []
    combinations = len(RSI_PERIODS) * len(RSI_LEVELS) * len(TP_SL_PAIRS)
    print(f"RSI・TP/SL設定: {combinations}パターン × LONG/SHORT")
    print(f"検証期間: {BACKTEST_START_DATE} ～ {BACKTEST_END_DATE}")
    # 5通貨を1つずつ読み込み、各設定を検証します。
    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 = load_candles(path)
        if len(candles) <= max(RSI_PERIODS) + 1:
            raise SystemExit(f"{symbol}のデータが不足しています: {len(candles)}本")
        print(f"[RUN] {symbol}: {len(candles)}本")
        all_rows.extend(run_symbol(candles, symbol, DEFAULT_OUTPUT_DIR))
    # 通貨別result.csvに加え、全結果を比較できるCSVも保存します。
    write_csv(DEFAULT_OUTPUT_DIR / "all_results.csv", all_rows, RESULT_COLUMNS)
    print(f"完了: {DEFAULT_OUTPUT_DIR}")
    print(f"実行時間: {time.perf_counter() - started:.2f}秒")
    return 0


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