前回の内容に引き続き、今回も仮想通貨botの開発状況をまとめていきます。
今回は、こちらの記事を参考にして「エントリーと手仕舞いを自動化するプログラム」を作ります。
この記事ではその前段階として「過去データを使ってロジックの検証をすること」がテーマです。
ロジックの定義
エントリーの条件は「陽線が３本連続した時」にしてあります。
そして、手仕舞いの条件は「現在の終値が、前回の終値を下回った時に、成り行きで売却する」と設定します。
つまり、「買いでエントリーしたらポジションが上がる限りは持ち続けて、下げに転じたらすぐに売る」という戦略を取ります。
そのために必要なことは以下の通り。
- 全体の流れを管理する
- 買い注文を出す
- 注文状況を管理する(確認と処理)
- 手仕舞いをする
- 処理をループさせる
- テスト検証
今回の学習では、部分ごとに学習を進めていきます。
全体の流れを管理する
flag = {
"buy_signal":0,
"order":False,
"position":False
}まずは、flagという変数を作って「買いシグナルの状況」「サーバーに出した注文の状況」「ポジション保有の有無」の情報を保有させます。
buy_singnalで設定しているのは、「基準は0だが、これが3になれば買い注文を出す」という指示です。ショート(売り注文)でもエントリーする場合は、別に設定する(sell_signalなど)必要があります。今回は省略します。
orderで設定しているのは「現在、サーバーに未約定の注文がなければ、処理を実行しないよ」という指示です。
もう少し正確に言うと、「(後ほど定義する)orderに当てはまる条件の時には、処理を実行するな」という意味です。
positionで設定しているのは「ポジションを保有しているなら、処理を実行しないよ」という指示です。
これも、詳しく言うと「(後ほど定義する)positionに当てはまる条件の時には、処理を実行するな」という意味です。
【参考記事】【Python入門】ブール型（Boolean）の用途と使い方を学ぼう！、Flagでの制御、ブール値(True、False)の使い方、組み込み型(PytholのDocs)
買い注文を出す
以下のコードは、買い注文を出すための関数です。
def buy_signal( data, last_data,flag ):
if flag["buy_signal"] == 0 and check_candle( data ):
flag["buy_signal"] = 1
elif falg["buy_signal"] == 1 and check_ascend( data,last_data ):
flag["buy_signal"] = 2
elif flag["buy_signal"] == 2 and check_ascend( data,last_data):
print("３本連続陽線 なので " + str(data["close_price"]) + "で 買い指し値")
＃注文コードを入れる
flag["buy_signal"] = 3
flag["order"] = True
else:
flag["buy_signal"] = 0
return flag上記のコードは、３本連続で陽線が続いたら買うという指示を出すためのものです。
まず気になったことは「=と==の違い」です。
調べてみると、=は代入演算子、==は比較演算子であることが分かりました。
- =の役割は、左辺の識別子(変数)に右辺のインスタンス(作ったもの)をバインドすること。
- ==の役割は、２つのオブジェクトの値が等しいかどうかを調べること。TrueかFalseかを確かめる文の中で出てくる。
注文コードは、#5の記事とこちらの記事で作成したものを使います。
詳しい内容は、次回以降の記事でまとめていきます。
【参考】Pythonの演算子まとめ、Python/変数と代入、インスタンス
注文状況の確認とその後の処理
以下のコードは、プログラムに則って出した指し値注文が通ったかどうかを確認するためのものです。
defから始めて、関数にしてあります。
def check_order( flag ):
#注文状況を確認して通っていたら以下を実行
#一定時間で注文が通っていなければキャンセルする
flag["order"] = False
flag["position"] = True
return flag指し値注文は出しただけでは通るかどうか分からないため、一定時間ごとに注文状況を確認する必要があります。
この関数では「注文が通っていた場合はflag["position"] をTrueに更新しなさい」という指示が書いてあります。
もう少し詳しく書くと「flag["order"] は処理を実行しない合図だぞ(Falseを代入しなさい)」「flag["position"]は処理を実行する合図だぞ(Trueを代入しなさい)」 ということを定めているということです。
今回のコードでは、#注文状況を確認して通っていたら以下を実行の部分のみを書いています。
これは、事前のテスト検証で買い注文は全て通ったものと見なして進めるためです。
本番のコードでは、#一定時間で注文が通っていなければキャンセルするも書き足します。
手仕舞いをする
以下のコードは、手仕舞い(決済)のためのもので、ポジションを保有している時に必要な指示の一部です。
これもdefで関数にしてあります。
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前回の終値を下回った場合は、「print(~~)の内容を書き出しなさい」と「 flag["position"] にFalseを代入(更新)しなさい」という指示を出しています。
これにより、終値が下がっていれば、決済の売り注文をサーバーに出して手仕舞いをして、買いシグナルを探して買い注文を出すプログラム(関数)が動くようになります。
この時点では「売り注文を出すためのプログラム」は、まだ書き込んでいません。
処理をループさせる
以下のコードは、全体の処理をループさせるためのものです。
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["opne_price"]
last_data["close_price"] = data["close_price"]
time.sleep(10)まずは最初の３行：「if flag["order"]〜〜」で「未約定の注文があるかどうか」を確認します。
具体的には、flag = check_order( flag )を実行することで、check_order()関数を呼び出して注文状況の管理につなげています。
そして、data = get_price()でdataを定義した上で、条件分岐を設定します。
「if data["close_time"] != last_data["close_time"]〜〜」では、「データを取得したタイミングで終値が異なる場合は、そのデータを書き出す」ように指示しています。
さらに、ポジションを持っている(flagがpositionである)の場合は、close_position()関数を呼び出して、手仕舞いのための条件を満たしているかどうかを確認するように指示しています。
一方、else〜〜の部分は、ポジションを保有していない時の指示です。
つまり、buy_signal()関数を呼び出して買いシグナルを探して、買い注文を出すプログラムを実行するように設定しています。
テスト検証
実行したコードは以下の通り。
ゼロ除算を避けるために、赤字の部分だけ修正しています。
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"] + 1)
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("３本連続で陽線 なので" + 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)実行の結果、プログラムが正常に動くことが確認できました。
0除算への対策は完璧なものではありませんが、戦略の実装や関数の組み合わせについては問題なさそうですね。
追記(2023/9/16)
こちらの記事のコメント欄で、ゼロ除算エラーへの対策が紹介されていました。
def check_candle( data )のところで最初に
if (data["high_price"]-data["low_price"]) == 0 : return Falseと書き込むことで「高値-安値」の結果が0の場合はFalseを返す様にする。
まとめ
今回は、テスト検証ということで、必要最低限のコードで組み立てています。
売買ロジックとコードとの関連性について理解が深まってきました。
同時に、わからないことや定義自体が曖昧なことも増えてきています。
今回のコードに関して言うと、「=と==はどう違うのか」「TrueとFalseはそれぞれ何を意味しているのか」「変数と代入」「プログラムを動かすための前提条件をどのように設定すると良いのか」「条件分岐の抜けもれを防ぐにはどうしたら良いのか」などを調べることになりました。
しかし、複雑そうに見えても、一つ一つ調べていけば答えが見えてきます。
また、自分自身が持っている言語体系で理解することも重要なのだと感じました。
本で調べたりネット上の解説を読んだりした上でコードを書き、そこで得た実感をもとに自分の言葉で再現する。
このプロセスを通して、本当の意味で理解することにつながります。
今後も時間をかけて少しずつ確実に理解を深めていきます。