pandasで行っている銘柄ごとの指標計算コードを高速化したい。

前提

JQuantsAPIを用いて、東証の銘柄からデータを取得し、それらを銘柄単位で
指標（ゴールデンクロスやレラティブストレングスなどの株価指標）を求め、
最終的にstreamlitで可視化するアプリケーションを作りたいと思っています。

実現したいこと

上記を行うため、まずは、JQuantsAPIを用いて株価データを取得（とりあえず２００日＋α分）し、特定の銘柄コード別に簡単な移動平均線系の指標を求めてcsv出力しようと思っています。

発生している問題・エラーメッセージ

ただ、上記を行う際、pandasを使って前処理を行っているのですが、単なる移動平均線の列追加だけでも10数分かかってしまいます。
他にもメトリクスを増やしたり、データ取得範囲を増やすこともしたいのですが、現状のコード実行速度では日々の運用が現実的ではなく、高速化をしたいと考えています。
最終的にはクラウドとかも考えないといけないのかもですが、まずは手元にある環境で高速に動かしたいので、よい対策があればご教示頂けないでしょうか。
また、合わせて、高速化を検討するうえで知っておいたほうが良い知識などありましたら、合わせてご教示いただけますと幸甚です。（ネットで見た感じだと、numba?とかを使うなども書いてあったのですが、いろいろなことが書いてあって、試してみたけどどれが効果的かわからず、出来ればそれほどライブラリに詳しくなくても、比較的手軽にできる方法がありがたいです。）

該当のソースコード

