Python编程学习：RSI三重峰值【通达信】动量结构同步指标

import pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport matplotlib.ticker as mtickerfrom matplotlib.pylab import mpl# 设置中文字体和负号显示mpl.rcParams['font.sans-serif'] = ['SimHei']mpl.rcParams['axes.unicode_minus'] = Falseclass LJIndicator:    def __init__(self, data):        """        Python编程学习        """        self.data = data.copy()        self.calculate_indicators()    def wma(self, series, period):        """加权移动平均"""        weights = np.arange(1, period + 1)        wma_values = np.zeros(len(series))        for i in range(period - 1, len(series)):            window = series[i - period + 1:i + 1]            wma_values[i] = np.sum(window * weights) / np.sum(weights)        # 前period-1个值用NaN填充        wma_values[:period - 1] = np.nan        return wma_values    def hhv(self, series, period):        """周期内最值"""        hhv_values = np.zeros(len(series))        for i in range(len(series)):            if i < period - 1:                hhv_values[i] = np.max(series[:i + 1]) if i >= 0 else series[i]            else:                hhv_values[i] = np.max(series[i - period + 1:i + 1])        return hhv_values    def sma(self, series, period):        """简单移动平均"""        return series.rolling(window=period, min_periods=1).mean()    def calculate_indicators(self):        """计算所有"""        df = self.data        # 原始变量计算        lijin1 = 6        lijin2 = df['close'].shift(1)  # REF(CLOSE,1)        # 计算RSI类型的指标        close_diff = df['close'] - lijin2        positive_diff = np.maximum(close_diff, 0)        abs_diff = np.abs(close_diff)        # SMA计算        sma_pos = self.sma(positive_diff, lijin1)        sma_abs = self.sma(abs_diff, lijin1)        # 避免除零        lijin3 = np.where(sma_abs != 0, sma_pos / sma_abs, 0)        # 计算周期长度        bars_count = np.arange(1, len(df) + 1)        period_length = np.where(bars_count 20, bars_count, 20)        # 计算lijin4        lijin3_ceil = np.ceil(lijin3 * 100)  # 乘以100以便ceiling函数更准确


    
        lijin4_values = np.zeros(len(df))        for i in range(len(df)):            period = int(period_length[i])            start_idx = max(0, i - period + 1)            window_values = lijin3_ceil[start_idx:i + 1] * df['high'].values[start_idx:i + 1]            lijin4_values[i] = np.max(window_values) if len(window_values) > 0 else 0        df['lijin4'] = lijin4_values        # 计算lijin5 (WMA 20)        df['lijin5'] = self.wma(df['lijin4'], 20)        # 计算lijin6        df['lijin6'] = 2 * (df['lijin4'] - df['lijin5'])        # 计算HHV指标        diff_4_5 = df['lijin4'] - df['lijin5']        df['lijin7'] = self.hhv(diff_4_5, 2)        df['lijin8'] = self.hhv(diff_4_5, 4)  # 2*2        df['lijin9'] = self.hhv(diff_4_5, 8)  # 2*4        # 计算趋势信号        df['lijin10'] = df['lijin7']        df['lijin11'] = df['lijin7']        df['lijin12'] = df['lijin10'].shift(1)        self.results = df        return df