简介TradeX-TPS 是一套面向股票量化交易与自动化操作场景的交易插件适配大智慧、通达信两大行情软件通过 TradeX.dll 或 Trade.dll 连接券商服务器实现买入、卖出、撤单、按预设比例清仓以及账户信息、持仓与可撤单信息的查询并可接入股票池与预警公式系统适合有一定软件使用基础、希望把交易动作脚本化的个人投资者与量化爱好者。资源包共 20 个文件约 14.71MB以 9 个 dll 动态库为核心配合 ini 配置、uid 授权、lic 许可、xml 参数、pdf 用户指南与 exe 服务端程序另附大智慧、通达信安装说明文本覆盖运行、配置与授权各环节。目前已有 3079 人学习下载。标准版免费读者可据此搭建本地交易通道理解插件与券商接口的对接方式并借助用户指南完成安装调试与常见问题排查。1. 交易插件 TradeX-TPS把行情终端的下单动作接进自己的策略里如果你每天盯盘用的是大智慧或通达信同时又有一套自己写的信号系统迟早会撞上同一个问题信号出来了手却还要切窗口、敲代码、填价格、点确认。TradeX-TPS 这类交易插件要解决的就是这段“最后一公里”。它本质是一个跑在本地的中间层向上对接行情终端的公式或预警输出向下对接券商柜台的下单接口把“看到信号”和“发出委托”之间的手工动作压缩成一次自动调用。适合两类人一是已经有稳定信号、想减少手动延迟的量化爱好者二是想把终端里的自选股、预警条件直接变成可执行指令的从业者。它不神秘但坑集中在接口适配和风控边界上。2. TradeX-TPS 的三种接入姿势公式桥、预警钩子、外部进程2.1 先想清楚你的信号从哪来大智慧和通达信本身是封闭的行情客户端它们不提供官方的通用交易 API。所以 TradeX-TPS 这类插件的第一件事是决定“信号怎么出来”。常见做法有三条路选哪条决定了后面所有代码的形态。第一条是公式桥。在终端里写一个自定义公式当条件满足时公式把股票代码、方向、价格写到一个本地文件或内存映射区。TradeX-TPS 作为独立进程去读这个区域。优点是改动最小不用碰终端安装目录缺点是轮询有延迟通常 200 毫秒到 1 秒且公式里能拿到的字段有限。第二条是预警钩子。利用终端自带的预警系统把预警结果输出到指定目录的文本文件。TradeX-TPS 监听文件变化。这条路比公式桥稳因为预警是终端原生功能不容易被版本更新打断但预警的触发频率和去重逻辑要自己控制否则同一只票会重复下单。第三条是外部进程直接算。完全不依赖终端公式自己用行情接口拉数据、算信号TradeX-TPS 只负责下单。灵活度最高但你就得自己维护行情源终端只剩看盘作用。我一般会推荐第二条给刚上手的人预警文件是纯文本格式固定调试时肉眼就能看出问题容易定位。2.2 一个最小可跑的预警文件监听器下面这段 Python 演示 TradeX-TPS 侧最核心的一环监听预警目录解析出新信号做基本去重后交给下单函数。它不依赖任何特定券商接口下单部分用伪代码占位你替换成自己的柜台调用即可。import os import time import json from collections import OrderedDict WATCH_DIR r./alert_out # 终端预警输出目录 PROCESSED OrderedDict() # 记录已处理信号防重复 MAX_KEEP 500 # 去重表上限防止内存无限涨 def parse_alert(line: str): 解析一行预警文本格式约定代码|方向|价格|时间 parts line.strip().split(|) if len(parts) ! 4: return None code, side, price, ts parts # 只接受合法方向和正数价格 if side not in (BUY, SELL): return None try: price float(price) except ValueError: return None if price 0: return None return {code: code, side: side, price: price, ts: ts} def signal_key(sig): 用代码方向分钟级时间做去重键避免同分钟重复触发 return f{sig[code]}_{sig[side]}_{sig[ts][:16]} def place_order(sig): 下单占位替换成你的柜台调用 print(f[ORDER] {sig[side]} {sig[code]} {sig[price]}) # 真实场景调用券商API检查返回码失败写日志 def main(): if not os.path.isdir(WATCH_DIR): os.makedirs(WATCH_DIR) while True: for fname in os.listdir(WATCH_DIR): fpath os.path.join(WATCH_DIR, fname) if not fname.endswith(.txt): continue with open(fpath, r, encodinggbk, errorsignore) as f: for line in f: sig parse_alert(line) if not sig: continue key signal_key(sig) if key in PROCESSED: continue PROCESSED[key] time.time() if len(PROCESSED) MAX_KEEP: PROCESSED.popitem(lastFalse) place_order(sig) # 处理完重命名避免下一轮重复读 os.rename(fpath, fpath .done) time.sleep(0.5) if __name__ __main__: main()逻辑说明parse_alert做了严格的字段校验这是血泪经验——终端预警文件偶尔会写入半行或空行不校验就会把脏数据送进下单函数。signal_key用分钟级时间做去重是因为预警系统在行情跳动快时可能对同一条件连续触发多次。PROCESSED用OrderedDict而不是普通 dict是为了在超过上限时按插入顺序淘汰最老的记录控制内存。os.rename把处理过的文件改名比维护已读偏移量更简单重启进程也不会重复消费。参数说明WATCH_DIR必须和终端预警设置里的输出目录一致注意终端可能用 GBK 编码写文件所以读取时指定encodinggbk。time.sleep(0.5)是轮询间隔太小浪费 CPU太大增加延迟0.5 秒对日线级别信号足够。MAX_KEEP根据你一天的最大信号量调整一般 500 够用。2.3 下单前的三道风控闸门信号解析出来不等于可以直接发单。TradeX-TPS 这类插件最容易翻车的地方就是把风控完全交给券商柜台。我一般会在插件里加三道闸门顺序执行任何一道不过就丢弃并写日志。第一道是单笔金额上限。用当前价格乘以委托数量超过阈值直接拒绝。第二道是当日累计委托次数。有些策略在震荡行情里会疯狂触发必须有个硬上限。第三道是标的白名单。只允许预设的股票池里的代码通过防止预警文件被误写或串号。MAX_AMOUNT 50000 # 单笔最大金额 MAX_ORDERS_PER_DAY 20 # 当日最大委托次数 WHITELIST {600000, 000001, 300750} # 示例白名单 order_count 0 def risk_check(sig, qty): global order_count amount sig[price] * qty if amount MAX_AMOUNT: return False, 单笔金额超限 if order_count MAX_ORDERS_PER_DAY: return False, 当日委托次数超限 if sig[code] not in WHITELIST: return False, 标的不在白名单 return True, ok这三道闸门不复杂但能挡掉绝大多数“手滑”和“信号抖动”导致的意外委托。注意order_count要在实际发单成功后才递增而不是在检查通过时递增否则失败的单子也会占用次数。3. 把 TradeX-TPS 接到券商柜台参数、字段与失败重试3.1 柜台接口的字段映射表不同券商的柜台接口字段名差异很大TradeX-TPS 的适配层核心工作就是做映射。下面这张表是我在多个项目里总结的常见字段对应关系左列是插件内部统一字段右列是柜台侧常见叫法。你拿到具体接口文档后按这个结构填一遍适配层就成型了。内部字段含义柜台常见字段名注意事项code证券代码stock_code / symbol注意带不带市场前缀side买卖方向entrust_bs / direction有的用 1/2有的用 B/Sprice委托价格entrust_price / price单位是元还是分要确认qty委托数量entrust_amount / volume股数不是手数price_type价格类型price_type限价/市价枚举不同映射层建议用配置字典而不是硬编码这样换柜台时只改配置不改逻辑。FIELD_MAP { code: stock_code, side: entrust_bs, price: entrust_price, qty: entrust_amount, price_type: price_type, } SIDE_MAP {BUY: 1, SELL: 2} # 按实际柜台枚举调整 def to_counter(sig, qty): return { FIELD_MAP[code]: sig[code], FIELD_MAP[side]: SIDE_MAP[sig[side]], FIELD_MAP[price]: sig[price], FIELD_MAP[qty]: qty, FIELD_MAP[price_type]: 0, # 0 通常代表限价 }逻辑说明FIELD_MAP把内部字段名和柜台字段名解耦SIDE_MAP处理方向枚举差异。to_counter输出一个扁平字典直接传给柜台 SDK 的下单函数。参数说明price_type的取值必须查你所用柜台的文档不同柜台对限价、市价、最优五档的枚举完全不同填错会导致委托被拒或变成市价单。3.2 失败重试与幂等处理柜台调用失败是常态网络抖动、柜台忙、字段校验不通过都会返回错误。TradeX-TPS 必须有重试机制但重试的前提是幂等——同一笔委托不能因为重试变成两笔。常见做法是给每笔委托生成一个本地唯一编号写进日志重试时带上同一个编号。如果柜台接口支持“委托编号”字段就把它传进去如果不支持就在重试前先查询一次当日委托确认没有已成交的同编号单子。import uuid import time def send_with_retry(counter_req, max_retry3): req_id str(uuid.uuid4())[:8] counter_req[local_id] req_id for attempt in range(max_retry): try: resp counter_api.place(counter_req) # 伪代码 if resp.get(code) 0: return True, req_id # 柜台明确拒绝不重试 if resp.get(code) in (1001, 1002): return False, resp.get(msg) except Exception as e: print(f第{attempt1}次失败: {e}) time.sleep(0.3 * (attempt 1)) # 退避 return False, 重试耗尽逻辑说明local_id是幂等键重试时保持不变。柜台返回明确业务错误码时不重试因为重试也不会成功只会浪费次数。退避时间用0.3 * (attempt 1)递增避免连续冲击柜台。参数说明max_retry设 3 次足够再多会拉长信号到委托的延迟反而失去插件意义。3.3 用日志把黑匣子打开TradeX-TPS 出问题时最怕的是“信号发了但没下单”或“下重复了”却查不到原因。我的习惯是三个日志分开写信号日志记录每一条解析出的信号原文和去重结果风控日志记录每一条被拒绝的信号及原因委托日志记录每一次柜台调用的请求、响应和耗时。三个日志都用追加模式按天切分。import logging from logging.handlers import TimedRotatingFileHandler def setup_logger(name, filename): logger logging.getLogger(name) logger.setLevel(logging.INFO) handler TimedRotatingFileHandler(filename, whenmidnight, backupCount7) handler.setFormatter(logging.Formatter(%(asctime)s %(message)s)) logger.addHandler(handler) return logger sig_log setup_logger(signal, logs/signal.log) risk_log setup_logger(risk, logs/risk.log) order_log setup_logger(order, logs/order.log)参数说明whenmidnight每天零点切分backupCount7保留一周。日志目录要提前建好否则 handler 初始化会报错。信号日志里建议把原始行也记下来方便和终端预警文件对照。4. 避坑与排查TradeX-TPS 落地时最容易翻车的五件事4.1 预警文件编码不对导致解析全空现象监听器跑起来后日志里一条信号都没有但预警目录里明明有文件。原因大智慧或通达信在某些系统区域设置下用 GBK 写文件而 Python 默认用 UTF-8 读split(|)之前就已经乱码了。解决读取时显式指定encodinggbk并且加errorsignore防止个别字节报错中断整个循环。如果还是空用十六进制查看器打开文件确认分隔符是不是全角竖线。4.2 同一信号重复下单现象一只票在几分钟内被连续买入多次。原因预警系统在条件持续满足时会反复触发而监听器只按文件去重没按信号内容去重。解决像 2.2 节那样用代码方向分钟级时间做去重键并且把去重表持久化到本地文件进程重启后也能恢复。注意去重键的时间粒度不要太细秒级会导致同一分钟内的正常多次信号被误杀。4.3 柜台返回“委托已接受”但实际未成交现象日志显示下单成功但持仓没变化。原因很多柜台接口的“成功”只代表委托已受理不代表已成交。限价单可能一直挂着或者被交易所拒绝。解决TradeX-TPS 不能只看下单返回码要增加一个成交查询循环每隔几秒查一次该委托的状态直到变成“已成”“部成”“已撤”为止。查询超时后写告警日志人工介入。4.4 价格字段单位搞错现象委托价格变成实际价格的 100 倍或 1/100。原因柜台接口有的用元有的用分文档里不一定写清楚。解决在适配层加一个断言委托价格必须在当日涨跌停范围内超出就拒绝并打日志。这个断言能挡住绝大多数单位错误。另外首次对接时先用一手、低价股做一笔真实委托验证别拿重仓股试。4.5 终端升级后预警目录结构变化现象某天终端自动更新后插件突然不工作了。原因终端升级可能改变预警输出路径、文件名规则或字段顺序。解决把预警目录和文件匹配规则做成配置项不要硬编码。同时监听器启动时先打印一次当前目录下的文件列表和首行内容方便快速判断格式是否变了。如果终端支持关闭自动更新手动控制升级时间。5. 进阶用状态机管理委托生命周期而不是一锤子买卖前面几章把信号到委托的链路跑通了但真实交易里一笔委托从发出到终结有好几个状态已报、部成、已成、已撤、废单。如果 TradeX-TPS 只负责“发出去”不管后续就会出现挂单一直占着资金、部分成交后策略不知道实际持仓、撤单时机错过等问题。我后来习惯在插件里加一个轻量状态机用一张本地表跟踪每笔委托的当前状态策略层只读这张表来做后续决策。状态机的核心是一个字典键是本地委托编号值包含柜台委托号、当前状态、已成交数量、最后更新时间。一个后台线程每隔固定间隔查询未终结的委托更新状态。状态迁移规则很简单已报可以变成部成、已成、已撤、废单部成可以变成已成或已撤已成和已撤是终态不再查询。import threading import time TERMINAL {已成, 已撤, 废单} orders {} # local_id - {counter_id, status, filled, updated} def poll_orders(): while True: for lid, info in list(orders.items()): if info[status] in TERMINAL: continue resp counter_api.query(info[counter_id]) # 伪代码 if not resp: continue info[status] resp[status] info[filled] resp[filled] info[updated] time.time() if info[status] in TERMINAL: order_log.info(f{lid} 终结 {info[status]} 成交{info[filled]}) time.sleep(2) threading.Thread(targetpoll_orders, daemonTrue).start()逻辑说明TERMINAL集合定义终态终态不再查询减少柜台压力。list(orders.items())在遍历时复制一份避免查询过程中字典被其他线程修改导致运行时错误。参数说明time.sleep(2)是查询间隔太短会触发柜台限流太长会导致状态更新滞后。对于日内策略2 秒可以接受对于高频场景需要改用柜台推送而不是轮询。状态机带来的好处是策略层可以问“我现在实际有多少可用资金、多少持仓”而不是靠猜。比如一个买入信号来了先查状态机里有没有未终结的买单占用资金有就跳过。这个判断能避免超买。另一个用法是撤单当信号反转时先查状态机找到对应的未终结委托发撤单请求等状态变成已撤后再发反向委托避免撤单和反向单之间的时间差导致仓位暴露。还有一个细节是部分成交的处理。如果一笔买单部成 300 股、还剩 700 股挂着信号消失需要撤单撤单后实际持仓是 300 股。状态机里的filled字段就是给策略层用的策略应该按filled而不是原始委托数量来更新自己的仓位视图。这个区别在震荡行情里很关键我见过不止一次因为按委托数量记账导致仓位对不上最后手动平仓才发现多了几百股。最后说一个验证方法在模拟环境里故意制造部分成交和撤单观察状态机的迁移日志是否和柜台查询结果一致。连续跑一周模拟状态机的记录和柜台对账单能对上再上实盘。这个习惯帮我省过不少后悔药。希望帮到你。本文还有配套的精品资源点击获取