FXの戦略を考えて、自動売買プログラムに実装【FXの自動売買プログラムをPythonで自作しよう⑦】

記事内に広告を含みます

この記事には広告を含む場合があります。

記事内で紹介する商品を購入することで、当サイトに売り上げの一部が還元されることがあります。

自動売買のメリットは、感情に影響されずに取引ができることだと思います。

実際にトレードをしていると、なかなか損切りができずに含み損を抱えたまま身動きが取れなくなり、トレードの機会を逃してしまうこともあります。

本記事では、次の2点について戦略を立てていきます。

資金管理
テクニカル分析

立てた戦略を元にプログラムに落とし込んでいきます。

資金管理
1. 2％ルール
2. リスクリワードレシオ
テクニカル分析
プログラムのフローチャート
サンプルコード
ポイントの解説
おわりに

目次へ

資金管理

FXの資金管理で重要な2％ルールとリスクリワード比率について説明します。

2％ルール

2％ルールとは、1回のトレードにおける損失を口座資金の2％以内に抑えるというものです。

例えば10万円の資金でトレードする場合、損失が2千円に達した時点で損切りを行います。

口座資産に変動があった場合は再計算を行い、一定割合のリスクを保つようにします。

これをプログラムで再現していきます。

リスクリワードレシオ

リスクリワードレシオとは、損失（リスク）と利益（リワード）の比率で、次の計算式で求めることができます。

リスクリワードレシオ=勝ちトレードの平均利益÷負けトレードの平均損失

つまり、リスクリワードレシオが1を超えると勝ちトレードの平均利益の方が負けトレードの平均損失よりも高く、1を下回ると負けトレードの平均損失の方が勝ちトレードの平均利益を上回るということです。

次のグラフは、リスクリワードレシオと勝率の関係をグラフ化したものです。（縦軸：リスクリワードレシオ、横軸：勝率）

出典：OANDA

リスクリワードレシオを1とすると、勝率が50％以上ないと損益がプラスにならないということになります。

リスクリワードレシオはコントロールできますが、勝率はコントロールできません。目標とするリスクリワードレシオを達成できる勝率の手法を見つけることが重要です。

理想的なリスクリワードレシオは投資手法によって変わりますが、一般的に順張りは勝率が低い傾向なのでリスクリワードレシオは高めに、逆張りは勝率が高い傾向なのでリスクリワードレシオは低めに設定するのが良いと思います。

テクニカル分析

テクニカル分析とは、過去の値動きから将来の値動きを分析する方法です。テクニカル分析に利用するインジケーターは、移動平均線（MA）やボリンジャーバンドが有名です。

インジケーターの計算は、プログラミングの得意とするところです。

まず、順張りと逆張りについて考えたいと思います。どちらを選択するかによって、有効なインジケーターが変わってきます。

順張りと逆張りの特徴は次のとおりです。

	順張り	逆張り
取引方法	トレンドに沿って取り引きをする	トレンドに逆らって取り引きをする
得意な相場	トレンド相場	レンジ相場
メリット	トレンドが発生すると大きな利益を期待できる	短期で利益を上げやすい
デメリット	ダマシが多い	トレンドが発生していると損失を広げる
有効なインジケーター	移動平均線、MACDなど	ボリンジャーバンド、RSIなど

それぞれ相反する特徴があり、プログラミングで相場を判断することは難しいと考えています。どちらを選択するかは、状況に応じて判断する必要があります。

今回はシンプルに、順張りで移動平均線のゴールデンクロスやデッドクロスをシグナルにしたロジックのプログラムを作成していきます。

プログラムのフローチャート

上記を踏まえて戦略のフローチャートは次のようにします。

フローチャートの流れ

ポジションの確認をして、有れば10秒待って繰り返す。無ければ次へ。
価格データを取得する。
移動平均線の計算をして、ゴールデンクロス又はデッドクロスの判定をする。
シグナルが発生していなければ②へ戻る。発生していれば次へ。
成行き注文を出す。
利確と損切りの価格を計算する。
利確と損切りの価格をOCO注文して、①に戻る。

⑦のOCO注文とは「One Cancels the Other」の略で、2つの注文を一度に発注しておき、どちらかが成立すると、もう一つが自動的にキャンセルになるという注文方法です。リスクリワードレシオを固定するために、利確と損切りの価格を計算して注文を出しておきます。

その後、ポジションが決済されるまで待機をして、ポジションが無くなれば②に移ります。

サンプルコード

上記のロジックでプログラムを作成すると次のようになります。

Python

