通达信数据 - 使用通达信量化TQ(tdx quant)(3) - 量化脚本程序流程
量化脚本程序流程
读取A股代码.txt 文件读取所有A股代码。
对每只股票进行分析计算
筛选出满足条件的股票
保存到c:\new_tdx_test\T0002\blocknew\LMSJ.blk
通达信TQ选股，调用通达信公式选股
import numpy as np
import pandas as pd
from tqcenter import tdxdata
import os
def 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 False
def 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 False
def 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,")source：wechat
通达信量化TQ支持的数据（股票和指数，K线数据，财务数据，交易数据）
支持调用通达信公式。
支持实盘交易