😤『国内FXでもゴトー日やりたい！』GMOコインAPIで自動売買を作る全手順 - Python + GCF で24時間稼働

ゴトー日判定 → GMOコイン 外国為替FX API → USD/JPY 売買
     ↑
Cloud Scheduler（定期実行）→ Cloud Functions（Python）

gotobi-gmo-fx/
├── main.py              # メインスクリプト
├── gmo_fx_client.py     # GMOコインFX API クライアント
├── gotobi_logic.py      # ゴトー日判定ロジック
└── requirements.txt     # 依存パッケージ

# gmo_fx_client.py
import hmac
import hashlib
import time
import requests
import json
from datetime import datetime

class GMOFXClient:
    """
    GMOコイン外国為替FX APIと通信するためのクライアントクラス。
    APIへのリクエスト、署名生成、各種データ取得・注文機能をカプセル化します。
    """

    def __init__(self, api_key: str, api_secret: str):
        """
        クライアントの初期化。APIキーとシークレットを設定します。
        """
        self.api_key = api_key
        self.api_secret = api_secret
        self.private_endpoint = 'https://forex-api.coin.z.com/private'
        self.public_endpoint = 'https://forex-api.coin.z.com/public'

    def _create_signature(self, timestamp: str, method: str, path: str, body: str = '') -> str:
        """
        APIリクエストに必要な署名を生成します。（Private APIでのみ使用）
        GMOコインのAPI仕様で定められたルールに従い、
        「タイムスタンプ + HTTPメソッド + パス + リクエストボディ」
        という文字列を、APIシークレットを使って暗号化（HMAC-SHA256）します。
        これにより、リクエストが改ざんされておらず、正規のユーザーからのものであることを証明します。
        """
        text = timestamp + method + path + body
        sign = hmac.new(
            bytes(self.api_secret.encode('ascii')),
            bytes(text.encode('ascii')),
            hashlib.sha256
        ).hexdigest()
        return sign

    def _get_timestamp(self) -> str:
        """
        APIリクエストに使用するUNIXタイムスタンプ（ミリ秒）を取得します。
        """
        return '{0}000'.format(int(time.mktime(datetime.now().timetuple())))

    def _private_request(self, method: str, path: str, body: dict = None) -> dict:
        """
        Private API（認証が必要なAPI）へのリクエストを送信する共通関数。
        署名の生成とヘッダーの設定を自動で行います。
        """
        timestamp = self._get_timestamp()
        body_str = '' if body is None else json.dumps(body)

        headers = {
            'API-KEY': self.api_key,
            'API-TIMESTAMP': timestamp,
            'API-SIGN': self._create_signature(timestamp, method, path, body_str),
            'Content-Type': 'application/json'
        }
        url = self.private_endpoint + 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_assets(self) -> dict:
        """
        資産残高（取引余力、証拠金維持率など）を取得します。
        """
        return self._private_request('GET', '/v1/account/assets')

    def get_positions(self, symbol: str = None) -> dict:
        """
        現在保有しているポジション（建玉）の一覧を取得します。
        """
        path = '/v1/openPositions'
        if symbol:
            path += f'?symbol={symbol}'
        return self._private_request('GET', path)

    def place_order(
        self, symbol: str, side: str, size: int, execution_type: str = 'MARKET', time_in_force: str = 'FAK'
    ) -> dict:
        """
        新規の成行注文を実行します。
        """
        path = '/v1/order'
        body = {
            'symbol': symbol, 'side': side, 'executionType': execution_type, 'size': str(size), 'timeInForce': time_in_force
        }
        return self._private_request('POST', path, body)

    def close_all_positions(self, symbol: str, side: str) -> list:
        """
        指定した通貨ペアのポジションをすべて決済します。
        """
        results = []
        positions = self.get_positions(symbol)
        if positions.get('status') != 0:
            return [{'error': positions}]
        for position in positions.get('data', {}).get('list', []):
            if (position['side'] == 'BUY' and side == 'SELL') or (position['side'] == 'SELL' and side == 'BUY'):
                result = self._private_request('POST', '/v1/closeOrder', {
                    'symbol': symbol, 'side': side, 'executionType': 'MARKET',
                    'size': str(position['size']), 'positionId': position['positionId'], 'timeInForce': 'FAK'
                })
                results.append(result)
        return results

# gotobi_logic.py
from datetime import datetime, timedelta
import calendar

def is_gotobi(target_date: datetime = None) -> bool:
    """
    指定された日付がゴトー日（五十日）に該当するかを判定します。
    """
    if target_date is None:
        target_date = datetime.now()
    day = target_date.day
    weekday = target_date.weekday()
    if day % 5 == 0 and weekday < 5:
        return True
    if weekday == 4:
        tomorrow = target_date + timedelta(days=1)
        day_after_tomorrow = target_date + timedelta(days=2)
        if tomorrow.month == target_date.month and tomorrow.day % 5 == 0:
            return True
        if day_after_tomorrow.month == target_date.month and day_after_tomorrow.day % 5 == 0:
            return True
    return False

