FXボット とりあえず作ってみた編 その3 ー ポジションの保有を確認

前回 からの続きでプロトタイプボットの概要設計の中から、全体の構造と最新のレートの取得を組み込みましたので、次は現在保有中のポジションがあるかどうかを確認する処理を追加してみます。

---

ポジション保有中の確認処理の追加

---

前提

今回作成するボットの前提として、動きをシンプルにするためにポジション保有中の場合は取引を行わない仕様になっています。 従いまして、最新のレートを取得しマーケットがオープンしていて取引可能な状態の時、まずはポジションの有無を確認し、ポジションがないのみ次のステップへ進むという流れになります。

フローチャート

1. 初期処理

ポジションの保有の有無を確認するためには、onadapyV20のendpoints.pricingというエンドポイントを使用しますので、頭の部分にimport行を足します。

import oandapyV20.endpoints.positions as positions

3. ポジションの確認処理

onadapyV20のendpoints.positionsを呼び出し、得られた情報からすでにポジションをもっているか確認します。
まずはどんな情報が得られるか一度出力してみます。

{
  "position": {
    "instrument": "USD_JPY",
    "long": {
      "units": "0",    👈
      "pl": "-0.0396",      
      "resettablePL": "-0.0396",
      "financing": "-0.0002",
      "dividendAdjustment": "0.0000",
      "guaranteedExecutionFees": "0.0000",
      "unrealizedPL": "0.0000"
    },
    "short": {
      "units": "0",   👈
      "pl": "-0.0620",
      "resettablePL": "-0.0620",
      "financing": "-0.0019",
      "dividendAdjustment": "0.0000",
      "guaranteedExecutionFees": "0.0000",
      "unrealizedPL": "0.0000"
    },
    "pl": "-0.1016",
    "resettablePL": "-0.1016",
    "financing": "-0.0021",
    "commission": "0.0000",
    "dividendAdjustment": "0.0000",
    "guaranteedExecutionFees": "0.0000",
    "unrealizedPL": "0.0000"
  },
  "lastTransactionID": "1632"
}

取得したデータを解析すると"position"キーの"long"と"short"の"units"部にポジションサイズが格納されているようですので、ここが両方とも０の時はポジションがないと判断します。 また、ポジションの有無の結果を"status"として返します。

これをもとに"def Position()"の中身を作成してみます。

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}
  

これで関数の中身ができました。

メイン

「2. 最新のレートを取得」の時と同様にポジションのあるなしによって結果をステータスとして関数から戻し、その結果によって次の処理に進むかどうかを決定します。

        r_pos = Position()             
        #ポジション無しの時次の処理へ
        if r_pos['status'] == "GO":   
          Signal()
          Order()

---

サンプルプログラム

---

これらの処理を追加したものをコーディングしてみました。（👈👇👆部分）

#外部モジュール
from oandapyV20 import API
import oandapyV20.endpoints.pricing as pricing
import oandapyV20.endpoints.positions as positions #👈ポジション確認用エンドポイントの追加
  
import json
import time
import datetime
    
#口座情報（自分の情報を入力）
TOKEN = ""
ID = ""
    
#取引通貨  
INSTRUMENT = "USD_JPY"
#レート桁数
DECIMALS = 3 
#Pip桁数
PIP_LOCATION = -2 

#最大許容スプレッド  
MAX_SPREAD_PIPS = 2 #Pips
    
#ループ回数
LOOP = 100 #回
#待機時間
WAIT = 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():  #シグナル判定（赤３黒３）
  pass
  
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":   
          Signal()
          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))

動かしてみると以下のようにポジションの有無も追加で表示されるようになりました。

{'status': 'GO', 'bid': '111.965', 'ask': '111.979', 'spread': 0.014}
継続-トレード可能
待機中  0        
{'status': 'GO', 'bid': '111.965', 'ask': '111.979', 'spread': 0.014}
継続-トレード可能
待機中  1        
{'status': 'GO', 'bid': '111.965', 'ask': '111.979', 'spread': 0.014}
継続-トレード可能
待機中  2        

---

まとめ

---

最新のレートを取得しそこで取引可能であれば現在の保有ポジションを確認し、ポジションがなければまた次の処理へ進むという部分まで完成しました。
次回はシグナルが発生しているかどうか確認する処理を作成していきたいと思います。

---