Web Lukes
📄

【Python】RSI逆張りボットを作る｜pandas-taでテクニカル指標計算

公開日: 2025/12/13更新日: 2025/12/13
#Python#pandas-ta#RSI#FX#自動売買#逆張り

前回は移動平均線クロスボット（順張り）を作りました。

今回はRSI（相対力指数）を使った逆張りボットをPythonで作ります。

pandas-taとは？

pandas-ta（pandas Technical Analysis）は、130種類以上のテクニカル指標を簡単に計算できるライブラリです。

RSI、MACD、ボリンジャーバンド、ATR...なんでもあります。

インストール

Bash
pip install pandas-ta

RSIの計算

pandas-taを使う場合

Python
import pandas as pd
import pandas_ta as ta


# DataFrameに対してRSIを計算
df["rsi"] = ta.rsi(df["close"], length=14)

1行で終わり！ これがpandas-taの魅力です。

自分で計算する場合（参考）

Python
def calculate_rsi(df, period=14):
    """RSIを自力で計算"""
    delta = df["close"].diff()


    gain = delta.where(delta > 0, 0)
    loss = (-delta).where(delta < 0, 0)


    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()


    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))


    return rsi

pandas-taを使えば、こんな計算を毎回書く必要がありません。

RSIシグナルの検出

Python
def detect_rsi_signal(df, oversold=30, overbought=70):
    """
    RSIシグナルを検出


    Returns:
        "buy": 売られすぎから回復（買いシグナル）
        "sell": 買われすぎから下落（売りシグナル）
        None: シグナルなし
    """
    if len(df) < 2:
        return None


    current_rsi = df["rsi"].iloc[-1]
    prev_rsi = df["rsi"].iloc[-2]


    if pd.isna(current_rsi) or pd.isna(prev_rsi):
        return None


    # 売られすぎゾーン（30以下）から回復
    if prev_rsi < oversold and current_rsi >= oversold:
        return "buy"


    # 買われすぎゾーン（70以上）から下落
    if prev_rsi > overbought and current_rsi <= overbought:
        return "sell"


    return None

完成版：RSI逆張りBot

Python
import os
import time
import hmac
import hashlib
import json
import requests
import pandas as pd
import pandas_ta as ta
from datetime import datetime
from dotenv import load_dotenv


load_dotenv()


# === 設定 ===
API_KEY = os.environ.get("GMO_FX_API_KEY")
API_SECRET = os.environ.get("GMO_FX_API_SECRET")
SYMBOL = "USD_JPY"
ORDER_SIZE = 100
RSI_PERIOD = 14
RSI_OVERSOLD = 30
RSI_OVERBOUGHT = 70
CHECK_INTERVAL = 60


# === APIヘルパー関数 ===
def create_signature(timestamp, method, path, body=""):
    text = timestamp + method + path + body
    return hmac.new(
        bytes(API_SECRET.encode("ascii")),
        bytes(text.encode("ascii")),
        hashlib.sha256
    ).hexdigest()


def get_timestamp():
    return str(int(time.time() * 1000))


def private_request(method, path, body=None):
    timestamp = get_timestamp()
    body_str = "" if body is None else json.dumps(body)


    headers = {
        "API-KEY": API_KEY,
        "API-TIMESTAMP": timestamp,
        "API-SIGN": create_signature(timestamp, method, path, body_str),
        "Content-Type": "application/json"
    }


    url = "https://forex-api.coin.z.com/private" + path


    if method == "GET":
        response = requests.get(url, headers=headers)
    elif method == "POST":
        response = requests.post(url, headers=headers, data=body_str)


    return response.json()


# === データ取得 ===
def get_klines(symbol, interval="1hour"):
    url = "https://forex-api.coin.z.com/public/v1/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "date": datetime.now().strftime("%Y%m%d")
    }


    response = requests.get(url, params=params)
    data = response.json()


    if data["status"] == 0:
        df = pd.DataFrame(data["data"])
        df["openTime"] = pd.to_datetime(df["openTime"])
        df["close"] = df["close"].astype(float)
        df["high"] = df["high"].astype(float)
        df["low"] = df["low"].astype(float)
        return df
    return None


# === テクニカル分析 ===
def calculate_indicators(df):
    """RSIを計算"""
    df["rsi"] = ta.rsi(df["close"], length=RSI_PERIOD)
    return df


def detect_rsi_signal(df):
    """RSIシグナルを検出"""
    if len(df) < 2:
        return None


    current_rsi = df["rsi"].iloc[-1]
    prev_rsi = df["rsi"].iloc[-2]


    if pd.isna(current_rsi) or pd.isna(prev_rsi):
        return None


    if prev_rsi < RSI_OVERSOLD and current_rsi >= RSI_OVERSOLD:
        return "buy"


    if prev_rsi > RSI_OVERBOUGHT and current_rsi <= RSI_OVERBOUGHT:
        return "sell"


    return None


