OANDA シストレ(oandapyV20) の始め方 【関数編３】

OANDAで、シストレ/APIを利用する際にお世話になる関数の紹介です。

OANDA シストレ(oandapyV20) の始め方 記事一覧

• 導入編
• 関数編１
• 関数編２
• 関数編３ ←本記事
• Pandas編（作成中）
• 実用編（作成中）

もくじ

• 関数の定時実行
• 文字列のLINE通知
• DataFrameを画像化しLINEに送る
• 処理時間の計測

関数の定時実行

登録した関数を、定時に実行します。python3 -m pip install schedule --userが事前に必要。下記の例では、毎時00分にjobという関数を実行。

import schedule
import time

def job():
    print("I'm working...")

for i in range(24):
    schedule.every().day.at("{:02}:00".format(i)).do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

文字列のLINE通知

文字列をLINEで通知をしてくれる。業務中にも、こっそりと取引の様子を見たい人向け。下記の下準備が必要です。

1. こちらからログインする
2. 右上の自分のユーザー名をクリックし、マイページへ
3. 「トークンを発行する」をクリック
4. トークン名を記入 (これは毎回、通知メッセージの最初に表示される)
5. 「1:1でLINE Notifyから通知を受け取る」を選択し、発行するをクリック
6. 表示されるトークンをメモする

import requests

def line_notify (message):
    line_notify_token = 'please_get_and_put_your_token'
    line_notify_api = 'https://notify-api.line.me/api/notify'
    payload = {'message': message}
    headers = {'Authorization': 'Bearer ' + line_notify_token}
    requests.post(line_notify_api, data=payload, headers=headers)
    print(message)

line_notify('テストメッセージ')

テストメッセージ
# (LINEには「[トークン名] テストメッセージ」の形式でメッセージが来る)

DataFrameを画像化しLINEに送る

DataFrameをLINEで見ようと思うと、LINE Notifyが添付は画像しかできない関係上、画像化してから送りつける必要があります。

事前にpython3 -m pip install matplotlibをしてあげた上で、下記のコマンドで対応できます。サイズ制限があるので、あまり大きい画像は送れません。

import matplotlib.pyplot as plt

def line_notify_image (df, image):
    fig, ax = plt.subplots(figsize=(3,3))
    ax.axis('off')
    ax.axis('tight')
    ax.table(cellText=df.values, colLabels=df.columns, loc='center', bbox=[0,0,1,1])
    plt.savefig(image)
    
    line_notify_token = 'REPLACE_YOUR_TOKEN_HERE'
    line_notify_api = 'https://notify-api.line.me/api/notify'
    payload = {'message': image}
    headers = {'Authorization': 'Bearer ' + line_notify_token}
    files = {'imageFile': open(image, "rb")}
    requests.post(line_notify_api, data=payload, headers=headers, files=files)
    print(image)

line_notify_image(df, 'test.png')

# 下記の画像がLINEに送信されるはずです
test.png

処理時間の計測

calc()で挟んだ部分の処理時間を計測 (こちらの記事を参照)。

start = 0
def calc ():
    global start
    if start == 0:
        start = time.time()
    else:
        print(int((time.time() - start) * 1000), 'ms')
        start = 0

calc()
for i in range(100):
    print('.')
calc()

.
.
.
.
.
.
.
.
.
.
11 ms