【Pythonコードあり】GMO FX APIでFX自動売買を実装！ローソク足取得からエントリーまで

from pathlib import Path

import backtest_config as strategy

# Falseの場合は注文内容を表示するだけで、実際の注文は行いません。
# 実際に注文する場合だけTrueへ変更してください。
ENABLE_REAL_ORDER = False

LIMIT_BUFFER_PIPS = 0.2
MAX_SPREAD_PIPS = 3.0

# GMO FXへ送る価格の小数桁数です。
# USD/JPYは3桁、EUR/USDなど別の通貨ペアでは、その銘柄に合う桁数へ変更してください。
PRICE_DECIMAL_PLACES = 3

ORDER_HISTORY_PATH = (
    Path(__file__).resolve().parent
    / "data"
    / "orders"
    / "order_history.csv"
)

# APIキーはコードへ直接書かず、環境変数から読み込みます。
API_KEY_ENV = "GMO_FX_API_KEY"
SECRET_KEY_ENV = "GMO_FX_SECRET_KEY"

import hashlib
import hmac
import json
import os
import ssl
import time
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import certifi

import backtest_config as strategy
import candle_updater
import entry_signal
import live_config as config
from order_recorder import save_order_record

# 最新レートなど、APIキーなしで利用できるPublic APIの接続先です。
PUBLIC_URL = "https://forex-api.coin.z.com/public"

# 建玉・注文の確認や発注に使用するPrivate APIの接続先です。
PRIVATE_URL = "https://forex-api.coin.z.com/private"

# HTTPS通信時にサーバー証明書を検証するため、certifiのCA証明書を使います。
SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where())

# APIから取得した時刻や売買判定の基準を日本時間へそろえます。
JST = timezone(timedelta(hours=9), name="JST")

# 使い方:
# 1. fetch_gmo_candles.pyで、判定に必要な過去のローソク足CSVを作成します。
# 2. live_config.pyで、実注文の有効・無効や最大スプレッドを設定します。
# 3. PowerShellなどでGMO-APIフォルダへ移動します。
# 4. 次のコマンドを実行します。
#    python live_trade.py
# 5. 初期設定ではデモ発注として、注文内容を表示・CSV記録するだけです。
# 6. 実注文する場合だけ、live_config.pyのENABLE_REAL_ORDERをTrueにします。
#
# このスクリプトが行う処理:
# ローソク足更新 → エントリー判定 → 最新レート確認 → IFDOCO注文作成 → 発注記録
#
# 注意:
# このサンプルは注文の送信までを扱います。LIMIT注文が実際に約定したか、
# TP・SLで決済されたか、損益がいくらになったかは確認しません。


def request_api(request):
    """GMO FX APIへリクエストを送り、data部分を返す。"""

    # タイムアウトを設定して、APIから応答がない状態で待ち続けることを防ぎます。
    with urlopen(request, context=SSL_CONTEXT, timeout=20) as response:
        # レスポンス本文はUTF-8のJSONなので、Pythonの辞書へ変換します。
        payload = json.loads(response.read().decode("utf-8"))

    # GMO FX APIは、正常に処理された場合にstatus=0を返します。
    if int(payload.get("status", -1)) != 0:
        raise RuntimeError(f"GMO FX APIエラー: {payload.get('messages')}")

    # 呼び出し元では、共通情報を除いたdata部分だけを扱います。
    return payload.get("data")


def public_get(path, params=None):
    """認証不要のPublic APIへGETリクエストを送る。"""

    # 検索条件がある場合だけ、辞書をURL用のクエリ文字列へ変換します。
    query = f"?{urlencode(params)}" if params else ""

    # Public APIのURLとAPIごとのパス・検索条件を連結します。
    return request_api(Request(PUBLIC_URL + path + query))


def private_get(path, params=None):
    """署名が必要なPrivate APIへGETリクエストを送る。"""

    # 建玉や注文を通貨ペアで絞り込むためのクエリ文字列を作ります。
    query = f"?{urlencode(params)}" if params else ""
    return private_request("GET", path, query=query)


def private_post(path, body):
    """署名が必要なPrivate APIへPOSTリクエストを送る。"""

    # 署名対象と送信内容が同じになるよう、注文内容を余分な空白のないJSONにします。
    body_text = json.dumps(body, separators=(",", ":"))
    return private_request("POST", path, body_text=body_text)


