简介面向量化交易与程序化交易开发者同花顺自动化交易接口项目代码以Python异步网络库tornado为核心通过同花顺网页下单接口完成多策略交易、指令排队与成交查询能有效解决手动下单效率低、多策略协同难的问题适合熟悉Python的中高级开发者用于实盘或仿真环境。压缩包仅7KB共6个文件包含2个py源码文件、2个csv数据文件、1个txt依赖列表与1个inscode配置覆盖主程序、环境配置、依赖清单与交易数据记录等模块可快速还原运行环境并理解接口调用结构。已有631人学习下载。除买入卖出及查询接口外还提供了长期运行机制、交易队列处理、券商接入注意事项和参数配置说明并包含环境配置与异常处理要点可直接作为同花顺量化自动交易项目的二次开发基础。1. 先说清楚这个标题到底在解决什么问题如果你曾试着让同花顺客户端自动执行“均线金叉就买入、死叉就卖出”大概率会在第一步卡住翻遍设置也找不到一个对外开放的交易接口。标题里写“同花顺自动化交易API”容易让人误以为官方开放了完整的程序化交易通道实际上官方侧重点在行情和资讯服务真正下单的口子都由开户券商掌握。于是这里就有了一个需求差你想要的是一套“能自动读取行情、自己计算信号、再走合法通道下单”的链路而“API”三个字只是这条链路的外壳。这篇文章会讲清楚怎么搭这条链路先把客户端产生的数据变成程序可读的原料再把策略做成独立进程最后用模拟盘验证执行通道。适合有 Python 基础、但没有券商量化接口权限、又想验证自写策略的散户。2. 第一层把同花顺客户端的行情数据变成程序能读的“原料”2.1 数据源怎么选日志文件、DDE 和券商行情的取舍做自动化交易第一步永远不是写策略而是解决“数据从哪来”。很多人一上来就去找“同花顺 API 文档”结果搜出来一堆第三方封装装完不是要登录就是要抓包稳定性全凭运气。我一般会按数据来源把方案分成三类先看需求再选。数据源延迟开发成本稳定性适合场景客户端日志导出秒级低读本地文件即可中等受客户端写盘策略影响个人盯盘、低频策略、打新监控DDE 动态数据交换毫秒级中需要处理 Windows 消息循环较高但依赖客户端保持前台运行高频盯盘、自定义指标实时刷新券商官方行情接口毫秒级中高需申请权限最高数据格式规范实盘量化、有券商量化权限互联网免费行情源秒级低但字段不全低接口变动频繁复盘、回测、非实时策略我推荐第一类作为起步方案原因是它不碰任何灰色协议只读本地文件且客户端本身就提供“日志导出”或“数据导出”功能。你在客户端里点一下导出程序再盯着导出文件读增量行就能以秒级延迟拿到行情。这套思路的边界也很清楚它拿到的是“客户端已经展示的数据”不是交易所原始行情所以适合分钟级策略和监控类任务。如果你以后要高频再换第三类。2.2 监控客户端日志文件的最小脚本确定了“读本地日志”之后第一个脚本要解决三件事定位文件、记住上次读到哪里、按固定间隔轮询。尽量不要用文件系统监听库比如 watchdog因为客户端在写日志时经常是“先写临时文件再替换”监听事件会漏掉或重复触发反而不如老老实实轮询。import os import time LOG_PATH rD:/ths/export/quote.log # 改成你本机导出的日志文件路径 POLL_INTERVAL 1.0 # 轮询间隔单位秒 def read_new_lines(path, last_offset): 读取文件从 last_offset 到当前末尾的新增内容 with open(path, r, encodinggbk, errorsignore) as f: f.seek(last_offset) new_text f.read() new_offset f.tell() return new_text, new_offset if __name__ __main__: # 启动时记录文件当前大小避免把历史旧数据全部读一遍 last_offset os.path.getsize(LOG_PATH) print(fmonitor start, offset{last_offset}) while True: try: current_size os.path.getsize(LOG_PATH) if current_size ! last_offset: text, last_offset read_new_lines(LOG_PATH, last_offset) if text: # 先打印最后200个字符确认格式后再接后续解析 print(text[-200:]) time.sleep(POLL_INTERVAL) except KeyboardInterrupt: break这段代码里os.path.getsize拿当前文件大小和上一次记录的偏移比较大小变化才去读增量避免每次空转。f.seek(last_offset)让文件指针直接跳到上次结束位置只读新增部分文件很大时也扛得住。这里两个关键参数LOG_PATH要按你本机实际路径改不同版本客户端存放位置不一样建议先在软件里手动导出一份再用文件资源管理器定位POLL_INTERVAL秒级够了设成 0.1 会更灵敏但会给磁盘带来不必要的读写压力而且策略往往不需要那么高频。这个脚本有几个容易翻车的细节。第一编码要用gbk国内行情软件日志默认是 ANSI 编码直接按utf-8读会抛解码异常第二errorsignore是保护手段个别行混进非法字符时不至于让整个脚本崩掉第三启动时记录last_offset很关键否则程序一重启会把整个历史文件重新读一遍后果是解析出一堆过期行情可能触发错误信号。2.3 解析日志行并写入 SQLite轮询脚本只能看到原始文本下一步是把它解析成结构化数据。不同版本客户端的日志格式有差异常见的一行大概是11:25:30.123 000001 15.250即“时间、股票代码、最新价”。我建议先手动打开日志文件看几行再照着写正则。import re import sqlite3 from datetime import datetime LINE_PATTERN re.compile( r(?Ptime\d{2}:\d{2}:\d{2}\.\d{3})\s r(?Pcode\d{6})\s r(?Pprice\d\.\d{3}) ) DB_PATH market.db conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS tick ( ts TEXT, code TEXT, price REAL, PRIMARY KEY (ts, code) ) ) def store_line(line): m LINE_PATTERN.search(line) if not m: return False today datetime.now().strftime(%Y-%m-%d) ts f{today} {m.group(time)} code m.group(code) price float(m.group(price)) conn.execute( INSERT OR IGNORE INTO tick (ts, code, price) VALUES (?, ?, ?), (ts, code, price), ) conn.commit() return True这个解析函数做了三件正经事用命名分组提取时间、代码、价格给时间补上日期前缀因为日志里往往只有时分秒利用PRIMARY KEY (ts, code)做唯一约束同一秒同一只股票只保留一条天然去重。INSERT OR IGNORE是配合唯一约束用的重复数据直接跳过而不是报错。这里有一个非常重要的取舍conn.commit()每写入一行就提交一次好处是断电丢数据风险小坏处是高频行情下磁盘 I/O 可能成为瓶颈。如果你监控的多只股票每秒都有数据建议改成“每收集 100 行或每 2 秒批量提交一次”能明显降低卡顿。这个优化不做也能跑但等你把股票池扩到几十只就会体会到什么叫“行情没卡你的脚本先卡了”。落到 SQLite 还有一个额外好处数据文件单一策略进程可以直接读同一个market.db。日志文件是文本多进程同时读没问题但策略进程如果直接去读客户端日志还要重复解析格式不如让专门的采集进程负责写库策略只关心“表里有什么新数据”。3. 第二层策略与模拟下单把信号变成订单3.1 为什么策略必须是独立进程采集和策略如果写在一个脚本里用不了多久就会踩坑。行情采集要保持固定频率读文件、写数据库而策略要计算指标、判断买卖点两者对时间的敏感度完全不一样。混在一起时策略里一句慢 SQL 就能拖住采集导致日志文件越积越多重启时又得从头读。更难受的是你想回测某段历史数据只能把整个脚本停下来改代码。所以我一般会把采集进程和策略进程分开采集跑一个终端窗口策略跑另一个互不干扰。另一个理由是职责边界。采集进程不需要关心怎么下单策略进程不需要关心日志文件编码。中间只通过market.db交换数据哪一层出问题就单独重启哪一层。这个设计不是花架子能省掉大量“改了一个循环导致行情断了一小时”的尴尬。3.2 一个最小双均线策略脚本策略脚本从 SQLite 里取最近一段时间的最低价序列计算短周期和长周期均线在短线上穿长线时给买入信号下穿时给卖出信号。注意这里直接用tick表里的价格序列做均线严格来说应该先把 tick 聚合成分钟线再计算否则行情剧烈波动时每笔报价都算一个点信号会频繁抖动。下面是简化版先跑通链路再谈精度。import sqlite3 import time DB_PATH market.db SHORT_WIN 5 LONG_WIN 10 MIN_TRADE_INTERVAL 60 # 同一只股票两次下单的最小间隔秒 LAST_TRADE {} def get_close_prices(code, limit200): conn sqlite3.connect(DB_PATH) rows conn.execute( SELECT ts, price FROM tick WHERE code? ORDER BY ts DESC LIMIT ?, (code, limit), ).fetchall() conn.close() rows.reverse() return [r[1] for r in rows] def sma(values, window): if len(values) window: return None return sum(values[-window:]) / window def check_signal(code): prices get_close_prices(code) if len(prices) LONG_WIN: return None short sma(prices, SHORT_WIN) long sma(prices, LONG_WIN) if short is None or long is None: return None now time.time() last LAST_TRADE.get(code, 0) if short long and now - last MIN_TRADE_INTERVAL: LAST_TRADE[code] now return BUY if short long and now - last MIN_TRADE_INTERVAL: LAST_TRADE[code] now return SELL return None if __name__ __main__: # 演示每5秒检查一次某只股票 while True: signal check_signal(000001) if signal: print(f{time.strftime(%H:%M:%S)} signal{signal}) time.sleep(5)这里的核心参数是SHORT_WIN和LONG_WIN5 和 10 只是默认值实际效果取决于你喂进去的数据粒度。如果你用 tick 级报价5 和 10 代表最近 5 笔与 10 笔波动很大如果先聚合迟分钟线5 和 10 才代表 5 分钟与 10 分钟均线。这个区别会直接决定策略跑出来是“频繁交易”还是“稳定趋势跟踪”。MIN_TRADE_INTERVAL60是保险丝防止同一个信号在 1 分钟内被触发几十次。很多新手做自动化交易第一批学费就交在这个地方日志里同样价格出现两次策略就傻傻地连下两单一天下来手续费比你赚的还多。还要提醒一点LAST_TRADE只是一个内存字典程序重启后清空。实际要做到好一点的防重应该把最后一笔下单时间记录在 SQLite 里重启后还能继续判断。这一步不难但属于典型的“看起来能跑跑两天就出问题”的隐患。3.3 模拟撮合与执行链路验证策略输出信号后不要急着想怎么连接券商。我先建议搭一个本地模拟撮合表把每一次信号都当成一次真实订单记录然后检查三个条件信号价格是否在当时行情区间内、下单间隔是否合理、最终虚拟成交记录是否符合预期。这个模拟步骤能筛掉大部分低级错误而且不需要承担实盘风险。字段示例作用signal_time2025-06-01 10:30:00记录信号生成时间code600000股票代码sideBUY买卖方向signal_price8.550信号触发时最新价valid_price_low8.500模拟允许成交的最低限价valid_price_high8.600模拟允许成交的最高限价statusFILLED状态待成交/已成交/拒单created_at2025-06-01 10:30:02订单创建时间模拟撮合的逻辑很简单把信号写入 orders 表后下一次行情进来时判断最新价是否落在valid_price_low和valid_price_high之间落进去就把状态置为FILLED否则一直挂单。这能验证策略对价格的敏感度和成交细节。等模拟盘连跑两周、统计的成交率和滑点都稳定之后再考虑通过券商官方接口做真实委托。这一步是“后悔药”所在能拦下大量裸跑的冲动。4. 常见问题与避坑数据掉线、编码、重复触发与合规边界4.1 日志文件“半天不更新”脚本还卡着现象采集脚本一直正常运行但market.db里的数据停在了某个时间点客户端界面明明还在跳数字文件大小却固定不变。原因多数情况下是客户端把日志路径改了或者日志写满了进入滚动写模式旧文件不再追加。也有可能是你导出后没开启“自动导出”功能手动导出的文件导完就不再更新这不算bug是机制。解决在客户端设置里找到日志导出相关选项开启按时间增量追加或者把导出频率设为“实时”。然后到日志文件所在目录看一眼确认文件名是否带日期后缀如果今天生成了新文件就要把采集脚本里的LOG_PATH指向新的路径。我习惯在脚本里加一段目录扫描逻辑取目录下最新修改的那个*.log文件而不是写死文件名能减少这类人工干预。4.2 中文股票名乱码正则匹配不到现象read_new_lines读出来的文本里股票名称显示成“浜崘”之类的乱码价格字段反而正常但正则匹配经常失败。原因客户端日志里混了 GBK 和 UTF-8 两种编码内容只用一个encodinggbk处理时某些中文字符被错误解码甚至在errorsignore下直接被丢弃导致整行长度变化。解决先打开日志文件用十六进制方式看一眼文件头确定实际编码。如果确认是混合编码不要一次性read()整个文件改成按行读取并用errorsreplace或errorsignore处理同时保证正则只需要匹配数字和字母字段比如用search而不是match只要价格部分完整就能成功。股票名乱码不影响交易信号不必强求完美解码。4.3 信号连续触发同一个价格买了一大堆现象策略在某一分钟里输出了十几个 BUY 信号模拟盘列表里全是一条条几乎相同价格的买单。原因均线计算用的数据粒度太细比如直接用 tick 报价序列计算而 tick 在短时间内来回穿均线策略窗口内反复出现“金叉-死叉-金叉”。MIN_TRADE_INTERVAL虽然限制了两次下单间隔 60 秒但如果第一次买入后价格下穿再上穿已经过了 60 秒就被判定为新信号。解决把 tick 数据先聚合成 1 分钟 K 线用分钟线产生的均线不要用裸 tick。聚合方法很简单从tick表里按code和每分钟分组取出该分钟内最后一条价格作为该分钟收盘价。同时把MIN_TRADE_INTERVAL设为至少 300 秒配合 K 线周期使用。这样信号量会大幅下降策略也更符合人工盯盘习惯。4.4 把“本地拿数据”理解成“官方API”走偏了合规现象有人觉得既然能从本地日志拿到行情就应该能用类似思路去截取下单请求甚至通过模拟点击替自己下单。原因把“自动化”理解成了“绕过官方接口”。本地日志读取本身风险低但模拟点击或截取网络请求替代交易属于灰色操作既可能违反软件使用条款也存在操作错乱导致真实资产损失的风险。安全红线非常清楚正规券商不会认可这种“另类通道”。解决正确做法是先确认你开户的券商是否提供量化接口或条件单功能。有就申请官方通道没有就退回到“自动化出信号 人工确认下单”的半自动模式。这个方案虽然不 100% 全自动但合规、稳定且能把精力聚焦在策略本身。自动化交易的核心价值在于“策略可复现、可回测”不一定非要无人值守全自动。4.5 客户端一升级所有路径和字段全废现象某天打开客户端看到升级提示点了之后采集脚本开始报错日志文件找不到了原来能匹配的正则一行都匹配不上。原因客户端版本升级会调整安装目录或日志输出结构这是最常见的外部变化。写死路径和正则是灵验的但也是脆弱的一次升级就让你之前调好的解析全部作废。解决把容易变化的信息抽到配置文件里比如config.ini采集脚本启动时读取路径、文件名模板、正则表达式。升级后只需要改配置重启不需要改代码。另外每次升级前给当前脚本和数据目录做一次备份升级后如果新字段不兼容还能用旧数据继续调试。这个习惯能让你少熬好几个夜。5. 进阶做法用历史回放校准策略再上真实通道模拟盘跑通后别急着连真实通道先用历史回放把策略“擦一遍”。具体做法很简单将过去几周导出的日志文件按日期重命名分批次放到采集脚本能监控的目录里让采集解析逻辑以快进的方式“重演”历史行情。然后观察策略信号输出的时间点、价格、信号频率对照当时真实走势确认没有出现“未来函数”。所谓未来函数就是策略偷偷用到了发生在这个时间点之后的数据在回放中最常见的表现是信号出现在最高点或最低点之后后知后觉得离谱。回放之后还要做参数校准。不要只盯着一组均线参数跑出漂亮曲线就觉得稳了换一段行情可能全变。我一般会取最近三周做滚动验证第一周定参第二周微调第三周验证三道关都过了才进入模拟下单。如果其中一段行情曲线明显失真说明参数过拟合重新回到“少参数、多规则”的方向。规则太复杂意味着依赖过多巧合每次市场风格切换都可能翻车。这属于玄学范畴但经历过的都懂越简单的规则在长时间里活得越久。再看风险边界。给策略加上三个硬性限制单只股票最大建仓占比不超过总资金的一定比例每日最大交易次数限制为几次连续亏损达到阈值就自动停止发信号。这三个限制应该写死在策略代码里而不是放在配置文件中靠自觉维护。它们就是你的刹车片没有刹车的策略开得越快碎得越快。我见过太多人先赢后输问题不是赢利能力不够而是风控模块没跟上。除了风险限制还要有运行监控。策略进程每 30 分钟检查一次orders表里是否有新的待成交订单如果长时间没有新订单且数据库也没有新行情写入就要自动拉起采集进程或发出提醒。我习惯把监控逻辑放在策略进程之外用系统自带的任务计划定时执行一个健康检查脚本避免主进程卡死后“没人收尸”。最后说一个我自己的教训以前总觉得“多跑一个品种多一份收益”于是同时监控十几只股票结果每个信号都想做复盘时发现大量时间花在解释为什么某只票不该买。后来我强制自己只保留 3 到 5 只自选信号多了宁愿错过也不追。这套自动化链路的价值在于帮你稳定执行纪律而不是帮你熬夜盯更多的屏幕。希望帮到你。本文还有配套的精品资源点击获取
