企业官网建设流程全解析

简介：面向量化交易开发者，这份项目代码提供了同花顺自动化交易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, timeout=timeout) # 很多网关对 GET 会直接返回 404, 但只要连接建立就说明链路是通的 logger.info(f"HTTP status code: {resp.status_code}") return True except Exception as exc: logger.error(f"Connectivity 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 = f"http://{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", json=payload, timeout=self.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}, timeout=self.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 下单与撤单：参数构造与异常返回处理

下单是整套系统里最不能出错的部分。参数错误、价格不合法、数量超限，轻则单子被拒，重则可能成交出意外仓位。代码包里封装了一个OrderManager：

import 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", json=payload, timeout=self.session.timeout ) data = resp.json() if data.get("code") != 0: # 把服务端返回的原因原样打出来, 方便排查 raise RuntimeError(f"Order 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}, timeout=self.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_code=signal["stock_code"], price=signal["price"], volume=signal["volume"], direction=direction ) self.daily_orders += 1 self.recent_signals.add(signal["signal_id"]) return True except Exception as exc: # 下单异常要打出来, 别默默吞掉 print(f"Order 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_volume

cash_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 线收盘时才出信号，错失了好价格。这个环节能帮你在上实盘之前就修正执行参数。

我还习惯做一次连续多日的“无人值守测试”：让程序在模拟环境下连跑三天，期间故意断一次网、杀一次进程，看恢复后能不能靠对账逻辑自洽。从那以后，我每次上线新策略都强制先跑一遍这个流程，确认日志干净、持仓一致，才会把资金切进去。希望这包代码能帮到一个像当年的我那样到处找自动交易入口的开发者；先把执行这层做扎实，策略本身的价值才真正发挥得出来。

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