FXボット とりあえず作ってみた編 その4 ー 売買シグナルのチェック
今までの回で全体の構造と最新のレートの取得とポジションの確認の処理をするところまで完成しましたので、次は売買シグナルが発生したかどうかを確認する処理を加えてみます。
売買シグナルの選定
売買の注文をいれるべきかどうか判断するには、どんな合図をきっかけにするかという事から決めなければなりません。テクニカル分析とはいってもどんなチャートをどのように使って判断するかいろいろな方法があります。もちろん複数組み合わたり、環境認識で時間軸で判断したりと様々な方法が考えられます。
まずは、ボットの基本部分を作ってきちんと動く事が目的になりますので、なるべく簡単なコーディングで表現できるものを選びたいところです。
とはいえ、ランダム関数で売買するか見送るかではつまりませんので、今回はろうそく足をもとに酒田五法のうちの「三兵」(赤三兵・黒三兵)を使って売買シグナルの判定としてみたいと思います。
「三兵」(赤三兵・黒三兵)とは
まずは「三兵」の説明になりますが、単純に言えば、ろうそく足が3本続けて同じ色になれば、トレンド発生しているという事で陽線が続けば「赤三兵」、陰線が続けば「黒三兵」となります。
いちおうトレンド系になりますので、「赤三兵」になったら買いサイン。「黒三兵」になったら売りサイン。という事になります。
ただ「三兵」とはいっても「先詰まり」「思案星」等々の派生形があり、これをすべて組み込んでいくと複雑になってしまいますので、今回は赤三兵・黒三兵のみを売買シグナルの判定材料にします。
シグナルの判定処理の追加
それでは、これを実際にプログラムとして書き込んでいきます。今回の範囲は下のフローチャートの赤い部分になります。
処理の概要
陽線陰線の確認ですからろうそく足が基本となりますのでまずはろうそく足の取得を行います。 次に取得したろうそく足が三兵の状態になっているかどうか調べるために、取得したデータをPandasのデータフレームに変換して、3回続けて陽線または陰線がでているか調べます。
ろうそく足の取得
初期処理
ろうそく足を取得するためには、あらたなエンドポイントが必要になりますのでonadapyV20のendpoints.instrumentsをインポートします。
またろうそく足を取得するのに必要な引数も後で変更しやすいように変数として定義しておきます。
import oandapyV20.endpoints.instruments as instruments
#ろうそく足取得用 COUNT = 10 #ろうそく足の取得本数 GRANULARITY = "M5" #時間足(5分)
ろうそく足の取得本数は多めにしてあります。三兵ですが最後の足はタイミングによってまだ閉じていない場合がありますので最低4つ必要です。
エンドポイントの呼び出し
必要な引数を渡して呼ぶだけです。
#引数セット params = { "count": COUNT, "granularity": GRANULARITY } #エンドポイント呼び出し r = instruments.InstrumentsCandles(instrument=INSTRUMENT, params=params) rv = api.request(r)
ろうそく足データをデータフレームへ
初期処理
ろうそく足をJSON形式で取得した後に計算や比較をしやすいようにデータフレームに変換しますので、同様にpandasをインポートしておきます。
import pandas as pd
データフレームへの変換
次にJSON形式で読み込まれたデータをpandasのjson_normalize関数を使えば簡単にデータフレームに移すことができます。
ついでにデータフレームのコラム名もわかりやすい形に変更しておきます。
#データフレームへの変換 df = pd.json_normalize(rv, record_path='candles', meta=['instrument', 'granularity'], sep='_') #コラム名の変更 df.columns = ['complete', 'volume', 'time_UTC', 'open', 'high', 'low', 'close', 'pair', 'ashi']
最新のろうそく足3本分の取得
次に最新のろうそく足から3本のみを取り出しますが、最後のろうそく足はまだ未完成の可能性がありますので、完成している足のみ3本を移します。
最後から3つですから"tail()"を使えば簡単に取り出せます。
#完成形のろうそく足を最後から3本分のみ取得 df = df[df['complete'] == True].tail(3)
始値・終値の差額の計算と陽線・陰線の判定
サーバから取得したデータは文字型なので陽線か陰線がでているか計算するために必要なコラムを数値型に変換します。
始値より終値が大きい時は陽線(1)、始値より終値が大きい時は陰線(-1)とします。(同じ時は0をセット)
0.1Pipでも差額があれば陽線・陰線の判定になりますが、とりあえずはこのままでいきます。
#計算用に属性を変更 df = df.astype({'open': float, 'close': float, 'high': float, 'low': float}) #3本分のろうそく足毎のトレンドの判定 df.loc[round(df['close'] - df['open'],3) > 0, 'trend'] = 1 #陽線(赤) df.loc[round(df['close'] - df['open'],3) < 0, 'trend'] = -1 #陰線(黒) df.loc[round(df['close'] - df['open'],3) == 0, 'trend'] = 0 #同じ
赤三兵・黒三兵の判定
最後に赤三兵または黒三兵になっているかの判定ですが、3つとも列の数値の合計が3であれば3回連続陽線で赤三兵、-3ならば3回陰線連続で黒三兵と考えられますので、
合計値が、3の時は買いサイン(数値1)、-3の時は売りサイン(数値-1)、それ以外の値の時はシグナルなし(数値-1)として戻り値にセットします。
次回の注文処理で使う必要があるので判定処理で使用したデータフレームも戻り値にセットしておきます。
#売買シグナルの判定 if df.trend.sum() == 3: #すべて陽線(上昇) signal = 1 print("買シグナル!") elif df.trend.sum() == -3: #すべて陰線(下降) signal = -1 print("売シグナル!") else: signal = 0 print("シグナルなし") print(df) #戻り値のセット return {'signal': signal, 'df': df} #👆#
メイン
前回と同様にシグナルのあるなしによって結果をステータスとして関数から戻し、その結果によって次の処理に進むかどうかを決定します。
r_Signal = Signal() if r_Signal['signal'] != 0: Order()
プログラムサンプル
これらの処理を追加したものをコーディングしてみました。(👈👇👆部分)
#外部モジュール from oandapyV20 import API import oandapyV20.endpoints.pricing as pricing import oandapyV20.endpoints.positions as positions import oandapyV20.endpoints.instruments as instruments #👈ろうそく足取得用エンドポイントの追加 import json import time import datetime import pandas as pd #👈 #口座情報(自分の情報を入力) TOKEN = "" ID = "" #取引通貨 INSTRUMENT = "USD_JPY" #レート桁数 DECIMALS = 3 #Pip桁数 PIP_LOCATION = -2 #最大許容スプレッド MAX_SPREAD_PIPS = 2 #Pips #ループ回数 LOOP = 100 #回 #待機時間 WAIT = 10 #秒 #👉ろうそく足取得用 COUNT = 10 #ろうそく足の取得本数 GRANULARITY = "M5" #時間足(5分) def CurrentRate(): #最新レートの取得 params = { "instruments": INSTRUMENT } r = pricing.PricingInfo(accountID=ID, params=params) rv = api.request(r) #スプレッドの計算 bid = rv['prices'][0]['closeoutBid'] ask = rv['prices'][0]['closeoutAsk'] spread = round(float(ask) - float(bid), DECIMALS) #トレード可能? if rv['prices'][0]['tradeable'] == True: max_spread = MAX_SPREAD_PIPS * (10 ** PIP_LOCATION) if spread < max_spread: status = "GO" else: #スプレッド拡大中 status = "SKIP" #クローズ/メンテ中 else: status = "STOP" #戻り値 return {'status': status, 'bid': bid, 'ask': ask, 'spread': spread} def Position(): #ポジションの確認処理追加 r = positions.PositionDetails(accountID=ID, instrument=INSTRUMENT) rv = api.request(r) if rv['position']['long']['units'] != "0" or rv['position']['short']['units'] != "0": print("ポジあり。待機") status = "SKIP" else: print("ポジなし。 継続") status = "GO" return {'status': status} def Signal(): #シグナル判定(赤3黒3) #👇# #ろうそく足の取得 #引数セット params = { "count": COUNT, "granularity": GRANULARITY } #エンドポイント呼び出し r = instruments.InstrumentsCandles(instrument=INSTRUMENT, params=params) rv = api.request(r) #データフレームへの変換 df = pd.json_normalize(rv, record_path='candles', meta=['instrument', 'granularity'], sep='_') #コラム名の変更 df.columns = ['complete', 'volume', 'time_UTC', 'open', 'high', 'low', 'close', 'pair', 'ashi'] #完成形のろうそく足を最後から3本分のみ取得 df = df[df['complete'] == True].tail(3) #計算用に属性を変更 df = df.astype({'open': float, 'close': float, 'high': float, 'low': float}) #3本分のろうそく足毎のトレンドの判定 df.loc[round(df['close'] - df['open'],DECIMALS) > 0, 'trend'] = 1 #陽線(赤) df.loc[round(df['close'] - df['open'],DECIMALS) < 0, 'trend'] = -1 #陰線(黒) df.loc[round(df['close'] - df['open'],DECIMALS) == 0, 'trend'] = 0 #同じ #売買シグナルの判定 if df.trend.sum() == 3: #すべて陽線(上昇) signal = 1 print("買シグナル!") elif df.trend.sum() == -3: #すべて陰線(下降) signal = -1 print("売シグナル!") else: signal = 0 print("シグナルなし") print(df) #戻り値のセット return {'signal': signal, 'df': df} #👆# def Order(): pass if __name__ == "__main__": try: api = API(access_token= TOKEN) for i in range(LOOP): #最新レート確認 r_rate = CurrentRate() print(r_rate) #次の処理 if r_rate['status'] == "GO": print("継続-トレード可能") #保有ポジションの確認 r_pos = Position() #ポジション無しの時 if r_pos['status'] == "GO": r_Signal = Signal() if r_Signal['signal'] != 0: #👈売買シグナルがでたら注文処理へ Order() #スプレッドが広すぎる elif r_rate['status'] == "SKIP": print("スキップ-スプレッド拡大中") #マーケットがクローズ(またはメンテ中) elif r_rate['status'] == "STOP": print("停止ーマーケットクローズ") break #ボット終了 else: print("停止ー予期せぬエラー発生") break #ボット終了 #次のサイクル print("待機中 ", i) time.sleep(WAIT) except Exception as e: print(e) finally: print("Botが停止しました。 UTC:", datetime.datetime.now(datetime.timezone.utc))
結果を確認しやすいように、データフレームもprint関数で印刷しておきます。
{'status': 'GO', 'bid': '113.814', 'ask': '113.828', 'spread': 0.014} 継続-トレード可能 ポジなし。 継続 シグナルなし complete volume time_UTC open high low close pair ashi trend 6 True 181 2021-10-27T19:20:00.000000000Z 113.795 113.796 113.772 113.794 USD_JPY M5 -1.0 7 True 318 2021-10-27T19:25:00.000000000Z 113.793 113.824 113.793 113.822 USD_JPY M5 1.0 8 True 524 2021-10-27T19:30:00.000000000Z 113.821 113.835 113.810 113.822 USD_JPY M5 1.0 待機中 0 {'status': 'GO', 'bid': '113.813', 'ask': '113.828', 'spread': 0.015} 継続-トレード可能 ポジなし。 継続 シグナルなし complete volume time_UTC open high low close pair ashi trend 6 True 181 2021-10-27T19:20:00.000000000Z 113.795 113.796 113.772 113.794 USD_JPY M5 -1.0 7 True 318 2021-10-27T19:25:00.000000000Z 113.793 113.824 113.793 113.822 USD_JPY M5 1.0 8 True 524 2021-10-27T19:30:00.000000000Z 113.821 113.835 113.810 113.822 USD_JPY M5 1.0 待機中 1
まとめ
今回は、シンプルにするためにひとつの手法のみを使用していますが、本格的に運用するには複数のものを組み合わせる事になると思いますのでここの部分は本来であればもっと複雑になると思いますが、 とりあえず売買シグナルの判定までは完成しましたので、次はいよいよ発注処理の作成になります。