简介THSTrader源代码包是一套面向量化交易研究者的Python交易系统实现覆盖行情获取、订单管理、策略回测与风险控制等核心环节适合有一定Python基础的开发者用来学习国内证券期货市场程序化交易的工程组织方式。压缩包共三百一十三个文件除两百九十九张用于演示界面效果的截图外还包含四个Python模块、三个Jupyter Notebook、两份Git忽略文件、许可证、说明文档及数据样本便于边查看运行效果边对照代码整包约三百四十七KB结构轻量紧凑。该资源已有一千一百七十三人浏览学习。阅读源码可以深入理解事件驱动模型、柜台API对接、行情数据解析与存储、多线程并行调度、订单生命周期管理以及日志与配置管理的实现细节同时验证码识别与数据集生成示例也能帮助体会实盘环境中的自动化处理思路。借助Notebook读者可以快速复现关键流程为后续自定义策略和性能优化打下基础。1. thstrader 是什么为什么要读它的源代码在量化交易里真正难的不是策略而是“通道”。同花顺客户端本身不带开放的编程接口而 thstrader 这个项目用 Python 重新实现了一个能连接同花顺本地交易服务的客户端把协议编解码、会话管理、行情订阅和委托下单全部摊开在源代码里。读它等于把券商柜台到本地客户端的这半条链路完整看了一遍。对做量化的 Python 工程师来说这份源代码比任何文档都有价值。它解决的核心问题是不依赖闭源的 DDE 或插件机制直接用 socket 与同花顺的本地服务通信实现行情、交易和账目查询。适合两类人一类是想把现有策略接到同花顺上跑实盘的人另一类是研究证券交易协议、想理解“盘口数据是怎么从行情服务到策略端”的人。本文不讲理论框架直接按源码结构把协议层、行情层、交易层讲透然后给一个能改的模板。2. 从 thstrader 源码拆出核心模块协议、行情、交易三层2.1 先读目录thstrader 的代码并不长拿到源码先整体看一遍目录。一个典型 thstrader 项目在社区流传的版本里结构大致是这样的ths_trader/ ├── bin/ │ └── main.py # 启动入口解析命令行参数 ├── config/ │ └── trading.json # 账号、服务器地址、交易参数 ├── core/ │ ├── client.py # 客户端主类对外暴露接口 │ ├── tcp.py # 底层 socket 通信与封包解析 │ ├── quote.py # 行情订阅与快照 │ ├── trader.py # 委托下单、撤单、查询 │ └── log.py # 日志与回执确认 ├── requirements.txt └── README.md先读core/client.py它决定了整个模块怎么协作# core/client.py 结构示意 class THSClient(object): def __init__(self, cfg): self.tcp TcpTransport(cfg[host], cfg[port]) self.quote QuoteService(self.tcp) self.trader TradeService(self.tcp) def login(self, user, pwd): resp self.tcp.request(login, useruser, pwdpwd) self.tcp.set_session(resp[sid]) return resp这段代码的逻辑说明核心是把三种职责拆成三个模块——TcpTransport只负责字节流的收发与封包QuoteService只维护行情订阅关系TradeService只处理委托结果。注意client.py中login成功后会把会话 ID 写回TcpTransport这是整个源码里最容易忽略的一行后面的行情、下单都要带着这个会话 ID否则服务端直接拒绝请求。还有一个细节值得留意社区流传的 thstrader 版本很多拿到代码后先确认tcp.py里封包长度字段是I还是H这决定了它兼容的是旧版还是新版协议直接决定你能不能连上当前安装的同花顺客户端。提示拿到源码后不要先去看bin里的启动脚本先读core/client.py它决定了整个模块怎么协作。2.2 通信层为什么 thstrader 直接读 socket 而不是引第三方 SDKthstrader 的协议层核心在tcp.py。它不依赖任何第三方 SDK而是直接通过 socket 连接本机同花顺客户端开启的本地端口。这种设计的原因在于同花顺本地服务是一个常驻进程它的接口是一个私有协议公开的 SDK 并不会提供。用 socket 直连本地服务是社区实现惯用的方案。协议层需要解决的最小问题是封包格式。常见的封装是 4 字节长度前缀 1 字节命令字 2 字节序列号 JSON 数据体。长度指整个封包字节数网络字节序按大端处理import socket import struct import json class TcpTransport: HEADER struct.Struct(IBH) # 长度 命令字 序列号 def __init__(self, host, port, timeout10): self.sock socket.create_connection((host, port), timeouttimeout) self.sock.settimeout(timeout) self.seq 0 def request(self, cmd, **params): self.seq (self.seq 1) 0xFFFF body json.dumps({cmd: cmd, seq: self.seq, **params}).encode(utf-8) packet self.HEADER.pack(len(body) 7, cmd, self.seq) body self.sock.sendall(packet) length, cmd_byte, seq self._read_header() payload self._recv_exact(length - 7) return json.loads(payload.decode(utf-8))参数说明HEADER用到了 Pythonstruct的IBHI为无符号 4 字节长度B为命令字H为序列号。packet的长度是len(body) 7因为 header 占用 7 字节。这里有一个容易出错的点命令字cmd必须是 0 到 255 之间的整数因此项目里往往用 0x01、0x02 这类枚举而不是字符串。请求发出后必须同步读回包因为本地服务不缓存请求读超时会直接影响下一次请求。这个封包设计决定了 thstrader 与同花顺客户端的每一个交互都要遵循一问一答的时序。如果读完这段代码还不理解为什么心跳用独立线程可以直接搜tcp.py里time.sleep的调用位置通常都是在单开线程里做 30 秒一次的心跳。2.3 行情层订阅快照与分时的数据格式quote.py负责把行情请求封装成上面的协议格式。它有两个关键方法subscribe用于向服务端登记关注列表snapshot用于拉取某个代码的实时快照。两者的差别在于subscribe是持续推送snapshot是拉取一次。class QuoteService: def __init__(self, transport): self._t transport self._watchlist set() def subscribe(self, code, market1): # market: 1 表示沪市0 表示深市 self._watchlist.add(code) resp self._t.request(0x10, codecode, marketmarket, freq0) if resp.get(code) ! 0: raise RuntimeError(f订阅失败: {code} - {resp}) return resp def snapshot(self, code, market1): resp self._t.request(0x11, codecode, marketmarket) return { last: resp[price], bid: resp[bid1], ask: resp[ask1], vol: resp[volume], ts: resp[timestamp], }逻辑说明subscribe返回的code字段与服务端约定一致0 表示成功。源码里freq0表示分时级推送若改成 1 则进入盘口逐笔注意逐笔推送对接收线程的处理能力要求更高。拉取快照时返回的字段名各版本不太一致常见字段是last、bid1、ask1读取前最好先打印一次resp看一下原始字段名避免盲目按文档取数。这段代码也揭示了一个重要边界thstrader 的行情是本地转发而不是直接连交易所行情数据到达延迟取决于同花顺客户端本身的刷新频率。2.4 交易层下单前必须完成的三个校验trader.py比quote.py复杂因为它涉及资金、持仓和订单状态。社区版 thstrader 的订单接口普遍需要至少三个校验可用资金检查、现价区间检查、最小交易数量检查class TradeService: MIN_QTY 100 # A股一手 PRICE_TICK 0.01 # 最小变动价位 def __init__(self, transport): self._t transport def buy(self, code, price, qty, market1): qty int(qty) if qty % self.MIN_QTY ! 0: raise ValueError(f{code} 买入数量需为 100 的整数倍当前 {qty}) if price 0 or round(price, 2) ! price: raise ValueError(price 必须为正数且最多两位小数) resp self._t.request(0x20, op0x01, codecode, priceprice, qtyqty, marketmarket) return {entrust_no: resp[entrust_no], status: resp[status]}参数说明op0x01表示买入卖出在源码里一般对应op0x02entrust_no是委托编号后续撤单必须以它为凭证。最重要的是qty的校验A股按手交易但同一份代码用在可转债上允许 10 张起购直接把MIN_QTY改成一个常量是社区版最常见的硬编码缺陷。如果你要把这段源码用于可转债需要把最小交易量改成按品种动态判断而不是全局常量。读交易层时建议把买入、卖出、撤单、查询持仓四个方法并列在屏幕上逐行对照它们的协议格式几乎一致只有op字段不同。理解了这一点后续在 gateway 层加一个统一的异常处理就很容易。3. 用 thstrader 代码在本地跑通最小行情与交易流程3.1 最小依赖安装与登录在把源码改造成策略之前先建立一个能本地运行的最小环境。thstrader 的依赖很少Python 3.8 以上基本只依赖标准库开发调试时可以加装pytest和requests前者用于写测试后者用于策略成交后的回调通知cd ths_trader python -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install -r requirements.txt python bin/main.py --config config/trading.json --probe--probe是源码自带的一个探测模式用来验证本地交易客户端端口是否在监听。执行后如果能输出端口号和版本信息说明本地服务可达。这里有一个最常见的失败场景本地同花顺客户端没有登录或者登录后关闭了自动交易服务TCP 连接会直接拒绝。probe模式本质上就是先connect再立即断开不产生任何业务报文。# bin/main.py --probe 关键片段 def probe(host, port): s socket.create_connection((host, port), timeout5) s.close() print(f[OK] {host}:{port} 端口可连接)逻辑说明probe不是简单的 ping它验证的是本地服务是否真正处于可接收连接的状态。如果端口通但probe失败常见原因是本机开了多个同花顺客户端连接落到了旧进程上。开工前把所有同花顺进程全部退出只保留一个最新客户端能避免一大批诡异问题。3.2 订阅自选股并输出实时报价登录之后下一步是订阅行情。把账号信息放入config/trading.json然后运行下面这段脚本它会在终端持续打印订阅标的的最新价、买一和卖一import time from core.client import THSClient cfg { host: 127.0.0.1, port: 8808, user: your_account, pwd: your_password, } client THSClient(cfg) client.login(cfg[user], cfg[pwd]) for code in [600519, 000001, 300750]: client.quote.subscribe(code) while True: for code in [600519, 000001, 300750]: snap client.quote.snapshot(code) print(f{code} last{snap[last]:.2f} bid{snap[bid]:.2f} , end) print(fask{snap[ask]:.2f} vol{snap[vol]}) time.sleep(3)逻辑说明subscribe只需要调用一次后续的snapshot才能拿到缓存值。这里的while True轮询周期设为 3 秒是因为 thstrader 的行情服务本身按秒级刷新轮询太快只是空耗 CPU。注意snapshot返回的vol单位是股而不是手打印时长整型即可。若想验证数据是不是实时变化对着行情软件里同一只股票看买一卖一即可。3.3 下一笔模拟买单与撤单行情通了之后验证交易闭环。先用很小的一笔比如 100 股测试买入流程成交后立刻撤单确认交易链路完整。如果你用的是模拟账户这一步不会产生真实资金变动entrust client.trader.buy(000001, price12.50, qty100) print(委托编号:, entrust[entrust_no]) print(委托状态:, entrust[status]) # 5 秒后查询委托若未成交则撤单 time.sleep(5) orders client.trader.query_orders() for order in orders: if order[entrust_no] entrust[entrust_no]: if order[status] 0: client.trader.cancel(entrust[entrust_no]) print(已撤单) else: print(部分或全部成交, status , order[status]) break提示这里的price12.50是硬编码的跑长周期脚本时必须先用snapshot读取当前最新价再按一定滑点偏移生成委托价否则会因为价格与盘口偏差过大被废单。这段代码的逻辑说明query_orders返回的status在不同版本里定义不一致常见的是用字符串0表示已报未成交1表示部分成交2表示全部成交。在把源码用于生产前务必先用一条模拟委托把状态枚举打印出来。另外一个关键点是cancel方法必须传入entrust_no也就是委托编号而不是股票代码这是社区用户特别容易搞混的参数。4. 把 thstrader 源码改造成自己的策略框架4.1 数据落盘先解决“盘中没记录”的痛点量化研究的第一步通常是积累数据。thstrader 的行情是订阅制若不主动保存收盘后什么都不会留下。常见做法是在订阅回调里把行情写入本地 CSV按日期分文件存储import csv from datetime import datetime class QuoteRecorder: def __init__(self, base_dirdata/): self.base_dir base_dir self.writers {} def write(self, code, snap): today datetime.now().strftime(%Y-%m-%d) filename f{self.base_dir}/{code}_{today}.csv if code not in self.writers: f open(filename, a, newline, encodingutf-8) writer csv.writer(f) self.writers[code] (f, writer) _, writer self.writers[code] writer.writerow([ snap[ts], snap[last], snap[bid], snap[ask], snap[vol] ])逻辑说明按日期分文件的好处是避免单个 CSV 文件越来越大回测时可以直接按日期切片。这里没有调用flush因为默认缓冲区足够应对秒级写入如果机器在盘中异常断电最多丢最后几行可接受。csv模块的writerow不会自动处理换行open时必须指定newline否则 Windows 上会出现空行这是排查落盘数据时最常见的坑。4.2 挂一个“MACD 底背离”自动买入任务社区里“MACD双底”这类条件单一直有热度。thstrader 改造为策略框架的核心不是重写指标而是把“订阅行情 → 计算指标 → 触发下单”串起来。下面是一个最小可用的 MACD 计算函数返回 DIF、DEA 和柱状值def macd(prices, fast12, slow26, signal9): # 标准 EMA 递推k 2 / (N 1) ema_fast, ema_slow [], [] for idx, p in enumerate(prices): if idx 0: ema_fast.append(p) ema_slow.append(p) else: kf 2 / (fast 1) ks 2 / (slow 1) ema_fast.append(p * kf ema_fast[-1] * (1 - kf)) ema_slow.append(p * ks ema_slow[-1] * (1 - ks)) dif [f - s for f, s in zip(ema_fast, ema_slow)] dea [] for idx, d in enumerate(dif): if idx 0: dea.append(d) else: kd 2 / (signal 1) dea.append(d * kd dea[-1] * (1 - kd)) hist [2 * (d - e) for d, e in zip(dif, dea)] return dif, dea, hist把指标计算挂进 thstrader 的轮询循环window [] while True: snap client.quote.snapshot(600519) window.append(snap[last]) if len(window) 120: window.pop(0) if len(window) 120: dif, dea, hist macd(window) # 检测最近两个低点的价格与 MACD 柱是否背离 if detect_divergence(window, hist): client.trader.buy(600519, snap[ask], 100) print(触发买入, snap[ask]) break time.sleep(3)参数说明window保存最近 120 个快照价约等于 6 分钟的分钟级数据。detect_divergence是你自己的判断函数返回True就触发买入买入价用卖一价snap[ask]以保证即时成交。这里用break结束进程避免同一次背离被重复触发。如果你要优化计算速度把macd函数替换成 numpy 向量化版本即可行情循环中每次重算 120 个样本的纯 Python 循环大约耗时几毫秒在单标的场景下完全可接受。4.3 三个必加的风控参数策略可以简单但风控不能省略。在 thstrader 源码的trader.py里加三个硬性检查可以在不改动协议层的条件下显著降低事故概率参数推荐值说明max_daily_buy3 次单日最大买入次数防止死循环反复下单max_position总资金 20%单标的最大持仓金额超出后 buy 直接抛异常min_sleep_sec1 秒两次委托之间最小间隔避免被本地服务限流# 在 trader.py 中插入的守护逻辑 class RiskGuard: def __init__(self, max_daily_buy3, max_position20000): self.max_daily_buy max_daily_buy self.max_position max_position self.daily_buy 0 def check(self, price, qty, today_orders): if self.daily_buy self.max_daily_buy: raise RuntimeError(超出当日买入次数上限) if price * qty self.max_position: raise RuntimeError(持仓金额超出上限) self.daily_buy 1参数说明max_daily_buy的计数在跨日时重置建议把重置逻辑放在订阅 tick 的time.time检查中。max_position的单位是金额而不是股数因为股价不同按金额限制才可比。min_sleep_sec的实现更简单在buy的入口记录time.time()两次调用间隔小于阈值就sleep补足。这三个参数组合起来能挡掉网络抖动导致重复下单这类高频事故。5. 读 thstrader 源码时最常见的 4 个调试问题5.1 端口通但连接立刻被断开用netstat确认监听端口后仍然连不上多数情况是本地同花顺客户端正在使用“极速交易”模式而本地服务端口没开。验证方法在同花顺客户端里重新登录一次交易账户再执行probe。若仍不行检查是否只有一个同花顺进程在跑多开进程会抢端口。另外调试时如果 IDE 提示“当前不会命中断点”先确认是不是连着之前启动的旧 Python 进程thstrader 的调试需要把整个解释器重启后重新连接不是简单刷新断点。5.2 订阅成功但没有行情推送subscribe返回成功后一直没有数据通常不是协议问题而是没有调用snapshot。thstrader 是“订阅 拉取”模型订阅只登记了关注列表真正取数要走snapshot。另外检查行情刷新频率设置如果客户端设置的是 5 秒刷新那snapshot返回的数据点和上一次一样是正常的。5.3 下单返回成功但持仓没变化委托返回了entrust_no不代表一定成交A股委托单可能挂在队列里。此时先查query_orders看status。如果status是“已报”说明没问题等待成交即可。如果状态是“废单”去查价格买入价必须落在当日涨跌停区间内并且是 0.01 元的整数倍。5.4 用 mock 对象在无行情环境验证策略逻辑做策略改造时不希望真连行情服务常见做法是给THSClient打补丁用一个返回固定值的 fake 对象替代quote和traderfrom unittest.mock import MagicMock client MagicMock() client.quote.snapshot.return_value {last: 12.5, bid: 12.49, ask: 12.51, vol: 12345} client.trader.buy.return_value {entrust_no: 20240001, status: 0} # 直接运行策略主循环观察触发逻辑是否按预期工作逻辑说明MagicMock会拦截所有属性访问因此client.quote.snapshot和client.trader.buy都能按设定返回。这种方法的意义在于行情波动和网络时序都被隔离了你验证的是策略本身的逻辑而不是链路。等到 mock 测试覆盖了背离触发、次数限制、金额限制三条路径后再去接真实行情出现问题时就能直接定位到通信层而不是策略层。本文还有配套的精品资源点击获取