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

Question

前提

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

実現したいこと

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

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

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

該当のソースコード

import os
from datetime import datetime

import japanize_matplotlib

import jquantsapi
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
from matplotlib.ticker import FormatStrFormatter

from tqdm import tqdm


# リフレッシュトークンが記載されているファイルを指定します
REFRESH_TOKEN_FILE_PATH = "jquantsapi-key.txt"
BUFFER_DATES = 10  # 計算のための余分な日付バッファ(とりあえず10日前後とっておく）
STORAGE_DIR_PATH = "marketdata"
CLOSE_COL = "AdjustmentClose"
RAW_STOCK_CODE = "Code"  # ベースデータの銘柄コード(ETF共存との関係で末尾に0が入って5桁になっているので後で加工必須)

def init_settings():
    """ もろもろの初期化処理(pandas, plot, 保存先

    :return:
    """
    # pandas の表示制限を調整します
    pd.set_option("display.max_rows", 1000)
    pd.set_option("display.max_columns", 1000)
    pd.set_option("display.width", 2000)

    # プロット用の設定をします
    sns.set(rc={'figure.figsize': (15, 10)})
    sns.set(font_scale=2)
    sns.set_style('whitegrid')
    japanize_matplotlib.japanize()

    # 保存先ディレクトリを指定します。
    os.makedirs(STORAGE_DIR_PATH, exist_ok=True)


# リフレッシュトークンを読み込むための関数を定義します
def get_refresh_token(refresh_token_file_path: str = REFRESH_TOKEN_FILE_PATH):
    with open(refresh_token_file_path, "r") as f:
        refresh_token = f.read()
    return refresh_token.rstrip().lstrip()


def get_stock_data(stock_range=200+BUFFER_DATES):
    """ データ取得+最初の加工処理 """
    now = pd.Timestamp.now(tz="Asia/Tokyo")  # timezone指定
    # stock_range = 200+BUFFER_DATES
    start_dt = now - pd.Timedelta(stock_range, unit="D")  # 計算用に10日分多めに指定が必要。
    end_dt = now
    price_file = f"{STORAGE_DIR_PATH}/price_{start_dt.strftime('%Y%m%d')}_{end_dt.strftime('%Y%m%d')}.csv.gz"
    if not os.path.isfile(price_file):  # 保存予定ファイルが存在したら取得しない。
        # jquantsAPIからのデータ取得処理
        print("指定区間のデータがないのでJquantsから読み取ります。少々お待ちください。")
        df_p = store_from_jquantsapi(start_dt, end_dt, price_file)  # start～endまで分のデータを保存
    else:
        # ファイルがあるときはデータを読み込みます
        print(f"file exists: {price_file}, loading")
        df_p = pd.read_csv(price_file, dtype="str")

    df_p.reset_index(drop=True, inplace=True)
    # 各列のデータ型を調整します
    df_p.loc[:, "Date"] = pd.to_datetime(df_p["Date"], format="%Y-%m-%d")
    df_p.loc[:, "Open"] = df_p["Open"].astype(np.float64)
    df_p.loc[:, "High"] = df_p["High"].astype(np.float64)
    df_p.loc[:, "Low"] = df_p["Low"].astype(np.float64)
    df_p.loc[:, "Close"] = df_p["Close"].astype(np.float64)
    df_p.loc[:, "Volume"] = df_p["Volume"].astype(np.float64)
    df_p.loc[:, "TurnoverValue"] = df_p["TurnoverValue"].astype(np.float64)
    df_p.loc[:, "AdjustmentFactor"] = df_p["AdjustmentFactor"].astype(np.float64)
    df_p.loc[:, "AdjustmentOpen"] = df_p["AdjustmentOpen"].astype(np.float64)
    df_p.loc[:, "AdjustmentHigh"] = df_p["AdjustmentHigh"].astype(np.float64)
    df_p.loc[:, "AdjustmentLow"] = df_p["AdjustmentLow"].astype(np.float64)
    df_p.loc[:, "AdjustmentClose"] = df_p["AdjustmentClose"].astype(np.float64)
    df_p.loc[:, "AdjustmentVolume"] = df_p["AdjustmentVolume"].astype(np.float64)

    return df_p


