【bitFlyer Gmail Python】 Tradingviewアラート注文BOT
こんにちは！Azulと申します。
犬は可愛い。
bitFlyerで使えるTradingviewのアラートを利用した、BOTのソースコードです。（Python）
なお、本noteにより発生した損益については、
いかなる場合でも当方では責任を負いません。
ご了承ください。
・Gmail API を使用
・メール既読にする事で受信済としています。
・Tradingviewからくるアラートのメールタイトルを条件に注文します。
・最初に実行する前にTradingview関連のメールは手動で既読にしといた方がいいかもしれない（誤注文を防ぐため）
Gmail API を使うには
Gmail APIを使うには、事前に有効化の作業が必要です。
Python のモジュールも必要なのでここらへん詳しく書いてくれてるページを参考にしてください。
https://qiita.com/orikei/items/73dc1ccc95d1872ab1cf
以下ソースコードです。
api_key と api_secret と line_notify_token はご自身のものに変更してください。
gmail_tradingview.py
from googleapiclient.discovery import build
from httplib2 import Http
from oauth2client import file, client, tools
from datetime import datetime, timedelta, timezone
from time import sleep
import hashlib
import hmac
import time
import requests
import json
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# urllb3
urllib3.disable_warnings(InsecureRequestWarning)
class gmail_api:
def __init__(self):
# self._SCOPES = 'https://www.googleapis.com/auth/gmail.readonly'
# 全ての権限を与える
self._SCOPES = 'https://mail.google.com/'
self._SUBJECT = 'Subject:TradingView '
def connect_gmail(self):
store = file.Storage('token.json')
creds = store.get()
if not creds or creds.invalid:
flow = client.flow_from_clientsecrets('credentials.json', self._SCOPES)
creds = tools.run_flow(flow, store)
service = build('gmail', 'v1', http=creds.authorize(Http()))
return service
def get_message_list(self,message_from):
# APIに接続
service = self.connect_gmail()
message_list = []
query = ' '
# from
query += 'From:' + message_from + ' '
# subject
query += self._SUBJECT
# 未読のメールのみを対象にする
query += 'is:unread'
# メールIDの一覧を取得する(10件)
message_id_list = service.users().messages().list(userId='me',maxResults=10,q=query).execute()
# 該当するメールが存在しない場合は、処理中断
if message_id_list['resultSizeEstimate'] == 0:
print("Message is not found")
return message_list
# メッセージIDを元に、メールの詳細情報を取得
for message in message_id_list['messages']:
row = {}
row['ID'] = message['id']
message_detail = service.users().messages().get(userId='me',id=message['id']).execute()
# メールを既読に変える
query = {"removeLabelIds": ["UNREAD"]}
service.users().messages().modify(userId="me", id=message['id'], body=query).execute()
for header in message_detail['payload']['headers']:
# 件名を取得する
if header['name'] == 'Subject':
row['Subject'] = header['value']
message_list.append(row)
return message_list
class bitflyer_trade:
def __init__(self):
self._api_key = "***************"
self._api_secret = "***************"
# connection pool
self._http = urllib3.PoolManager()
# 成行
def market_order(self, side, size):
retry = 0
while True:
if retry >= 30:
print('[{}]の成行注文に{}回失敗しました。エラーとして処理を終了します。'.format(side, int(retry)))
return None
try:
data = {
"product_code":"FX_BTC_JPY",
"child_order_type":"MARKET",
"side" : side, "size": float(size),
"minute_to_expire":1,
}
# POST
response,res_status = self.private_api_post('/v1/me/sendchildorder',data)
print(response)
if res_status == 200:
print('[{0}]の成行注文に成功しました。'.format(side))
break
else:
print(response["error_message"])
retry += 1
print('[{}]の成行注文に{}回失敗しました。5秒後に再度実行します。'.format(side, int(retry)))
sleep(5)
except Exception as e:
retry += 1
print(e)
print('[{}]の成行注文に{}回失敗しました。5秒後に再度実行します。'.format(side, int(retry)))
sleep(5)
return True
# 口座残高を取得
def getcollateral(self):
collateral = 0
open_position_pnl = 0
try:
response,res_status = self.private_api_get('/v1/me/getcollateral')
if res_status == 200:
collateral = response["collateral"]
open_position_pnl = response["open_position_pnl"]
else:
print('証拠金の取得に失敗しました。証拠金チェックを終了します。（変数エラー）')
except Exception as e:
print(e)
print('証拠金の取得に失敗しました。証拠金チェックを終了します。（APIエラー）')
return collateral, open_position_pnl
# プライベートAPI POST
def private_api_get(self, endpoint):
method = 'GET'
access_time = str(time.time())
access_text = access_time + method + endpoint
access_sign = hmac.new(bytes(self._api_secret.encode('ascii')), bytes(access_text.encode('ascii')), hashlib.sha256).hexdigest()
auth_header = {
'ACCESS-KEY': self._api_key,
'ACCESS-TIMESTAMP': access_time,
'ACCESS-SIGN': access_sign,
'Content-Type': 'application/json',
'Accept-Encoding': 'gzip, deflate'
}
url = 'https://api.bitflyer.com' + endpoint
response = self._http.request('GET', url, headers=auth_header, timeout=5.0)
data = ''
status = response.status
if(len(response.data) > 0):
data = json.loads(response.data.decode("utf-8"))
return data, status
# プライベートAPI POST
def private_api_post(self, endpoint, body):
body = json.dumps(body)
method = 'POST'
access_time = str(time.time())
access_text = access_time + method + endpoint + body
access_sign = hmac.new(bytes(self._api_secret.encode('ascii')), bytes(access_text.encode('ascii')), hashlib.sha256).hexdigest()
auth_header = {
'ACCESS-KEY': self._api_key,
'ACCESS-TIMESTAMP': access_time,
'ACCESS-SIGN': access_sign,
'Content-Type': 'application/json',
'Accept-Encoding': 'gzip, deflate'
}
url = 'https://api.bitflyer.com' + endpoint
response = self._http.request('POST', url, body=body, headers=auth_header, timeout=5.0)
data = ''
status = response.status
if(len(response.data) > 0):
data = json.loads(response.data.decode("utf-8"))
return data, status
class line_notify:
def __init__(self):
self._line_notify_token = "***************"
self._line_notify_api = 'https://notify-api.line.me/api/notify'
# connection pool
self._http = urllib3.PoolManager()
def post(self, message):
payload = {'message': message}
headers = {'Authorization': 'Bearer ' + self._line_notify_token}
try:
requests.post(self._line_notify_api, data=payload, headers=headers)
except:
pass
if __name__ == '__main__':
gmail = gmail_api()
bitflyer_trade = bitflyer_trade()
line_notify = line_notify()
lot_size = 0.01
position = 0
trandline_position = 0
collateral = 0
open_position_pnl = 0
collateral, open_position_pnl = bitflyer_trade.getcollateral();
while True:
# from が tradingview のメールを取得する
messages = gmail.get_message_list(message_from='noreply@tradingview.com')
# 結果を出力
for message in messages:
# 件名からシグナルを抜き出す
signal = ''
if 'TradingViewアラート' in message['Subject']:
signal = message['Subject'].replace('TradingViewアラート: ', '')
elif 'TradingView Alert' in message['Subject']:
signal = message['Subject'].replace('TradingView Alert: ', '')
message = 'SIGNAL: {}'.format(signal)
print(message)
# 売シグナルなら
if signal == 'SHORT':
order_lot = lot_size
# 既に買があればドテンする
if position != 0 and position > 0:
order_lot = round((lot_size * abs(position)) + lot_size, 8)
result = bitflyer_trade.market_order('SELL', order_lot)
if result is not None:
message = 'SHORT ENTRY: {}'.format(order_lot)
print(message)
line_notify.post(message)
position -= ( order_lot / lot_size )
# 買シグナルなら
elif signal == 'LONG':
order_lot = lot_size
# 既に売があればドテンする
if position != 0 and position < 0:
order_lot = round((lot_size * abs(position)) + lot_size, 8)
result = bitflyer_trade.market_order('BUY', order_lot)
if result is not None:
message = 'LONG ENTRY: {}'.format(order_lot)
print(message)
line_notify.post(message)
position += ( order_lot / lot_size )
print('POS: {} SIZE {} COL {} PNL {}'.format(position, ( lot_size * abs(position) ), int(collateral), int(open_position_pnl) ))
# 5秒sleep
sleep(5)