def is_buy_time(current_time: datetime = None) -> bool:
    """
    買いエントリーの時間帯（日本時間 2:00 ～ 9:54）かどうかを判定します。
    """
    if current_time is None:
        current_time = datetime.now()
    hour, minute = current_time.hour, current_time.minute
    return (2 <= hour <= 8) or (hour == 9 and minute <= 54)

def is_sell_time(current_time: datetime = None) -> bool:
    """
    売りエントリーの時間帯（日本時間 9:55 ～ 14:59）かどうかを判定します。
    """
    if current_time is None:
        current_time = datetime.now()
    hour, minute = current_time.hour, current_time.minute
    return (hour == 9 and minute >= 55) or (10 <= hour <= 14)

def is_close_time(current_time: datetime = None) -> bool:
    """
    ポジションを決済する時間帯（日本時間 15:00以降）かどうかを判定します。
    """
    if current_time is None:
        current_time = datetime.now()
    return current_time.hour >= 15

def get_action(current_time: datetime = None) -> str:
    """
    現在の時刻情報に基づいて、実行すべきアクションを決定します。
    """
    if current_time is None:
        current_time = datetime.now()
    if current_time.weekday() >= 5 or not is_gotobi(current_time):
        return 'WAIT'
    if is_buy_time(current_time):
        return 'BUY'
    elif is_sell_time(current_time):
        return 'SELL'
    elif is_close_time(current_time):
        return 'CLOSE'
    else:
        return 'WAIT'

# main.py
import os
import logging
from datetime import datetime
from gmo_fx_client import GMOFXClient
from gotobi_logic import get_action, is_gotobi
from google.cloud import firestore

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

SYMBOL = 'USD_JPY'
ORDER_SIZE = 100

db = firestore.Client()
POSITION_DOC_REF = db.collection('bot_state').document('gotobi_position')

def save_position_state(state: str):
    try:
        POSITION_DOC_REF.set({'position': state, 'timestamp': firestore.SERVER_TIMESTAMP})
        logger.info(f"ポジション状態をFirestoreに保存: {state}")
    except Exception as e:
        logger.error(f"ポジション状態の保存に失敗: {e}")

def load_position_state() -> str:
    try:
        doc = POSITION_DOC_REF.get()
        if doc.exists:
            state = doc.to_dict().get('position')
            logger.info(f"Firestoreからポジション状態を読込: {state}")
            return state
        return None
    except Exception as e:
        logger.error(f"ポジション状態の読込に失敗: {e}")
        return None

def run_trading_bot():
    current_position = load_position_state()
    api_key = os.environ.get('GMO_FX_API_KEY')
    api_secret = os.environ.get('GMO_FX_API_SECRET')
    if not api_key or not api_secret:
        logger.error('APIキーが設定されていません')
        return {'status': 'error', 'message': 'API keys not configured'}

    client = GMOFXClient(api_key, api_secret)
    now = datetime.now()
    action = get_action(now)
    logger.info(f'現在時刻: {now}, ゴトー日: {is_gotobi(now)}, 判定アクション: {action}, 現在ポジション: {current_position}')
    result = {'timestamp': str(now), 'action': action, 'executed': False}

    try:
        if action == 'BUY' and current_position != 'LONG':
            if current_position == 'SHORT':
                logger.info('既存ショートポジションを決済')
                client.close_all_positions(SYMBOL, 'BUY')
            logger.info(f'ロングエントリー実行: {SYMBOL} {ORDER_SIZE}通貨')
            response = client.place_order(SYMBOL, 'BUY', ORDER_SIZE)
            if response.get('status') == 0:
                save_position_state('LONG')
                result['executed'] = True
        elif action == 'SELL' and current_position != 'SHORT':
            if current_position == 'LONG':
                logger.info('既存ロングポジションを決済')
                client.close_all_positions(SYMBOL, 'SELL')
            logger.info(f'ショートエントリー実行: {SYMBOL} {ORDER_SIZE}通貨')
            response = client.place_order(SYMBOL, 'SELL', ORDER_SIZE)
            if response.get('status') == 0:
                save_position_state('SHORT')
                result['executed'] = True
        elif action == 'CLOSE' and current_position is not None:
            side_to_close = 'SELL' if current_position == 'LONG' else 'BUY'
            logger.info(f'{current_position} ポジションを決済 (実行サイド: {side_to_close})')
            client.close_all_positions(SYMBOL, side_to_close)
            save_position_state(None)
            result['executed'] = True
        else:
            logger.info('取引条件を満たさないため待機')
    except Exception as e:
        logger.error(f'取引処理中に予期せぬエラーが発生: {e}')
        result['error'] = str(e)
    return result

def gotobi_trading(request):
    return run_trading_bot()

if __name__ == '__main__':
    from dotenv import load_dotenv
    load_dotenv()
    print(run_trading_bot())

# requirements.txt
functions-framework==3.*
requests>=2.28.0
python-dotenv>=1.0.0
google-cloud-firestore>=2.13.0