Python编程学动量自习化:通李信真正波段王底指标源代码

import pandas as pdimport numpy as npdefcalculate_SMA(series, period, weight=1):"""计算加权平滑移动平均"""    sma = np.zeros(len(series))if len(series) == 0:return sma    sma[0] = series[0]for i in range(1, len(series)):        sma[i] = (weight * series[i] + (period - weight) * sma[i-1]) / periodreturn smadefcalculate_indicators(df):# 计算量化对冲套利指标    df['price'] = (2 * df['close'] + df['high'] + df['low']) / 4    df['MA30'] = df['price'].rolling(window=30).mean()    df['temp'] = (df['close'] - df['MA30']) / df['MA30'] * 100    df['量化对冲套利'] = df['temp'].ewm(span=3, adjust=False).mean()# 计算KDJ指标    low_min = df['low'].rolling(window=9).min().fillna(0)    high_max = df['high'].rolling(window=9).max().fillna(0)    rsv = ((df['close'] - low_min) / (high_max - low_min + 1e-12)) * 100    rsv = rsv.clip(0, 100)  # 限制在0-100范围# 计算K、D值    K = calculate_SMA(rsv.values, 3, 1)    D = calculate_SMA(K, 3, 1)    J = 3 * np.array(K) - 2 * np.array(D)    df['K'] = K    df['D'] = D    df['J'] = J# 计算lijinzz指标for period in [6, 12, 24]:        df[f'MA{period}'] = df['close'].rolling(window=period).mean()    df['lijinzz1'] = (df['close'] - df['MA6']) / df['MA6'] * 100    df['lijinzz2'] = (df['close'] - df['MA12']) / df['MA12'] * 100    df['lijinzz3'] = (df['close'] - df['MA24']) / df['MA24'] * 100    df['lijinzz4'] = (df['lijinzz1'] + 2*df['lijinzz2'] + 3*df['lijinzz3']) / 6    df['lijinzz5'] = df['lijinzz4'].rolling(window=3).mean()# 计算机会信号    df['J_prev'] = df['J'].shift(1)    cross_j_0 = (df['J'] > 0) & (df['J_prev'] <= 0)    df['机会信号'] = cross_j_0 & (df['lijinzz5'] <= -7)# 计算RSI指标    df['LC'] = df['close'].shift(1)    delta = df['close'] - df['LC']    gain = np.where(delta > 0, delta, 0)    loss = np.where(delta < 0, -delta, 0)    avg_gain = calculate_SMA(gain, 6, 1)    avg_loss = calculate_SMA(loss, 6, 1)    rs = avg_gain / (avg_loss + 1e-12)    df['RSI'] = 100 - (100 / (1 + rs))# 计算预警信号    rsi_prev = df['RSI'].shift(1)    df['预警信号'] = (df['RSI'] < 80) & (rsi_prev >= 80)return dfif __name__ == "__main__":    df = pd.DataFrame(data)# 计算指标    df = calculate_indicators(df)# 获取最近100天的信号    recent_signals = df.tail(100)# 打印信号结果    print("机会信号出现位置：")    print(recent_signals[recent_signals['机会信号']])    print("\n预警信号出现位置：")    print(recent_signals[recent_signals['预警信号']])