tdxquant

name	tdxquant
description	TdxQuant（通达信量化）全能开发助手。基于 tqcenter Python API 帮助用户完成量化策略开发全流程： 数据获取、选股策略、回测验证、实时监控、条件预警、公式调用、板块管理、交易下单。 触发场景：用户提到"量化策略"、"选股"、"回测"、"TdxQuant"、"通达信量化"、"tqcenter"、 "K线"、"因子"、"板块"、"金叉死叉"、"实时监控"、"行情订阅"、"预警"、"subscribe"、 "盯盘"、"涨停"、"tq.get_"、"tq.send_"、"tq.formula_"、"FN字段"、"GP字段"、 "price_df"、"get_full_tick"、"下单"、"买入"、"卖出"、"撤单"、"委托"、"持仓"、 "order_stock"、"stock_account"、"自动交易"、"实盘交易"等关键词时使用。

TdxQuant 全能开发助手

环境前提

• Python 3.7+（建议 3.13），64位
• 需预先启动通达信金融终端（支持 TQ 策略功能）
• tqcenter.py 位于通达信安装目录 PYPlugins\user\ 下，导入前须将该路径加入 sys.path

初始化与路径配置

自动获取通达信安装目录（推荐）

import sys, os, winreg

# 从注册表获取通达信安装目录
key_path = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\通达信金融终端64"
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path) as key:
    tdx_root, _ = winreg.QueryValueEx(key, "InstallLocation")

# 必须用 insert(0, ...) 确保优先加载通达信目录的 tqcenter.py
sys.path.insert(0, os.path.join(tdx_root, 'PYPlugins', 'user'))

from tqcenter import tq
tq.initialize(__file__)

手动指定路径（已知安装目录时）

import sys
sys.path.insert(0, 'E:/App/new_tdx_test64/PYPlugins/user')

from tqcenter import tq
tq.initialize(__file__)

关闭连接

tq.close()  # 手动关闭，程序退出时也会自动调用

初始化注意事项

• tq.initialize(__file__) 中 __file__ 作为策略唯一标识，同一路径不能同时运行两个实例
• ErrorId='12'：已有同名策略运行（会打印警告但不报错）
• ErrorId='6'/'7'：连接断开，会自动触发重新初始化
• 使用 sys.path.insert(0, ...) 而非 append()，避免加载到其他同名模块
• tqcenter.py 会在其上上级目录（PYPlugins/）自动定位 TPythClient.dll

策略开发流程

1. 初始化连接
2. 确认股票池来源（生成选股/回测代码前，先确认用户要从哪个板块选股，明确板块名称和 block_type 参数；如果用户没指定，推荐沪深300或中证500作为起点）
3. 获取数据（K线/快照/财务/板块成份股）
4. 信号计算（均线/因子/公式）
5. 执行操作（回测/预警/写入板块/交易下单）
6. 结果输出（print_to_tdx/send_message）
7. 如果是回测策略，询问用户是否需要生成配套的可视化脚本（收益曲线、回撤图等）

常用策略模式

选股入板块

from tqcenter import tq
tq.initialize(__file__)

codes = tq.get_stock_list_in_sector('通达信88')
df = tq.get_market_data(
    field_list=['Close'], stock_list=codes,
    start_time='20250101', end_time='',
    dividend_type='front', period='1d', fill_data=True
)
close_df = tq.price_df(df, 'Close', column_names=codes)
# ... 信号逻辑 ...
tq.send_user_block(block_code='MYBK', stocks=selected_list, show=True)

回测策略（配合 vectorbt）

import vectorbt as vbt
from tqcenter import tq
tq.initialize(__file__)

df = tq.get_market_data(
    field_list=['Close', 'Open'], stock_list=['688318.SH'],
    start_time='20240101', end_time='20250101',
    dividend_type='front', period='1d', fill_data=True
)
close_df = tq.price_df(df, 'Close', column_names=['688318.SH'])
open_df = tq.price_df(df, 'Open', column_names=['688318.SH'])

ma = vbt.MA.run(close_df, window=5).ma
entries = close_df.vbt.crossed_above(ma).shift(1).fillna(False).astype(bool)
exits = close_df.vbt.crossed_below(ma).shift(1).fillna(False).astype(bool)

pf = vbt.Portfolio.from_signals(
    close=close_df, entries=entries, exits=exits,
    price=open_df, init_cash=100000, fees=0.0003,
    freq='D', size_granularity=100
)
print(pf.stats())

