株価の1分足をリアルタイムでデータベースに保存します

証券会社の取引ツールを使用して1分足をほぼリアルタイムに保存します。楽天証券のマーケットスピード２のRSSを使用するとエクセルにデータを表示できることを利用します。kabuステーションのAPIなどでも可能かと思います。また、板情報や歩み値などにも応用がきくと思います。

まずはマーケットスピード２にログインしてエクセルに関数を入力します。簡単のためセルA1に次のRSS関数を入力します。これはマウスによる操作で簡単に入力できます。

=RssChart($A$2:$G$2, I2, "1M", 100)

セルI2にある銘柄コードの日付、時刻、始値、高値、安値、終値、出来高の1分足を取得するもので、1分すると下の行にデータが追加されていきます。100は表示本数のことです。

1分足が表示されたので、chatgptの力を借りて、これらのデータをデータベースにどのように書き込んでいったらよいかを聞いています。データベースはmysqlやMariaDBとします。なんども試行錯誤して対話を続けましたが期待した回答が得られませんでした。うまくいったと思っても、2回目はエラーが出たり、なぜか書き込まれるデータがエクセルに無い値であるなどして戸惑いました。しかし、書き込めたことはできたので諦めず考えます。ない値が書き込まれたこと自体が不明なので、Sheet2を作成して、関数が入ったSheet1の内容をそのままイコールでコピペすることにしました。これによりSheet2には式が入っていない状態を作り出すことができます。極力余分なものはない状態を作ります。

そしてchatgptと対話を続けて修正に修正を重ねてできたコードが下記になります。コードはpythonを使用しており、まったくわかりません。

import pandas as pd
import mysql.connector
from openpyxl import load_workbook

# エクセルファイルを読み込む
file_path = "C:/youtube/mysqlg.xlsm"
wb = load_workbook(file_path, read_only=True, data_only=True, keep_vba=True)
ws = wb['Sheet2']

# シートからデータを読み込んでPandas DataFrameに変換する
data = ws.values
columns = next(data)
df = pd.DataFrame(data, columns=columns)

# 最後の "--------" と "0" の行を無視するためにデータフレームをフィルタリング
df = df[(df['時刻'] != "--------") & (df['時刻'] != "0") & (df['時刻'] != 0)]

# 日付と時刻の列を結合して、datetime型に変換する
df['datetime'] = pd.to_datetime(df['日付'] + ' ' + df['時刻'], format='%Y/%m/%d %H:%M')

# 不要な列を削除する
df = df.drop(['日付', '時刻'], axis=1)

# NaNを削除する
df = df.dropna()

# 各列のデータ型を変換する
df['datetime'] = pd.to_datetime(df['datetime'], format='%Y-%m-%d %H:%M:%S')
df['time'] = df['datetime'].dt.strftime('%H:%M')
df['open'] = pd.to_numeric(df['始値'], errors='coerce')
df['high'] = pd.to_numeric(df['高値'], errors='coerce')
df['low'] = pd.to_numeric(df['安値'], errors='coerce')
df['close'] = pd.to_numeric(df['終値'], errors='coerce')
df['volume'] = pd.to_numeric(df['出来高'], errors='coerce')

# MySQLデータベースに接続する
cnx = mysql.connector.connect(
    host='192.168.26.10',
    user='root',
    password='password',
    database='db202305'
)
cursor = cnx.cursor()

# テーブルを作成する
create_table_query = '''
CREATE TABLE IF NOT EXISTS market_data (
id INT AUTO_INCREMENT PRIMARY KEY,
time CHAR(5),
datetime DATETIME,
open DECIMAL(10, 2),
high DECIMAL(10, 2),
low DECIMAL(10, 2),
close DECIMAL(10, 2),
volume INT
);
'''
cursor.execute(create_table_query)
cnx.commit()

# データをインサートする
for index, row in df.iterrows():
    insert_query = '''
    INSERT INTO market_data (time, datetime, open, high, low, close, volume)
    VALUES (%s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
    time = %s,
    open = %s,
    high = %s,
    low = %s,
    close = %s,
    volume = %s;
    '''
    data_tuple = (
        row["time"],
        row["datetime"],
        row["open"],
        row["high"],
        row["low"],
        row["close"],
        row["volume"],
        row["time"],
        row["open"],
        row["high"],
        row["low"],
        row["close"],
        row["volume"]
    )
    cursor.execute(insert_query, data_tuple)
cnx.commit()

# 接続を閉じる
cursor.close()
cnx.close()

このコードにエクセルファイルの場所やデータベース情報を入力します。エクセルファイルの拡張子が.xlsmとなっておりマクロが使用されていますが、最初は使用していませんでした。なぜ必要になったかはあとから説明します。マクロもまったくわかりませんがchatgptの力を借りました。上のコードはマクロがなくてもコマンドプロンプトで正常に動作します。またエクセルファイルとこのコードはcドライブのyoutubeというフォルダにあります。

cd\
cd youtube
python excel_to_mysql.py