import requests
import json
import hmac
import hashlib
import time
from datetime import datetime
from datetime import timedelta
import pandas as pd

apiKey = 'YOUR_API_KEY'
secretKey = 'YOUR_SECRET_KEY'

# -----設定項目
interval = '1hour'  # 使用する時間軸
get_price_period = 10  # 四本値を取得する日数
MA_short_period = 5  # 移動平均線の短期の日数
MA_long_period = 15  # 移動平均線の長期の日数
order_size = '10000'  # 注文数量
trade_risk = 0.02  # 1トレードあたりで許容できる口座の損失（％）
risk_reward_ratio = 2  # リスクリワードレシオ


# 成行注文を出す関数
def market_order():
    global flag

    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://forex-api.coin.z.com/private'
            path = '/v1/order'
            reqBody = {
                'symbol': 'USD_JPY',
                'side': flag['order']['side'],
                'size': order_size,
                'executionType': 'MARKET'
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))
            print(json.dumps(res.json(), indent=2))

            return

        except Exception as e:
            print(f'action=market_order error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# OCO決済を出す関数
def oco_order(limit, stop):
    global flag

    if flag['order']['side'] == 'BUY':
        side = 'SELL'

    elif flag['order']['side'] == 'SELL':
        side = 'BUY'

    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://forex-api.coin.z.com/private'
            path = '/v1/closeOrder'
            reqBody = {
                'symbol': 'USD_JPY',
                'side': side,
                'size': order_size,
                'executionType': 'OCO',
                'limitPrice': str(limit),
                'stopPrice': str(stop)
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))
            print(json.dumps(res.json(), indent=2))

            return

        except Exception as e:
            print(f'action=market_order error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# ポジションを確認する関数
def check_positions():
    global flag

    timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
    method = 'GET'
    endPoint = 'https://forex-api.coin.z.com/private'
    path = '/v1/positionSummary'

    text = timestamp + method + path
    sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()
    parameters = {
        'symbol': 'USD_JPY'
    }

    headers = {
        'API-KEY': apiKey,
        'API-TIMESTAMP': timestamp,
        'API-SIGN': sign
    }

    res = requests.get(endPoint + path, headers=headers, params=parameters)
    positions = eval(json.dumps(res.json(), indent=2))
    if not positions['data']['list']:
        flag['position']['price'] = 0.0
        flag['position']['lot'] = 0
        flag['position']['side'] = None
        flag['position']['exist'] = False

        print('現在ポジションは存在しません')

        return

    flag['position']['price'] = float(positions['data']['list'][0]['averagePositionRate'])
    flag['position']['lot'] = float(positions['data']['list'][0]['sumPositionSize'])
    flag['position']['side'] = positions['data']['list'][0]['side']
    flag['position']['exist'] = True

    return


# 四本値を取得する関数
def get_price_data():
    while True:
        try:
            price_data = []
            response_data = []

            if (datetime.today().hour == 0 or
                    datetime.today().hour == 1 or
                    datetime.today().hour == 2 or
                    datetime.today().hour == 3 or
                    datetime.today().hour == 4 or
                    datetime.today().hour == 5):
                today = datetime.today().date()

            else:
                today = datetime.today().date() + timedelta(days=1)

            for i in range(get_price_period):
                period = i - get_price_period
                date = (today + timedelta(days=period)).strftime('%Y%m%d')
                endPoint = 'https://forex-api.coin.z.com/public'
                path = '/v1/klines?symbol=USD_JPY&priceType=ASK&interval=' + interval + '&date=' + date

                response = requests.get(endPoint + path)
                response_data.append(response.json())

            for j in range(len(response_data)):
                for i in response_data[j]['data']:
                    price_data.append({'openTime': i['openTime'],
                                       'openTime_dt': datetime.fromtimestamp(int(i['openTime'][:-3]))
                                      .strftime('%Y/%m/%d %H:%M'),
                                       'open': i['open'],
                                       'high': i['high'],
                                       'low': i['low'],
                                       'close': i['close']})

            print('TIME： ' + str(price_data[-1]['openTime_dt']) + ' H： ' + str(price_data[-1]['high']) + ' L： ' +
                  str(price_data[-1]['low']) + ' C： ' + str(price_data[-1]['close']))

            df_price_data = pd.DataFrame(price_data)

            return df_price_data

        except Exception as e:
            print(f'action=get_price_data error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# 資産残高を取得する関数
def get_collateral():
    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'GET'
            endPoint = 'https://forex-api.coin.z.com/private'
            path = '/v1/account/assets'

            text = timestamp + method + path
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.get(endPoint + path, headers=headers)
            resp = eval(json.dumps(res.json(), indent=2))
            collateral = int(float(resp['data']['equity']))
            print('現在のアカウント残高は{}円です'.format(collateral))

            return collateral

        except Exception as e:
            print(f'action=get_collateral error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# 移動平均線のGC、DCを判定してエントリー注文を出す関数
def entry_signal(df):
    global flag

    df['MA-short'] = df['open'].rolling(MA_short_period).mean()
    df['MA-long'] = df['open'].rolling(MA_long_period).mean()

    if df.iat[-2, 6] < df.iat[-2, 7] and df.iat[-1, 6] > df.iat[-1, 7]:
        flag['order']['side'] = 'BUY'
        print('ゴールデンクロスが発生しました。')

    elif df.iat[-2, 6] > df.iat[-2, 7] and df.iat[-1, 6] < df.iat[-1, 7]:
        flag['order']['side'] = 'SELL'
        print('デッドクロスが発生しました。')

    else:
        flag['order']['side'] = None

    return flag


# 利確と損切りの価格を計算する関数
def calculate_settlement_price():
    global flag

    profitable_price = 0
    stop_loss_price = 0
    collateral = get_collateral()

    if flag['position']['side'] == 'BUY':
        profitable_price = round(flag['position']['price'] + (collateral * trade_risk * risk_reward_ratio / int(order_size)), 3)
        stop_loss_price = round(flag['position']['price'] - (collateral * trade_risk / int(order_size)), 3)

    elif flag['position']['side'] == 'SELL':
        profitable_price = round(flag['position']['price'] - (collateral * trade_risk * risk_reward_ratio / int(order_size)), 3)
        stop_loss_price = round(flag['position']['price'] + (collateral * trade_risk / int(order_size)), 3)

    return profitable_price, stop_loss_price


# メイン処理
flag = {
    'order': {
        'side': None,
    },
    'position': {
        'exist': False,
        'side': None,
        'price': 0.0,
        'lot': 0,
    },
}

check_positions()

while True:

    # ポジションがある場合
    if flag['position']['exist'] is True:
        time.sleep(10)
        check_positions()
        continue

    # ポジションがない場合
    else:
        price = get_price_data()
        flag = entry_signal(price)

        # シグナルが発生した時
        if flag['order']['side'] is not None:
            market_order()
            time.sleep(5)
            check_positions()
            limit_price, stop_price = calculate_settlement_price()
            oco_order(limit_price, stop_price)
            time.sleep(3600)

        else:
            time.sleep(10)

ポイントの解説

少し長いので、ポイントとなるところの解説をしていきます。

必要なライブラリのインポートと設定項目

Python

import requests
import json
import hmac
import hashlib
import time
from datetime import datetime
from datetime import timedelta
import pandas as pd

apiKey = 'YOUR_API_KEY'
secretKey = 'YOUR_SECRET_KEY'

# 設定項目
interval = '1hour'  # 使用する時間軸
get_price_period = 5  # 四本値を取得する日数
MA_short_period = 5  # 移動平均線の短期の日数
MA_long_period = 15  # 移動平均線の長期の日数
order_size = '10000'  # 注文数量
trade_risk = 0.02  # 1トレードあたりで許容できる口座の損失（％）
risk_reward_ratio = 2  # リスクリワードレシオ

最初に変更する可能性のある項目をまとめておくと、後々の設定変更が楽になります。ここの設定は必要に応じて変更してみてください。

OCO注文を出す関数

Python

# OCO決済を出す関数
def oco_order(limit, stop):
    global flag

    if flag['order']['side'] == 'BUY':
        side = 'SELL'

    elif flag['order']['side'] == 'SELL':
        side = 'BUY'

    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://forex-api.coin.z.com/private'
            path = '/v1/closeOrder'
            reqBody = {
                'symbol': 'USD_JPY',
                'side': side,
                'size': order_size,
                'executionType': 'OCO',
                'limitPrice': str(limit),
                'stopPrice': str(stop)
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))
            print(json.dumps(res.json(), indent=2))

            return

        except Exception as e:
            print(f'action=market_order error={e}')
            print('10秒後にリトライします')
            time.sleep(10)

OCO注文を出す場合は、executionTypeをOCOに設定します。さらに、指値価格をlimitPrice、逆指値価格をstopPriceに設定します。

四本値を取得する関数

Python

# 四本値を取得する関数
def get_price_data():
    while True:
        try:
            price_data = []
            response_data = []

            if (datetime.today().hour == 0 or
                    datetime.today().hour == 1 or
                    datetime.today().hour == 2 or
                    datetime.today().hour == 3 or
                    datetime.today().hour == 4 or
                    datetime.today().hour == 5):
                today = datetime.today().date()

            else:
                today = datetime.today().date() + timedelta(days=1)

            for i in range(get_price_period):
                period = i - get_price_period
                date = (today + timedelta(days=period)).strftime('%Y%m%d')
                endPoint = 'https://forex-api.coin.z.com/public'
                path = '/v1/klines?symbol=USD_JPY&priceType=ASK&interval=' + interval + '&date=' + date

                response = requests.get(endPoint + path)
                response_data.append(response.json())

            for j in range(len(response_data)):
                for i in response_data[j]['data']:
                    price_data.append({'openTime': i['openTime'],
                                       'openTime_dt': datetime.fromtimestamp(int(i['openTime'][:-3]))
                                      .strftime('%Y/%m/%d %H:%M'),
                                       'open': i['open'],
                                       'high': i['high'],
                                       'low': i['low'],
                                       'close': i['close']})

            print('TIME： ' + str(price_data[-1]['openTime_dt']) + ' H： ' + str(price_data[-1]['high']) + ' L： ' +
                  str(price_data[-1]['low']) + ' C： ' + str(price_data[-1]['close']))

            df_price_data = pd.DataFrame(price_data)

            return df_price_data

        except Exception as e:
            print(f'action=get_price_data error={e}')
            print('10秒後にリトライします')
            time.sleep(10)

GMOコインのPublic APIでは1日単位でしか取得できなかったので、for文で必要な日数を繰り返し取得して、ひとつにまとめています。そしてデータを扱いやすいようにpandas の DataFrameに変換しています。

また、GMOコインの仕様で、0時から5時の間は最新の1時間足が取得できなかったので、こちらはif文で取得する条件を変えています。

エントリー注文を出す関数

Python

# 移動平均線のGC、DCを判定してエントリー注文を出す関数
def entry_signal(df):
    global flag

    df['MA-short'] = df['open'].rolling(MA_short_period).mean()
    df['MA-long'] = df['open'].rolling(MA_long_period).mean()

    if df.iat[-2, 6] < df.iat[-2, 7] and df.iat[-1, 6] > df.iat[-1, 7]:
        flag['order']['side'] = 'BUY'
        print('ゴールデンクロスが発生しました。')

    elif df.iat[-2, 6] > df.iat[-2, 7] and df.iat[-1, 6] < df.iat[-1, 7]:
        flag['order']['side'] = 'SELL'
        print('デッドクロスが発生しました。')

    else:
        flag['order']['side'] = None

    return flag

移動平均線は、Pandas の関数である rollingで簡単に計算することができます。

ゴールデンクロスやデッドクロスの判定方法は、短期と長期の価格の比較が入れ替わったタイミングがゴールデンクロス又はデッドクロスが発生したタイミングとなります。

利確と損切りの価格を計算する関数

Python

# 利確と損切りの価格を計算する関数
def calculate_settlement_price():
    global flag

    profitable_price = 0
    stop_loss_price = 0
    collateral = get_collateral()

    if flag['position']['side'] == 'BUY':
        profitable_price = round(flag['position']['price'] + (collateral * trade_risk * risk_reward_ratio / int(order_size)), 3)
        stop_loss_price = round(flag['position']['price'] - (collateral * trade_risk / int(order_size)), 3)

    elif flag['position']['side'] == 'SELL':
        profitable_price = round(flag['position']['price'] - (collateral * trade_risk * risk_reward_ratio / int(order_size)), 3)
        stop_loss_price = round(flag['position']['price'] + (collateral * trade_risk / int(order_size)), 3)

    return profitable_price, stop_loss_price

損切り価格は、一回のトレードの損失が資産の2%になるように計算しています。

利確価格は、リスクリワードレシオが2になるようになるように計算しています。

メインの処理

Python

# メイン処理
flag = {
    'order': {
        'side': None,
    },
    'position': {
        'exist': False,
        'side': None,
        'price': 0.0,
        'lot': 0,
    },
}

check_positions()

while True:

    # ポジションがある場合
    if flag['position']['exist'] is True:
        time.sleep(10)
        check_positions()
        continue

    # ポジションがない場合
    else:
        price = get_price_data()
        flag = entry_signal(price)

        # シグナルが発生した時
        if flag['order']['side'] is not None:
            market_order()
            time.sleep(5)
            check_positions()
            limit_price, stop_price = calculate_settlement_price()
            oco_order(limit_price, stop_price)
            time.sleep(3600)

        else:
            time.sleep(10)