def private_request(method, path, query="", body_text=""):
    """環境変数のAPIキーで署名し、Private APIへリクエストを送る。"""

    # APIキーをソースコードへ直接書かず、live_config.pyで指定した環境変数から取得します。
    api_key = os.environ.get(config.API_KEY_ENV)
    secret_key = os.environ.get(config.SECRET_KEY_ENV)
    if not api_key or not secret_key:
        raise RuntimeError("GMO FXのAPIキーを環境変数へ設定してください。")

    # GMO Private APIのPOST制限を考慮し、連続発注にならないよう間隔を空けます。
    if method == "POST":
        time.sleep(1.2)

    # API-TIMESTAMPには、現在時刻をミリ秒単位のUNIX時刻で指定します。
    timestamp = str(int(time.time() * 1000))

    # GMO FXの仕様に従い、時刻・HTTPメソッド・パス・本文を順番に連結します。
    # GETのクエリ文字列は署名対象へ含めません。
    sign_text = timestamp + method + path + body_text

    # シークレットキーと署名対象文字列から、HMAC-SHA256署名を作成します。
    signature = hmac.new(
        secret_key.encode("ascii"), sign_text.encode("ascii"), hashlib.sha256
    ).hexdigest()

    # Private APIの認証に必要なAPIキー・時刻・署名をヘッダーへ設定します。
    headers = {
        "API-KEY": api_key,
        "API-TIMESTAMP": timestamp,
        "API-SIGN": signature,
    }

    # JSON本文を送信するPOSTリクエストでは、データ形式もヘッダーへ指定します。
    if body_text:
        headers["Content-Type"] = "application/json"

    # GETでは本文を付けず、POSTではJSON文字列をUTF-8へ変換して送信します。
    request = Request(
        PRIVATE_URL + path + query,
        data=body_text.encode("utf-8") if body_text else None,
        headers=headers,
        method=method,
    )
    return request_api(request)


def latest_ticker(symbol):
    """指定した通貨ペアの最新tickerを取得する。"""

    # ticker APIは複数通貨ペアを返すため、設定した通貨ペアだけを探します。
    for row in public_get("/v1/ticker") or []:
        if row.get("symbol") == symbol:
            return row

    # 対象がなければ、後続処理で誤った価格を使わないよう処理を止めます。
    raise RuntimeError(f"{symbol}の最新レートが見つかりません。")


def list_items(data):
    """GMO FX APIの一覧レスポンスをリストへそろえる。"""

    # APIによってdataが直接リストで返る場合は、そのまま利用します。
    if isinstance(data, list):
        return data

    # data内のlistに明細が入る形式にも対応します。
    if isinstance(data, dict):
        return data.get("list", [])

    # データがない場合も、呼び出し元が件数を安全に確認できるよう空リストを返します。
    return []


def format_price(value):
    """価格をコンフィグで指定した小数桁数の文字列に変換する。"""

    # floatのround()ではなくDecimalを使い、指定した小数桁数で四捨五入します。
    decimal_places = config.PRICE_DECIMAL_PLACES
    quantize_unit = Decimal("1").scaleb(-decimal_places)
    rounded = Decimal(str(value)).quantize(quantize_unit, ROUND_HALF_UP)
    return f"{rounded:.{decimal_places}f}"

def make_ifd_oco_order(ask, now):
    """買いLIMITと、利益確定・損切りを組み合わせたIFDOCO注文を作る。"""

    # ロング（買い）はASKを基準にします。
    # 注文を約定しやすくするため、現在のASKに0.2pipsを加えた買い指値を作ります。
    # この0.2pipsはlive_config.pyで変更できます。
    entry_price = ask + config.LIMIT_BUFFER_PIPS * strategy.PIP_SIZE

    # エントリー価格を基準に、コンフィグのpips幅からTP・SL価格を計算します。
    take_profit_price = (
        entry_price + strategy.TAKE_PROFIT_PIPS * strategy.PIP_SIZE
    )
    stop_loss_price = entry_price - strategy.STOP_LOSS_PIPS * strategy.PIP_SIZE

    # clientOrderIdは、自分のスクリプトから送った注文を識別するためのIDです。
    # このサンプルでは、移動平均線を使う戦略を表す「SMA」と実行日時を組み合わせます。
    return {
        "symbol": strategy.SYMBOL,
        "clientOrderId": "SMA" + now.strftime("%Y%m%d%H%M%S"),

        # IFDOCOでは、エントリーと決済条件を1回のAPIリクエストでまとめて登録します。

        # 一次注文: 新しく買いポジションを作るLIMIT注文です。
        "firstSide": "BUY",
        "firstExecutionType": "LIMIT",
        "firstSize": str(strategy.POSITION_SIZE_UNITS),
        "firstPrice": format_price(entry_price),

        # 二次注文: 買ったポジションを決済するための注文です。
        # 一次注文が約定した後、利益確定または損切りのどちらかで売却します。
        # どちらかが成立すると、もう一方はOCOの仕組みで処理されます。
        "secondSize": str(strategy.POSITION_SIZE_UNITS),
        "secondLimitPrice": format_price(take_profit_price),
        "secondStopPrice": format_price(stop_loss_price),
    }

