auカブコム証券API リアルタイムtickデーターからローソク足を作成する。「Pythonで株価分析」サンプルコード付き。 | 独学・Pythonで株式投資（旧トレードステーションのある生活）株式投資分析・自動売買で生活を豊かに

auカブコム証券APIから取得した、リアルタイムtickデーターからローソク足を作成する方法を解説します。

Pythonのサンプルコードも掲載していますので、Pythonによる株価分析や自動売買をチャレンジされている方のお役に立てれば幸いです。

Python初心者向けに、細かい解説をいれた動画をYouTubeにアップしました。

「サンプルコードを見ても、何が何だか分からん！！」という方は、このページ後半に動画へのリンクを貼っていますのでご覧下さい。

このページのサンプルコードで、下のような5分足チャートを作成します。

auカブコム証券APIからPUSH配信されたリアルタイムデータを、エディターで表示させている画像がこちら。

エディター画面の下半分に雪崩のようにデーターが押し寄せる様子が、画像で伝わるでしょうか？

１秒間に複数件のデーターを受信しています。

そのデーターの内１件がこちら。

{‘OverSellQty’: 2915500.0, ‘UnderBuyQty’: 3582200.0, ‘TotalMarketValue’: 35444310256850.0, ‘MarketOrderSellQty’: 0.0, ‘MarketOrderBuyQty’: 0.0, ‘BidTime’: ‘2022-01-04T09:28:01+09:00’, ‘AskTime’: ‘2022-01-04T09:28:01+09:00’, ‘Exchange’: 1, ‘ExchangeName’: ‘東証１部’, ‘TradingVolume’: 10131500.0, ‘TradingVolumeTime’: ‘2022-01-04T09:28:01+09:00’, ‘VWAP’: 2165.2455, ‘TradingValue’: 21937184450.0, ‘BidQty’: 2800.0, ‘BidPrice’: 2172.5, ‘BidSign’: ‘0101’, ‘Sell1’: {‘Price’: 2172.5, ‘Qty’: 2800.0, ‘Sign’: ‘0101’, ‘Time’: ‘2022-01-04T09:28:01+09:00’}, ‘Sell2’: {‘Price’: 2173.0, ‘Qty’: 3700.0}, ‘Sell3’: {‘Price’: 2173.5, ‘Qty’: 6000.0}, ‘Sell4’: {‘Price’: 2174.0, ‘Qty’: 33400.0}, ‘Sell5’: {‘Price’: 2174.5, ‘Qty’: 20500.0}, ‘Sell6’: {‘Price’: 2175.0, ‘Qty’: 58100.0}, ‘Sell7’: {‘Price’: 2175.5, ‘Qty’: 7100.0}, ‘Sell8’: {‘Price’: 2176.0, ‘Qty’: 16000.0}, ‘Sell9’: {‘Price’: 2176.5, ‘Qty’: 6700.0}, ‘Sell10’: {‘Price’: 2177.0, ‘Qty’: 27300.0}, ‘AskQty’: 3100.0, ‘AskPrice’: 2171.5, ‘AskSign’: ‘0101’, ‘Buy1’: {‘Price’: 2171.5, ‘Qty’: 3100.0, ‘Sign’: ‘0101’, ‘Time’: ‘2022-01-04T09:28:01+09:00’}, ‘Buy2’: {‘Price’: 2171.0, ‘Qty’: 4000.0}, ‘Buy3’: {‘Price’: 2170.5, ‘Qty’: 4900.0}, ‘Buy4’: {‘Price’: 2170.0, ‘Qty’: 8100.0}, ‘Buy5’: {‘Price’: 2169.5, ‘Qty’: 5500.0}, ‘Buy6’: {‘Price’: 2169.0, ‘Qty’: 8900.0}, ‘Buy7’: {‘Price’: 2168.5, ‘Qty’: 17000.0}, ‘Buy8’: {‘Price’: 2168.0, ‘Qty’: 20200.0}, ‘Buy9’: {‘Price’: 2167.5, ‘Qty’: 15200.0}, ‘Buy10’: {‘Price’: 2167.0, ‘Qty’: 17600.0}, ‘Symbol’: ‘7203’, ‘SymbolName’: ‘トヨタ自動車’, ‘CurrentPrice’: 2172.5, ‘CurrentPriceTime’: ‘2022-01-04T09:28:01+09:00’, ‘CurrentPriceChangeStatus’: ‘0057’, ‘CurrentPriceStatus’: 1, ‘CalcPrice’: 2172.5, ‘PreviousClose’: 2105.5, ‘PreviousCloseTime’: ‘2021-12-30T00:00:00+09:00’, ‘ChangePreviousClose’: 67.0, ‘ChangePreviousClosePer’: 3.18, ‘OpeningPrice’: 2158.0, ‘OpeningPriceTime’: ‘2022-01-04T09:03:00+09:00’, ‘HighPrice’: 2184.0, ‘HighPriceTime’: ‘2022-01-04T09:13:08+09:00’, ‘LowPrice’: 2154.5, ‘LowPriceTime’: ‘2022-01-04T09:03:01+09:00’, ‘SecurityType’: 1}