Python
1import os
2from datetime import datetime
3
4import japanize_matplotlib
5
6import jquantsapi
7import matplotlib.pyplot as plt
8import numpy as np
9import pandas as pd
10import seaborn as sns
11from matplotlib.ticker import FormatStrFormatter
12
13from tqdm import tqdm
14
15
16# リフレッシュトークンが記載されているファイルを指定します
17REFRESH_TOKEN_FILE_PATH = "jquantsapi-key.txt"
18BUFFER_DATES = 10  # 計算のための余分な日付バッファ(とりあえず10日前後とっておく）
19STORAGE_DIR_PATH = "marketdata"
20CLOSE_COL = "AdjustmentClose"
21RAW_STOCK_CODE = "Code"  # ベースデータの銘柄コード(ETF共存との関係で末尾に0が入って5桁になっているので後で加工必須)
22
23def init_settings():
24    """ もろもろの初期化処理(pandas, plot, 保存先
25
26    :return:
27    """
28    # pandas の表示制限を調整します
29    pd.set_option("display.max_rows", 1000)
30    pd.set_option("display.max_columns", 1000)
31    pd.set_option("display.width", 2000)
32
33    # プロット用の設定をします
34    sns.set(rc={'figure.figsize': (15, 10)})
35    sns.set(font_scale=2)
36    sns.set_style('whitegrid')
37    japanize_matplotlib.japanize()
38
39    # 保存先ディレクトリを指定します。
40    os.makedirs(STORAGE_DIR_PATH, exist_ok=True)
41
42
43# リフレッシュトークンを読み込むための関数を定義します
44def get_refresh_token(refresh_token_file_path: str = REFRESH_TOKEN_FILE_PATH):
45    with open(refresh_token_file_path, "r") as f:
46        refresh_token = f.read()
47    return refresh_token.rstrip().lstrip()
48
49
50def get_stock_data(stock_range=200+BUFFER_DATES):
51    """ データ取得+最初の加工処理 """
52    now = pd.Timestamp.now(tz="Asia/Tokyo")  # timezone指定
53    # stock_range = 200+BUFFER_DATES
54    start_dt = now - pd.Timedelta(stock_range, unit="D")  # 計算用に10日分多めに指定が必要。
55    end_dt = now
56    price_file = f"{STORAGE_DIR_PATH}/price_{start_dt.strftime('%Y%m%d')}_{end_dt.strftime('%Y%m%d')}.csv.gz"
57    if not os.path.isfile(price_file):  # 保存予定ファイルが存在したら取得しない。
58        # jquantsAPIからのデータ取得処理
59        print("指定区間のデータがないのでJquantsから読み取ります。少々お待ちください。")
60        df_p = store_from_jquantsapi(start_dt, end_dt, price_file)  # start～endまで分のデータを保存
61    else:
62        # ファイルがあるときはデータを読み込みます
63        print(f"file exists: {price_file}, loading")
64        df_p = pd.read_csv(price_file, dtype="str")
65
66    df_p.reset_index(drop=True, inplace=True)
67    # 各列のデータ型を調整します
68    df_p.loc[:, "Date"] = pd.to_datetime(df_p["Date"], format="%Y-%m-%d")
69    df_p.loc[:, "Open"] = df_p["Open"].astype(np.float64)
70    df_p.loc[:, "High"] = df_p["High"].astype(np.float64)
71    df_p.loc[:, "Low"] = df_p["Low"].astype(np.float64)
72    df_p.loc[:, "Close"] = df_p["Close"].astype(np.float64)
73    df_p.loc[:, "Volume"] = df_p["Volume"].astype(np.float64)
74    df_p.loc[:, "TurnoverValue"] = df_p["TurnoverValue"].astype(np.float64)
75    df_p.loc[:, "AdjustmentFactor"] = df_p["AdjustmentFactor"].astype(np.float64)
76    df_p.loc[:, "AdjustmentOpen"] = df_p["AdjustmentOpen"].astype(np.float64)
77    df_p.loc[:, "AdjustmentHigh"] = df_p["AdjustmentHigh"].astype(np.float64)
78    df_p.loc[:, "AdjustmentLow"] = df_p["AdjustmentLow"].astype(np.float64)
79    df_p.loc[:, "AdjustmentClose"] = df_p["AdjustmentClose"].astype(np.float64)
80    df_p.loc[:, "AdjustmentVolume"] = df_p["AdjustmentVolume"].astype(np.float64)
81
82    return df_p
83
84
85def store_from_jquantsapi(st_date, end_date, price_file):
86    """ JquantsAPIからのデータ取得処理(指定した範囲の名称となるファイル(price_file)がなければ保存)
87
88    :param st_date:
89    :param end_date:
90    :param price_file:
91    :return:
92    """
93    # 株価情報を取得します (データ取得に約数分待ちます)
94    df = None
95    if end_date.hour < 19:
96        # データ更新時間前の場合は日付を1日ずらします。
97        end_date -= pd.Timedelta(1, unit="D")
98    if not os.path.isfile(price_file):
99        df = jqapi.get_price_range(start_dt=st_date, end_dt=end_date)
100        df.to_csv(price_file, compression="gzip", index=False)
101        print(f"save file: {price_file}")
102    return df
103
104
105def add_stock_metrics(df):
106    sc_list = df[RAW_STOCK_CODE].unique()
107    concat_df = pd.DataFrame([], columns=df.columns)
108    for sc in tqdm(sc_list):
109
110        # if int(sc) / 10 > 1500:  # テスト的に一部銘柄コードのデータだけ動かしたいときの処理
111        #     break
112        df_filter = df.query(f"{RAW_STOCK_CODE} == @sc").copy()
113        tmp_df = add_ma_dev_rate(df_filter)
114        concat_df = pd.concat([concat_df, tmp_df])
115    return concat_df
116
117
118def add_ma_dev_rate(df, short=25, middle=75, long=200):
119    """ 移動平均および乖離率、パーフェクトオーダーかどうかをメトリクスとして追加する。 """
120    df[f"SMA{short}"] = df[CLOSE_COL].rolling(window=short).mean()
121    df[f"SMA{middle}"] = df[CLOSE_COL].rolling(window=middle).mean()
122    df[f"SMA{long}"] = df[CLOSE_COL].rolling(window=long).mean()
123
124    # 移動平均線乖離率
125    df[f"SMA{short}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{short}"]) / df[f"SMA{short}"] * 100
126    df[f"SMA{middle}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{middle}"]) / df[f"SMA{middle}"] * 100
127    df[f"SMA{long}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{long}"]) / df[f"SMA{long}"] * 100
128
129    df["PerfectOrder"] = np.where((df[f"SMA{short}"] > df[f"SMA{middle}"]) &
130                                  (df[f"SMA{middle}"] > df[f"SMA{long}"]), 1, 0)
131
132    return df
133
134# ガター内の緑色のボタンを押すとスクリプトを実行します。
135if __name__ == '__main__':
136    init_settings()  # pandasやプロット、保存先の初期設定を行う。
137
138    # リフレッシュトークンをファイルから読み込みます
139    # https://application.jpx-jquants.com/menuから取得/更新。
140    refresh_token = get_refresh_token()
141
142    # J-Quants APIのクライアントクラスを初期化します
143    jqapi = jquantsapi.Client(refresh_token=refresh_token)
144
145    # 過去dataを取得(保存期間から開始日終了日を決定し、そのファイル名でデータを保存
146    df_p = get_stock_data()
147
148    df_p = add_stock_metrics(df_p)
149    df_p.to_csv("stock_metrics_result.csv")
150

