import pandas as pd import numpy as np from datetime import datetime, timedelta from .config import DATE_FORMAT class Indicator: def __init__(self) -> None: pass @staticmethod def get_all_rsi(rsi_df: pd.DataFrame, periods: int = 14, smooth_k: int = 3, smooth_d: int = 3) -> pd.DataFrame: rsi_df["stoch_rsi"] = \ Indicator.stoch_rsi(rsi_df["rsi"], periods) rsi_df["stoch_rsi_smooth_k"] = \ Indicator.stoch_rsi_smooth_k(rsi_df["stoch_rsi"], smooth_k) rsi_df["stoch_rsi_smooth_d"] = Indicator.stoch_rsi_smooth_d( rsi_df["stoch_rsi_smooth_k"], smooth_d ) return rsi_df @staticmethod def stoch_rsi(rsi: pd.Series, periods: int = 14) -> pd.Series: ma, mi = ( rsi.rolling(window=periods).max(), rsi.rolling(window=periods).min(), ) return (rsi - mi) * 100 / (ma - mi) @staticmethod def stoch_rsi_smooth_k(stoch_rsi: pd.Series, k: int = 3) -> pd.Series: return stoch_rsi.rolling(window=k).mean() @staticmethod def stoch_rsi_smooth_d(stoch_rsi_k: pd.Series, d: int = 3) -> pd.Series: return stoch_rsi_k.rolling(window=d).mean() @staticmethod def generate_dates(start_date: datetime.date, num_days: int, format=DATE_FORMAT): lst_date = [] current_date = start_date for i in range(num_days): current_date += timedelta(days=1) while current_date.weekday() >= 5: current_date += timedelta(days=1) lst_date.append(current_date.strftime(format)) return lst_date @staticmethod def get_ichimoku_cloud( df: pd.DataFrame, conversion_period=9, base_period=26, span_b_period=52, displacement=26, ) -> pd.DataFrame: space_displacement = np.full(displacement, np.nan) tenkan_sen = ( df["high"].rolling(window=conversion_period).max() + df["low"].rolling(window=conversion_period).min() ) / 2 kijun_sen = ( df["high"].rolling(window=base_period).max() + df["low"].rolling(window=base_period).min() ) / 2 senkou_span_a = (tenkan_sen + kijun_sen) / 2 senkou_span_b = ( df["high"].rolling(window=span_b_period).max() + df["low"].rolling(window=span_b_period).min() ) / 2 chikou_span = df["close"].shift(-displacement) last_date = datetime.strptime(df["time"].iloc[-1], DATE_FORMAT) lst_date = Indicator.generate_dates(last_date, displacement) time = np.concatenate((df["time"], lst_date)) tenkan_sen = np.concatenate((tenkan_sen, space_displacement)) kijun_sen = np.concatenate((kijun_sen, space_displacement)) senkou_span_a = np.concatenate((space_displacement, senkou_span_a)) senkou_span_b = np.concatenate((space_displacement, senkou_span_b)) chikou_span = np.concatenate((chikou_span, space_displacement)) data_dict = { "time": time, "tenkan_sen": tenkan_sen, "kijun_sen": kijun_sen, "senkou_span_a": senkou_span_a, "senkou_span_b": senkou_span_b, "chikou_span": chikou_span, "tenkan_kijun": tenkan_sen - kijun_sen, "kumo_cloud": senkou_span_a - senkou_span_b } return pd.DataFrame(data_dict)