简介面向量化交易开发者这份项目代码提供了同花顺自动化交易WEB下单接口API的完整实现基于tornado异步网络库可支持多策略交易指令进入队列并逐条执行适用于需要程序化下单、持仓查询及成交确认的自动化交易场景。压缩包内共6个文件包含2个Python脚本核心交易逻辑与API配置、2个csv文件交易数据与工作队列记录、1个txt说明文档及1个inscode运行配置整体仅7KB结构精简、便于快速部署和二次修改。目前已有631人学习下载。代码覆盖了环境依赖、券商接入、同花顺平台设置、交易服务参数配置等关键环节并演示了买入、卖出及成交查询接口的调用方式对长期运行的交易队列与注意事项也有针对性处理。初学者可据此搭建自己的量化交易框架有经验者也能从中获得异步交易调度的实现思路。1. 同花顺自动化交易API把它拆成可复现的 Python 模块做量化的人大多有过这种经历策略信号在收盘后跑得好好的一回测也漂亮可一接实盘就卡在下单那一步。手动敲单慢不说还容易手滑找第三方柜台接口又要过券商资质审核。这套同花顺自动化交易 API 的代码包解决的正是这个环节用 Python 直接对接同花顺客户端的行情、委托、持仓查询能力把从信号产生到订单回报的整条链路串起来。它适合已经能写策略、但卡在交易执行层的开发者以及想快速搭建自动化交易框架、又不想从头造轮子的个人量化项目。2. 选型与准备为什么绕开券商接口直接走同花顺通道2.1 三条技术路线的对比自己抓包、券商官方 API、同花顺接口我在动手之前把常见的接入方式横向比过一遍。你会发现这不是简单的“哪个好用”而是“哪个能落地”。第一种是自己抓客户端的网络包解析私有协议再构造请求。这条路最硬核加密、压缩、动态字段、服务端校验每一项都能耗掉大量业余时间。抓来的协议还可能因为客户端升级就失效属于典型的高投入、高维护方案适合把全部精力砸在基础设施上的人不适合还有策略要迭代的量化开发者。第二种是券商官方 API。它的优点是合规、稳定缺点也明显个人投资者往往拿不到接口权限或者说要满足资金门槛、交易经验、考试等一系列硬性条件。即便拿到了通常也只能交易自家券商的账户换一个券商就要重新对接一套接口。对于只做模拟盘验证、或者账户分散在多个券商的人来说这权限就是瓶颈。第三种就是这套代码走的路子在同花顺客户端之上做一层 Python 封装。客户端本身已经完成了登录、风控、协议解析这些脏活代码包只需要负责“读状态、发指令”把精力集中在策略层。这种方式也适合多券商场景因为客户端本身能登录不同券商的账户。选型结论很直接如果你没有券商 API 权限又不想陷入协议逆向的泥潭同花顺通道的性价比是最高的。2.2 安装依赖与最小验证代码这套代码包的依赖不多核心就三块网络请求库、数据解析库、日志库。常见做法是用requests做 HTTP 通信pandas做行情数据处理loguru或者标准库logging做日志记录。安装命令如下pip install requests pandas loguru装完依赖先别急着跑策略我习惯先用一段最小代码验证环境能不能连通。所谓“连通”是指本机到同花顺服务端的网络链路是通的并能正确解析回包import requests from loguru import logger def check_connectivity(server_url: str, timeout: int 5) - bool: 验证到同花顺服务端的连通性。 参数说明: server_url: 服务端地址, 一般是客户端配置里的行情或交易网关地址 timeout: 超时时间, 单位秒, 视本机网络波动情况调整 try: resp requests.get(server_url, timeouttimeout) # 很多网关对 GET 会直接返回 404, 但只要连接建立就说明链路是通的 logger.info(fHTTP status code: {resp.status_code}) return True except Exception as exc: logger.error(fConnectivity check failed: {exc}) return False if __name__ __main__: # 这里换成你本机客户端配置里的网关地址 check_connectivity(http://127.0.0.1:8899)这段代码的逻辑很直白发一个 HTTP 请求只要没有抛超时或连接拒绝就说明环境没问题。timeout参数我一般设 5 秒因为有些网关第一次握手会慢本地网络环境差的可以放宽到 10 秒但超过 10 秒还不行就要查防火墙或代理了。2.3 配置文件设计把账号、地址、轮询频率拆出去我踩过最大的坑之一就是把参数硬编码在代码里来回改最后改到连自己都忘了哪套配置对应哪种环境。好的做法是把所有环境相关的东西放进一个config.yaml代码里只负责读取server: host: 127.0.0.1 port: 8899 timeout: 5 trade: account_id: your_account_id max_position_ratio: 0.8 # 单只股票最大仓位占比 retry_times: 3 # 下单失败自动重试次数 retry_interval: 2 # 重试间隔, 单位秒 strategy: enable_trade: false # 策略总开关, 调试时务必关掉 max_orders_per_day: 20 # 单日最大委托笔数, 防止失控enable_trade这个开关我强烈建议保留首次调试时保持false只走行情和信号计算流程确认没有误报信号后再打开。max_orders_per_day是另一道保险策略出现高频死循环时它能直接截断当天的委托总量。3. 登录与会话保持核心代码模块拆解3.1 会话管理登录、心跳、自动重连同花顺接口不同于一般 REST API它更接近“长连接 会话状态”。也就是说你需要先建立一个会话得到会话 ID后续所有操作都带着这个 ID 走并且要周期性发心跳包保持活跃。这套代码里把这块做成了一个独立的Session类import time import requests class TradingSession: 同花顺交易会话管理: 登录、心跳、重连 def __init__(self, host: str, port: int, timeout: int 5): self.base_url fhttp://{host}:{port} self.timeout timeout self.session_id None self.session requests.Session() # 复用 TCP 连接, 减少握手开销 def login(self, account_id: str, password: str) - bool: 登录网关, 拿到会话 ID 常见做法是把账号密码放请求体, 返回体里携带 session_id payload { account_id: account_id, password: password } resp self.session.post(f{self.base_url}/login, jsonpayload, timeoutself.timeout) data resp.json() if data.get(code) 0: self.session_id data[data][session_id] return True return False def heartbeat(self) - bool: 发心跳包, 维持会话活性; 返回 False 说明会话可能已过期 if not self.session_id: return False resp self.session.post( f{self.base_url}/heartbeat, json{session_id: self.session_id}, timeoutself.timeout ) return resp.json().get(code) 0 def keep_alive_loop(self, interval: int 30): 每 interval 秒发一次心跳, 断线后自动重连 while True: time.sleep(interval) if not self.heartbeat(): # 这里可以触发重连逻辑, 我这里只记录日志 print(Heartbeat lost, attempting to reconnect...) # 重连的具体代码略, 一般重新调用 login 即可参数上要注意心跳间隔不要设太短否则服务端可能判定为“高频请求”而主动断开也不宜超过服务端的空闲超时上限常见做法是设 30 到 60 秒。requests.Session()在这里不是必需品但它会复用底层 TCP 连接在频繁发请求的场景下能减少握手时间体感上快不少。3.2 下单与撤单参数构造与异常返回处理下单是整套系统里最不能出错的部分。参数错误、价格不合法、数量超限轻则单子被拒重则可能成交出意外仓位。代码包里封装了一个OrderManagerimport time class OrderManager: def __init__(self, session: TradingSession): self.session session def place_order(self, stock_code: str, price: float, volume: int, direction: str buy, order_type: str limit) - str: 提交委托 参数说明: stock_code: 股票代码, 如 600000 price: 委托价格, 限价单必填; 市价单可传 0 volume: 委托数量, A股一手是 100 股, 传入的必须是 100 的整数倍 direction: buy 或 sell order_type: limit 限价单 / market 市价单 返回: 委托编号, 后续用它查状态或撤单 if volume 0 or volume % 100 ! 0: raise ValueError(A股委托数量必须是 100 的整数倍) payload { session_id: self.session.session_id, stock_code: stock_code, price: price if order_type limit else 0, volume: volume, direction: direction, order_type: order_type } resp self.session.session.post( f{self.session.base_url}/order, jsonpayload, timeoutself.session.timeout ) data resp.json() if data.get(code) ! 0: # 把服务端返回的原因原样打出来, 方便排查 raise RuntimeError(fOrder rejected: {data.get(message)}) return data[data][order_id] def cancel_order(self, order_id: str) - bool: 按委托编号撤单 resp self.session.session.post( f{self.session.base_url}/cancel, json{session_id: self.session.session_id, order_id: order_id}, timeoutself.session.timeout ) return resp.json().get(code) 0这块代码最值得说的两点一是数量和价格的校验放在本地做掉而不是等服务端返回错误再处理能减少无效请求二是order_type的区分限价单必须传价格市价单传 0这个规则各平台略有差异接入不同客户端时要先拿模拟环境试一遍。3.3 查询委托状态与持仓轮询节奏怎么设自动交易要想不变成“盲人摸象”就必须有可靠的回报查询。委托提交之后会不会成交、成交了多少、剩余多少都要靠查询接口拿数据。代码包里用的是轮询模式def query_order_status(order_id: str, max_wait: int 30, interval: int 3): 轮询委托状态, 直到终态或超时 参数说明: max_wait: 最长等待时间, 单位秒 interval: 轮询间隔, 单位秒; 太短容易触发限流, 太长则响应迟钝 elapsed 0 while elapsed max_wait: resp requests.post( f{session.base_url}/order/status, json{session_id: session.session_id, order_id: order_id} ) data resp.json()[data] status data[status] # pending / filled / partial / cancelled if status in (filled, cancelled): return data elapsed interval time.sleep(interval) return {status: timeout, order_id: order_id}轮询间隔是这里最容易犯错的参数。设为 1 秒内短时间大量订单会触发限流设为 10 秒以上急跌的时候可能来不及止损。我一般取 3 秒作为一个折中值如果单量每天都很大再把间隔调小同时增加单日委托上限来对冲风险。4. 从策略信号到真实下单完整落地链路4.1 一个可工作的策略回调框架拿到这包代码后大多数人直接想做的事是“把我的策略信号接进去”。要实现这一点最好策略端和交易端之间用一个简单的事件回调做解耦。代码包里给出的框架长这样class TradeExecutor: 接收策略信号, 执行交易动作 def __init__(self, order_manager: OrderManager, config: dict): self.om order_manager self.config config self.daily_orders 0 # 当日已委托笔数 self.recent_signals set() # 记录最近已处理信号, 防重复 def on_signal(self, signal: dict) - bool: 策略回调入口 signal 的格式, 用的时候按你自己的策略输出对齐: { stock_code: 600000, action: buy, # buy / sell / hold price: 12.34, # 期望委托价, 0 表示市价 volume: 200, # 需要保证是 100 的整数倍 signal_id: mad_20250101_001 # 唯一信号标识 } # 检查重复信号, 避免同一信号被重复执行 if signal[signal_id] in self.recent_signals: return False # 总开关控制 if not self.config[strategy][enable_trade]: return False # 单日委托笔数保护 if self.daily_orders self.config[strategy][max_orders_per_day]: return False action signal[action] if action hold: return False direction buy if action buy else sell try: order_id self.om.place_order( stock_codesignal[stock_code], pricesignal[price], volumesignal[volume], directiondirection ) self.daily_orders 1 self.recent_signals.add(signal[signal_id]) return True except Exception as exc: # 下单异常要打出来, 别默默吞掉 print(fOrder failed: {exc}) return False这个框架最大的价值在“防重复”和“总开关”这两层。策略在分钟级别重复输出同一个信号太常见了如果不去重一天下来委托笔数会爆炸式增长。signal_id我用的是“策略名 日期 序号”拼接能保证每天从零开始计数。4.2 持仓与资金同步下单前先算可买数量很多新手犯的错误是策略信号说买 500 股就直接下单完全不管账户里还有没有资金。代码包里提供了一个PortfolioManager在委托前先校验class PortfolioManager: def __init__(self, session: TradingSession): self.session session def get_available_cash(self) - float: 查询账户可用资金 resp requests.post( f{self.session.base_url}/account/balance, json{session_id: self.session.session_id} ) return resp.json()[data][available_cash] def get_position(self, stock_code: str) - int: 查询某只股票的当前持仓数量 resp requests.post( f{self.session.base_url}/position/query, json{session_id: self.session.session_id, stock_code: stock_code} ) return resp.json()[data][volume] def calc_buyable_volume(self, stock_code: str, price: float, cash_ratio: float 0.9) - int: 根据可用资金和仓位上限计算可买股数 cash self.get_available_cash() max_amount cash * cash_ratio raw_volume int(max_amount / (price * 100)) * 100 # 往下取整到 100 的倍数 return raw_volumecash_ratio这里取 0.9 是为了留一点缓冲防止计算时还没把手续费和滑点算进去导致拒绝。实际执行时如果价格波动导致资金差一点我还习惯在委托前用最新价按 0.85 的比例再压一次确保万无一失。4.3 异常恢复程序中断后的仓位对账实盘最怕的不是策略亏钱而是程序挂掉后你不知道自己到底有什么仓位。代码包在启动时会先执行一次“对账”把本地记录的持仓和服务器端返回的持仓做一次全量比对def reconcile_positions(local_positions: dict, remote_positions: dict) - list: 对比本地记录和服务端持仓, 返回差异清单 参数说明: local_positions: 本地持久化的持仓 {600000: 2000, 000001: 500} remote_positions: 查询接口返回的持仓, 格式相同 返回: 差异项列表, 每一项是个 dict: {stock_code: ..., local: 2000, remote: 0, diff: -2000} all_codes set(local_positions.keys()) | set(remote_positions.keys()) diff [] for code in all_codes: lv local_positions.get(code, 0) rv remote_positions.get(code, 0) if lv ! rv: diff.append({ stock_code: code, local: lv, remote: rv, diff: rv - lv }) return diff为什么要做这一步因为你永远不知道上一次断电的时候是“下了单没回报”还是“回报收到了但没写入本地”。你在拿到任何一套自动交易代码之后第一步都应该先跑一次对账确认代码里的持仓数据不是空中楼阁。5. 避坑与常见问题排查五条实打实的踩坑记录5.1 客户端未登录导致接口抛“无效会话”现象是登录接口返回code 0拿到session_id后不到两分钟再下单就报“无效会话”或session expired。一开始以为是代码的问题排查了半天才发现是同花顺客户端本身没点“登录”按钮服务端只接受了网关握手但并没有建立真实交易通道。原因是我的代码里只调用了本地网关的登录接口这个接口能成功只能说明本机和网关通信正常而网关是否完成券商端验证取决于客户端界面状态。解决方法是把客户端的自动登录打开并且在下单前先调用一次查询接口探活确认会话真的能用再执行交易动作。5.2 委托价格字段传成了浮点数被拒单有一次做限价单价格直接传了12.34看起来没毛病但服务端返回“price invalid”。排查下来发现它要求价格字段是整数也就是“以分为单位”需要传1234而不是12.34。这是各平台之间最常见的差异点没有统一标准。解决方案是在place_order里加一层价格单位转换如果传入的是元就乘 100 转成分如果传入的已经是分就不动。本来想写自动化判断但后来发现有的券商接口返回价格是元有的返回是分干脆在配置里加了一个price_unit字段显式指定单位不猜。5.3 撤单报成功但订单仍然成交撤单接口返回code 0但几分钟后查委托状态发现这单居然成交了。开始以为撤单功能有 Bug后来复盘才发现是撤单和成交之间出现了“时间窗”。如果订单已经部分成交剩余部分在撤单请求到达之前刚好也成交了撤单接口当然会返回成功。解决办法不是改撤单逻辑而是撤单后立刻按order_id查询最终状态并对比成交数量。如果发现“已经全量成交”就把这次撤单记录成“无效撤单”不要影响策略的持仓计算。从那以后我的所有撤单路径都会强制带上一次终态查询。5.4 轮询太频繁被服务端限流刚开始跑的时候我在循环里 200 毫秒查一次委托状态结果跑了不到十分钟所有请求都开始返回“too many requests”连普通查询都进不去了。这说明网关或者服务端对单个会话的请求频率有硬限制。解决方式是把查询间隔统一改成 3 秒并且把“批量查询”替代“逐单查询”。也就是用一个接口一次把当天的所有委托拉回来在本地做状态过滤而不是一笔单子一笔单子去查。只查那些状态还在中途的单子数量少的时候用 1 秒间隔也可以。5.5 策略在非交易日也疯狂“下单”周五收盘后忘了关程序结果周六早上起来一看日志策略还在反复提交委托然后全被服务端拒绝。原因是我的策略没有做交易日历判断。服务端虽然会拒单但日志会被刷爆信号计数也会被搞乱。解决方法是引入一个交易时段的判断函数先判断“今天是不是交易日”再看“当前时间是否在 9:30-15:00 之间”。这两个条件缺一不可。注意节假日后的调休日有的交易所有交易有的没有不能只看星期几就直接判断最好用服务端提供的交易日历接口来拉取而不是本地写死日期。6. 把自动交易跑稳的验证技巧影子交易与订单回放拿到代码包并跑通基础功能后先别急着上实盘。“能下单”和“下单能成交且不影响持仓”是两回事。我常用两个验证手法来测试整套系统效果都很直接。第一个是影子交易。把同一个策略信号同时打到两个执行通道一个走模拟账号一个走实盘账号。这样你会很清楚——策略信号的成交滑点是多少、模拟盘和实盘在限价单上成交概率差多少、走到同一个价位的延迟差异有多大。比如模拟盘 5 秒就成交的单子实盘可能要挂 30 秒还在排队又比如模拟盘总能按限价成交实盘却经常滑价。这些信息只有影子交易能给到你。第二个是针对历史行情的“回放测试”。我写了一个简单的回放脚本把某只股票过去一个月的分钟行情逐条读进来喂给策略模块让它生成信号再把信号送入订单模块。整个过程不开真实下单只记录下“如果这单发出当时的盘口能不能成交”。这个回放和常规回测不同的地方在于它观察的是订单执行模拟而不是策略盈亏def replay_execution(history_rows: list, executor: TradeExecutor) - dict: 按历史行情回放策略信号, 模拟成交判定 参数说明: history_rows: 每行包含 {time: 09:31, price: 12.34, volume: 123400, bid1: 12.30, ask1: 12.35} executor: 策略执行器, 可传入模拟模式 返回: 统计信息: {signals: n, filled: m, avg_slippage: x} filled 0 slippage_list [] for row in history_rows: signal generate_signal(row) # 你自己的策略信号生成函数 if not signal: continue since_last_row signal # 简单的成交判定: 买单价 ask1, 或 卖单价 bid1, 才算可成交 if signal[action] buy and signal[price] row[ask1]: filled 1 slippage_list.append(signal[price] - row[ask1]) elif signal[action] sell and signal[price] row[bid1]: filled 1 slippage_list.append(row[bid1] - signal[price]) avg_slippage sum(slippage_list) / len(slippage_list) if slippage_list else 0 return {signals: len(slippage_list), filled: filled, avg_slippage: avg_slippage}跑完回放后如果填单率低于 60%先别急着说策略好用说明你的委托价设得离盘口太远或者你总是在一根 K 线收盘时才出信号错失了好价格。这个环节能帮你在上实盘之前就修正执行参数。我还习惯做一次连续多日的“无人值守测试”让程序在模拟环境下连跑三天期间故意断一次网、杀一次进程看恢复后能不能靠对账逻辑自洽。从那以后我每次上线新策略都强制先跑一遍这个流程确认日志干净、持仓一致才会把资金切进去。希望这包代码能帮到一个像当年的我那样到处找自动交易入口的开发者先把执行这层做扎实策略本身的价值才真正发挥得出来。本文还有配套的精品资源点击获取