简介同花顺自动化交易API项目代码是一套面向量化交易场景的Python开源工具基于Tornado异步网络库实现帮助交易者通过同花顺WEB下单接口完成多策略自动买卖与成交查询解决手动下单效率低、策略执行不稳定等痛点适用于个人量化策略实盘执行及程序化交易学习。压缩包共6个文件包含2个Python脚本API调用封装与交易主逻辑、2个CSV文件交易记录与工作队列数据、1个TXT环境说明和1个inscode配置整体仅7KB结构简洁便于直接对照修改。目前已有631人下载学习适合具备一定Python基础的量化入门者也能为专业开发者提供轻量级参考。内容覆盖环境配置、券商接入、服务参数设定、买入卖出及查询接口调用方式并给出长期运行与交易队列的设计思路同时提醒了指令发送时机、网络稳定性等注意事项通过配置参数参考与队列管理示例读者可快速搭建自己的自动交易框架理解Tornado异步并发与逐条执行队列的配合方式。这些要点可帮助使用者在实盘部署时减少中断风险。1. 同花顺自动化交易API先搞清这几条路再动手想用“同花顺自动化交易API”跑A股策略的开发者搜索时都会卡在同一个地方官方没有提供一个统一的、面向个人用户的开放接口社区里流传的自动化交易项目代码几乎全是围绕客户端做封装。这不是一个拿到SDK就能跑的问题而是一个“选哪条路线、怎么把本地链路拼通”的工程问题。本文要解决的就是让你在动手前先建立选型判断再把一套最小可运行的框架搭出来最后落到实盘不翻车的排查方法上。适合有一定Python基础、想把自己的交易想法自动化的个人开发者或小团队如果你追求极低延迟、要跑高频那这类客户端级方案并不合适本文也能帮你省下时间。2. 跑通第一笔模拟下单从选型到最小可运行框架2.1 先认清三条路机构版接口、客户端文件监控、本地封装同花顺自动化交易API这个标题落到实际项目里通常对应三条技术路线。第一条是官方机构版接口面向券商资管、私募等机构需要单独申请和资质审核普通个人拿不到且文档和代码都不公开。第二条是客户端文件监控利用客户端导出的持仓、资金、成交回报等本地文件程序监听文件变化来感知账户状态再用自动化操作或本地接口完成下单。第三条是本地封装由第三方用某种方式把客户端的协议或界面操作封装成Python接口。选择建议很直接个人和小团队选“客户端文件监控 本地下单封装”的组合这是社区项目里最常见、也最容易被接手的一条路线。它牺牲了一部分延迟和稳定性但换来的是上手快、依赖少、能看清每一步在做什么。机构版接口虽然稳但“拿不到”本身就是最大的门槛。真正让你熬夜排错的反而不是下单逻辑而是登录态怎么保持、文件什么时候落盘、轮询节奏怎么设这些会在后面几章重点展开。2.2 用Python搭出最小下单框架无论你拿到的是哪种封装最终代码结构都差不多登录、查持仓、下单、查状态。先把骨架搭出来用统一的方法名封住底层差异后面换数据源或换下单通道时只改一个模块就行。# trade_client.py # 最小可运行的交易客户端骨架直接把底层通道替换成你实际拿到的接口即可 import time import json from pathlib import Path class TradeClient: 统一封装登录/下单/查询底层通道可以是文件监控、本地HTTP或模拟点击 def __init__(self, config_path: str): self.config json.loads(Path(config_path).read_text()) self.session None self.account self.config[account] def _call(self, action: str, payload: dict) - dict: # 实际项目里这一步会去发HTTP请求、写指令文件或驱动界面操作。 # 返回值统一为 {code: 0, msg: ok, data: {...}} # 这里留空方便你直接接入自己拿到的通道实现。 raise NotImplementedError def login(self) - dict: # 登录成功后保存会话令牌后续下单和查询都要带上 resp self._call(login, { user: self.config[user], password: self.config[password], source: self.config.get(source, client), }) if resp.get(code) 0: self.session resp[data][token] return resp def order(self, symbol: str, direction: str, price_type: str, price: float, volume: int) - dict: # 通用下单方法direction 取 buy/sellprice_type 取 limit/market if not self.session: raise RuntimeError(not logged in) payload { symbol: symbol, direction: direction, price_type: price_type, price: price, volume: volume, session: self.session, } return self._call(order, payload) def query_position(self, symbol: str ) - dict: # 查询持仓symbol 传空时返回全部持仓 return self._call(query_position, { symbol: symbol, session: self.session, })逻辑说明骨架把登录、下单、持仓查询收敛成三个方法核心是_call这个通道入口。你拿到的项目代码差别基本都集中在_call内部——有的项目从这里去读客户端生成的指令文件有的项目在这里调用封装好的本地接口。把业务层和通道层分开最大的好处是排错时只需要盯_call的返回值和日志不需要把整个交易逻辑翻一遍。参数说明price_type决定走限价还是市价A股市价委托有特定支持条件实盘前一定要用自己的账户先验证一遍限价单volume必须是100的整数倍否则会返回数量错误session字段一定不要遗漏很多第一次跑通的人都会因为登录态没传到下单请求里白白排错一晚上。2.3 验证下单状态不能只看返回码下单接口返回“成功”不代表委托就进了交易所它可能只是“本地已接收”。完整的落盘确认要看委托状态和成交回报这是最容易踩坑的地方。# verify_order.py # 轮询委托状态直到出现最终态已撤、已成、已废或超时 import time def wait_order_filled(client, order_id: str, timeout: float 15.0) - dict: 轮询查询委托状态返回最终状态字典 start time.time() while time.time() - start timeout: resp client.query_order(order_id) if resp.get(code) ! 0: time.sleep(0.5) continue status resp[data][status] # pending/filled/canceled/rejected if status in (filled, canceled, rejected): return resp[data] time.sleep(0.3) raise TimeoutError(forder {order_id} not reached final state within {timeout}s)逻辑说明这个循环不是无脑 sleep而是每次查询后对状态做分类。pending继续等filled、canceled、rejected都是最终态直接返回结果。超时抛出异常让上层策略不要拿着一个不确定状态的订单继续做仓位统计。参数说明timeout设 15 秒是因为A股连续竞价时段委托通常在百毫秒到几秒内就能轮到回报超过 15 秒要么是网络通道异常要么是客户端没有正常连接柜台。轮询间隔 0.3 秒是比较折中的节奏不会把客户端或接口通道打爆也不会让用户等太久。3. 看懂接口协议登录态、委托与查询的核心字段3.1 登录态管理如何避免频繁掉线自动化交易项目里最难缠的不是下单逻辑而是登录态。很多同花顺自动化交易API项目源码里登录成功后会返回一个 token但 token 的有效期往往只有几分钟到几小时不等。如果不做保活处理就会出现“上午跑得好好的下午第一笔订单突然报错未登录”的情况。常见做法是启动一个后台心跳任务周期性地调用保活接口或重新登录。心跳间隔要参考你拿到的接口文档但 30 到 60 秒是多数项目的保守区间。间隔太短会给对方服务造成不必要的压力间隔太长又容易在关键时刻掉链子。心跳任务还要记录每次执行结果连续失败超过 3 次就主动触发重新登录而不是等下一次下单时才发现会话失效。# heartbeat.py # 登录态保活定时心跳失败连续超过3次则强制重登 import time import threading class SessionKeeper(threading.Thread): def __init__(self, client, interval: float 30.0): super().__init__(daemonTrue) self.client client self.interval interval self.failed_count 0 self.stop_flag False def run(self): while not self.stop_flag: time.sleep(self.interval) resp self.client.heartbeat() if resp.get(code) ! 0: self.failed_count 1 if self.failed_count 3: self.client.login() # 自动重登 self.failed_count 0 else: self.failed_count 0逻辑说明心跳线程独立于交易主循环不阻塞下单流程。failed_count的关键作用是把偶发网络抖动和真正失活区分开单次失败不重登连续 3 次失败才触发重登避免频繁重登导致客户端弹窗或验证码。参数说明interval按 30 秒起步实际项目里如果发现 token 有效期只有 5 分钟就把间隔调到 60 秒或更短原则是“心跳周期远小于有效期”。如果重登会触发验证码或短信确认那这个方案就不适合全自动跑需要在日志里标记人工介入点。3.2 委托下单与撤单的字段设计下单和撤单是自动化交易里最重要的两个核心操作字段设计直接决定对接成败。我列出一份常见的核心字段清单你在读项目源码时按这张表去核对底层实现是否完整。场景字段名类型说明委托买入symbolstring6位股票代码如 600000------------委托买入directionstringbuy 买入 / sell 卖出------------委托买入price_typestringlimit 限价 / market 市价------------委托买入pricefloat委托价格限价单必填精确到0.01元------------委托买入volumeint委托数量必须是100的整数倍------------委托买入business_typestring普通买卖 / 信用交易默认普通------------撤单order_idstring要撤销的委托编号------------撤单sessionstring登录令牌防止身份校验失败------------委托查询order_idstring按编号查传空则查当日全部委托关键的坑有两个。一是限价单的价格边界A股有涨跌停限制委托价超出当日涨停价或跌停价会被柜台直接拒绝返回的错误码还各不相同有的报“价格超出范围”有的报“非法参数”。二是撤单必须先查状态如果委托已经全部成交撤单会返回“不允许撤单”或“已撤”项目代码里必须对这类非零返回码做兼容处理不能一律当成异常中断策略。3.3 行情数据K线与盘口怎么和策略对接多数自动化交易项目都会同时拉行情数据不然策略没有输入信号。行情接入和交易通道通常是两套独立接口这点在项目代码里要分清。# quote_client.py # 行情客户端拉日K和最新盘口快照 import requests class QuoteClient: def __init__(self, base_url: str): self.base_url base_url.rstrip(/) def get_kline(self, symbol: str, period: str daily, count: int 120) - list: 获取K线period 支持 daily/weekly/minute resp requests.get(f{self.base_url}/kline, params{ symbol: symbol, period: period, count: count, }, timeout5) resp.raise_for_status() data resp.json() return data[data][items] # 每项含时间戳/open/high/low/close/volume def get_snapshot(self, symbol: str) - dict: 获取最新盘口返回最新价、买一卖一、五档量 resp requests.get(f{self.base_url}/snapshot, params{ symbol: symbol, }, timeout5) resp.raise_for_status() data resp.json() return data[data]逻辑说明行情接口和交易接口分开封装互不引用。这样做的好处是行情断了不影响交易模块交易登录态失效也不影响策略继续拉盘口数据排错。K线接口拉回来的是原始列表策略模块再接一层转成 DataFrame 或自己的内部结构。参数说明period取daily时count一般不少于 60保证均线类指标有足够的历史窗口minute是高频轮询场景用的但要注意行情接口通常有访问频率限制每秒超过 2 次很容易被临时封禁。盘口快照get_snapshot适合做市价单的参考价不建议直接拿它算指标因为它是瞬时数据会有跳动。4. 把策略信号接进交易系统联动代码与参数设计4.1 策略信号转下单最简信号处理器策略和交易通道的衔接层是整个项目代码里最容易写乱的地方。常见错误是把策略逻辑和下单逻辑揉在一起导致无法单独测试策略也无法单独模拟下单。我一般会在两个模块之间加一个信号处理函数负责把策略输出的文字信号转成标准下单指令。# signal_handler.py # 将策略信号统一转换为下单指令并在下单前做一轮基础合法性检查 class SignalHandler: def __init__(self, client): self.client client self.last_order_id None def handle(self, signal: dict): signal 结构: {symbol: 600000, action: buy, volume: 200, price_type: limit, price: 12.35} symbol signal[symbol] action signal[action] volume signal[volume] # 基础检查数量、方向、是否有未完成的大额委托 if action not in (buy, sell): raise ValueError(funsupported action: {action}) if volume % 100 ! 0: raise ValueError(fvolume must be multiple of 100: {volume}) direction buy if action buy else sell resp self.client.order( symbolsymbol, directiondirection, price_typesignal.get(price_type, limit), pricesignal.get(price, 0), volumevolume, ) if resp.get(code) 0: self.last_order_id resp[data][order_id] return resp逻辑说明handle方法做的事情就是把策略字典翻译成下单调用。真正的策略信号来源很杂可能是均线交叉、布林带突破也可能是外部脚本推送的消息但不管来源进到这个函数时必须已经是标准化结构。last_order_id会在后面的状态轮询和撤单流程里用到。参数说明price_type缺省为limit这是安全考量。市价单虽然省事但在流动性差的标的上容易以意想不到的价格成交限价单被拒时最多是不成交不会造成极端滑点。price缺省为 0 也会在下单前被接口拒绝这里有意识地让失败早发生不让无效委托进到交易通道。4.2 轮询节奏与超时三个必调的参数任何基于同花顺自动化交易API的实测项目里轮询节奏都是血泪经验密集区。轮询太快客户端或服务端会拒绝连接报错还很难查轮询太慢下单后成交回报延迟太大策略还以为没成交就重复下单。第一个必调参数是策略信号轮询间隔interval。常见做法是 0.5 到 3 秒取一个值短线日内策略取 0.5 到 1 秒中低频策略取 1 到 3 秒。低于 0.3 秒就非常容易触发限流或者把客户端界面卡死。第二个必调参数是下单请求超时timeout。我一般会设 10 秒。网络抖动时一次请求 15 秒才返回如果你设了 5 秒超时就会误判失败然后重发可能造成重复委托。第三个必调参数是重试次数retry。重试不能无限重试更不能无脑重试。常见做法是同一笔委托最多重试 2 次并且重试前先查一次委托状态确认上一笔没有成交再重发。4.3 仓位与风控检查下单前必须过的一道闸策略给信号风控给否决权。没有风控闸门的自动化交易项目迟早会在一笔异常信号上亏掉前面所有利润。风控检查放在信号处理器和下单接口之间拦截逻辑要简单直接不做复杂计算。# risk_guard.py # 下单前风控闸门单笔上限、当日累计上限、最小资金余量 class RiskGuard: def __init__(self, max_single_volume: int, max_daily_volume: int): self.max_single max_single_volume self.max_daily max_daily_volume self.daily_volume 0 def check(self, symbol: str, direction: str, volume: int, price: float) - bool: 返回 True 表示允许下单False 表示拒绝 if volume self.max_single: return False if self.daily_volume volume self.max_daily: return False # 资金余量检查买入按价格*数量*1.003估算留出手续费缓冲 if direction buy: need price * volume * 1.003 if need self.available_cash: return False return True逻辑说明这个闸门是纯内存状态重启后daily_volume会清零所以实盘项目还要加一层磁盘或数据库持久化避免运行中途重启导致当日累计限额失效。available_cash需要在每次资金查询后更新不能在下单时才去查否则并发场景容易误判。参数说明max_single_volume的典型经验值是单笔不超过总资金的 20% 对应股数具体数值按你的资金量折算。1.003是估算手续费和滑点的乘数A股佣金和印花税实际比例每个人不同留 0.3% 的缓冲比较稳妥。买入资金余量检查是风控闸门里最重要的一条它防止“账户余额不足导致委托被柜台拒绝”这类重复报错。5. 实盘不翻车的排查清单常见报错与踩坑5.1 登录态悄悄失效报错滞后现象策略运行两小时后第一笔下单返回成功但持仓查询显示没有新增持仓日志里也没有报错。原因是下单接口在会话失效后仍返回了“本地受理”状态真正写进柜台时被拒绝。这种情况在文件监控类项目里尤其常见因为下单指令先落成本地文件客户端读取后才去执行等发现失败已经过了好几秒。排查方法是把所有下单返回值里的委托编号都保存下来然后定时用委托查询接口核对状态。解决方向是把登录保活心跳间隔调短同时在每次下单前先做一次轻量的会话校验校验失败就重登后再下单不要拿过期令牌直接发委托。5.2 涨跌停价格被拒误判接口故障现象某只股票开盘涨停自动化程序尝试以涨停价买入返回错误码是“非法参数”或“价格超出范围”。新手很容易当成接口不稳定反复重试结果触发风控锁。原因大概率是限价单价格触达了涨跌停边界柜台拒收。解决方法是下单前先拉一次最新价和昨收价计算当日涨停价和跌停价委托价超过边界就降级处理。降级策略常见有两种一是改挂单价格为边界价二是放弃这笔买入记录信号但不下单。不要尝试用市价单绕过部分股票涨停时市价单也未必能成交而且容易在开板时打到奇怪的价格。5.3 轮询太快卡界面接口不是无限调用现象策略跑起来后客户端界面操作明显变卡鼠标移动都跟不上再过一会儿程序直接无响应。查看日志发现轮询间隔设了 300 毫秒短时间内发出了上千次查询请求。原因很直接客户端或第三方通道不是为高频轮询设计的界面渲染和行情推送都跑在同一线程里频繁查询把资源占满了。解决方法是把轮询间隔调到 1 秒以上同时把盘口快照和交易状态查询分流到不同线程降低单线程压力。如果必须高频获取行情优先用独立的行情接口而不是靠反复查询交易接口来间接获取数据。5.4 撤单与成交查询的时序问题现象策略发现委托挂了 3 秒没成交发撤单指令撤单返回成功但过了几秒又看到持仓增加了说明这笔单子撤单前已经成交只是回报延迟。原因在于撤单指令和成交回报之间存在时间差客户端或柜台可能先处理了成交后收到撤单请求。解决方法是撤单前先查一次委托状态如果已经处于“部分成交”或“全部成交”就不再发撤单指令改为记录成交结果。撤单后也要再轮询一次最终状态确认是“已撤”而不是“已成”。5.5 程序崩溃后恢复先查状态再重新下单现象程序运行中手动关闭重启后策略认为“上次的买单可能没成交”于是重新发了一笔买单。结果发现原买单其实已经成交导致重复买入持仓数量翻倍。解决方法是启动时先跑一次“当日委托查询”把未完成状态的订单全部拉出来然后决定是撤单还是继续等待再让策略进入正常信号循环。这一步是恢复流程的第一道防线必须在策略逻辑启动之前执行。日志里也要记录每次启动时的未完成订单列表方便事后核对持仓变化来源。6. 进阶技巧用回放和风控表验证整个自动化链路6.1 回放让历史行情驱动整个决策链路把实盘接口切到“只读模式”用历史K线数据触发策略信号再把信号输入信号处理器但不真正下单而是打印出“如果下单会以什么价格成交、成交多少”。这一步能验证策略逻辑、字段映射、风控闸门三者之间的衔接是否正确。常见做法是把历史K线按时间逐条推送进策略模块每推一条就跑一次完整链路最后对比模拟成交记录和手工核算结果。6.2 先跑通只读模式再开实盘下单权回放通过后我习惯用模拟账户跑至少一周的只读加模拟下单模式。所谓只读模式就是所有查询接口真实调用所有下单接口都打到模拟环境或打到一个会被废弃的本地指令文件。这段时间重点观察登录态是否稳定、轮询节奏是否合理、风控闸门有没有误杀正常信号。很多在回放阶段暴露不出来的问题都会在这几天浮出水面比如盘中时段接口响应变慢、午间休市时心跳任务报错等。6.3 留好运行日志出问题看时间线日志是自动交易系统最后一道保障。至少要记录每个时间点的订单ID、信号来源、委托参数、返回状态、成交回报以及心跳任务的每次执行结果。出问题时不要只盯报错信息要看同一时间段内行情变化、委托状态和风控记录三条线对不上的地方。我自己跑自动化交易项目有个习惯每天收盘后花五分钟看一眼日志里有没有“异常但不致命”的记录比如某次心跳失败后自动重登比如某笔委托因价格越界被风控拦下。这些单次看都不重要但连续几天出现同样的苗头就说明某个参数需要调整。技术方案再稳也架不住运营细节的疏忽把这个排查习惯坚持下来比多写一百行代码都管用。希望帮到你。本文还有配套的精品资源点击获取