def print_order(order):
    """APIへ送る主な注文内容を、日本語で確認できるように表示する。"""

    print("注文内容")
    print(f"  通貨ペア: {order['symbol']}")
    print(f"  売買方向: 買い（{order['firstSide']}）")
    print(f"  注文方法: IFDOCO / {order['firstExecutionType']}")
    print(f"  注文数量: {order['firstSize']}")
    print(f"  エントリー価格: {order['firstPrice']}")
    print(f"  利益確定価格: {order['secondLimitPrice']}")
    print(f"  損切り価格: {order['secondStopPrice']}")

def make_order_record(
    now,
    signal,
    order,
    mode,
    status,
    message,
    root_order_id="",
):
    """発注結果と注文内容を、CSVへ保存できる形にまとめる。"""

    return {
        "recorded_at": now.isoformat(),
        "mode": mode,
        "status": status,
        "symbol": order["symbol"],
        "side": order["firstSide"],
        "size": order["firstSize"],
        "entry_price": order["firstPrice"],
        "take_profit_price": order["secondLimitPrice"],
        "stop_loss_price": order["secondStopPrice"],
        "signal_time": signal["close_time"].isoformat(),
        "client_order_id": order["clientOrderId"],
        "root_order_id": root_order_id,
        "message": message,
    }


def root_order_id_from_result(result):
    """IFDOCOレスポンスの3注文に共通する親注文IDを取り出す。"""

    # IFDOCOのレスポンスには、一次注文・利益確定・損切りの情報が
    # リストで返ります。その先頭から、共通のrootOrderIdを取得します。
    if isinstance(result, list) and result and isinstance(result[0], dict):
        return str(result[0].get("rootOrderId", ""))
    return ""


def main():
    # すべての日時判定をそろえるため、現在時刻を日本時間で1回だけ取得します。
    now = datetime.now(JST)
    print(f"処理開始日時: {now.isoformat()}")

    # 作成済みの15分足・1時間足・4時間足・日足CSVへ最新データを追加します。
    print("\n【ローソク足データの更新】")
    candle_updater.update_candle_csvs(now)

    # 更新したCSVを使い、確定済みの最新15分足で買い条件を判定します。
    # シグナルがない場合は、発注処理へ進まず終了します。
    print("\n【エントリー条件の判定】")
    signal = entry_signal.find_latest_signal(now)
    if signal is None:
        print("買いシグナルはありません。")
        return

    # 最新tickerから、買値のASKと売値のBIDを取得します。
    # tickerはPublic APIなので、デモ発注ではAPIキーなしで確認できます。
    print("\n【最新レートの取得】")
    ticker = latest_ticker(strategy.SYMBOL)
    ask = float(ticker["ask"])
    bid = float(ticker["bid"])
    print(f"ASK: {format_price(ask)}")
    print(f"BID: {format_price(bid)}")

    # ASKとBIDの差をpipsへ変換します。
    # スプレッドが設定値より広いと、不利な価格で取引する可能性があるため見送ります。
    spread_pips = (ask - bid) / strategy.PIP_SIZE
    print(f"スプレッド: {spread_pips:.2f} pips")
    if spread_pips > config.MAX_SPREAD_PIPS:
        print(f"スプレッドが広いため見送ります: {spread_pips:.2f} pips")
        return

    # 最新ASKを基準にIFDOCO注文を作り、画面に表示します。
    print("\n【IFDOCO注文の作成】")
    order = make_ifd_oco_order(ask, now)
    print_order(order)

    # デモ発注または実注文の処理を先に行い、結果を変数へ入れます。
    # 注文履歴CSVへの保存は、この後の共通処理で1回だけ行います。
    root_order_id = ""
    order_error = None
    result = None

    # ENABLE_REAL_ORDERがFalseの場合は、GMO FXへ注文を送りません。
    if not config.ENABLE_REAL_ORDER:
        mode = "DEMO"
        status = "DEMO"
        message = "実際の注文は送信していません。"
    else:
        # 実注文時だけ、Private APIで現在の建玉と未約定注文を確認します。
        # どちらかが存在する場合は、ポジションや注文が重なるのを避けるため見送ります。
        positions = list_items(
            private_get("/v1/openPositions", {"symbol": strategy.SYMBOL})
        )
        orders = list_items(
            private_get("/v1/activeOrders", {"symbol": strategy.SYMBOL})
        )
        print(f"現在の建玉: {len(positions)}件")
        print(f"未約定注文: {len(orders)}件")
        if positions or orders:
            print("既存の建玉または未約定注文があるため見送ります。")
            return

        mode = "LIVE"

        # 条件をすべて通過したため、Private APIへIFDOCO注文を送信します。
        try:
            result = private_post("/v1/ifoOrder", order)
            status = "SUBMITTED"
            message = "GMO FXへ注文を送信しました。"
            root_order_id = root_order_id_from_result(result)
        except Exception as error:
            # 通信エラーでもGMO側が注文を受理している可能性があります。
            # 同じ注文を再送信する前に、GMO側の注文一覧を確認してください。
            status = "ERROR"
            message = str(error)
            order_error = error

    # 発注結果が決まったため、CSVへ保存するデータを作ります。
    print("\n【注文内容・送信結果の記録】")
    record = make_order_record(
        now,
        signal,
        order,
        mode=mode,
        status=status,
        message=message,
        root_order_id=root_order_id,
    )
    save_order_record(config.ORDER_HISTORY_PATH, record)
    print(f"保存先: {config.ORDER_HISTORY_PATH}")

    if order_error is not None:
        raise order_error

    if mode == "DEMO":
        print("デモ発注を注文履歴CSVへ記録しました。実注文は行っていません。")
    else:
        # SUBMITTEDは注文受付を表し、LIMIT注文が約定したEXECUTEDとは異なります。
        print("実際の注文をGMO FXへ送信し、注文履歴CSVへ記録しました。")
        print("GMO FXからの注文結果:", result)


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