调用通达信公式

# 单只股票
tq.formula_set_data_info(stock_code='688318.SH', stock_period='1d', count=100, dividend_type=1)
result = tq.formula_zb(formula_name='MACD', formula_arg='12,26,9')

# 批量选股
xg = tq.formula_process_mul_xg(
    formula_name='UPN', formula_arg='3',
    stock_list=['688318.SH','600519.SH'],
    stock_period='1d', count=5, dividend_type=1
)

实时监控与预警

import json, time
from tqcenter import tq
tq.initialize(__file__)

triggered = set()

def on_update(data_str):
    try:
        code = json.loads(data_str).get('Code')
        if code in triggered:
            return
        snap = tq.get_market_snapshot(stock_code=code)
        now, pre = float(snap['Now']), float(snap['LastClose'])
        if pre <= 0: return
        rise = (now - pre) / pre * 100
        if rise > 5.0:
            triggered.add(code)
            tq.unsubscribe_hq(stock_list=[code])
            from datetime import datetime
            tq.send_warn(
                stock_list=[code],
                time_list=[datetime.now().strftime("%Y%m%d%H%M%S")],
                price_list=[str(now)], close_list=[str(pre)],
                volum_list=[snap.get('Volume','0')],
                bs_flag_list=['0'], warn_type_list=['0'],
                reason_list=[f'涨幅{rise:.2f}%突破5%'], count=1
            )
    except Exception as e:
        print(f"回调异常: {e}")

codes = tq.get_stock_list_in_sector('通达信88')[:100]
tq.subscribe_hq(stock_list=codes, callback=on_update)
while True:
    time.sleep(1)

财务选股（基本面筛选）

from tqcenter import tq
tq.initialize(__file__)

# 获取沪深300成份股
codes = tq.get_stock_list('23')

# 获取最新财务数据：ROE(FN197)、营收增长率(FN183)、资产负债率(FN210)
fd = tq.get_financial_data_by_date(
    stock_list=codes,
    field_list=['Fn197', 'Fn183', 'Fn210'],
    year=0, mmdd=0  # 0表示最新
)

# 筛选 ROE > 15% 且 营收增长 > 10% 且 资产负债率 < 60%
selected = []
for code, data in fd.items():
    try:
        roe = float(data.get('FN197', 0))
        rev_growth = float(data.get('FN183', 0))
        debt_ratio = float(data.get('FN210', 100))
        if roe > 15 and rev_growth > 10 and debt_ratio < 60:
            selected.append(code)
    except (ValueError, TypeError):
        continue

print(f"筛选出 {len(selected)} 只股票")
tq.send_user_block(block_code='CWXG', stocks=selected, show=True)

行业轮动选股

from tqcenter import tq
tq.initialize(__file__)

# 获取所有行业板块
sectors = tq.get_stock_list('16', list_type=1)  # 研究行业一级

# 获取各板块近5日涨幅，筛选前3
import pandas as pd
sector_perf = {}
for s in sectors:
    code = s['Code']
    try:
        bk = tq.get_bkjy_value(stock_list=[code], field_list=['BK1'], start_time='', end_time='')
        if code in bk and not bk[code].empty:
            sector_perf[code] = float(bk[code].iloc[-1].get('BK1', 0))
    except Exception:
        continue

top3 = sorted(sector_perf, key=sector_perf.get, reverse=True)[:3]

# 从前3行业中获取成份股
all_codes = []
for sec in top3:
    all_codes.extend(tq.get_stock_list_in_sector(sec))

# ... 对 all_codes 做进一步选股/回测 ...

交易下单（实盘/模拟盘）

from tqcenter import tq
from tqcenter import tqconst
tq.initialize(__file__)

# 第一步：获取账户句柄（所有交易操作的前置步骤）
myAccount = tq.stock_account(account="1190008847", account_type="STOCK")
if myAccount < 0:
    print("获取账户句柄失败，请检查客户端是否已登录")
    exit()

# 第二步：查询可用资金
asset = tq.query_stock_asset(account_id=myAccount)
cash = float(asset['Cash'])
print(f"可用资金: {cash:.2f}")

# 第三步：查询当前持仓
positions = tq.query_stock_positions(account_id=myAccount)
pos_dict = {p['Code']: p for p in positions}