# === 取引 ===
def get_positions():
    result = private_request("GET", f"/v1/openPositions?symbol={SYMBOL}")
    if result["status"] == 0:
        return result["data"]["list"]
    return []


def place_order(side):
    body = {
        "symbol": SYMBOL,
        "side": side,
        "executionType": "MARKET",
        "size": str(ORDER_SIZE),
        "timeInForce": "FAK"
    }
    result = private_request("POST", "/v1/order", body)


    if result["status"] == 0:
        print(f"✅ 注文成功: {side} {ORDER_SIZE} {SYMBOL}")
        return True
    else:
        print(f"❌ 注文エラー: {result}")
        return False


def close_all_positions(side):
    positions = get_positions()
    for pos in positions:
        if (pos["side"] == "BUY" and side == "SELL") or \
           (pos["side"] == "SELL" and side == "BUY"):
            body = {
                "symbol": SYMBOL,
                "side": side,
                "executionType": "MARKET",
                "size": pos["size"],
                "positionId": pos["positionId"],
                "timeInForce": "FAK"
            }
            private_request("POST", "/v1/closeOrder", body)
            print(f"✅ 決済: {pos['side']} {pos['size']}")


# === メインループ ===
def run_bot():
    print("=" * 50)
    print(f"🤖 Python RSI Reversal Bot 起動")
    print(f"   通貨ペア: {SYMBOL}")
    print(f"   RSI期間: {RSI_PERIOD}")
    print(f"   売られすぎ: {RSI_OVERSOLD} / 買われすぎ: {RSI_OVERBOUGHT}")
    print("=" * 50)


    current_position = None
    can_buy = True
    can_sell = True


    while True:
        try:
            # ローソク足データを取得
            df = get_klines(SYMBOL)
            if df is None:
                print("データ取得失敗。リトライ...")
                time.sleep(10)
                continue


            # RSIを計算
            df = calculate_indicators(df)


            # 現在のRSI
            current_rsi = df["rsi"].iloc[-1]


            # RSIゾーンに入ったらフラグをリセット
            if current_rsi < RSI_OVERSOLD:
                can_buy = True
            if current_rsi > RSI_OVERBOUGHT:
                can_sell = True


            # シグナル検出
            signal = detect_rsi_signal(df)


            # 状態表示
            rsi_status = "売られすぎ" if current_rsi < RSI_OVERSOLD else \
                        "買われすぎ" if current_rsi > RSI_OVERBOUGHT else "通常"


            print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                  f"RSI: {current_rsi:.1f} ({rsi_status}) | "
                  f"Signal: {signal or '-'} | "
                  f"Position: {current_position or 'None'}")


            # ポジションがあれば何もしない
            if len(get_positions()) > 0:
                current_position = "ACTIVE"
                time.sleep(CHECK_INTERVAL)
                continue


            current_position = None


            # シグナルに応じて売買
            if signal == "buy" and can_buy:
                print("🔔 RSI売られすぎから回復！買いエントリー")
                place_order("BUY")
                can_buy = False


            elif signal == "sell" and can_sell:
                print("🔔 RSI買われすぎから下落！売りエントリー")
                place_order("SELL")
                can_sell = False


        except Exception as e:
            print(f"エラー: {e}")


        time.sleep(CHECK_INTERVAL)


if __name__ == "__main__":
    run_bot()

pandas-taで使える他のテクニカル指標

MACD

Python
macd = ta.macd(df["close"])
df["macd"] = macd["MACD_12_26_9"]
df["macd_signal"] = macd["MACDs_12_26_9"]
df["macd_hist"] = macd["MACDh_12_26_9"]

ボリンジャーバンド

Python
bbands = ta.bbands(df["close"], length=20, std=2)
df["bb_upper"] = bbands["BBU_20_2.0"]
df["bb_middle"] = bbands["BBM_20_2.0"]
df["bb_lower"] = bbands["BBL_20_2.0"]

ATR（Average True Range）

Python
df["atr"] = ta.atr(df["high"], df["low"], df["close"], length=14)

ストキャスティクス

Python
stoch = ta.stoch(df["high"], df["low"], df["close"])
df["stoch_k"] = stoch["STOCHk_14_3_3"]
df["stoch_d"] = stoch["STOCHd_14_3_3"]

一括で計算

Python
# 全部まとめて計算！
df.ta.strategy("All")

複合条件の例

RSI単体ではダマシが多いので、複数の指標を組み合わせます。

RSI + 移動平均線フィルター

Python
def detect_filtered_signal(df):
    """RSI + MAフィルター"""


    # MAを計算
    df["ma200"] = ta.sma(df["close"], length=200)


    current_rsi = df["rsi"].iloc[-1]
    prev_rsi = df["rsi"].iloc[-2]
    current_price = df["close"].iloc[-1]
    ma200 = df["ma200"].iloc[-1]


    # 200MAより上にいる時は買いのみ
    if current_price > ma200:
        if prev_rsi < 30 and current_rsi >= 30:
            return "buy"


    # 200MAより下にいる時は売りのみ
    if current_price < ma200:
        if prev_rsi > 70 and current_rsi <= 70:
            return "sell"


    return None

