Python编程学动量自动化：通达信真正波段王顶底指标源代码

import pandas as pd
import numpy as np

def calculate_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]) / period
    return sma

def calculate_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 df

if __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['预警信号']])