# 第四步：下单买入
order_res = tq.order_stock(
    account_id=myAccount,
    stock_code="688318.SH",
    order_type=tqconst.STOCK_BUY,
    order_volume=200,
    price_type=tqconst.PRICE_MY,
    price=160.0
)
print(order_res)
# Value=1: 实盘待确认  Value=2: 模拟盘直接成功  Value=0: 失败

# 第五步：查询委托状态
orders = tq.query_stock_orders(account_id=myAccount, stock_code="688318.SH")
for o in orders:
    print(f"委托编号:{o['Wtbh']} 状态:{o['Status']} 成交:{o['CjVol']}/{o['WtVol']}")

# 第六步：撤单（如需要）
if orders:
    cancel_res = tq.cancel_order_stock(
        account_id=myAccount,
        stock_code=orders[0]['Code'],
        order_id=orders[0]['Wtbh']
    )
    print(cancel_res)

选股信号 + 自动下单（完整流程）

from tqcenter import tq
from tqcenter import tqconst
tq.initialize(__file__)

# 获取账户
myAccount = tq.stock_account(account="", account_type="STOCK")  # 空=当前登录账户

# 查可用资金
asset = tq.query_stock_asset(account_id=myAccount)
cash = float(asset['Cash'])

# 查当前持仓，避免重复买入
positions = tq.query_stock_positions(account_id=myAccount)
held_codes = {p['Code'] for p in positions if float(p['TotalVol']) > 0}

# 选股逻辑（示例：MACD金叉）
codes = tq.get_stock_list_in_sector('通达信88')
buy_list = []
for code in codes:
    if code in held_codes:
        continue
    tq.formula_set_data_info(stock_code=code, stock_period='1d', count=50, dividend_type=1)
    result = tq.formula_zb(formula_name='MACD', formula_arg='12,26,9')
    dif = result['Data']['DIF']
    dea = result['Data']['DEA']
    if len(dif) >= 2 and dif[-2] < dea[-2] and dif[-1] > dea[-1]:
        buy_list.append(code)

# 等额分配资金下单
if buy_list:
    per_stock_cash = cash / len(buy_list)
    for code in buy_list:
        snap = tq.get_market_snapshot(stock_code=code)
        price = float(snap['Now'])
        volume = int(per_stock_cash / price / 100) * 100  # 整手
        if volume >= 100:
            res = tq.order_stock(
                account_id=myAccount,
                stock_code=code,
                order_type=tqconst.STOCK_BUY,
                order_volume=volume,
                price_type=tqconst.PRICE_MY,
                price=price
            )
            print(f"下单 {code}: {volume}股 @ {price}, 结果: {res}")

常用辅助函数

详见 market-data.md 中的 price_df 和 get_full_tick 章节。

• tq.price_df(df, 'Close', column_names=codes) — 将 get_market_data 返回值转为宽表（行=日期, 列=股票代码）
• tq.get_full_tick(stock_code) — 获取完整 tick 数据，返回字段与 get_market_snapshot 类似，常用于订阅回调

关键约束与常见陷阱

交易下单

• 必须先获取句柄： 所有交易函数（order_stock/query_stock_asset/query_stock_positions/query_stock_orders/cancel_order_stock）调用前，必须先调用 tq.stock_account() 获取 account_id，且返回值 ≥ 0 才有效（0 也是有效句柄，< 0 才是失败）
• 实盘 vs 模拟盘返回值不同： order_stock 返回 Value=1 表示实盘待用户确认（非失败），Value=2 表示模拟盘直接成功，Value=0 才是失败。代码中判断下单结果时注意区分
• 实盘自动下单需券商开通： 默认实盘下单会弹窗让用户确认，要实现全自动交易需联系券商开通支持 TQ 的版本
• A股 T+1 限制： 今日买入的股票 CanUseVol 为 0，当日不可卖出。卖出前应检查 query_stock_positions 返回的 CanUseVol
• 委托数量必须为整手（100股）： 科创板/北交所最低 200 股，买入时需按 int(cash / price / 100) * 100 取整
• 委托查询仅限当日： query_stock_orders 只能查当日委托，无法查历史委托
• 返回值均为字符串： query_stock_asset/query_stock_positions 返回的数值字段（Balance、Cash、Cbj、TotalVol 等）均为 str 类型，计算前需 float() 或 int() 转换

数据获取