試したこと

ネットで高速化の情報を調査した。

補足情報（FW/ツールのバージョンなど）

ここにより詳細な情報を記載してください。

8524ba23

2023/01/16 00:58

> 単なる移動平均線の列追加だけでも10数分かかってしまいます。これはすでにデータはファイルに保存済み（API呼出を含まない）さらにはデータ読込後（純粋にadd_stock_metrics関数の呼び出しのみ）の時間でしょうか。であれば行数（銘柄数×日数？）といった対象データの規模感を提示ください。

H.K2

2023/01/16 10:57

これはすでにデータはファイルに保存済み（API呼出を含まない）さらにはデータ読込後（純粋にadd_stock_metrics関数の呼び出しのみ）の時間でしょうか。 API呼び出しは含みません。ただし、関数呼び出しのみの時間ではないです。 (APIで読みだしたものはcsv保存しており、それを読みだしてからの時間となります。）行数については、4288銘柄で210日程度(200日移動平均を算出する＋バッファdata）となり、９０万行ほどとなります。

行動規範の内容に同意します

回答1件

ベストアンサー

移動平均のwindowサイズが大きいのが原因と推測されます。
データを時間方向に間引いてから、小さいwindowサイズで計算をするのがよいかと思います。

また、numpy.lib.stride_tricks.sliding_window_view を用いると多少は改善が期待できます。
https://numpy.org/devdocs/reference/generated/numpy.lib.stride_tricks.sliding_window_view.html

投稿2023/01/16 06:06

fpfpfp

総合スコア55

H.K2

2023/01/16 11:03

コメントありがとうございます。＞データを時間方向に間引いてから、小さいwindowサイズで計算をするのがよいかと思います。これはデータの精度は少々落ちても日足を週足で変えてしまうということでしょうか。（そのまま間引いてしまうとボラティリティの大きい銘柄は特性が変わってしまうように思うので、週足ベースでのHigh,Low,Open,Closeでの算出？）＞また、numpy.lib.stride_tricks.sliding_window_view を用いると多少は改善が期待できます。ありがとうございます。試してみます。

fpfpfp

2023/01/17 04:06

> そのまま間引いてしまうとボラティリティの大きい銘柄は特性が変わってしまうように思うのでそのまま間引くの意味でしたが、ご指摘のとおり不適切ですね。 > データの精度は少々落ちても日足を週足で変えてしまうこちらが妥当かと思います。 add_ma_dev_rate の引数を小さい値にして、時間計測をしてみることをおすすめします。あと、np.convolve のほうが適切かも https://numpy.org/doc/stable/reference/generated/numpy.convolve.html

H.K2

2023/01/17 23:53

ご回答ありがとうございます！確認するの遅くなりました、すみません。なるほど。ご指摘の内容で夜にでも試してみます。ありがとうございます！

行動規範の内容に同意します