BTCFXのエントリーと手仕舞い(決済)をpythonで自動化しよう

さて、前回の記事で作った「赤三兵のローソク足パターンで買いシグナルを出すpythonコード」を改良して、エントリーと手仕舞いまでを自動化するコードを書いてみましょう。

手仕舞いの条件を決める

今回は以下のようなシンプルなルールにしましょう。
1.現在の終値が、前回の終値を下回ったら手仕舞う

手仕舞いのロジック

今回は、赤三兵のローソク足パターンを、上昇トレンドの序盤を捉える買いシグナルとして使います。そのため、買いエントリーした後、そのまま素直に上がり続ければポジションを持ち続け、1度でも終値が下げて引けたら、そこで成行で手仕舞うようにします。

手仕舞いのロジックとしては、「利確」と「損切」にわけて、それぞれの値幅で決める方法もあります。ですが、今回ははじめてのBOT作成の練習なので、損切りと利確とで条件を分けることはせず、共通の条件で一律に手仕舞うことにします。

Pythonコードの解説

今回の記事の構成としては、まずポイントとなるpythonコードを部分的に解説していき、最後に全体のコードを示す、という流れにします。

全体の流れを管理する方法

まず flag という変数を作り、そこに「買いシグナルの状況」「サーバーに出した注文の状況」「ポジション保有の有無」などの情報を保有させます。

flag = {
	"buy_signal":0,
	"order":False,
	"position":False
}

buy_signal は「何本連続で陽線が続いているか?」を示すフラッグで、これが「3」になれば買い注文を出すことにします。今回の記事では省略しますが、もしショートでもエントリーしたい場合は、sell_signal という要素を新しくフラッグに追加します。

order は「現在サーバーに未約定の注文があるかどうか?」を管理するためのフラッグです。position は、「現在ポジション(建玉)を保有しているかどうか?」を管理するためのフラッグです。

買い注文の関数

まずは「価格データからシグナルをチェックして、買いシグナルが点灯していれば買い注文を出す」という部分の自作の関数を作ります。以下のようになります。

def buy_signal( data,last_data,flag ):
	if flag["buy_signal"] == 0 and check_candle( data ):
		flag["buy_signal"] = 1
	elif flag["buy_signal"] == 1 and check_candle( data )  and check_ascend( data,last_data ):
		flag["buy_signal"] = 2
	elif flag["buy_signal"] == 2 and check_candle( data )  and check_ascend( data,last_data ):
		print("3本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
		# ここに買い注文のコードを入れる
		
		flag["buy_signal"] = 3
		flag["order"] = True
	else:
		flag["buy_signal"] = 0
	return flag

「赤三兵」というローソク足パターン(3本連続の陽線)を判定するために、buy_signal()という関数を使います。これは今まで解説していた「While文」の中身のロジックを外に出して、関数にまとめただけです。

・買いシグナルを出すpythonコード

check_candle() の条件を満たす陽線が1本現れたら、flag[“buy_signal”] に 1 を代入します。これが2本続けば、flag[“buy_signal”]=2、3本続けば、flag[“buy_signal”]=3 にして、Bitflyerのサーバーに指値で買い注文を出します。

今回は、先に過去データで検証をしたいので、まだ実際に注文を出すコードは書きません。が、ここに注文を出すコードを入れれば、注文を出せます。注文を出すためのコードは以下の記事で説明しています。

・BitflyerにCCXTで注文を出す方法

注文状況を管理する関数

def check_order( flag ):
	
	# 注文状況を確認して通っていたら以下を実行
	# 一定時間で注文が通っていなければキャンセルする

	flag["order"] = False
	flag["position"] = True
	return flag

今回の自動売買ロジックでは、買いシグナルが点灯したら、その時点での終値付近でBitflyerのサーバーに指値注文を出します。

指値注文は刺さるかどうかがわかりませんので、10秒おきに注文状況を確認し、このcheck_order() で注文が通ったかどうかを確認します。注文が通っていた場合は、flag[“position”] を True に更新します。これで初めて手仕舞いのための関数が動作する仕組みです。

また、もし一定時間で注文が通っていなければ、注文をキャンセルするようにします。この部分もあとで check_order() 関数に書き足しますが、テスト検証の時点では、買い指値はすべて「刺さったもの」として進めたいので、上記のように書いておきます。

手仕舞いのための関数

def close_position( data,last_data,flag ):
	if data["close_price"] < last_data["close_price"]:
		print("前回の終値を下回ったので" + str(data["close_price"]) + "で決済")
		flag["position"] = False
	return flag

ポジションを保有している場合には、以下の close_position() 関数が呼ばれるようにしておきます。

このclose_position() 関数の中では、以下のif文を使い、前回の終値と今回の終値を比較します。

if data["close_price"] < last_data["close_price"]:

もし終値が下がっていれば、決済のために売り注文をサーバーに出します。そして、flag["position"] の値を False に更新します。これによって、また先頭の「買いシグナルを探して買い注文を出す関数」が動くようになります。

全体のループ文

さて、このように「買い注文(エントリー)の関数」「注文の約定をチェックする関数」「決済をする関数」の3つが準備できたところで、全体の流れを管理するためのループ処理を作ります。

while True:		
	if flag["order"]:
		flag = check_order( flag )

	data = get_price(60,i)	
	if data["close_time"] != last_data["close_time"]:
		print_price( data )
		
		if flag["position"]:
			flag = close_position( data,last_data,flag )			
		else:
			flag = buy_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)

