前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。
前回までの学習で「bitFlyerで三兵のシグナルを検証して自動売買を行うbot」を完成させることができました。
今回は、サーバーエラー対策のコードを組み込んで、安定して稼働するbotに仕上げていきます。
参考にしたのはこちらの記事です。
例外処理
今回の目的は「bitFlyerでサーバーエラーが発生した時の対処法をコードに組み込む」ことです。
エラーが出てしまうと、botは止まってしまいます。
そこで、想定されるエラー(=例外)を処理することが必要になるのです。
例外処理には様々な方法がありますが、今回は「try,exept」を使います。
基本形は以下の通り。
try:
#リクエスト通信を書く
except 例外名 as e:
#エラーが返ってきたときの処理を書く
「e」はエラーを示すオブジェクト名です。「except 例外名 as 変数名」という形をとることでエラーメッセージの内容を確認することができます。この変数には「e」「err」が用いられることが多いです。
なお、spyderエディタではエラーが発生した時にコンソールにエラーの種類が表示されるので便利です。
【参考】例外処理とは、例外処理 【exception handling】、Pythonの例外処理(try, except, else, finally)、コンソール
例外処理が必要なタイミング
今回のコードでエラーが発生する可能性があるのは「APIを使ってリクエスト通信をするタイミング」です。
つまり「外部サーバーと通信する部分」ということになります。
具体的には、以下の通り。
- Cryptowatchから価格データを取得する
- bitflyerに買い・売り注文を出す
- bitflyerから未約定の注文一覧を取得する
- bitflyerから建玉の一覧を取得する
この4カ所に例外処理を行うコードを組み込むことで、安定して稼働し続けるbotを完成させることができます。
データを取得する時の例外処理
まずは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] }
ここに「try,except」を組み込んで例外処理ができるようにします。
具体的には「Cryptowatchから価格が取得できなかった場合は、10秒待機してから価格取得をやり直しなさい」という指示を組み込みます。
修正後のコードは以下の通り。赤文字の部分が変更点です。
def get_price(min,i):
while True:
try:
response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc", params = { "periods" : 60 })
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.RequiestException as e:
print("Cryptowatchの価格取得でエラー発生 : ",e)
print("10秒待機してやり直します")
time.sleep(10)
まず、以下のコードが何のためのものか調べました。
response.raise_for_status()
これはtry-except文において、エラーを吐き出すための準備です。
【参考】Pythonのrequestsライブラリで学ぶHTTPエラー処理の基本、Python の requests モジュールについて、Python API通信時の例外処理、ステータスコードの使い方(初心者向け)
次に気になったのは、以下の部分。
print("Cryptowatchの価格取得でエラー発生 : ",e)
なぜ「,e」を最後に付けているのかを調べてみました。
その理由は「エラーの理由も書き出しなさい」と指示するためです。どのようなエラーが発生したのかが分かると例外処理に繋げやすくなります。
【参考】【Invalid Syntax】Python のよくある基本的なエラーと確認方法まとめ、
注文を出す時の例外処理
注文を出す時にはCCXTライブラリを経由してbitFlyerのAPIと通信しているため、ここでも例外処理が必要です。
元のコードは以下の通り。
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)
これを「try-except文」で修正したものが以下のコードです。
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)
基本的な形は、先ほどのコードと同じです。
ここでは「break」を入れることで、「while文を抜けなさい」という指示を追加しています。
【参考】図解!Python whileループからbreakで抜ける方法を徹底解説!、図解!Python while文のループ処理を徹底解説!
これを踏襲して、外部サーバーと通信する前提で書いてあるコードには全て「try-except文」で修正を入れていきます。
完成版のコード
完成版のコードは以下の通り。
赤字部分で0除算エラーの対策をしています。
後からコードを書き込んだため、インデント調整をする必要がありました。
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 ):
if data["high_price"] == data["low_price"]:
return False
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("3本連続で陽線 なので" + 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("3本連続で陰線 なので" + 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)
APIキーとAPIシークレットを入力してコードを実行すると、無事機能することが確認できました。
まとめ
ここまで約2週間かかりました。
ロジックを実装する基本的な部分をなんとか理解することができました。
また、分からないことを調べる方法やどうやったら求める答えに辿り着くことができるのかも分かってきました。
具体的には「pythonのDocsを読む」「初級者用の技術書を読む」「ネット検索をする」「ChatGPTにエラー対策をさせる」「システムトレードの本を読む」などの方法が挙げられます。
いずれの場合も「課題に対して仮説を持って調べる」ことが重要なのだと感じます。
目的意識を持ち、問題となっていることを細かい要素に分解して、問題界越の方法を探る。これを続けていきます。
今回のコードを実行し続けるには、PCをつけっぱなしにする必要があったり、APIのコスト制限に引っかかってbotが止まったりする可能性があります。
つまり、書いたコードを安定して稼働させるための環境が必要です。
そこで次回は、botを24時間稼働させるためのレンタルサーバー利用について学んでいきます。