import csv
import json
import ssl
import time
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import certifi

import backtest_config as strategy

# 更新するローソク足の時間足です。
# 1時間足以下は日単位、4時間足以上は年単位でGMO APIから取得します。
CANDLE_INTERVALS = ("15min", "1hour", "4hour", "1day")

# CSVに出力する列です。
CSV_COLUMNS = ["timestamp_jst", "open", "high", "low", "close"]

# GMO FXのPublic APIの接続先です。
PUBLIC_URL = "https://forex-api.coin.z.com/public"

# certifiが持っている証明書一覧を使って、HTTPS通信の証明書を検証します。
# Windows環境でCERTIFICATE_VERIFY_FAILEDになる場合の対策です。
SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where())

# GMO APIの時刻はUTCのミリ秒で返るため、CSVにはJSTへ変換して保存します。
JST = timezone(timedelta(hours=9), name="JST")


def public_get(path, params=None):
    """GMO FXのPublic APIへGETリクエストを送る。"""

    # APIに渡すパラメータがある場合は、URL用のクエリ文字列へ変換します。
    query = f"?{urlencode(params)}" if params else ""

    # Public APIの接続先、APIごとのパス、クエリ文字列をつないでURLを作ります。
    request = Request(PUBLIC_URL + path + query)

    # GMO APIへアクセスして、返ってきたJSON文字列をPythonの辞書に変換します。
    # 20秒以内に応答がない場合は、待ち続けずタイムアウトにします。
    with urlopen(request, context=SSL_CONTEXT, timeout=20) as response:
        payload = json.loads(response.read().decode("utf-8"))

    # GMO APIのstatusが0以外なら、エラーとして処理を止めます。
    if int(payload.get("status", -1)) != 0:
        raise RuntimeError(f"GMO FX APIエラー: {payload.get('messages')}")

    # payload["data"]に、APIから取得したデータが入っています。
    return payload.get("data")


def fetch_klines(symbol, price_type, interval, date_key):
    """指定した日付のローソク足を取得し、CSV用の形式へ変換する。"""

    # GMO APIへ連続でアクセスしすぎないように、1秒待ってから呼び出します。
    time.sleep(1.0)

    # 通貨ペア、価格種別、時間足、日付を指定してローソク足を取得します。
    # date_keyは、1時間足以下では"20260102"、4時間足以上では"2026"の形式です。
    rows = public_get(
        "/v1/klines",
        {
            "symbol": symbol,
            "priceType": price_type,
            "interval": interval,
            "date": date_key,
        },
    )

    # APIのレスポンスから、CSVに保存するローソク足をこのリストへまとめます。
    result = []
    for row in rows or []:
        # openTimeはUTCのミリ秒なので、日時へ変換してから日本時間へ直します。
        timestamp = datetime.fromtimestamp(
            int(row["openTime"]) / 1000, tz=timezone.utc
        ).astimezone(JST)

        # CSVに必要な開始日時と四本値だけを取り出します。
        result.append(
            {
                "timestamp_jst": timestamp.isoformat(timespec="seconds"),
                "open": row["open"],
                "high": row["high"],
                "low": row["low"],
                "close": row["close"],
            }
        )

    # CSVへ追加できる形式に変換したローソク足を返します。
    return result


