通达信指标解析与Python重构：构建情绪反转因子公式学习

import pandas as pd
import numpy as np

def calculate_technical_indicators(df):
    """
    计算通达信指标解析与Python重构：构建情绪反转因子公式学习
    """
    # 参数设置
    短期周期 = 6
    中期周期 = 12
    CCI周期 = 14
    随机周期 = 9
    RSI平滑周期1 = 3
    RSI平滑周期2 = 3
    BIAS周期 = 6
    
    # 复制数据框以避免修改原始数据
    data = df.copy()
    
    # 计算MACD指标
    data['EMA12'] = data['close'].ewm(span=12).mean()
    data['EMA26'] = data['close'].ewm(span=26).mean()
    data['离差值'] = data['EMA12'] - data['EMA26']
    data['讯号线'] = data['离差值'].ewm(span=9).mean()
    data['MACD柱状线'] = (data['离差值'] - data['讯号线']) * 2
    
    # 计算随机KD指标
    data['最低最低价'] = data['low'].rolling(window=随机周期).min()
    data['最高最高价'] = data['high'].rolling(window=随机周期).max()
    data['随机原始值'] = (data['close'] - data['最低最低价']) / (data['最高最高价'] - data['最低最低价']) * 100
    data['随机K值'] = data['随机原始值'].rolling(window=RSI平滑周期1).mean()
    data['随机D值'] = data['随机K值'].rolling(window=RSI平滑周期2).mean()
    data[


    
'随机J值'] = 3 * data['随机K值'] - 2 * data['随机D值']
    
    # 计算CCI指标
    data['典型价'] = (data['high'] + data['low'] + data['close']) / 3
    data['典型价MA'] = data['典型价'].rolling(window=CCI周期).mean()
    data['平均绝对偏差'] = data['典型价'].rolling(window=CCI周期).apply(lambda x: np.mean(np.abs(x - np.mean(x))))
    data['CCI指标'] = (data['典型价'] - data['典型价MA']) / (0.015 * data['平均绝对偏差'])
    
    # 计算RSI指标
    data['昨收价'] = data['close'].shift(1)
    data['价格变动'] = data['close'] - data['昨收价']
    data['上涨幅度'] = np.where(data['价格变动'] > 0, data['价格变动'], 0)
    data['下跌幅度'] = np.where(data['价格变动'] 0, np.abs(data['价格变动']), 0)
    
    data['RSI短期_上涨'] = data['上涨幅度'].rolling(window=短期周期).mean()
    data['RSI短期_下跌'] = data['下跌幅度'].rolling(window=短期周期).mean()
    data['RSI短期'] = data['RSI短期_上涨'] / data['RSI短期_下跌'] * 100
    
    data['RSI中期_上涨'] = data['上涨幅度'].rolling(window=中期周期).mean()
    data['RSI中期_下跌'] = data['下跌幅度'].rolling(window=中期周期).mean()
    data['RSI中期'] = data['RSI中期_上涨'] / data['RSI中期_下跌'] * 100
    
    # 计算威廉指标
    data['10日最高价'] = data['high'].rolling(window=10).max()
    data['10日最低价'] = data['low'].rolling(window=10).min()
    data['威廉指标'] = 100 * (data['10日最高价'] - data['close']) / (data['10日最高价'] - data['10日最低价'])
    
    # 计算乖离率
    data['MA_BIAS'] = data['close'].rolling(window=BIAS周期).mean()
    data['乖离率'] = (data['close'] - data['MA_BIAS']) / data['MA_BIAS'] * 100
    
    


    
