【シストレ開発#15】「テクニカル分析」チャート表示機能追加！

print(self.df)
           OHLC_open  OHLC_high  OHLC_low  OHLC_close
Date
2021/04/01       ...        ...       ...         ...
2021/04/02       ...        ...       ...         ...
2021/04/03       ...        ...       ...         ...
2021/04/04       ...        ...       ...         ...
2021/04/05       ...        ...       ...         ...
       ...       ...        ...       ...         ...

sma = self.df["OHLC_close"].rolling(window=SPAN).mean()

ema = self.df["OHLC_close"].ewm(span=SPAN).mean()

wma = self.df["OHLC_close"].rolling(window=SPAN).apply(self._calc_wma)

def _calc_wma(self, prices： pd.Series) -> float：
    weights = np.arange(len(prices)) + 1
    wma = np.sum(weights * prices) / weights.sum()
    return wma

# ---------- 基準線 ----------
max = self.df["OHLC_high"].rolling(window=SPAN_BASE).max()
min = self.df["OHLC_low"].rolling(window=SPAN_BASE).min()
base = (max + min) / 2

# ---------- 転換線 ----------
max = self.df["OHLC_high"].rolling(window=SPAN_CONV).max()
min = self.df["OHLC_low"].rolling(window=SPAN_CONV).min()
conv = (max + min) / 2

# ---------- 先行スパン線A ----------
series = (base + conv) / 2
span_a = series.shift(ICHMK_SPN_OFS)

# ---------- 先行スパン線B ----------
max = self.df["OHLC_high"].rolling(window=SPAN_SPAN_B).max()
min = self.df["OHLC_low"].rolling(window=SPAN_SPAN_B).min()
series = (max + min) / 2
span_b = series.shift(ICHMK_SPN_OFS)

# ---------- 遅行線 ----------
lag = self.df["OHLC_close"].shift(-ICHMK_LAG)

# ---------- 平均値 ----------
base = self.df["OHLC_close"].rolling(window=SPAN).mean()

# ---------- 標準偏差線 ----------
sigma = self.df["OHLC_close"].rolling(window=SPAN).std(ddof=0)

sigma_plus1 = base + sigma       # +1σ
sigma_minus1 = base - sigma      # -1σ
sigma_plus2 = base + sigma * 2   # +2σ
sigma_minus2 = base - sigma * 2  # -2σ
sigma_plus3 = base + sigma * 3   # +3σ
sigma_minus3 = base - sigma * 3  # -3σ

rsi_sma = self.df["OHLC_close"].rolling(window=SPAN).apply(self._calc_rsi_by_sma)

def _calc_rsi_by_sma(self, prices： pd.Series) -> float：
    sumofup = prices[prices > 0].sum()
    sumofdown = -prices[prices < 0].sum()
    return sumofup / (sumofup + sumofdown) * 100

# ---------- MACD線 ----------
ema_s = self.df["OHLC_close"].ewm(span=MACD_EMA_S, adjust=False).mean()
ema_l = self.df["OHLC_close"].ewm(span=MACD_EMA_L, adjust=False).mean()
macd = ema_s - ema_l

# ---------- シグナル線 ----------
sig = macd.ewm(span=MACD_EMA_SIG).mean()

sr = self.df["OHLC_close"]

# --------------- %K ---------------
per_k = sr.rolling(window=STCHA_K).apply(self._calc_stochastics_k)

# --------------- %D ---------------
self.df["numer"] = sr.rolling(window=STCHA_D).apply(self._calc_stochastics_d_numer)
self.df["denom"] = sr.rolling(window=STCHA_D).apply(self._calc_stochastics_k_denom)
per_d = sr.rolling(window=STCHA_D).apply(self._calc_stochastics_d)

# --------------- %Slow D ---------------
per_sd = per_d.rolling(window=STCHA_SD).mean()

def _calc_stochastics_k(self, sr： pd.Series) -> float：
  df = self.df.loc[sr.index]
  latest_close = df["OHLC_close"][-1]
  max_high = df["OHLC_high"].max()
  min_low = df["OHLC_low"].min()
  k = (latest_close - min_low) / (max_high - min_low)
  return k * 100

def _calc_stochastics_d_numer(self, sr： pd.Series) -> float：
  df = self.df.loc[sr.index]
  latest_close = df["OHLC_close"][-1]
  min_low = df["OHLC_low"].min()
  return latest_close - min_low

def _calc_stochastics_k_denom(self, sr： pd.Series) -> float：
  df = self.df.loc[sr.index]
  max_high = df["OHLC_high"].max()
  min_low = df["OHLC_low"].min()
  return max_high - min_low

def _calc_stochastics_d(self, sr： pd.Series) -> float：
  df = self.df.loc[sr.index]
  numer = df["numer"].sum()
  denom = df["denom"].sum()
  return (numer / denom) * 100