日本発祥のテクニカル指標「一目均衡表」を通知してみた【Python】

17 min read読了の目安(約15500字

背景

キノコードさんの動画を見て株式のテクニカル分析に挑戦してみました。
※単なる「やってみた」系です!💦

先人たちの知恵をお借りするなどして解決できたことを、この場をお借りして感謝するとともに、大変恐縮ですが 自分のメモ として、こちらへまとめておきます。

キノコードさんの動画

【Pythonでファイナンス分析(株・FX)】日本発祥のテクニカル指標「一目均衡表」の作成方法

https://www.youtube.com/watch?v=xihK3A40pZ8&list=TLGGkNHe6lV6rF0wOTA2MjAyMQ

環境

(開発)
Windows 10 64bit (20H2)
Python 3.9.5
VS Code

手順

1. 各種ライブラリのインストール

最初に躓いたのが テクニカル指標を簡単に生成可能な TA-Lib というライブラリのインストール。
解決方法は👇に共有しておきます。

https://zenn.dev/whitecat_22/articles/9e30d88e31c3ec

2. 他

その他は動画を見ながら進めれば、あら簡単!できあがり! ( ´艸`)

株価は動画と同じく、^N225(日経平均株価)です。

それじゃぁ、あんまりなので、最終的に作成したグラフを添付して、SlackやTwitterへ通知してみましたよ~。

通知結果

  • Slack

  • Twitter

作成したコードは下記になります。ご参考まで。

handler.py
try:
    #from notifiers import unzip_requirements
    import unzip_requirements
except ImportError:
    print('Import Error - unzip_requirements')
    pass
except Exception as e:
    print(e)
    pass

import csv
import datetime
import os
from os.path import join, dirname
import mplfinance as mpf
from pandas_datareader import data
import pandas as pd
from dotenv import load_dotenv
from dateutil.relativedelta import relativedelta

import matplotlib.pyplot as plt
import talib as ta

from notifiers import slack
from notifiers import twitter

import json
import logging

# settins for logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Load env variants
dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)
stock_code = os.environ.get("STOCK_CODE")

# Get today's date for getting the stock price and csv&image filename
today = datetime.date.today()

# tmp directory is present by default on Cloud Functions, so guard it
if not os.path.isdir('/tmp'):
    os.mkdir('/tmp')

FILENAME = '%s.csv' % str(today)


def generate_stock_chart_image():
    """
    Generate a six-month stock chart image with mplfinance
    """
    dataframe = pd.read_csv(
        f"/tmp/{str(today)}.csv", index_col=0, parse_dates=True)
    # The return value `Date` from yahoofinance is sorted by asc, so change it to desc for plot
    dataframe = dataframe.sort_values('Date')
    date = dataframe.index

    # 基準線
    high = dataframe['High']
    low = dataframe['Low']

    max26 = high.rolling(window=26).max()
    min26 = low.rolling(window=26).min()

    dataframe['basic_line'] = (max26 + min26) / 2

    dataframe.tail()

    plt.figure(figsize=(16, 6))
    plt.plot(dataframe['basic_line'], label='basic')
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.legend()
    plt.grid()
    #plt.show()

    # 転換線
    high9 = high.rolling(window=9).max()
    low9 = low.rolling(window=9).min()

    dataframe['turn_line'] = (high9 + low9) / 2

    dataframe.tail()

    plt.figure(figsize=(16, 6))
    plt.plot(dataframe['basic_line'], label='basic')
    plt.plot(dataframe['turn_line'], label='turn')
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.legend()
    plt.grid()
    #plt.show()

    # 雲形
    dataframe['span1'] = (dataframe['basic_line'] + dataframe['turn_line']) / 2

    high52 = high.rolling(window=52).max()
    low52 = low.rolling(window=52).min()

    dataframe['span2'] = (high52 + low52) / 2

    dataframe.tail()

    plt.figure(figsize=(16, 6))
    plt.plot(dataframe['basic_line'], label='basic')
    plt.plot(dataframe['turn_line'], label='turn')
    plt.fill_between(date, dataframe['span1'], dataframe['span2'],
                     facecolor="gray", alpha=0.5, label="span")
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.legend()
    plt.grid()
    #plt.show()

    # 遅行線
    dataframe['slow_line'] = dataframe['Adj Close'].shift(-25)

    dataframe.head()

    plt.figure(figsize=(16, 6))
    plt.plot(dataframe['basic_line'], label='basic')
    plt.plot(dataframe['turn_line'], label='turn')
    plt.fill_between(date, dataframe['span1'], dataframe['span2'],
                     facecolor="gray", alpha=0.5, label="span")
    plt.plot(dataframe['slow_line'], label='slow')
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.legend()
    plt.grid()
    #plt.show()

    # ボリンジャーバンド用のdataframe追加
    dataframe["upper"], dataframe["middle"], dataframe["lower"] = ta.BBANDS(
        dataframe['Adj Close'], timeperiod=25, nbdevup=2, nbdevdn=2, matype=0)
    dataframe.tail()

    # ボリンジャーバンドプロット
    apds = [mpf.make_addplot(dataframe['upper'], color='g'),
            mpf.make_addplot(dataframe['middle'], color='b'),
            mpf.make_addplot(dataframe['lower'], color='r')
            ]

    # MACD用のdataframe追加
    dataframe['macd'], dataframe['macdsignal'], dataframe['macdhist'] = ta.MACD(
        dataframe['Adj Close'], fastperiod=12, slowperiod=26, signalperiod=9)
    dataframe.tail()

    # RSIデータフレーム追加
    dataframe["RSI"] = ta.RSI(dataframe["Adj Close"], timeperiod=25)
    dataframe.tail()

    # 基準線、転換線、雲、遅行線の追加
    apds = [mpf.make_addplot(dataframe['upper'], color='g'),
            mpf.make_addplot(dataframe['middle'], color='b'),
            mpf.make_addplot(dataframe['lower'], color='r'),
            mpf.make_addplot(dataframe['macdhist'], type='bar',
                             width=1.0, panel=1, color='gray', alpha=0.5, ylabel='MACD'),
            mpf.make_addplot(dataframe['RSI'], panel=2,
                             type='line', ylabel='RSI'),
            mpf.make_addplot(dataframe['basic_line']),  # 基準線
            mpf.make_addplot(dataframe['turn_line']),  # 転換線
            mpf.make_addplot(dataframe['slow_line']),  # 遅行線
            ]

    # 保存
    mpf.plot(dataframe, type='candle', addplot=apds, figsize=(30, 10), style='sas',
             volume=True, volume_panel=3, panel_ratios=(5, 2, 2, 1), savefig=f"/tmp/{str(today)}.png")

    # ローソク足
    mpf.plot(dataframe, type='candle', figsize=(16, 6),
             style='sas', xrotation=0, volume=True, addplot=apds)

    fig, ax = mpf.plot(dataframe, type='candle', figsize=(16, 9),
                       style='sas', xrotation=0, volume=True, addplot=apds, returnfig=True,
                       volume_panel=3, panel_ratios=(5, 2, 2, 1),
                       fill_between=dict(
                           y1=dataframe['span1'].values, y2=dataframe['span2'].values, alpha=0.5, color='gray'),
                       savefig=f"/tmp/{str(today)}.png"
                       )
    #plt.show()
    labels = ["basic", "turn", "slow", "span"]
    ax[0].legend(labels)
    """
    mpf.plot(dataframe, type='candle', figratio=(12, 4),
             volume=True, mav=(5, 25), style='sas',
             savefig=f"/tmp/{str(today)}.png")
    """

def generate_csv_with_datareader():
    """
    Generate a csv file of OHLCV with date with yahoofinance API
    """
    # 株価推移の開始日を指定(6ヶ月を指定)
    start_date = today - relativedelta(months=6)

    # yahoofinanceのライブラリ経由でAPIを叩く(stock_codeは環境変数で株コードを指定)
    df = data.DataReader(stock_code, 'yahoo', start_date, today)
    df = df[['High', 'Low', 'Open', 'Close', 'Adj Close', 'Volume']]
    df.tail()

    # APIで取得したデータを一旦CSVファイルにする
    df = df.sort_values(by='Date', ascending=False)
    df.to_csv(f"/tmp/{str(today)}.csv")
    # print(df)


def lambdahandler(event, context):
    """
    lambda_handler
    """
    logging.info(json.dumps(event))

    print('event: {}'.format(event))
    print('context: {}'.format(context))
    """
    The main function that will be executed when this Python file is executed
    """
    generate_csv_with_datareader()
    generate_stock_chart_image()

    with open(f"/tmp/{str(today)}.csv", 'r', encoding="utf-8") as file:
        # Skip header row
        reader = csv.reader(file)
        header = next(reader)
        for i, row in enumerate(csv.DictReader(file, header)):
            # Send only the most recent data to Slack notification
            if i == 0:
                slack.Slack(today, row).post()
                twitter.Twitter(today, row).post()

    return {
        'statusCode': 200,
        'body': 'ok'
    }


if __name__ == "__main__":
    print(lambdahandler(event=None, context=None))
notifiers/slack.py
"""
json: Format the data to be sent by the SLack API into JSON
requests: HTTP client
"""
import os
from os.path import join, dirname
from dotenv import load_dotenv
# Import WebClient from Python SDK (github.com/slackapi/python-slack-sdk)
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

dotenv_path = join(dirname(__file__), '.env')
#load_dotenv(dotenv_path)
load_dotenv(verbose=True)
# WebClient insantiates a client that can call API methods
# When using Bolt, you can use either `app.client` or the `client` passed to listeners.
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
# ID of channel that you want to upload file to
load_dotenv(dotenv_path)
token = os.environ.get("SLACK_BOT_TOKEN")
channel_id = os.environ.get("SLACK_CHANNEL_ID")
stock_code = os.environ.get("STOCK_CODE")

class Slack():
    """
    Notification Class to configure the settings for the Slack API
    """
    def __init__(self, date, ohlcv):
        self.__date = date
        self.text = self.__format_text(ohlcv)

    @property
    def date(self):
        """
        Property of date to be displayed in Slack text
        :return: Date
        """
        return self.__date

    def __format_text(self, ohlcv):
        """
        Create params data for sending Slack notification with API.
        :param dict[str, str, str, str, str, str] ohlcv:
        :type ohlcv: {
            'Date': '2020-12-29',
            'Open': '7620',
            'High': '8070',
			'Low': '7610',
			'Close': '8060',
			'Volume': '823700'
        }
        :return: String
        """
        text = f"本日は{self.date.strftime('%Y年%m月%d日')}です。\n" \
               f"取得可能な最新日付の株価情報をお知らせします。 \n\n"\
               f"*銘柄* {str(stock_code)}\n" \
               f"*日付* {str(ohlcv['Date'])}\n" \
               f"*始値* {float(ohlcv['Open'])}\n" \
			   f"*高値* {float(ohlcv['High'])}\n" \
			   f"*安値* {float(ohlcv['Low'])}\n" \
			   f"*終値* {float(ohlcv['Close'])}\n" \
			   f"*出来高* {float(ohlcv['Volume'])}"
        return text

    def post(self):
        """
        POST request to Slack file upload API
        API docs: https://slack.com/api/files.upload
        """
        # The name of the file you're going to upload
        file = open(f"/tmp/{str(self.date)}.png", 'rb')
        title = f"{str(self.date)}.png"
        # Call the files.upload method using the WebClient
        # Uploading files requires the `files:write` scope
        try:
            client.files_upload(
                channels=channel_id,
                initial_comment=self.text,
                file=file,
                title=title
            )
        except Exception as e:
            print(e)
notifiers/twitter.py
"""
json: Format the data to be sent by the Twitter API into JSON
requests: HTTP client
"""
import os
from os.path import join, dirname
from dotenv import load_dotenv
import tweepy


dotenv_path = join(dirname(__file__), '.env')
#load_dotenv(dotenv_path)
load_dotenv(verbose=True)

load_dotenv(dotenv_path)
# 各種twitterのKeyをセット CONSUMER_KEY, CONSUMER_SECRET, ACCESS_KEY, ACCESS_KEY_SECRET
CONSUMER_KEY = os.environ.get('CONSUMER_KEY')
CONSUMER_SECRET = os.environ.get('CONSUMER_SECRET')
ACCESS_KEY = os.environ.get('ACCESS_KEY')
ACCESS_KEY_SECRET = os.environ.get('ACCESS_KEY_SECRET')

stock_code = os.environ.get("STOCK_CODE")

# tweepyの設定
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_KEY, ACCESS_KEY_SECRET)
api = tweepy.API(auth)


class Twitter():
    """
    Notification Class to configure the settings for the Twitter API
    """
    def __init__(self, date, ohlcv):
        self.__date = date
        self.text = self.__format_text(ohlcv)

    @property
    def date(self):
        """
        Property of date to be displayed in Slack text
        :return: Date
        """
        return self.__date

    def __format_text(self, ohlcv):
        """
        Create params data for sending Twitter notification with API.
        :param dict[str, str, str, str, str, str] ohlcv:
        :type ohlcv: {
            'Date': '2020-12-29',
            'Open': '7620',
            'High': '8070',
			'Low': '7610',
			'Close': '8060',
			'Volume': '823700'
        }
        :return: String
        """
        text = f"本日は{self.date.strftime('%Y年%m月%d日')}です。\n" \
               f"取得可能な最新日付の株価情報をお知らせします。 \n\n"\
               f"銘柄  {str(stock_code)}\n" \
               f"日付  {str(ohlcv['Date'])}\n" \
               f"始値  {float(ohlcv['Open'])}\n" \
			   f"高値  {float(ohlcv['High'])}\n" \
			   f"安値  {float(ohlcv['Low'])}\n" \
			   f"終値  {float(ohlcv['Close'])}\n" \
			   f"出来高  {float(ohlcv['Volume'])}"
        return text

    def post(self):
        """
        POST request to Twitter API
        API docs: https://developer.twitter.com/en/docs/twitter-api/api-reference-index
        """
        # The name of the file you're going to upload
        file = open(f"/tmp/{str(self.date)}.png", 'rb')
        title = f"{str(self.date)}.png"
        # Call the files.upload method using the WebClient
        # Uploading files requires the `files:write` scope
        try:
            file_names = ['/tmp/' + title, ]
            media_ids = []
            for filename in file_names:
                res = api.media_upload(filename)
                media_ids.append(res.media_id)
            # tweet with multiple images
            api.update_status(status= self.text + "\n\n" + title, media_ids=media_ids)
        except Exception as e:
            print(e)

参考

対象動画の書き起こしブログ:

https://kino-code.com/python_finance-02/

(編集後記)

株価や為替、暗号資産の値動きなどの分析・予測には様々な指標が用いられます。
多くの指標がありますが、ご自身の投資目的や経験に基づいて、最適な指標を選ぶことが求められます。いろいろな指標の算出を試してみるのも新たな発見があり、面白いと思います。
テクニカル分析によって、投資に興味を持った方はぜひ予測や、実際の投資にもチャレンジしてみてはいかがでしょうか。

私も機械学習について理解を深めて、予測まで漕ぎ着けたいと思います。


●課題

Lambda での定期実行としたく、コンテナイメージからのデプロイを試みていますが、関数を実行すると、TA-Libを呼び出す際にエラーが発生してしまいます。

libta_lib.so.0: cannot open shared object file: No such file or directory

⇒ DockerFileの指定方法が悪く、TA-Lib内のruntimeファイルをインストールできていないのではと推測しています。引き続き、原因究明・解決に努めます。

{
  "errorMessage": "Unable to import module 'handler': libta_lib.so.0: cannot open shared object file: No such file or directory",
  "errorType": "Runtime.ImportModuleError",
  "stackTrace": []
}
START RequestId: eabdc8a8-a22f-4e07-88b4-dda1582b5bb6 Version: $LATEST
Import Error - unzip_requirements
OpenBLAS WARNING - could not determine the L2 cache size on this system, assuming 256k
[WARNING]    2021-06-06T07:01:46.095Z        Matplotlib created a temporary config/cache directory at /tmp/matplotlib-i7nui6jm because the default path (/home/sbx_user1051/.config/matplotlib) is not a writable directory; it is highly recommended to set the MPLCONFIGDIR environment variable to a writable directory, in particular to speed up the import of Matplotlib and to better support multiprocessing.
[ERROR] Runtime.ImportModuleError: Unable to import module 'handler': libta_lib.so.0: cannot open shared object file: No such file or directory
Traceback (most recent call last):
END RequestId: eabdc8a8-a22f-4e07-88b4-dda1582b5bb6
REPORT RequestId: eabdc8a8-a22f-4e07-88b4-dda1582b5bb6    Duration: 4599.39 ms    Billed Duration: 4600 ms    Memory Size: 1024 MB    Max Memory Used: 80 MB