手順を追って解説します。

■サンプルコードを動作させる条件■

・auカブコム証券の口座を開設し、API使用設定が完了していること

・kabuステーションを起動し、ログイン済みであること

・Pythonが使用できること(*^_^*)

株価ストリームデーターをデーターベースに保存

株価ストリームデーター取得

リアルタイムに配信される株価のストリームデーターを取得します。

auカブコム証券APIのサンプルコードを流用して、関数を作ります。

ストリームデーターを取得する関数がこちら。

##### 株価ストリームデーター取得専用．PUSH配信開始 #####
def ats_websocket_get_tick_data_only():
    # kabuステーションのサーバから通知が来た時に呼ばれる
    def on_message(ws, message):
        print('--- RECV MSG. --- ')
        print('===json print===')
        # json.loads() でJSONドキュメントを含んだ文字列をPythonオブジェクトへ変換する
        content = json.loads(message)
        print(content)

        #### 作成した関数で、SqliteデーターベースにPUSH配信されたデーターを格納する
        Store_pushdata_in_sqlite(content)

        return content

    def on_error(ws, error):
        print('--- ERROR --- ')
        print(error)

    def on_close(ws):
        print('--- DISCONNECTED --- ')

    # WebSoketが実行されると最初に呼ばれる関数
    def on_open(ws):
        print('--- CONNECTED --- ')
        def run(*args):
            while (True):
                line = sys.stdin.readline()
                # != は、等価でないとTrue、等価だとFalseを返す。
                if line != '':
                    print('closing...')
                    ws.close()
        _thread.start_new_thread(run, ())

    url = 'ws://localhost:18080/kabusapi/websocket'
    # websocket.enableTraceの引数をFalseからTrueにすると、
    # websocket - clientモジュールのログが出力され、最初のHTTPを利用したコネクション確立のやりとりが確認出来る
    # websocket.enableTrace(True)
    ws = websocket.WebSocketApp(url,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.on_open = on_open
    ws.run_forever()

##### 株価ストリームデーター取得専用．PUSH配信開始 #####

def ats_websocket_get_tick_data_only():

# kabuステーションのサーバから通知が来た時に呼ばれる

def on_message(ws, message):

print('--- RECV MSG. --- ')

print('===json print===')

# json.loads() でJSONドキュメントを含んだ文字列をPythonオブジェクトへ変換する

content = json.loads(message)

print(content)

#### 作成した関数で、SqliteデーターベースにPUSH配信されたデーターを格納する

Store_pushdata_in_sqlite(content)

return content

def on_error(ws, error):

print('--- ERROR --- ')

print(error)

def on_close(ws):

print('--- DISCONNECTED --- ')

# WebSoketが実行されると最初に呼ばれる関数

def on_open(ws):

print('--- CONNECTED --- ')

def run(*args):

while (True):

line = sys.stdin.readline()

# != は、等価でないとTrue、等価だとFalseを返す。

if line != '':

print('closing...')

ws.close()

_thread.start_new_thread(run, ())

url = 'ws://localhost:18080/kabusapi/websocket'

# websocket.enableTraceの引数をFalseからTrueにすると、

# websocket - clientモジュールのログが出力され、最初のHTTPを利用したコネクション確立のやりとりが確認出来る

# websocket.enableTrace(True)

ws = websocket.WebSocketApp(url,

on_message=on_message,

on_error=on_error,

on_close=on_close)

ws.on_open = on_open

ws.run_forever()

PUSH配信されたデーターを保存するDB作成

PUSH配信されたデーターを保存するDBを作成します。

今回はSQLAalchmyを利用して、SQLiteに保存します。

##### ＜＜SQLAlchemy＞＞ で株価tickをデーターベースに保存 ####

#### Connecting（接続）
# SQLiteのデーターベースを作成します。
#必要なライブラリー
# from sqlalchemy import create_engine
# engine = create_engine('使用するDB名:///パス・ファイル名', echo=True)
engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

#### 基本（Bace）クラスを作成
# declarative_base（）関数を使用して基本クラスを作成します。
#必要なライブラリー
# from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

#### クラスを継承してテーブルを定義する
#必要なライブラリー
# from sqlalchemy import Column
# from sqlalchemy import DateTime
# from sqlalchemy.types import Float
# from sqlalchemy.types import Integer
# from sqlalchemy.types import String
class StorePushdataInDf(Base):
    __tablename__ = 'stream_data'
    id = Column(Integer, primary_key=True, autoincrement=True)  # 主キー・autoincrement設定
    time = Column(DateTime, nullable=False)
    get_time = Column(DateTime, nullable=False)
    content = Column(String(64), nullable=True)
    name = Column(String(64), nullable=True)
    content_Symbol = Column(String(64), nullable=True)
    content_SymbolName = Column(String(64), nullable=True)
    content_CurrentPrice = Column(Float)
    open = Column(Float)
    close = Column(Float)
    high = Column(Float)
    low = Column(Float)
    volume = Column(Integer)

####このコードでデーターベースが作成される####
Base.metadata.create_all(engine)


#### SqliteデーターベースにPUSH配信されたデーターを格納する ####
def Store_pushdata_in_sqlite(content):
    ####### ＜＜sqlalchemy＞＞でデーターを書き込む ######
    # 必要なライブラリー
    # from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=engine)
    #### セッションオブジェクトを作る ####
    session = Session()
    # 現在の日付時刻を取得
    now = datetime.datetime.now()
    # print(now)
    # PUSH配信された情報から　'Symbol'　'SymbolName'　"CurrentPrice"　を抽出する
    content_Symbol = content['Symbol']
    content_SymbolName = content['SymbolName']
    content_CurrentPrice = content["CurrentPrice"]

    content = str(content)
    candle_data1 = StorePushdataInDf(time=now,get_time=now, name=content, content_Symbol=content_Symbol,
                                     content_SymbolName=content_SymbolName, content_CurrentPrice=content_CurrentPrice)
    # print(candle_data1.time)

    ######　セッションにａｄｄする ######
    session.add(candle_data1)

    ##### コミットして初めて永続化決定 ########
    session.commit()
    print('DBに保存しました。')

##### ＜＜SQLAlchemy＞＞で株価tickをデーターベースに保存 ####

#### Connecting（接続）

# SQLiteのデーターベースを作成します。

#必要なライブラリー

# from sqlalchemy import create_engine

# engine = create_engine('使用するDB名:///パス・ファイル名', echo=True)

engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

#### 基本（Bace）クラスを作成

# declarative_base（）関数を使用して基本クラスを作成します。

#必要なライブラリー

# from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

#### クラスを継承してテーブルを定義する

#必要なライブラリー

# from sqlalchemy import Column

# from sqlalchemy import DateTime

# from sqlalchemy.types import Float

# from sqlalchemy.types import Integer

# from sqlalchemy.types import String

class StorePushdataInDf(Base):

__tablename__ = 'stream_data'

id = Column(Integer, primary_key=True, autoincrement=True) # 主キー・autoincrement設定

time = Column(DateTime, nullable=False)

get_time = Column(DateTime, nullable=False)

content = Column(String(64), nullable=True)

name = Column(String(64), nullable=True)

content_Symbol = Column(String(64), nullable=True)

content_SymbolName = Column(String(64), nullable=True)

content_CurrentPrice = Column(Float)

open = Column(Float)

close = Column(Float)

high = Column(Float)

low = Column(Float)

volume = Column(Integer)

####このコードでデーターベースが作成される####

Base.metadata.create_all(engine)

#### SqliteデーターベースにPUSH配信されたデーターを格納する ####

def Store_pushdata_in_sqlite(content):

####### ＜＜sqlalchemy＞＞でデーターを書き込む ######

# 必要なライブラリー

# from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

#### セッションオブジェクトを作る ####

session = Session()

# 現在の日付時刻を取得

now = datetime.datetime.now()

# print(now)

# PUSH配信された情報から　'Symbol'　'SymbolName'　"CurrentPrice"　を抽出する

content_Symbol = content['Symbol']

content_SymbolName = content['SymbolName']

content_CurrentPrice = content["CurrentPrice"]

content = str(content)

candle_data1 = StorePushdataInDf(time=now,get_time=now, name=content, content_Symbol=content_Symbol,

content_SymbolName=content_SymbolName, content_CurrentPrice=content_CurrentPrice)

# print(candle_data1.time)

######　セッションにａｄｄする ######

session.add(candle_data1)

##### コミットして初めて永続化決定 ########

session.commit()

print('DBに保存しました。')

スレッド（並列処理）として実行

スレッド（並列処理）として実行させます。

if __name__ == '__main__':
    # スレッドを走らせる(並列処理で実行)
    # tagetで走らせる関数を渡す。
    t1 = threading.Thread(target=ats_websocket_get_tick_data_only)
    # スレッド1を走らせる
    t1.start()

if __name__ == '__main__':

# スレッドを走らせる(並列処理で実行)

# tagetで走らせる関数を渡す。

t1 = threading.Thread(target=ats_websocket_get_tick_data_only)

# スレッド1を走らせる

t1.start()

データー取得からDB保存まで全体のコード

データー取得からDB保存まで全体のコードがこちらです。

import datetime
import json

# websocket関数に使用する（auカブコム証券APIサンプルコードそのまま）
import sys
import websocket
import _thread

# スレッドを使用する（並列処理）
import threading

# PUSH配信されたデーターを保存するDB作成に使用する（
# sqlalchmyのライブラリ（後ほど説明します）
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy.types import Float
from sqlalchemy.types import Integer
from sqlalchemy.types import String
from sqlalchemy.orm import sessionmaker


###### PUSH配信されたデーターを保存するDB作成 ######

##### ＜＜SQLAlchemy＞＞ で株価tickをデーターベースに保存 ####

#### Connecting（接続）
# SQLiteのデーターベースを作成します。
#必要なライブラリー
# from sqlalchemy import create_engine
# engine = create_engine('使用するDB名:///パス・ファイル名', echo=True)
engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

#### 基本（Bace）クラスを作成
# declarative_base（）関数を使用して基本クラスを作成します。
#必要なライブラリー
# from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

#### クラスを継承してテーブルを定義する
#必要なライブラリー
# from sqlalchemy import Column
# from sqlalchemy import DateTime
# from sqlalchemy.types import Float
# from sqlalchemy.types import Integer
# from sqlalchemy.types import String
class StorePushdataInDf(Base):
    __tablename__ = 'stream_data'
    id = Column(Integer, primary_key=True, autoincrement=True)  # 主キー・autoincrement設定
    time = Column(DateTime, nullable=False)
    get_time = Column(DateTime, nullable=False)
    content = Column(String(64), nullable=True)
    name = Column(String(64), nullable=True)
    content_Symbol = Column(String(64), nullable=True)
    content_SymbolName = Column(String(64), nullable=True)
    content_CurrentPrice = Column(Float)
    open = Column(Float)
    close = Column(Float)
    high = Column(Float)
    low = Column(Float)
    volume = Column(Integer)

####このコードでデーターベースが作成される####
Base.metadata.create_all(engine)


#### SqliteデーターベースにPUSH配信されたデーターを格納する ####
def Store_pushdata_in_sqlite(content):
    ####### ＜＜sqlalchemy＞＞でデーターを書き込む ######
    # 必要なライブラリー
    # from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=engine)
    #### セッションオブジェクトを作る ####
    session = Session()
    # 現在の日付時刻を取得
    now = datetime.datetime.now()
    # print(now)
    # PUSH配信された情報から　'Symbol'　'SymbolName'　"CurrentPrice"　を抽出する
    content_Symbol = content['Symbol']
    content_SymbolName = content['SymbolName']
    content_CurrentPrice = content["CurrentPrice"]

    content = str(content)
    candle_data1 = StorePushdataInDf(time=now,get_time=now, name=content, content_Symbol=content_Symbol,
                                     content_SymbolName=content_SymbolName, content_CurrentPrice=content_CurrentPrice)
    # print(candle_data1.time)

    ######　セッションにａｄｄする ######
    session.add(candle_data1)

    ##### コミットして初めて永続化決定 ########
    session.commit()
    print('DBに保存しました。')


##### 株価ストリームデーター取得専用．PUSH配信開始 #####
def ats_websocket_get_tick_data_only():
    # kabuステーションのサーバから通知が来た時に呼ばれる
    def on_message(ws, message):
        print('--- RECV MSG. --- ')
        print('===json print===')
        # json.loads() でJSONドキュメントを含んだ文字列をPythonオブジェクトへ変換する
        content = json.loads(message)
        print(content)

        #### 作成した関数で、SqliteデーターベースにPUSH配信されたデーターを格納する
        Store_pushdata_in_sqlite(content)

        return content

    def on_error(ws, error):
        print('--- ERROR --- ')
        print(error)

    def on_close(ws):
        print('--- DISCONNECTED --- ')

    # WebSoketが実行されると最初に呼ばれる関数
    def on_open(ws):
        print('--- CONNECTED --- ')
        def run(*args):
            while (True):
                line = sys.stdin.readline()
                # != は、等価でないとTrue、等価だとFalseを返す。
                if line != '':
                    print('closing...')
                    ws.close()
        _thread.start_new_thread(run, ())

    url = 'ws://localhost:18080/kabusapi/websocket'
    # websocket.enableTraceの引数をFalseからTrueにすると、
    # websocket - clientモジュールのログが出力され、最初のHTTPを利用したコネクション確立のやりとりが確認出来る
    # websocket.enableTrace(True)
    ws = websocket.WebSocketApp(url,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.on_open = on_open
    ws.run_forever()


if __name__ == '__main__':
    # スレッドを走らせる(並列処理で実行)
    # tagetで走らせる関数を渡す。
    t1 = threading.Thread(target=ats_websocket_get_tick_data_only)
    # スレッド1を走らせる
    t1.start()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

import datetime

import json

# websocket関数に使用する（auカブコム証券APIサンプルコードそのまま）

import sys

import websocket

import _thread

# スレッドを使用する（並列処理）

import threading

# PUSH配信されたデーターを保存するDB作成に使用する（

# sqlalchmyのライブラリ（後ほど説明します）

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column

from sqlalchemy import DateTime

from sqlalchemy.types import Float

from sqlalchemy.types import Integer

from sqlalchemy.types import String

from sqlalchemy.orm import sessionmaker

###### PUSH配信されたデーターを保存するDB作成 ######

##### ＜＜SQLAlchemy＞＞で株価tickをデーターベースに保存 ####

#### Connecting（接続）

# SQLiteのデーターベースを作成します。

#必要なライブラリー

# from sqlalchemy import create_engine

# engine = create_engine('使用するDB名:///パス・ファイル名', echo=True)

engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

#### 基本（Bace）クラスを作成

# declarative_base（）関数を使用して基本クラスを作成します。

#必要なライブラリー

# from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

#### クラスを継承してテーブルを定義する

#必要なライブラリー

# from sqlalchemy import Column

# from sqlalchemy import DateTime

# from sqlalchemy.types import Float

# from sqlalchemy.types import Integer

# from sqlalchemy.types import String

class StorePushdataInDf(Base):

__tablename__ = 'stream_data'

id = Column(Integer, primary_key=True, autoincrement=True) # 主キー・autoincrement設定

time = Column(DateTime, nullable=False)

get_time = Column(DateTime, nullable=False)

content = Column(String(64), nullable=True)

name = Column(String(64), nullable=True)

content_Symbol = Column(String(64), nullable=True)

content_SymbolName = Column(String(64), nullable=True)

content_CurrentPrice = Column(Float)

open = Column(Float)

close = Column(Float)

high = Column(Float)

low = Column(Float)

volume = Column(Integer)

####このコードでデーターベースが作成される####

Base.metadata.create_all(engine)

#### SqliteデーターベースにPUSH配信されたデーターを格納する ####

def Store_pushdata_in_sqlite(content):

####### ＜＜sqlalchemy＞＞でデーターを書き込む ######

# 必要なライブラリー

# from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

#### セッションオブジェクトを作る ####

session = Session()

# 現在の日付時刻を取得

now = datetime.datetime.now()

# print(now)

# PUSH配信された情報から　'Symbol'　'SymbolName'　"CurrentPrice"　を抽出する

content_Symbol = content['Symbol']

content_SymbolName = content['SymbolName']

content_CurrentPrice = content["CurrentPrice"]

content = str(content)

candle_data1 = StorePushdataInDf(time=now,get_time=now, name=content, content_Symbol=content_Symbol,

content_SymbolName=content_SymbolName, content_CurrentPrice=content_CurrentPrice)

# print(candle_data1.time)

######　セッションにａｄｄする ######

session.add(candle_data1)

##### コミットして初めて永続化決定 ########

session.commit()

print('DBに保存しました。')

##### 株価ストリームデーター取得専用．PUSH配信開始 #####

def ats_websocket_get_tick_data_only():

# kabuステーションのサーバから通知が来た時に呼ばれる

def on_message(ws, message):

print('--- RECV MSG. --- ')

print('===json print===')

# json.loads() でJSONドキュメントを含んだ文字列をPythonオブジェクトへ変換する

content = json.loads(message)

print(content)

#### 作成した関数で、SqliteデーターベースにPUSH配信されたデーターを格納する

Store_pushdata_in_sqlite(content)

return content

def on_error(ws, error):

print('--- ERROR --- ')

print(error)

def on_close(ws):

print('--- DISCONNECTED --- ')

# WebSoketが実行されると最初に呼ばれる関数

def on_open(ws):

print('--- CONNECTED --- ')

def run(*args):

while (True):

line = sys.stdin.readline()

# != は、等価でないとTrue、等価だとFalseを返す。

if line != '':

print('closing...')

ws.close()

_thread.start_new_thread(run, ())

url = 'ws://localhost:18080/kabusapi/websocket'

# websocket.enableTraceの引数をFalseからTrueにすると、

# websocket - clientモジュールのログが出力され、最初のHTTPを利用したコネクション確立のやりとりが確認出来る

# websocket.enableTrace(True)

ws = websocket.WebSocketApp(url,

on_message=on_message,

on_error=on_error,

on_close=on_close)

ws.on_open = on_open

ws.run_forever()

if __name__ == '__main__':

# スレッドを走らせる(並列処理で実行)

# tagetで走らせる関数を渡す。

t1 = threading.Thread(target=ats_websocket_get_tick_data_only)

# スレッド1を走らせる

t1.start()

アプリを作成

ローソク足を表示させるために、簡単なアプリを作成しました。

ブラウザ上にアプリケーションを作成するライプラリー「Streamlit」をインポートします。

#streamlit　ブラウザ上にアプリケーションを作成するライプラリーをインポート
import streamlit as st


st.markdown('# 5分足チャートを描画します。')

ip_extracted_symbol = st.text_input('チャート描画する証券コードを入力',value="167030019")
'入力された証券コード:',ip_extracted_symbol

if st.button('tickデーターから5分足チャートを描画します。'):
    import pandas as pd
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column
    from sqlalchemy import DateTime
    from sqlalchemy.types import Float
    from sqlalchemy.types import Integer
    from sqlalchemy.types import String
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import desc

    ### 接続
    # engine = create_engine('sqlite:///sample.db', echo=True)
    # engine = create_engine('sqlite:///C:/Stock_price_data/push20200104.db', echo=False)
    engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

    ### テーブルとクラスのマッピング
    # ORMでは、DBのテーブルとPythonのクラスをマッピングさせます。
    # declarative_base()はそのベースとなるクラスで、これを継承してテーブルに対応するクラスを作成します。

    Base = declarative_base()


    class StorePushdataInDf(Base):
        __tablename__ = 'stream_data'
        id = Column(Integer, primary_key=True, autoincrement=True)  # 主キー・autoincrement設定
        time = Column(DateTime, nullable=False)
        get_time = Column(DateTime, nullable=False)
        content = Column(String(64), nullable=True)
        name = Column(String(64), nullable=True)
        content_Symbol = Column(String(64), nullable=True)
        content_SymbolName = Column(String(64), nullable=True)
        content_CurrentPrice = Column(Float)
        open = Column(Float)
        close = Column(Float)
        high = Column(Float)
        low = Column(Float)
        volume = Column(Integer)


    ####このコードでデーターベースが作成される####
    # Base.metadata.create_all(engine)

    ### セッションの確立
    Session = sessionmaker(bind=engine)
    session = Session()

    ### SELECTする

    #入力された銘柄コードを代入
    extracted_symbol = ip_extracted_symbol

    sql_statement = session.query(
        StorePushdataInDf).order_by(desc(
        StorePushdataInDf.time)).filter(StorePushdataInDf.content_Symbol == extracted_symbol).limit(10000).statement

    print(f'sql:ここから→{sql_statement}←ここまで')

    df = pd.read_sql_query(sql=sql_statement, con=engine)

    # DBと切断
    session.close()

    # エンジン破棄
    engine.dispose()

    df.set_index("time", inplace=True)
    print(df)

    print(df.describe())
    # print(df.index)

    df_ohlc = df['content_CurrentPrice'].resample('5T').ohlc()


    df.to_excel('tick_data.xls')
    df_ohlc.to_excel('tick_data_ohlc.xls')
    print(df_ohlc)

    # 日付がインデックスだとplotlyで描画されないようなので、インデックスから外す
    df_ohlc_ri = df_ohlc.reset_index()

    #pandasで時系列データのOHLC（四本値）を算出・ダウンサンプリング
    #https://note.nkmk.me/python-pandas-resampling-ohlc/

    import plotly.graph_objects as go

    # 初期化
    fig5 = go.Figure()

    st.write('ローソク足に移動平均線を追加')

    fig5 = go.Figure(data=[go.Candlestick(x=df_ohlc_ri['time'],
                                          open=df_ohlc_ri['open'],
                                          high=df_ohlc_ri['high'],
                                          low=df_ohlc_ri['low'],
                                          close=df_ohlc_ri['close'],
                                          name=ip_extracted_symbol)])

    # チャートのタイトルを表示
    fig5.update_layout(title_text=f'銘柄コード：{ip_extracted_symbol}')

    #スライドバー非表示
    fig5.update_layout(xaxis_rangeslider_visible=False)

    # fig5.show()
    st.plotly_chart(fig5, use_container_width=True)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

#streamlit　ブラウザ上にアプリケーションを作成するライプラリーをインポート

import streamlit as st

st.markdown('# 5分足チャートを描画します。')

ip_extracted_symbol = st.text_input('チャート描画する証券コードを入力',value="167030019")

'入力された証券コード:',ip_extracted_symbol

if st.button('tickデーターから5分足チャートを描画します。'):

import pandas as pd

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column

from sqlalchemy import DateTime

from sqlalchemy.types import Float

from sqlalchemy.types import Integer

from sqlalchemy.types import String

from sqlalchemy.orm import sessionmaker

from sqlalchemy import desc

### 接続

# engine = create_engine('sqlite:///sample.db', echo=True)

# engine = create_engine('sqlite:///C:/Stock_price_data/push20200104.db', echo=False)

engine = create_engine('sqlite:///C:/Stock_price_data/push20220104.db', echo=False)

### テーブルとクラスのマッピング

# ORMでは、DBのテーブルとPythonのクラスをマッピングさせます。

# declarative_base()はそのベースとなるクラスで、これを継承してテーブルに対応するクラスを作成します。

Base = declarative_base()

class StorePushdataInDf(Base):

__tablename__ = 'stream_data'

id = Column(Integer, primary_key=True, autoincrement=True) # 主キー・autoincrement設定

time = Column(DateTime, nullable=False)

get_time = Column(DateTime, nullable=False)

content = Column(String(64), nullable=True)

name = Column(String(64), nullable=True)

content_Symbol = Column(String(64), nullable=True)

content_SymbolName = Column(String(64), nullable=True)

content_CurrentPrice = Column(Float)

open = Column(Float)

close = Column(Float)

high = Column(Float)

low = Column(Float)

volume = Column(Integer)

####このコードでデーターベースが作成される####

# Base.metadata.create_all(engine)

### セッションの確立

Session = sessionmaker(bind=engine)

session = Session()

### SELECTする

#入力された銘柄コードを代入

extracted_symbol = ip_extracted_symbol

sql_statement = session.query(

StorePushdataInDf).order_by(desc(

StorePushdataInDf.time)).filter(StorePushdataInDf.content_Symbol == extracted_symbol).limit(10000).statement

print(f'sql:ここから→{sql_statement}←ここまで')

df = pd.read_sql_query(sql=sql_statement, con=engine)

# DBと切断

session.close()

# エンジン破棄

engine.dispose()

df.set_index("time", inplace=True)

print(df)

print(df.describe())

# print(df.index)

df_ohlc = df['content_CurrentPrice'].resample('5T').ohlc()

df.to_excel('tick_data.xls')

df_ohlc.to_excel('tick_data_ohlc.xls')

print(df_ohlc)

# 日付がインデックスだとplotlyで描画されないようなので、インデックスから外す

df_ohlc_ri = df_ohlc.reset_index()

#pandasで時系列データのOHLC（四本値）を算出・ダウンサンプリング

#https://note.nkmk.me/python-pandas-resampling-ohlc/

import plotly.graph_objects as go

# 初期化

fig5 = go.Figure()

st.write('ローソク足に移動平均線を追加')

fig5 = go.Figure(data=[go.Candlestick(x=df_ohlc_ri['time'],

open=df_ohlc_ri['open'],

high=df_ohlc_ri['high'],

low=df_ohlc_ri['low'],

close=df_ohlc_ri['close'],

name=ip_extracted_symbol)])

# チャートのタイトルを表示

fig5.update_layout(title_text=f'銘柄コード：{ip_extracted_symbol}')

#スライドバー非表示

fig5.update_layout(xaxis_rangeslider_visible=False)

# fig5.show()

st.plotly_chart(fig5, use_container_width=True)

アプリで表示された様子がこちら。

「tickデーターから5分足チャートを描画します。」ボタンを押すと、その時点までtickデーターを集計してローソク足を描画します。

まとめ

auカブコム証券APIから、リアルタイムに配信される株価のストリームデーターからローソク足を作成する方法を解説しました。

説明不足で、わかりにくい部分が多いと思いましたので、詳細を解説した動画をアップしました。

このページ後半の動画リンクからご覧下さい。

コードを出来るだけシンプルにする為に、ボタンを押すたびに最新の1千件のデーターを取得してローソク足を描画しています。

tickデーターをデーターベースに保存する際に、分足に情報を整理してから保存すれば、ローソク足描画速度は改善されます。

皆さん色々試してみてください。

動画で解説

この記事を書いた人

あしおゆたか

株式投資歴22年のシステムトレーダー（←主な取引手法）

机上の理論ではなく、実体験に基づいた記事作りをモットーにしています。

スポーツクラブに毎週2日～3日通い、サウナ後の暴飲暴食が趣味。

◇主な投資対象
日本株式
株式ＥＴＦ（上場投資信託）
日経225先物
日経225先物オプション

◇運営者情報はこちら