def store_from_jquantsapi(st_date, end_date, price_file):
    """ JquantsAPIからのデータ取得処理(指定した範囲の名称となるファイル(price_file)がなければ保存)

    :param st_date:
    :param end_date:
    :param price_file:
    :return:
    """
    # 株価情報を取得します (データ取得に約数分待ちます)
    df = None
    if end_date.hour < 19:
        # データ更新時間前の場合は日付を1日ずらします。
        end_date -= pd.Timedelta(1, unit="D")
    if not os.path.isfile(price_file):
        df = jqapi.get_price_range(start_dt=st_date, end_dt=end_date)
        df.to_csv(price_file, compression="gzip", index=False)
        print(f"save file: {price_file}")
    return df


def add_stock_metrics(df):
    sc_list = df[RAW_STOCK_CODE].unique()
    concat_df = pd.DataFrame([], columns=df.columns)
    for sc in tqdm(sc_list):

        # if int(sc) / 10 > 1500:  # テスト的に一部銘柄コードのデータだけ動かしたいときの処理
        #     break
        df_filter = df.query(f"{RAW_STOCK_CODE} == @sc").copy()
        tmp_df = add_ma_dev_rate(df_filter)
        concat_df = pd.concat([concat_df, tmp_df])
    return concat_df


def add_ma_dev_rate(df, short=25, middle=75, long=200):
    """ 移動平均および乖離率、パーフェクトオーダーかどうかをメトリクスとして追加する。 """
    df[f"SMA{short}"] = df[CLOSE_COL].rolling(window=short).mean()
    df[f"SMA{middle}"] = df[CLOSE_COL].rolling(window=middle).mean()
    df[f"SMA{long}"] = df[CLOSE_COL].rolling(window=long).mean()

    # 移動平均線乖離率
    df[f"SMA{short}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{short}"]) / df[f"SMA{short}"] * 100
    df[f"SMA{middle}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{middle}"]) / df[f"SMA{middle}"] * 100
    df[f"SMA{long}_乖離率"] = (df[CLOSE_COL] - df[f"SMA{long}"]) / df[f"SMA{long}"] * 100

    df["PerfectOrder"] = np.where((df[f"SMA{short}"] > df[f"SMA{middle}"]) &
                                  (df[f"SMA{middle}"] > df[f"SMA{long}"]), 1, 0)

    return df

# ガター内の緑色のボタンを押すとスクリプトを実行します。
if __name__ == '__main__':
    init_settings()  # pandasやプロット、保存先の初期設定を行う。

    # リフレッシュトークンをファイルから読み込みます
    # https://application.jpx-jquants.com/menuから取得/更新。
    refresh_token = get_refresh_token()

    # J-Quants APIのクライアントクラスを初期化します
    jqapi = jquantsapi.Client(refresh_token=refresh_token)

    # 過去dataを取得(保存期間から開始日終了日を決定し、そのファイル名でデータを保存
    df_p = get_stock_data()

    df_p = add_stock_metrics(df_p)
    df_p.to_csv("stock_metrics_result.csv")

試したこと

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

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

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

Comments

[can110]

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

[H.K2]

これはすでにデータはファイルに保存済み（API呼出を含まない）
さらにはデータ読込後（純粋にadd_stock_metrics関数の呼び出しのみ）の時間でしょうか。
API呼び出しは含みません。ただし、関数呼び出しのみの時間ではないです。
(APIで読みだしたものはcsv保存しており、それを読みだしてからの時間となります。）

行数については、4288銘柄で210日程度(200日移動平均を算出する＋バッファdata）となり、９０万行ほどとなります。

Answer 1

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

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

Comments

[H.K2]

コメントありがとうございます。

＞データを時間方向に間引いてから、小さいwindowサイズで計算をするのがよいかと思います。
これはデータの精度は少々落ちても日足を週足で変えてしまうということでしょうか。
（そのまま間引いてしまうとボラティリティの大きい銘柄は特性が変わってしまうように思うので、週足ベースでのHigh,Low,Open,Closeでの算出？）

＞また、numpy.lib.stride_tricks.sliding_window_view を用いると多少は改善が期待できます。
ありがとうございます。試してみます。

[fpfpfp]

> そのまま間引いてしまうとボラティリティの大きい銘柄は特性が変わってしまうように思うので
そのまま間引くの意味でしたが、ご指摘のとおり不適切ですね。

> データの精度は少々落ちても日足を週足で変えてしまう
こちらが妥当かと思います。

add_ma_dev_rate の引数を小さい値にして、時間計測をしてみることをおすすめします。
あと、np.convolve のほうが適切かも
https://numpy.org/doc/stable/reference/generated/numpy.convolve.html

[H.K2]

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