def read_csv(path):
    """既存のローソク足CSVを読み込む。"""

    # 初回実行などでCSVがまだ存在しない場合は、空のデータとして扱います。
    if not path.exists():
        return []

    # utf-8-sigを指定し、BOM付き・BOMなしのどちらのCSVも読み込めるようにします。
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        return list(csv.DictReader(file))


def update_candle_csvs(now):
    """作成済みCSVへ、最新のローソク足を重複なく追加する。"""

    # GMOの1時間足以下の日付キーは、JST午前6時に切り替わります。
    # 午前0時から6時までは前日をAPI上の基準日として扱います。
    business_date = (now - timedelta(hours=6)).date()
    print("ローソク足CSVの差分更新を開始します。")

    # CANDLE_INTERVALSで指定した時間足を1つずつ更新します。
    for interval in CANDLE_INTERVALS:
        # 時間足に対応するCSVの保存先をbacktest_config.pyから取得します。
        path = strategy.CANDLE_CSV_PATHS[interval]

        # 保存済みのローソク足を読み込み、新しく取得した足を後から追加します。
        rows = read_csv(path)
        print(f"[{interval}] のデータ取得中...")

        if interval in ("15min", "1hour"):
            # 1時間足以下は日ごとに取得するため、CSVの最終日時を取得開始日にします。
            # CSVが空の場合は、現在の基準日から取得します。
            start_date = (
                datetime.fromisoformat(rows[-1]["timestamp_jst"]).date()
                if rows
                else business_date
            )

            # 日付境界付近の足を取りこぼさないよう、前日から再取得します。
            start_date = min(start_date, business_date) - timedelta(days=1)

            # APIへ渡す日付を"20260102"のようなYYYYMMDD形式で1日ずつ作ります。
            date_keys = [
                (start_date + timedelta(days=offset)).strftime("%Y%m%d")
                for offset in range((business_date - start_date).days + 1)
            ]
        else:
            # 4時間足以上は年ごとに取得するため、CSVの最終日時から開始年を決めます。
            # CSVが空の場合は、現在の基準年だけを取得します。
            start_year = (
                datetime.fromisoformat(rows[-1]["timestamp_jst"]).year
                if rows
                else business_date.year
            )

            # APIへ渡す日付を"2026"のようなYYYY形式で1年ずつ作ります。
            date_keys = [
                str(year) for year in range(start_year, business_date.year + 1)
            ]

        # 作成した日付または年を順番にAPIへ渡し、ローソク足を取得します。
        for date_key in date_keys:
            fetched_rows = fetch_klines(
                strategy.SYMBOL,
                strategy.PRICE_TYPE,
                interval,
                date_key,
            )

            # 既存のローソク足へ、今回取得したローソク足を追加します。
            rows.extend(fetched_rows)

        # 再取得した同じ開始時刻の足を1本にまとめます。
        # 同じ時刻が複数ある場合は、後からAPIで取得した新しい内容を残します。
        unique_rows = {row["timestamp_jst"]: row for row in rows}

        # CSVが古い足から新しい足の順になるよう、開始日時で並べ替えます。
        rows = sorted(unique_rows.values(), key=lambda row: row["timestamp_jst"])

        # 保存先フォルダが存在しない場合は、親フォルダも含めて作成します。
        path.parent.mkdir(parents=True, exist_ok=True)

        # 更新後の全ローソク足でCSVを上書きします。
        with path.open("w", encoding="utf-8", newline="") as file:
            writer = csv.DictWriter(file, fieldnames=CSV_COLUMNS)

            # 1行目へ列名を書き、その後へローソク足をまとめて書き込みます。
            writer.writeheader()
            writer.writerows(rows)

    print("ローソク足CSVの差分更新が完了しました。")


def main():
    # ローソク足の日付判定を日本時間で行うため、現在時刻をJSTで取得します。
    now = datetime.now(JST)

    # 設定されたすべての時間足について、CSVへ最新データを追加します。
    update_candle_csvs(now)
    print("15分足・1時間足・4時間足・日足のCSVを更新しました。")


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