上記のWhile文の中では、まず最優先で「未約定の注文がサーバーにあるかどうか?」を、flag["order"] を使って確認します。もし未約定の注文があれば、先ほど作った check_order() 関数を使って、Bitflyerのサーバーに注文状況を問い合わせます。必要があれば、ここで注文のキャンセルもします。

以下の部分です。

if flag["order"]:
	flag = check_order( flag )

次に10秒おきにCryptowatchのAPIを使用して最新の価格データを取得します。

そして1分足の価格データ(例)に更新があったかどうかを確認し、もし更新があった場合は、日時や価格をprint()で表示します。また「現在ポジションを保有しているかどうか?」を flag["position"] で確認し、その結果に応じて条件分岐します。

if flag["position"]:
	flag = close_position( data,last_data,flag )			
else:
	flag = buy_signal( data,last_data,flag )	

ポジションを保有している場合

if flag["position"] でポジションを保有中かどうかを確認します。もし保有中であれば、先ほど作ったclose_position() 関数を呼んで、手仕舞いのための条件を満たしているかどうかを確認します。

手仕舞いの条件を満たしていれば、売りの決済注文を出します。満たしていなければ、また10秒待機して次のループ処理に戻ります。

ポジションを保有していない場合

if flag["position"] でポジションを保有中かどうかを確認して、もしポジションを保有していなければ、最初に作った buy_signal() 関数を実行し、買いシグナルが点灯するかどうかを確認します。買いシグナル(3本連続の陽線)が点灯すれば、サーバーに買い注文を出します。

以上が、全体のロジックです。

テスト検証のためのコード

以下は、過去データのテスト検証のための、pythonコードの全文です。

ここでいう過去データの検証とは、リアルタイムの足ではなく、直近の500件の確定足を使って「売買シグナルが点灯するかどうか?」などの挙動を確認するための検証のことです。勝率等の検証(いわゆるバックテスト)のことではありません。詳しくは以下を読んでください。

・過去の価格データを使った検証

※ サンプルコードでは1分足のデータを使っていますが、60の部分を300に変更すれば5分足、1800に変更すれば30分足、3600に変更すれば1時間足に修正できます。実際に売買BOTを動かすときは、最新価格の取得の遅延、約定までの時間などを考慮するので、(この売買ロジックの場合)1分足は適していないと思います。

import requests
from datetime import datetime
import time

response = requests.get("https://api.cryptowat.ch/markets/bitflyer/btcfxjpy/ohlc",params = { "periods" : 60 })