# 计算ADX指标
    data['TR'] = np.maximum(
        data['high'] - data['low'],
        np.maximum(
            np.abs(data['high'] - data['close'].shift(1)),
            np.abs(data['low'] - data['close'].shift(1))
        )
    )
    data['真实波幅'] = data['TR'].rolling(window=CCI周期).sum()
    
    data['最高差'] = data['high'] - data['high'].shift(1)
    data['最低差'] = data['low'].shift(1) - data['low']
    
    data['上升方向量'] = np.where(
        (data['最高差'] > 0) & (data['最高差'] > data['最低差']),
        data['最高差'],
        0
    )
    data['上升方向量'] = data['上升方向量'].rolling(window=CCI周期).sum()
    
    data['下降方向量'] = np.where(
        (data['最低差'] > 0) & (data['最低差'] > data['最高差']),
        data['最低差'],
        0
    )
    data['下降方向量'] = data['下降方向量'].rolling(window=CCI周期).sum()
    
    data['上升方向线'] = data['上升方向量'] * 100 / data['真实波幅']
    data['下降方向线'] = data['下降方向量'] * 100 / data['真实波幅']
    
    data['DX'] = np.abs(data['下降方向线'] - data['上升方向线']) / (data['下降方向线'] + data['上升方向线']) * 100
    data['平均趋向指数'] = data['DX'].rolling(window=CCI周期).mean()
    data['趋向指数均值'] = (data['平均趋向指数'] + data['平均趋向指数'].shift(CCI周期)) / 2
    
    # 计算资金流量指标
    data['资金流典型价'] = (data['high'] + data['low'] + data['close']) / 3
    data['资金流额'] = data['资金流典型价'] * data['volume']
    
    data['资金流正'] = np.where(
        data['资金流典型价'] > data['资金流典型价'].shift(1),
        data['资金流额'],


    

        0
    )
    data['资金流正和'] = data['资金流正'].rolling(window=CCI周期).sum()
    
    data['资金流负'] = np.where(
        data['资金流典型价'] < data['资金流典型价'].shift(1),
        data['资金流额'],
        0
    )
    data['资金流负和'] = data['资金流负'].rolling(window=CCI周期).sum()
    
    data['资金流量比'] = data['资金流正和'] / data['资金流负和']
    data['资金流量指标'] = 100 - (100 / (1 + data['资金流量比']))
    
    # 计算动态买卖气指标
    data['动态买方'] = np.where(
        data['open'] <= data['open'].shift(1),
        0,
        np.maximum(data['high'] - data['open'], data['open'] - data['open'].shift(1))
    )
    data['动态卖方'] = np.where(
        data['open'] >= data['open'].shift(1),
        0,
        np.maximum(data['open'] - data['low'], data['open'].shift(1) - data['open'])
    )
    
    data['动态买方和'] = data['动态买方'].rolling(window=23).sum()
    data['动态卖方和'] = data['动态卖方'].rolling(window=23).sum()
    
    data['动态买卖气'] = np.where(
        data['动态买方和'] > data['动态卖方和'],
        (data['动态买方和'] - data['动态卖方和']) / data['动态买方和'],
        np.where(
            data['动态买方和'] == data['动态卖方和'],
            0,
            (data['动态买方和'] - data['动态卖方和']) / data['动态卖方和']
        )
    )
    
    # 计算动量指标
    data['动量线'] = data['close'] - data['close'].shift(20)
    data['动量均线'] = data['动量线'].rolling(window=60).mean()
    
    # 计算震荡量指标
    data['MA20'] = data['close'


    
].rolling(window=20).mean()
    data['震荡量'] = 100 * (data['close'] - data['MA20'])
    data['震荡均线'] = data['震荡量'].ewm(span=6).mean()
    
    # 计算多空线指标
    data['多空价'] = (3 * data['close'] + data['low'] + data['open'] + data['high']) / 6
    
    # 计算加权多空线
    weights = np.array([20, 19, 18, 17, 16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1])
    for i in range(len(data)):
        if i >= 20:
            window = data['多空价'].iloc[i-20:i]
            data.loc[data.index[i], '多空线'] = np.sum(weights * window.values) / 210
        else:
            data.loc[data.index[i], '多空线'] = np.nan
    
    data['多空均线'] = data['多空线'].rolling(window=20).mean()
    
    # 计算通达信指标解析与Python重构：构建情绪反转因子公式学习所有条件
    data['条件1'] = (data['离差值'] > data['讯号线']) & (data['离差值'].shift(1) <= data['讯号线'].shift(1))
    data['条件2'] = (data['随机K值'] > data['随机D值']) & (data['随机K值'].shift(1) <= data['随机D值'].shift(1)) & (data['随机K值'] -20)
    data['条件3'] = (data['CCI指标'] -90) & (data['CCI指标'] > -100)
    data['条件4'] = (data['RSI短期'] > data['RSI中期']) & (data['RSI短期'].shift(1) <= data['RSI中期'].shift(1)) & (data['RSI短期'] 50)


    

    data['条件5'] = (data['威廉指标'].shift(1) > 90) & (data['威廉指标'] 70)
    data['条件6'] = (data['乖离率'] > -5) & (data['乖离率'].shift(1) <= -5)
    
    data['条件7'] = (
        (data['上升方向线'].shift(1) < data['下降方向线'].shift(1)) &
        (data['上升方向线'].shift(1) < data['平均趋向指数'].shift(1)) &
        (data['上升方向线'].shift(1) < data['趋向指数均值'].shift(1)) &
        (data['上升方向线'] > data['下降方向线']) &
        (data['上升方向线'] > data['平均趋向指数']) &
        (data['上升方向线'] > data['趋向指数均值'])
    )
    
    data['条件8'] = (data['资金流量指标'] > 20) & (data['资金流量指标'].shift(1) <= 20)
    data['条件9'] = (data['动量线'] > data['动量均线']) & (data['动量线'].shift(1) <= data['动量均线'].shift(1))
    data['条件10'] = data['动态买卖气'] -0.8
    
    # 计算所有W组合条件
    conditions = [
        '条件1', '条件2', '条件3', '条件4', '条件5', 
        '条件6', '条件7', '条件8', '条件9', '条件10'
    ]
    
    # 生成所有可能的3条件组合
    from itertools import combinations
    w_conditions = []
    for i, combo in enumerate(combinations(conditions, 3), 1):
        w_name = f'W{i}'
        data[w_name] = data[combo[0]] & data[combo[1]] & data[combo[2]]
        w_conditions.append(w_name)
    
    # 计算信号条件
    data['信号条件'] = False
    for w in w_conditions:
        data[


    
'信号条件'] = data['信号条件'] | data[w]
    
    # 生成信号值
    data['信号值'] = np.where(data['信号条件'], 1, 0)
    
    # 计算趋势指标
    data['HHV55'] = data['high'].rolling(window=55).max()
    data['LLV55'] = data['low'].rolling(window=55).min()
    data['RSV'] = (data['close'] - data['LLV55']) / (data['HHV55'] - data['LLV55']) * 100
    data['SMA1'] = data['RSV'].rolling(window=5).mean()
    data['SMA2'] = data['SMA1'].rolling(window=3).mean()
    data['趋势赋值'] = 3 * data['SMA1'] - 2 * data['SMA2']
    data['趋势线'] = data['趋势赋值'].ewm(span=3).mean() - 10
    
    # 计算火焰山指标
    data['近期最低'] = data['low'].rolling(window=30).min()
    data['波动幅度'] = np.abs(data['low'] - data['low'].shift(1)).rolling(window=3).mean() / \
                    np.maximum(data['low'] - data['low'].shift(1), 0).rolling(window=3).mean() * 20
    data['波动极值'] = data['波动幅度'].rolling(window=30).max() * 20
    
    data['火焰山'] = np.where(
        (data['low'] <= data['近期最低']) & (data['趋势线'] 40),
        (data['波动幅度'] + data['波动极值']),
        0
    )
    data['火焰山'] = data['火焰山'].ewm(span=3).mean() / 618
    
    return data