RSI + ボリンジャーバンド

Python
def detect_bb_rsi_signal(df):
    """RSI + ボリンジャーバンド"""


    bbands = ta.bbands(df["close"], length=20, std=2)
    df["bb_lower"] = bbands["BBL_20_2.0"]
    df["bb_upper"] = bbands["BBU_20_2.0"]


    current_price = df["close"].iloc[-1]
    current_rsi = df["rsi"].iloc[-1]


    # 価格がBB下限 かつ RSI30以下 → 買い
    if current_price < df["bb_lower"].iloc[-1] and current_rsi < 30:
        return "buy"


    # 価格がBB上限 かつ RSI70以上 → 売り
    if current_price > df["bb_upper"].iloc[-1] and current_rsi > 70:
        return "sell"


    return None

MQL4との比較

RSI計算

MQL4:

MQL4
double rsi = iRSI(NULL, 0, 14, PRICE_CLOSE, 0);

Python:

Python
df["rsi"] = ta.rsi(df["close"], length=14)

MQL4は1つの値を取得、Pythonは全履歴をまとめて計算。

Pythonのメリット

  1. 分析しやすい：pandasでデータ操作が簡単
  2. 指標が豊富：pandas-taで130種類以上
  3. 組み合わせが簡単：複合条件を自由に作れる
  4. バックテストしやすい：backtestingライブラリと連携

次のステップ

RSI逆張りボットができました！

次は、24時間安定稼働させるためのVPS（クラウド）運用を解説します。

📚 次に読む記事: VPSでFXボットを24時間稼働させる方法

自分のPCをつけっぱなしにしなくても、クラウド上でボットを動かし続ける方法を学びましょう！

🎁おすすめツール・サービス

FX自動売買VPSなら50%OFF

キャンペーンで月1,790円〜。MT4/MT5プリインストール済み、自動復旧機能付き。14日間無料お試し。

シンクラウドデスクトップ for FX
ゴトー日EAを探しているならソースコード付

4年で資金4.7倍のバックテスト実績。ソースコード付きで自由にカスタマイズ可能。

ゴトー日EA Ultimate版（3,980円）

※上記リンクにはプロモーション（広告）が含まれています

🔮
MEMBERSHIP

量子相場謀略｜ホソノP

ブログでは書けない本音のトレード戦略とEA開発の裏側を公開。 データに基づいた検証で、再現性のある手法だけをお届けします。

✓EAソースコード公開
✓バックテスト生データ共有
✓リアルタイム相場分析
✓メンバー限定記事
メンバーシップに参加する※ noteアカウントで簡単登録
ポストする@hosono_p をフォロー

📚あわせて読みたい

📄

【Python FXボット入門】MT4を使わない自動売買の始め方

Pythonで直接FX取引する自動売買ボットの作り方。MT4に依存しない、自由度の高いシステム開発の第一歩。GMOコインやOANDAのAPIを使った実践的な入門ガイド。

FX自動売買2025/12/13
📄

【Python】移動平均線クロスボットを作る｜pandasでテクニカル分析

Pythonとpandasを使って移動平均線クロス戦略のFXボットを作ります。ローソク足データの取得から、MA計算、クロス検出、自動売買まで完全解説。

FX自動売買2025/12/13
📄

【Python】損切り・リスク管理の自動化｜トレーリングストップの実装

FXボットのリスク管理を自動化する方法を解説。固定損切り、トレーリングストップ、ブレイクイーブン、ATRベースの動的損切りまで、Pythonで実装します。

FX自動売買2025/12/13
📄

【コピペOK】RSI逆張りEAの作り方｜売られすぎ・買われすぎで自動売買

RSIが30以下で買い、70以上で売りという逆張りEAを作ります。移動平均線クロス（順張り）との違いも解説。コード全文公開、初心者でも理解できます。

FX自動売買2025/12/13
📄

【検証】ゴトー日ショート戦略×テクニカルフィルター｜22種類試して判明した衝撃の事実

「RSIが買われすぎだから売り！」それ、逆効果でした。323回のゴトー日ショートに22種類のテクニカルフィルターを適用した結果、ロングとは全く違う傾向が判明。ADXレンジ×MACDで勝率70%超えの組み合わせを発見！

FX自動売買2026/1/19
📄

📊【検証】RSI70/30クロス戦略は使えるのか？「過熱後の冷却」を狙う手法を統計検定

RSIが70を超えた後10本以内に70を下回った時に売り、30を下回った後10本以内に30を上回った時に買い。この「過熱後の冷却タイミング」を狙う戦略は有効？28,000本のデータで検証した結果、4時間足の買いシグナルに光明が！

FX自動売買2025/12/19