前回の記事の続きです。
前回までで、Bitflyerで自動的に売買シグナルを判定してエントリーし、さらに決済（利確・損切）まで自動で行うBOTを作ることができました。しかしこの自動売買BOTを実践で長時間、安定的に動かし続けるためには、「例外処理」が必要になります。
例外処理とは
簡単にいうと「サーバーエラー」の対策のことです。
自動売買BOTでは、pythonのプログラミングコード（構文や文法）自体に問題がなくても、外部のサーバーが原因で止まってしまうことがありえます。例えば、Bitflyerに買い注文を出したものの、Bitflyerのサーバーが混雑していてエラーになってしまったり、一時的にダウンしていて応答がなくなるような場合です。
このような場合に何も対策をしていないと、自動売買BOT自体もエラーを吐いて止まってしまいます。例えば、以下のような感じです。
↑ pythonで買い注文を出したところ、Bitflyerのサーバーから「500 Internal Server Error」（サーバーエラー）が返ってきたため、そこでプログラムの実行が停止されてしまった画面です。
キチンと例外処理をしていれば、外部サーバーから思わぬエラーが返ってきても、そこで自動売買BOTが停止することはなくなります。自動売買BOTがポジションを持ったままエラーを吐いて止まってしまうと、知らない間に損失が膨らみ続ける危険もあります。
特にBitflyerはサーバーから頻繁に500エラーが返ってくるため、きちんと例外処理を書くことが大切です。
例外処理の基本的な書き方
例外処理の基本的な構文は以下です。
try: #リクエスト通信 except 例外名 as e: #エラーが返ってきたときの処理
エラー処理が発生する可能性があるのは、リクエスト通信が必要な場面です。具体的にはAPIを使用する場面ですね。前回の記事で作成した自動売買BOTの例でいえば、以下のような場面です。
・Cryptowatchから価格データを取得する
・Bitflyerに買い・売り注文を出す
・Bitflyerから未約定の注文一覧を取得する
・Bitflyerから建玉の一覧を取得する
上記の場面では、外部サーバーとの通信が発生するため、すべて例外処理を書かなければなりません。逆にいえば、通信が発生する箇所にすべて例外処理を書けば、通信エラーで自動売買BOTが停止することはなくなります。
具体的なコード例
例えば、前回のCryptowatchから価格データを取得する関数に例外処理を入れてみましょう。前回までのコードは以下でした。
修正前
# Cryptowatchから価格を取得する関数
def get_price(min,i):
response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc", params = { "periods" : 60 })
data = response.json()
return { "close_time" : data["result"][str(min)][i][0],
"open_price" : data["result"][str(min)][i][1],
"high_price" : data["result"][str(min)][i][2],
"low_price" : data["result"][str(min)][i][3],
"close_price": data["result"][str(min)][i][4] }
ここに例外処理を入れてみましょう。
例外処理
# Cryptowatchから価格を取得する関数
def get_price(min,i):
while True:
try:
response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc", params = { "periods" : 60 }, timeout = 5)
response.raise_for_status()
data = response.json()
return { "close_time" : data["result"][str(min)][i][0],
"open_price" : data["result"][str(min)][i][1],
"high_price" : data["result"][str(min)][i][2],
"low_price" : data["result"][str(min)][i][3],
"close_price": data["result"][str(min)][i][4] }
except requests.exceptions.RequestException as e:
print("Cryptowatchの価格取得でエラー発生 : ",e)
print("10秒待機してやり直します")
time.sleep(10)
上記のコードでは、正常にデータを取得できれば価格データを返し、もし通信エラーなどの異常が発生した場合は、10秒間待機した後で、データの取得をやり直すようにプログラムしています。そのため、全体をwhile文で囲んでいます。
「try:」の部分で、requests()というリクエスト通信用の関数を使い、「except requests.exceptions.RequestException as e:」の部分で、通信エラーがあった場合の処理を記述しています。エラーがあった場合は、print(“Cryptowatchの価格取得でエラー発生”, e)でエラーの内容も表示するようにしています。
実行結果の例
これを実行すると、Cryptowatchのサーバーから通信エラーが返ってきても、BOTは停止することなく自動的に10秒待機して通信をやり直してくれます。
Bitflyerへの注文の例外処理
同じように、Bitflyerのサーバーと通信する箇所にも例外処理を入れてみましょう。この記事の自動売買BOTは、すべてCCXTライブラリ経由でBitflyerと通信しています。そのため、例外処理はCCXTライブラリで用意されている「ccxt.BaseError」を使います。
ここでは、手仕舞いの成行注文を出す箇所のコードを使って説明します。
修正前
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='market',
side='sell',
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["position"]["exist"] = False
time.sleep(30)
ここに同じように例外処理を入れると、以下のようになります。
例外処理
while True:
try:
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='market',
side='sell',
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["position"]["exist"] = False
time.sleep(30)
break
except ccxt.BaseError as e:
print("BitflyerのAPIでエラー発生",e)
print("注文の通信が失敗しました。30秒後に再トライします")
time.sleep(30)
通信に失敗した場合に、何度も自動でやり直しをさせたい場合は、このように「While文」で囲みます。While文で書くときは、「break」を書くのを忘れないようにしてください。breakは、成功したらwhile文を抜ける、という意味です。
やり直しをさせる必要がない場合は、While文で囲む必要はありません。
実行結果の例
同じように、すべての通信箇所に例外処理を書いておくとBitflyerの注文が通りにくい時間帯でも、BOTがエラーで停止することはなくなります。
「例外名」を自分で調べる方法
さきほどの構文の「except 例外名 as e:」の部分を、なぜ「except requests.exceptions.RequestException」と書くのか？疑問に思った方もいるかもしれません。これは、requests()ライブラリの仕様でそう決まっているからです。
つまりrequests()ライブラリを開発してくれた人たちが、そういう名前のエラー（例外）をあらかじめ用意してくれている、ということです。
例外の名前の部分は使用するライブラリによって違います。例えば、CCXTライブラリで例外処理を書くときは、「except ccxt.BaseError as e:」と書きます。大抵の場合は、「ライブラリ名 error handling」「ライブラリ名 例外処理」などで検索すると、何らかのヒントが見つかります。
例）
requests 例外処理
CCXT error handling
例外処理を入れた完成版のpythonコード
さて、それでは前回の記事のコードに例外処理を入れたバージョンを作ってみましょう！ 以下がpythonコードの全文です。
import requests
from datetime import datetime
import time
import ccxt
bitflyer = ccxt.bitflyer()
bitflyer.apiKey = '**********'
bitflyer.secret = '**********'
# Cryptowatchから価格を取得する関数
def get_price(min,i):
while True:
try:
response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc", params = { "periods" : 60 }, timeout = 5)
response.raise_for_status()
data = response.json()
return { "close_time" : data["result"][str(min)][i][0],
"open_price" : data["result"][str(min)][i][1],
"high_price" : data["result"][str(min)][i][2],
"low_price" : data["result"][str(min)][i][3],
"close_price": data["result"][str(min)][i][4] }
except requests.exceptions.RequestException as e:
print("Cryptowatchの価格取得でエラー発生 : ",e)
print("10秒待機してやり直します")
time.sleep(10)
# 時間と始値・終値を表示する関数
def print_price( data ):
print( "時間： " + datetime.fromtimestamp(data["close_time"]).strftime('%Y/%m/%d %H:%M') + " 始値： " + str(data["open_price"]) + " 終値： " + str(data["close_price"]) )
# 各ローソク足が陽線・陰線の基準を満たしているか確認する関数
def check_candle( data,side ):
realbody_rate = abs(data["close_price"] - data["open_price"]) / (data["high_price"]-data["low_price"])
increase_rate = data["close_price"] / data["open_price"] - 1
if side == "buy":
if data["close_price"] < data["open_price"] : return False
elif increase_rate < 0.0003 : return False
elif realbody_rate < 0.5 : return False
else : return True
if side == "sell":
if data["close_price"] > data["open_price"] : return False
elif increase_rate > -0.0003 : return False
elif realbody_rate < 0.5 : return False
else : return True
# ローソク足が連続で上昇しているか確認する関数
def check_ascend( data,last_data ):
if data["open_price"] > last_data["open_price"] and data["close_price"] > last_data["close_price"]:
return True
else:
return False
# ローソク足が連続で下落しているか確認する関数
def check_descend( data,last_data ):
if data["open_price"] < last_data["open_price"] and data["close_price"] < last_data["close_price"]:
return True
else:
return False
# 買いシグナルが出たら指値で買い注文を出す関数
def buy_signal( data,last_data,flag ):
if flag["buy_signal"] == 0 and check_candle( data,"buy" ):
flag["buy_signal"] = 1
elif flag["buy_signal"] == 1 and check_candle( data,"buy" ) and check_ascend( data,last_data ):
flag["buy_signal"] = 2
elif flag["buy_signal"] == 2 and check_candle( data,"buy" ) and check_ascend( data,last_data ):
print("３本連続で陽線 なので" + str(data["close_price"]) + "で買い指値を入れます")
flag["buy_signal"] = 3
try:
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='limit',
side='buy',
price= data["close_price"],
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["order"]["exist"] = True
flag["order"]["side"] = "BUY"
time.sleep(30)
except ccxt.BaseError as e:
print("Bitflyerの注文APIでエラー発生",e)
print("注文が失敗しました")
else:
flag["buy_signal"] = 0
return flag
# 売りシグナルが出たら指値で売り注文を出す関数
def sell_signal( data,last_data,flag ):
if flag["sell_signal"] == 0 and check_candle( data,"sell" ):
flag["sell_signal"] = 1
elif flag["sell_signal"] == 1 and check_candle( data,"sell" ) and check_descend( data,last_data ):
flag["sell_signal"] = 2
elif flag["sell_signal"] == 2 and check_candle( data,"sell" ) and check_descend( data,last_data ):
print("３本連続で陰線 なので" + str(data["close_price"]) + "で売り指値を入れます")
flag["sell_signal"] = 3
try:
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='limit',
side='sell',
price= data["close_price"],
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["order"]["exist"] = True
flag["order"]["side"] = "SELL"
time.sleep(30)
except ccxt.BaseError as e:
print("BitflyerのAPIでエラー発生",e)
print("注文が失敗しました")
else:
flag["sell_signal"] = 0
return flag
# 手仕舞いのシグナルが出たら決済の成行注文を出す関数
def close_position( data,last_data,flag ):
if flag["position"]["side"] == "BUY":
if data["close_price"] < last_data["close_price"]:
print("前回の終値を下回ったので" + str(data["close_price"]) + "あたりで成行で決済します")
while True:
try:
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='market',
side='sell',
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["position"]["exist"] = False
time.sleep(30)
break
except ccxt.BaseError as e:
print("BitflyerのAPIでエラー発生",e)
print("注文の通信が失敗しました。30秒後に再トライします")
time.sleep(30)
if flag["position"]["side"] == "SELL":
if data["close_price"] > last_data["close_price"]:
print("前回の終値を上回ったので" + str(data["close_price"]) + "あたりで成行で決済します")
while True:
try:
order = bitflyer.create_order(
symbol = 'BTC/JPY',
type='market',
side='buy',
amount='0.01',
params = { "product_code" : "FX_BTC_JPY" })
flag["position"]["exist"] = False
time.sleep(30)
break
except ccxt.BaseError as e:
print("BitflyerのAPIでエラー発生",e)
print("注文の通信が失敗しました。30秒後に再トライします")
time.sleep(30)
return flag
# サーバーに出した注文が約定したかどうかチェックする関数
def check_order( flag ):
try:
position = bitflyer.private_get_getpositions( params = { "product_code" : "FX_BTC_JPY" })
orders = bitflyer.fetch_open_orders(
symbol = "BTC/JPY",
params = { "product_code" : "FX_BTC_JPY" })
except ccxt.BaseError as e:
print("BitflyerのAPIで問題発生 : ",e)
else:
if position:
print("注文が約定しました！")
flag["order"]["exist"] = False
flag["order"]["count"] = 0
flag["position"]["exist"] = True
flag["position"]["side"] = flag["order"]["side"]
else:
if orders:
print("まだ未約定の注文があります")
for o in orders:
print( o["id"] )
flag["order"]["count"] += 1
if flag["order"]["count"] > 6:
flag = cancel_order( orders,flag )
else:
print("注文が遅延しているようです")
return flag
# 注文をキャンセルする関数
def cancel_order( orders,flag ):
try:
for o in orders:
bitflyer.cancel_order(
symbol = "BTC/JPY",
id = o["id"],
params = { "product_code" : "FX_BTC_JPY" })
print("約定していない注文をキャンセルしました")
flag["order"]["count"] = 0
flag["order"]["exist"] = False
time.sleep(20)
position = bitflyer.private_get_getpositions( params = { "product_code" : "FX_BTC_JPY" })
if not position:
print("現在、未決済の建玉はありません")
else:
print("現在、まだ未決済の建玉があります")
flag["position"]["exist"] = True
flag["position"]["side"] = position[0]["side"]
except ccxt.BaseError as e:
print("BitflyerのAPIで問題発生 ： ", e)
finally:
return flag
# ここからメイン
last_data = get_price(60,-2)
print_price( last_data )
time.sleep(10)
flag = {
"buy_signal":0,
"sell_signal":0,
"order":{
"exist" : False,
"side" : "",
"count" : 0
},
"position":{
"exist" : False,
"side" : ""
}
}
while True:
if flag["order"]["exist"]:
flag = check_order( flag )
data = get_price(60,-2)
if data["close_time"] != last_data["close_time"]:
print_price( data )
if flag["position"]["exist"]:
flag = close_position( data,last_data,flag )
else:
flag = buy_signal( data,last_data,flag )
flag = sell_signal( data,last_data,flag )
last_data["close_time"] = data["close_time"]
last_data["open_price"] = data["open_price"]
last_data["close_price"] = data["close_price"]
time.sleep(10)
Bitflyerのサーバーと通信する部分（買い注文・売り注文・手仕舞いの成行注文・キャンセル・注文一覧の取得・建玉一覧の取得）には、すべて例外処理を書いています。面倒ですが、これがBOTを安定的に動かす上では重要になります。
完成！
これで「最初の自動売買BOTを作ってみる！」の章は完成です。
この章を読んだ方は、以下のことができるようになったはずです。
１．数字で定義できる売買ロジックを考える
２．ローソク足の実体やヒゲの長さをシグナルにする
３．ローソク足のパターン（三兵）をシグナルにする
４．過去の価格データを使ってシグナルをテストする
５．売買シグナルで自動的にエントリーし、手仕舞いするBOTを作る
６．サーバーエラーで停止しないように例外処理を書く
ここまで勉強したことを応用してカスタマイズすれば、「３本以上連続で陰線だが、徐々に実体が短くなって出来高も減っている」という局面で、「実体が短くて長い下ヒゲの付いた陽線（ハンマー）が現れたら買いでエントリーする」というような、より実践的な売買ロジックも作れるでしょう。
※ ちなみに、これは「出来高・価格分析の完全ガイド」という本で紹介されている売買ロジックです。興味がある方は練習で実装してみてください。
次回以降の記事
次以降の章では、もう少し高度な「移動平均線」「MACD」などのトレンドフォロー指標、「RSI」「ストキャスティクス」などのオシレーターをpythonで実装する方法を解説します。
自動売買BOTなどのシステムトレードは、もともとチャートパターン（例えば、Wボトムやエリオット波動などの視覚的なもの）を認識するよりも、MACDヒストグラムやRSIなどの数値指標で売買シグナルを判断する方が得意です。そのため、自動売買BOTの精度を上げるためには、テクニカル指標を理解して実装できるようになることが必須です。
次の記事では、自動売買BOTと相性のいいテクニカル指標とその実装方法を解説していきます。