また必要なライブラリが無いと正常に動作しないのでインストールしておきます。さらにこのコードには日本語が含まれているのでエンコードの種類をutf8にして保存します。普段、プログラムはやっていないのでメモ帳や秀丸エディタを使用しました。

例
pip install openpyxl

このコードは下記をことを加味して作成してあります。

１、エクセルの最終行がスクショのようななっているので対策済み

２、最終行の数値でない文字列の上のデータは、次のデータが記入され始めるまで動的であるため、これを保存すると違うデータが入る問題を対策
３、多少はエクセルを開いていなくとも、RSS関数で過去のデータが表示できるので重複せずそれを取り込む対策
４、エクセルのRSS関数で表示されている時刻や数値の表示形式はすべて標準になっているのでデータベースに書き込む際に適切な表示形式にする

次は1分ごとにエクセルを監視して、データに変更があったらデータベースに書き込むようにします。これに関してもchatgptに聞いてみます。watchdogというライブラリを提案してもらい正常に動作しました。しかし、watchdogではRSSの数式を感知できないようです。というより、このライブラリはフォルダの変更を監視するものであるようです。タスクスケジューラを使用して1分ごとにスクリプトを実行してもデータは書き込まれませんでした。どうやら手動で上書き保存すればうまくいくようです。先のコードはせっかく正常に動作しているので、コードを変更することなく実現する方法を考えます。

エクセルの自動保存の間隔を1分にして、手動で保存しなくてもいいようにします。メニューのファイルからオプションをクリックします。複数、設定項目があり、「保存」で設定をすることができます。しかし、ファイルの保存場所をOneDriveしか選べません。回避する方法はあるみたいですが、エクセルファイルをOneDriveに移動して試してみます。なお、次のコードもchatgptに聞いて用意しました。これを先に作成したスクリプトと同じフォルダに、適当な名前を付けて保存しておきます。完成したスクリプトに変更を加えると動作しなくなるかもしれないのでこのようにしました。

import time
import os
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ExcelFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.xlsx') or event.src_path.endswith('.xls'):
            print(f'変更があったエクセルファイル: {event.src_path}')
            subprocess.run(['python', 'excel_to_mysql.py'])

if __name__ == "__main__":
    directory_to_watch = "C:\Users\minok\OneDrive - ja"  # エクセルファイルがあるディレクトリのパスに置き換えてください
    event_handler = ExcelFileHandler()
    observer = Observer()
    observer.schedule(event_handler, directory_to_watch, recursive=False)
    observer.start()
    print("監視開始")

    try:
        while True:
            time.sleep(60)  # 1 分ごとに監視
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

PermissionError: [Errno 13] Permission denied: ‘C:\\Users\\minok\\OneDrive – ja\\mysqlg.xlsx’

エラーが発生しました。アクセスを拒否されているみたいです。Excelファイルが別のプログラムによって開かれている場合、このエラーになる可能性があるらしいです。

OneDriveには誰もが保存するわけではないので次の方法を試すことにします。ローカルに保存してあるファイルにエクセルのVBAを使用して1分ごとに保存するというものです。標準モジュールにchatgptで教えてもらったコードを書きます。

Sub AutoSave()
    Application.DisplayAlerts = False
    ThisWorkbook.Save
    Application.DisplayAlerts = True
    Application.OnTime Now + TimeValue("00:01:00"), "AutoSave"
End Sub

ファイルはマクロありになったので上書き保存しようとするとメッセージが出ますので「いいえ」を選択し、拡張子を.xlsmにして保存します。

後で作成した監視用のスクリプトは、エクセルがローカルにあることと、拡張子が変更になったことを考慮して変更しておきます。

        if event.src_path.endswith('.xlsm') or event.src_path.endswith('.xls'):
            print(f'変更があったエクセルファイル: {event.src_path}')
            subprocess.run(['python', 'excel_to_mysql.py'])

if __name__ == "__main__":
    directory_to_watch = "C:/youtube"  # エクセルファイルがあるディレクトリのパスに置き換えてください

それではもう一度コマンドプロンプトで確認してみましょう。そしてエクセルでマクロを実行することも忘れないようにします。

cd\
cd youtube
python monitor_excel.py

マクロにより上書き保存されると、データベースにデータが書き込まれるはずです。スクリプトでもメッセージがでるようになっています。

本当にデータが書き込まれたか確認します。直近のデータが表示されればいいので、それなりのSQLの構文を入力します。

SELECT m.*
FROM market_data m
INNER JOIN (
    SELECT datetime
    FROM market_data
    ORDER BY datetime DESC
    LIMIT 1,5 -- 最終行の直前から5行分取得する
) AS last_rows ON m.datetime = last_rows.datetime;

せっかくですので、このデータをGrafanaで可視化してみます。日本時間でデータを取得したので9時間ずれています。さらに、最初のデータではなく直近のデータを取得することで、Grafanaでもすぐに確認できるようにします。

SELECT datetime - INTERVAL 9 HOUR, open, high, low, close, volume FROM db202305.market_data ORDER BY datetime DESC LIMIT 500;

Grafanaで上記のコードを入力すると確認することができました。