251203通达信官方量化(TDX Quant)(3)—实践<六脉神剑选股>

import numpy as npimport pandas as pdfrom tqcenter import tdxdataimport osdef ema(series, window):    """计算指数移动平均"""    return series.ewm(span=window, adjust=False).mean()def sma(series, window, k=1):    """计算简单移动平均"""    return series.rolling(window=window).mean()def llv(series, window):    """计算周期内最低值"""    return series.rolling(window=window).min()def hhv(series, window):    """计算周期内最高值"""    return series.rolling(window=window).max()def ma(series, window):    """计算简单移动平均"""    return series.rolling(window=window).mean()def ref(series, periods):    """引用若干周期前的数据"""    return series.shift(periods)def six_pulse_sword_signal(close, high, low):    """    计算六脉神剑选股信号    返回值: True表示满足买入条件，False表示不满足    """    # MACD指标    diff = ema(close, 8) - ema(close, 13)    dea = ema(diff, 5)    a1 = diff > dea        # KDJ指标    rsv1 = (close - llv(low, 8)) / (hhv(high, 8) - llv(low, 8)) * 100    k = sma(rsv1, 3, 1)    d = sma(k, 3, 1)    a2 = k > d        # RSI指标    lc = ref(close, 1)    rsi1 = sma(np.maximum(close - lc, 0), 5, 1) / sma(np.abs(close - lc), 5, 1) * 100    rsi2 = sma(np.maximum(close - lc, 0), 13, 1) / sma(np.abs(close - lc), 13, 1) * 100    a3 = rsi1 > rsi2        # LWR指标    rsv = -(hhv(high, 13) - close) / (hhv(high, 13) - llv(low, 13)) * 100    lwr1 = sma(rsv, 3, 1)    lwr2 = sma(lwr1, 3, 1)    a4 = lwr1 > lwr2        # BBI指标    bbi = (ma(close, 3) + ma(close, 6) + ma(close, 12) + ma(close, 24)) / 4    a5 = close > bbi        # MTM指标    mtm = close - ref(close, 1)    mms = 100 * ema(ema(mtm, 5), 3) / ema(ema(np.abs(mtm), 5), 3)    mmm = 100 * ema(ema(mtm, 13), 8) / ema(ema(np.abs(mtm), 13), 8)    a6 = mms > mmm        # 判断当前是否满足所有条件且前一天不满足    current_condition = a1 & a2 & a3 & a4 & a5 & a6    previous_condition = ref(current_condition, 1)        # 涨买入信号：当前满足所有条件且前一天不满足    buy_signal = current_condition & (previous_condition == False)        # 返回最新的信号    return buy_signal.iloc[-1] if len(buy_signal) > 0 else Falsedef save_to_block_file(selected_stocks, block_file_path):    """    将选股结果保存到通达信板块文件    沪市: 1 + 6位代码    深市: 0 + 6位代码    北京: 2 + 6位代码    """    try:        # 确保目录存在        os.makedirs(os.path.dirname(block_file_path), exist_ok=True)                # 打开文件准备写入（会清空原有内容）        with open(block_file_path, 'w', encoding='gbk') as f:            for stock in selected_stocks:                # 解析股票代码                if stock.endswith('.SH'):                    # 沪市                    code = '1' + stock[:6]                elif stock.endswith('.SZ'):                    # 深市                    code = '0' + stock[:6]                else:                    # 默认处理（如果有北京市场或其他）                    code = stock[:7]  # 保留原始格式的前7位                                f.write(code + '\n')                print(f"选股结果已保存到 {block_file_path}")        return True    except Exception as e:        print(f"保存选股结果到文件时出错: {e}")        return Falsedef select_stocks_six_pulse_sword():    """根据六脉神剑策略选股"""    # 初始化连接    tdxdata.initialize(__file__)        try:        # 从A股代码.txt文件中读取股票列表        print("正在读取A股代码文件...")        stock_list = []        # 使用脚本所在目录的绝对路径来定位A股代码.txt文件        script_dir = os.path.dirname(os.path.abspath(__file__))        file_path = os.path.join(script_dir, 'A股代码.txt')        print(f"尝试从路径读取文件: {file_path}")                with open(file_path, 'r', encoding='utf-8') as f:            for line in f:                if '|' in line:                    code, name = line.split('|')                    name = name.strip()  # 去除换行符                                        # 剔除ST和*ST股票                    if name.startswith('ST') or name.startswith('*ST'):                        continue                                        # 转换代码格式以适配tdxdata接口                    if code.startswith('SH'):                        stock_code = code[2:] + '.SH'                    elif code.startswith('SZ'):                        stock_code = code[2:] + '.SZ'                    else:                        stock_code = code                    stock_list.append(stock_code)                print(f"共读取到{len(stock_list)}只A股股票")                # 存储符合条件的股票        selected_stocks = []                # 对所有股票进行分析        total_stocks = len(stock_list)        print(f"\n开始分析全部{total_stocks}只股票...")                for i, stock in enumerate(stock_list):            try:                # 获取最近60个交易日的日线数据（为了有足够的数据计算指标）                data = tdxdata.get_market_data(                    field_list=['Open', 'High', 'Low', 'Close'],                    stock_list=[stock],                    period='1d',                    count=60,                    dividend_type='none'                )                                if not data or len(data) == 0:                    continue                                # 转换为DataFrame                df = pd.DataFrame({                    'open': data['Open'][stock],                    'high': data['High'][stock],                    'low': data['Low'][stock],                    'close': data['Close'][stock]                })                                # 确保数据足够                if len(df) < 30:                    continue                                # 计算六脉神剑信号                signal = six_pulse_sword_signal(df['close'], df['high'], df['low'])                                if signal:                    selected_stocks.append(stock)                                # 显示进度（每500只股票显示一次）                if (i + 1) % 500 == 0 or i == total_stocks - 1:                    print(f"已处理{i + 1}/{total_stocks}只股票，已发现{len(selected_stocks)}只满足条件的股票...")                                except Exception as e:                # 忽略个别股票的错误                print(f"处理股票{stock}时出错: {e}")                continue                print(f"选股完成，共选出{len(selected_stocks)}只符合条件的股票")        return selected_stocks            except FileNotFoundError as e:        print(f"文件未找到: {e}")        print("请确保A股代码.txt文件存在于脚本所在目录")        return []    except Exception as e:        print(f"执行选股过程中发生错误: {e}")        import traceback        traceback.print_exc()        return []    finally:        # 关闭连接        try:            tdxdata.close()        except:            pass# 主程序入口if __name__ == "__main__":    print("开始执行六脉神剑选股策略...")    # 打印当前工作目录    import os    print(f"当前工作目录: {os.getcwd()}")    print(f"脚本所在目录: {os.path.dirname(os.path.abspath(__file__))}")        try:        selected = select_stocks_six_pulse_sword()        if selected:            print("\n符合六脉神剑买入条件的股票列表:")            for stock in selected:                print(stock)                            # 保存到通达信板块文件            # 使用基于脚本位置的绝对路径            script_dir = os.path.dirname(os.path.abspath(__file__))            block_file_path = os.path.join(script_dir, "..", "..", "T0002", "blocknew", "LMSJ.blk")            save_to_block_file(selected, block_file_path)                        # 输出选股结果供通达信使用            result = "XG," + "|".join([f"{i}#{stock}" for i, stock in enumerate(selected)])            print(f"\n选股结果代码: {result}")        else:            print("未发现符合条件的股票")            print("XG,")                        # 即使没有选股结果，也创建空的板块文件            # 使用基于脚本位置的绝对路径            script_dir = os.path.dirname(os.path.abspath(__file__))            block_file_path = os.path.join(script_dir, "..", "..", "T0002", "blocknew", "LMSJ.blk")            save_to_block_file([], block_file_path)    except Exception as e:        print(f"程序执行过程中发生错误: {e}")        import traceback        traceback.print_exc()        print("XG,")