☰
TradeX-TPS交易插件:从行情信号到券商下单的自动化实践
2026/10/14 8:38:07 网站建设 项目流程

简介: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", encoding="gbk", errors="ignore") 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(last=False) 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 编码写文件,所以读取时指定encoding="gbk"。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/S
price委托价格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_retry=3): 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"第{attempt+1}次失败: {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, when="midnight", backupCount=7) 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")

参数说明:when="midnight"每天零点切分,backupCount=7保留一周。日志目录要提前建好,否则 handler 初始化会报错。信号日志里建议把原始行也记下来,方便和终端预警文件对照。

4. 避坑与排查:TradeX-TPS 落地时最容易翻车的五件事

4.1 预警文件编码不对导致解析全空

现象:监听器跑起来后,日志里一条信号都没有,但预警目录里明明有文件。原因:大智慧或通达信在某些系统区域设置下用 GBK 写文件,而 Python 默认用 UTF-8 读,split("|")之前就已经乱码了。解决:读取时显式指定encoding="gbk",并且加errors="ignore"防止个别字节报错中断整个循环。如果还是空,用十六进制查看器打开文件确认分隔符是不是全角竖线。

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(target=poll_orders, daemon=True).start()

逻辑说明:TERMINAL集合定义终态,终态不再查询,减少柜台压力。list(orders.items())在遍历时复制一份,避免查询过程中字典被其他线程修改导致运行时错误。参数说明:time.sleep(2)是查询间隔,太短会触发柜台限流,太长会导致状态更新滞后。对于日内策略,2 秒可以接受;对于高频场景,需要改用柜台推送而不是轮询。

状态机带来的好处是策略层可以问“我现在实际有多少可用资金、多少持仓”,而不是靠猜。比如一个买入信号来了,先查状态机里有没有未终结的买单占用资金,有就跳过。这个判断能避免超买。另一个用法是撤单:当信号反转时,先查状态机找到对应的未终结委托,发撤单请求,等状态变成已撤后再发反向委托,避免撤单和反向单之间的时间差导致仓位暴露。

还有一个细节是部分成交的处理。如果一笔买单部成 300 股、还剩 700 股挂着,信号消失需要撤单,撤单后实际持仓是 300 股。状态机里的filled字段就是给策略层用的,策略应该按filled而不是原始委托数量来更新自己的仓位视图。这个区别在震荡行情里很关键,我见过不止一次因为按委托数量记账导致仓位对不上,最后手动平仓才发现多了几百股。

最后说一个验证方法:在模拟环境里故意制造部分成交和撤单,观察状态机的迁移日志是否和柜台查询结果一致。连续跑一周模拟,状态机的记录和柜台对账单能对上,再上实盘。这个习惯帮我省过不少后悔药。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询