• get_market_data 单次最多 24000 条，分钟线需分批获取
• get_market_data 返回 {field: DataFrame(行=时间, 列=股票代码)}，已经是常规格式。tq.price_df() 用于从返回的 dict 中提取单个字段为独立 DataFrame，并自动处理日期索引和缺失值
• subscribe_hq 最多订阅 100 条
• send_warn 的 reason_list 每个元素最多 25 汉字；注意参数名是 volum_list（非 volume_list）
• 复权类型有两套写法：行情 API 用字符串 'none'/'front'/'back'；公式 API 用整数 0/1/2
• 周期：1m 5m 15m 30m 1h/60m 1d 1w 1mon 1q 1hy 1y tick

数据可用性陷阱

• get_more_info 的资金流字段（Zjl 主买净额、Zjl_HB 主力净流入、TotalBVol/TotalSVol 等）仅盘中实时有效，盘后返回值全为 0。如需历史资金流数据，目前 API 无直接支持
• get_gpjy_value 的部分 GP 字段（如 GP02 龙虎榜、GP06 陆股通等）需要在通达信客户端手动下载对应数据包后才能返回有效值，否则返回 None。使用前建议先用单只股票测试字段是否有数据

公式调用陷阱

• formula_set_data / formula_set_data_info 的 count 最大 24000
• 批量公式（formula_process_mul_xg/zb）无需提前 set_data
• count 与 end_time 互斥（重要）： formula_process_mul_xg/zb 中，当 count > 0 时，end_time 参数会被忽略，实际返回的是最新的 N 根 K 线的计算结果。如需指定时间范围，必须设 count=0 并同时传 start_time + end_time
• 批量 vs 单只返回结构不同（重要）：
  • 单只调用 formula_zb 返回：{'Data': {'指标名': [值列表]}}
  • 单只调用 formula_xg 返回：{'Data': {'条件名': [值列表]}}
  • 批量调用 formula_process_mul_zb 返回：{stock_code: {'指标名': [值列表]}}
  • 批量调用 formula_process_mul_xg 返回：{stock_code: {'条件名': [值列表]}}
  • 注意：批量返回的外层 key 是股票代码，不是 'Data'
• 条件选股 vs 技术指标公式区分： 通达信公式分为技术指标（formula_zb）和条件选股（formula_xg）两种类型。区分方法：技术指标公式输出连续数值序列（如 MACD 输出 DIF/DEA/MACD），条件选股公式输出 0/1 布尔信号。如果不确定公式类型，可以先用 formula_zb 尝试，如果报错或返回空再换 formula_xg
• SMA 收敛问题： 使用含 SMA（递归移动平均）的指标（如 KDJ、RSI、WR 等）进行批量计算时，count 设置过小会导致初始值不准确。经验参考：日线建议 count ≥ 250（约一年交易日），周线建议 count ≥ 120。如果对精度要求高，可设 count=-1 从头计算，但会显著增加耗时

故障排除

现象	原因与解决
"TQ数据接口初始化失败"	确保通达信客户端已运行并登录；用 sys.path.insert(0, ...) 而非 append()；确认 PYPlugins/ 下有 TPyth.dll 和 TPythClient.dll
ErrorId='12'	已有同名策略运行（同一 __file__ 路径），关闭旧实例或换文件名
菜单一直显示"正在开启TQ策略..."	检查是否有防火墙弹窗，允许访问即可
指标值前面是 None	count 不够覆盖公式最大回溯参数，如 MA(C,20) 至少需要 20 根 K 线
send_warn 报 ValueError	确认 count > 0；确认各 list 长度 ≥ count；确认 price_list/close_list/volum_list 元素为纯数字字符串
批量公式选股结果比客户端少	count 太小，增大到覆盖公式所需的最大回溯 K 线数
Module Not Found	确认 PYPlugins/user 路径正确，使用绝对路径

API 参考（按需加载）

查询具体接口参数和字段时，参见对应文件：

• 行情与基础数据 — get_market_data, get_market_snapshot, get_stock_info, get_more_info 等
• 财务与交易数据 — get_financial_data(FN1-584), get_gpjy_value(GP01-46), get_bkjy_value, get_scjy_value, get_gp_one_data 等
• 板块管理与公式调用 — get_stock_list, get_stock_list_in_sector, send_user_block, formula_zb/xg/exp, formula_process_mul 等
• 交易下单 — stock_account, order_stock, query_stock_asset, query_stock_positions, query_stock_orders, cancel_order_stock 等
• 订阅与通知 — subscribe_hq, send_warn, send_message, send_file, print_to_tdx 等
• 常量枚举 — 市场后缀, 复权类型, K线周期, market参数值