PythonでGMO FX APIから最新ローソク足を取得する方法｜既存CSVへ不足分を追加

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()

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()