def get_price(min,i):
	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] }

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 ):
	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 data["close_price"] < data["open_price"] : return False
	elif increase_rate < 0.0005 : 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 buy_signal( data,last_data,flag ):
	if flag["buy_signal"] == 0 and check_candle( data ):
		flag["buy_signal"] = 1
	elif flag["buy_signal"] == 1 and check_candle( data )  and check_ascend( data,last_data ):
		flag["buy_signal"] = 2
	elif flag["buy_signal"] == 2 and check_candle( data )  and check_ascend( data,last_data ):
		print("3本連続で陽線 なので" + str(data["close_price"]) + "で買い指値")
		# ここに買い注文のコードを入れる
		
		flag["buy_signal"] = 3
		flag["order"] = True
		
	else:
		flag["buy_signal"] = 0
	return flag


# 手仕舞いのシグナルが出たら決済の成行注文を出す関数
def close_position( data,last_data,flag ):

	if data["close_price"] < last_data["close_price"]:
		print("前回の終値を下回ったので" + str(data["close_price"]) + "で決済")
		flag["position"] = False
	return flag


# サーバーに出した注文が約定したか確認する関数
def check_order( flag ):
	
	# 注文状況を確認して通っていたら以下を実行
	# 一定時間で注文が通っていなければキャンセルする
	
	flag["order"] = False
	flag["position"] = True
	return flag


# ここからメイン処理
last_data = get_price(60,0)
print_price( last_data )

flag = {
	"buy_signal":0,
	"sell_signal":0,
	"order":False,
	"position":False
}
i = 1

while i<500:		
	if flag["order"]:
		flag = check_order( flag )

	data = get_price(60,i)
	if data["close_time"] != last_data["close_time"]:
		print_price( data )
		
		if flag["position"]:
			flag = close_position( data,last_data,flag )			
		else:
			flag = buy_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"]
		i += 1	

	time.sleep(0)

実行結果

このpythonコードを実行すると以下のようになります。

今回の例では、直近の1分足のデータを500件分チェックしたところ、以下の2カ所でシグナルが点灯していました。

1箇所目

・2018/3/22 1時45分
・買い 992000円
・決済 993631円

2箇所目

・2018/3/22 3時08分
・買い 980925円
・決済 982266円

実際のチャートを見てみると、以下の部分でシグナルが点灯していたことがわかります。

1つ目の矢印が「赤三兵」での買いシグナルが出たローソク足、2つ目の矢印が「前回の終値を下回った」ことで決済シグナルの出たローソク足です。なおこのテストでは、前述のように「買い指値はすぐに刺さったもの」として検証しています。

注意

繰り返しますが、今回の自動売買BOTは「はじめて作る方」のための練習用であり、統計的に勝率を検証したものではありません。今回の例ではたまたま利益が出ていますが、もっと広範にテストすれば、このままのロジックでは全く勝てないことがわかると思います。

勝率を検証する方法などは今後、解説していく予定ですが、今回は単に「自分の設計したとおりの場面でエントリー・決済をしているか?」を確認するためにテストをしている、と考えてください。

次回記事:実際に自動売買ができる試作版のBOTを作る!

「BTCFXのエントリーと手仕舞い(決済)をpythonで自動化しよう」への5件のフィードバック

  1. 先日Pythonの基礎を終えてからこちらのチュートリアルを参考にしてBOTを制作に取り掛かっています。

    わかりやすくて助かります。

    1. コメントありがとうございます。
      そういっていただけると嬉しいです! 今後もよろしくお願いします。

  2. 珍しいパターンですが、データ取得時の「高値-安値」の結果が0(値段差異がない)になると
    値を0で割ろうとして「ZeroDivisionError: division by zero」が発生するため、その場合、このコードではエラーになります。稀な例ですが。

    応急回避策としては
    def check_candle( data )
    のところで最初に
    if (data[“high_price”]-data[“low_price”]) == 0 : return False
    とかで「高値-安値」の結果が0の場合はFalseを返す様にしてあげればとりあえずは機能するかと。

    ※本来はもっと違う処理を考えないといけないと思いますが、とりあえずスクリプトを動かすためだけの非実用的な対処になります。

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です