pythonでstockindicatorsライブラリを使う

SMAとかならTA-Libでいいんだけど、Zig-Zagを使ってみたかったので入れてみた。
うーん。デフォルトでPandasやNumpyに対応していなくて、quetes_listに変換する必要があり、処理件数が多いとパフォーマンス的にちょっと厳しい。

Guide and Pro tips | Stock Indicators for Python

.NET 8.0 (Linux、macOS、Windows) をダウンロードする

sudo apt-get update
sudo apt-get install -y dotnet-sdk-8.0
pip install stock-indicators

import pandas as pd
filepath = "~/FX_binary/Data/usdjpy-m2-bid-2022-01-01-2022-12-31.csv"
data = pd.read_csv(filepath)

# timestamp列を日時に変換
def arrange(df):
    df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
    df = df.set_index('date')
    del df['timestamp']
    df.columns = [col.title() for col in df.columns]
    return df
data = arrange(data)
data = data[data['Volume']!=0]

from stock_indicators import indicators
from stock_indicators import EndType
from stock_indicators.indicators.common import Quote
quetes_list = [
    Quote(date, open, high, low, close, volume)
    for date, open, high, low, close, volume
    in zip(data.index, data['Open'], data['High'], data['Low'], data['Close'], data['Volume'])
]
results = indicators.get_zig_zag(quetes_list, EndType.HIGH_LOW, 0.04)

import numpy as np
zigzag_high = []
zigzag_low = []
zigzag_type = []
zigzag = []
for r in results:
    zigzag_high.append(r.retrace_high)
    zigzag_low.append(r.retrace_low)
    zigzag_type.append(r.point_type)
    zigzag.append(r.zig_zag)

zigzag_high = np.array(zigzag_high)
zigzag_low = np.array(zigzag_low)
zigzag= np.array(zigzag)

data['zigzag_type'] = zigzag_type
data['zigzag_H'] = np.nan
data['zigzag_L'] = np.nan
data['zigzag_H'] = data['High'].mask(data['zigzag_type'] != 'H')
data['zigzag_L'] = data['Low'].mask(data['zigzag_type'] != 'L')
data['prev_H'] = data['zigzag_H'].copy().ffill()
data['prev_L'] = data['zigzag_L'].copy().ffill()

import plotly.graph_objects as go

dfpl = data[-8000:-4000]
fig = go.Figure(data=[go.Candlestick(x=dfpl.index,
                open=dfpl['Open'],
                high=dfpl['High'],
                low=dfpl['Low'],
                close=dfpl['Close'])])
fig.add_scatter(x=dfpl.index, y=dfpl['zigzag_H'],
                mode="markers", name="zigzag_H", marker=dict(size=5, color="violet"))
fig.add_scatter(x=dfpl.index, y=dfpl['zigzag_L'],
                mode="markers", name="zigzag_L", marker=dict(size=5, color="yellowgreen"))
fig.add_scatter(x=dfpl.index, y=dfpl['prev_H'],
                mode="markers", name="prev_H", marker=dict(size=4, symbol="line-ew-open", color="violet"))
fig.add_scatter(x=dfpl.index, y=dfpl['prev_L'],
                mode="markers", name="prev_L", marker=dict(size=4, symbol="line-ew-open", color="yellowgreen"))

fig.update_layout(xaxis_rangeslider_visible=False)
fig.show()