简介这是一套基于Python开发的华泰证券通达信客户端自动化交易工具面向量化初学者、个人投资者及金融IT实践者解决手动盯盘下单效率低、响应慢的问题。工具支持同时监控5只股票依据价格与时间双重条件自动触发买卖指令单次下单耗时低于2秒适配Windows 10 64位环境亦兼容32位Python依赖pywin32与tushare实现界面交互与行情获取。资源包为166KB的ZIP压缩文件共含12个文件包括核心脚本pyautotrade.pyw、封装Windows API的winguiauto.py、说明文档README.md、LICENSE协议文件以及8张配置界面与交易流程截图如trading.png、setting1–7.png直观呈现操作逻辑与UI适配要点。目前已有3227人学习下载读者可直接部署运行获得一套轻量、可调试、带完整界面交互逻辑的实盘级自动化下单原型涵盖从条件判断、窗口控制到订单提交的全流程代码实现。1. 为什么一个“能自动下单”的股票脚本比写十个爬虫更难交付你手头有一份行情数据、一个券商接口文档、一段用 Python 调用 TDX通达信本地数据库的代码甚至还能用pywinauto模拟点击交易软件——但只要答辩老师问一句“如果委托单发出去了没成交系统怎么知道撤单失败了谁来兜底连续三次下单超时是网络问题还是账号被风控”绝大多数课设/毕设就卡在这儿功能看似跑通实则没有状态闭环无法进入真实交易逻辑验证环节。这不是算法题而是一套带状态机、异常熔断、日志可追溯、操作可回滚的轻量级交易执行框架。pyautotrade_tdx这个项目不追求高频、不对接 Level-2 行情、不碰两融或期权它专注解决一个具体场景在 A 股个人投资者常用环境Windows 通达信客户端 某主流券商柜台下实现「信号触发 → 本地行情确认 → 模拟下单 → 成交反馈 → 日志归档」的最小可行闭环。它适合需要交一份「有业务语义、有容错设计、有验收依据」的课程设计或毕业设计的学生也适合某公司内部投研小组想快速验证策略逻辑又不愿采购整套量化平台的工程师。下面这整套方案就是我带过三个类似模拟项目后沉淀下来的、能直接答辩、能现场演示、能经得起老师逐行追问的落地结构。2. 总体架构设计三层结构与四个核心模块的职责划分这套方案不是“写个 while True: check_signal()”的脚本堆砌而是按工业级轻量系统思路拆解数据层 → 策略层 → 执行层 → 监控层。四层之间通过明确定义的数据契约通信避免胶水代码泛滥也方便后续替换模块比如把 TDX 换成 Tushare或把 pywinauto 换成券商官方 API。2.1 数据层TDX 本地数据库的稳定读取与缓存机制通达信的.tdx文件本质是二进制结构直接解析易出错。本方案采用pytdx库v1.97作为标准读取器而非自行解析。关键点在于不实时读盘而用内存缓存 定时刷新既规避文件锁冲突又降低 I/O 压力。# data_loader.py from pytdx.hq import TdxHq_API from pytdx.exhq import TdxExHq_API import time from threading import Lock class TDXDataCache: def __init__(self, ip127.0.0.1, port7709): self.api TdxHq_API(heartbeatFalse) self.ip ip self.port port self._cache {} self._lock Lock() self._last_update 0 self._update_interval 30 # 秒级缓存避免频繁连接 def connect(self): try: if not self.api.connect(self.ip, self.port): raise ConnectionError(TDX Hq API connect failed) except Exception as e: raise ConnectionError(fTDX connection error: {e}) def get_security_quotes(self, market, codes): 批量获取最新行情带缓存命中判断 cache_key f{market}_{,.join(codes)} now time.time() with self._lock: if (cache_key in self._cache and now - self._cache[cache_key][timestamp] self._update_interval): return self._cache[cache_key][data] # 缓存失效重新拉取 try: quotes self.api.get_security_quotes([(market, code) for code in codes]) if quotes is None: raise ValueError(TDX get_security_quotes returned None) result {q[code]: q for q in quotes} with self._lock: self._cache[cache_key] { data: result, timestamp: now } return result except Exception as e: # 缓存未命中时返回上一次有效缓存降级 with self._lock: if cache_key in self._cache: return self._cache[cache_key][data] raise e提示pytdx的get_security_quotes是通达信客户端后台服务暴露的协议接口必须确保通达信 PC 版已启动且开启“允许远程控制”设置 → 系统设置 → 其他 → 允许远程控制。这是学生最容易忽略的前置条件导致connect()一直超时却查不出原因。该模块设计要点连接复用TdxHq_API实例全局单例避免反复建连缓存降级网络异常时返回旧缓存保障策略层不因数据源抖动而中断市场编码映射market0表示深市1表示沪市需在调用前统一转换避免传错导致空数据异常分类捕获区分ConnectionError网络/服务未启、ValueError协议返回异常、KeyError代码不存在为上层提供明确错误类型。2.2 策略层信号生成与过滤的可插拔设计策略不写死在主循环里而是定义为独立类实现统一接口execute()返回标准化信号字典。这样便于单元测试、AB 对比、以及答辩时快速切换策略演示。# strategy/base_strategy.py from abc import ABC, abstractmethod from typing import Dict, List, Optional class BaseStrategy(ABC): abstractmethod def execute(self, quotes: Dict[str, dict]) - List[Dict]: 输入{code: {price: xx, vol: yy, ...}} 格式行情 输出[{code: 000001, action: buy, volume: 100, price: 10.25}, ...] pass # strategy/macd_crossover.py class MACDCrossOver(BaseStrategy): def __init__(self, short_period12, long_period26, signal_period9): self.short_p short_period self.long_p long_period self.signal_p signal_period def execute(self, quotes: Dict[str, dict]) - List[Dict]: signals [] for code, q in quotes.items(): # 此处简化实际应调用 talib 计算 MACD此处仅示意逻辑分支 if not all(k in q for k in [price, open, high, low]): continue # 示例规则收盘价上穿 5 日均线且 MACD 柱由负转正 ma5 (q[open] q[high] q[low] q[price]) / 4 # 简化均值 if q[price] ma5 and q.get(macd_hist, 0) 0: signals.append({ code: code, action: buy, volume: 100, price: round(q[price] * 1.002, 2), # 挂单价略高于现价 reason: MACD金叉价格突破MA5 }) return signals注意策略类中禁止直接调用print()或写文件——所有日志必须走统一logging模块否则答辩时无法证明信号生成过程可审计。reason字段是答辩加分项体现业务理解深度。该模块设计要点输入输出契约严格行情字典字段名、信号字典结构全项目统一避免执行层解析失败策略隔离每个策略文件只含一个类无全局变量支持热加载答辩时可现场改参数 reload价格精度处理A 股价格最小变动单位为 0.01 元round(price, 2)是硬性要求否则券商接口会拒单Volume 合法性校验需检查是否为 100 的整数倍A 股最小交易单位否则下单直接失败。2.3 执行层Pywinauto 模拟交易的精准控制与状态反馈这是整个方案最易翻车的模块。pywinauto不是万能鼠标录制器它依赖窗口句柄、控件 ID、文本匹配三重定位。本方案采用“窗口树遍历 文本模糊匹配 控件ID fallback”三级定位策略并内置状态等待与超时熔断。# executor/tdx_executor.py from pywinauto import Application from pywinauto.timings import wait_until, TimeoutError import re class TDXExecutor: def __init__(self, app_pathrC:\TdxW\TdxW.exe): self.app_path app_path self.app None self.main_dlg None def launch_or_connect(self): try: # 尝试连接已运行进程 self.app Application(backenduia).connect(pathself.app_path) except Exception: # 启动新进程 self.app Application(backenduia).start(self.app_path) # 等待主窗口出现标题含通达信 self.main_dlg self.app.window(title_re.*通达信.*) wait_until(30, 0.5, lambda: self.main_dlg.exists()) self.main_dlg.set_focus() def _find_trade_window(self): # 通达信交易窗口标题通常为网上交易V6或XX证券 trade_wins self.app.windows(title_re.*网上交易.*|.*证券.*, found_index0) if not trade_wins: raise RuntimeError(未找到交易窗口请确认已登录交易软件) return trade_wins[0] def place_order(self, code: str, action: str, volume: int, price: float) - Dict: 返回标准化结果字典含 status, order_id, message, timestamp result { status: failed, order_id: , message: , timestamp: time.time() } try: trade_dlg self._find_trade_window() # 定位买卖按钮通达信常用文本为买入/卖出位置相对固定 buy_btn trade_dlg.child_window(title买入, control_typeButton, found_index0) sell_btn trade_dlg.child_window(title卖出, control_typeButton, found_index0) # 输入代码框通常在左上角文本含证券代码 code_edit trade_dlg.child_window(title_re.*证券代码.*|.*代码.*, control_typeEdit).wrapper_object() price_edit trade_dlg.child_window(title_re.*委托价格.*|.*价格.*, control_typeEdit).wrapper_object() vol_edit trade_dlg.child_window(title_re.*委托数量.*|.*数量.*, control_typeEdit).wrapper_object() # 清空并填入 code_edit.set_text(code) price_edit.set_text(str(price)) vol_edit.set_text(str(volume)) # 点击对应按钮 if action buy: buy_btn.click_input() else: sell_btn.click_input() # 等待弹窗提示成功/失败 success_msg trade_dlg.child_window(title_re.*成功.*|.*已提交.*, control_typeText) fail_msg trade_dlg.child_window(title_re.*失败.*|.*错误.*|.*拒绝.*, control_typeText) wait_until(10, 0.3, lambda: success_msg.exists() or fail_msg.exists()) if success_msg.exists(): result[status] success result[message] success_msg.window_text() # 从提示中提取订单号通达信常为订单号123456789 match re.search(r订单号[:]\s*(\d), success_msg.window_text()) result[order_id] match.group(1) if match else else: result[message] fail_msg.window_text() if fail_msg.exists() else 未知错误 except TimeoutError: result[message] 操作超时未在10秒内检测到交易结果弹窗 except Exception as e: result[message] f执行异常{str(e)} return result提示backenduia是关键通达信新版使用 UI Automation 协议若用默认win32后端大部分控件无法识别。务必在Application()初始化时显式指定。该模块设计要点控件定位健壮性不用绝对坐标易因分辨率/缩放失效而用title_re模糊匹配 control_type类型约束状态等待非轮询wait_until()内置指数退避比time.sleep()更可靠订单号提取通达信弹窗文本格式不统一用正则提取比字符串切片更鲁棒失败信息结构化result[message]必须包含可读错误原文方便答辩时展示“系统如何诊断问题”。2.4 监控层日志归档、成交核对与简易风控开关监控不是加个print()而是构建可回溯的操作证据链。本方案采用“操作日志 成交查询日志 风控事件日志” 三分离设计全部写入本地 SQLite 数据库支持按日期、代码、状态快速检索。# monitor/db_manager.py import sqlite3 from datetime import datetime class DBManager: def __init__(self, db_pathtrade_log.db): self.db_path db_path self.init_db() def init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS operation_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, code TEXT NOT NULL, action TEXT NOT NULL, volume INTEGER NOT NULL, price REAL NOT NULL, status TEXT NOT NULL, order_id TEXT, message TEXT, strategy TEXT ) ) conn.execute( CREATE TABLE IF NOT EXISTS deal_check_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, order_id TEXT NOT NULL, code TEXT NOT NULL, dealt_volume INTEGER, dealt_price REAL, status TEXT NOT NULL, check_source TEXT -- tdx_local or broker_api ) ) conn.commit() def log_operation(self, record: dict): with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO operation_log (timestamp, code, action, volume, price, status, order_id, message, strategy) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?), ( datetime.now().strftime(%Y-%m-%d %H:%M:%S), record[code], record[action], record[volume], record[price], record[status], record.get(order_id, ), record.get(message, ), record.get(strategy, unknown) ) ) conn.commit()注意SQLite 是单文件嵌入式数据库无需额外服务完美适配课设部署场景。但需在项目启动时检查写权限——若运行在受限账户如学校机房db_path应指向用户目录os.path.expanduser(~/trade_log.db)否则首次写入会静默失败。该模块设计要点日志即证据每笔委托、每次成交核对、每个风控触发都落库答辩时可导出 CSV 展示完整流水时间戳标准化统一用datetime.now()生成避免系统时区差异字段语义清晰check_source区分本地 TDX 查询 vs 券商接口查询体现多源验证意识风控开关外置在配置文件中设max_daily_orders5执行层每次下单前查库统计当日已下单数超限则跳过并记风控日志。3. 关键实现细节与避坑指南那些让答辩老师眼前一亮的细节这个模块不是罗列“注意事项”而是直击学生在调试、演示、答辩中最可能当场卡壳的 4 个血泪现场。每一条都来自真实翻车记录解决方案经过三次以上项目验证。3.1 现象pywinauto找不到“买入”按钮child_window(title买入)返回空列表原因通达信交易窗口存在多层嵌套且部分券商定制版将按钮文本改为“买入(B)”或“B 买入”纯字符串匹配失效更隐蔽的是窗口未激活时UIA 后端可能无法枚举子控件。解决第一步强制激活窗口trade_dlg.set_focus()后加trade_dlg.wait(ready, timeout5)第二步放宽匹配title_re.*买入.*|.*B.*|.*buy.*并增加found_index0防止匹配到菜单栏第三步终极兜底用trade_dlg.descendants(control_typeButton)获取所有按钮再用btn.window_text()循环比对是否含“买”字中文环境。3.2 现象委托单显示“已提交”但 TDX 本地成交查询查不到日志里dealt_volume始终为 0原因TDX 本地数据库的成交表vipdoc\sh\lday\或sz\lday\更新有延迟通常滞后 1~3 分钟而你的程序在下单后立即查库自然为空。这不是程序 bug而是数据源特性。解决在deal_check_log表中增加check_delay_sec字段记录本次查询距下单的时间差实现“延迟重查”逻辑首次查不到时记录statuspending启动后台线程每隔 60 秒重查一次最多查 5 次覆盖 5 分钟窗口答辩时可演示下单后日志显示pending1 分钟后自动更新为filled体现对真实市场数据特性的尊重。3.3 现象策略信号频繁触发但实际只成交了一半operation_log中大量statusfailedmessage却是空字符串原因pywinauto的click_input()在窗口被其他程序遮挡时会静默失败message未被捕获更常见的是通达信弹窗出现时焦点不在交易窗口导致child_window()查找不到控件。解决在place_order()开头强制trade_dlg.set_focus()并在click_input()后加time.sleep(0.3)防止操作过快所有except Exception as e:必须包含result[message] f执行异常{type(e).__name__}: {str(e)}杜绝空消息增加“操作录像”开关启用时用pywinauto.keyboard.send_keys({PRTSC})截图保存命名含时间戳答辩时可展示“失败时刻的屏幕状态”。3.4 现象程序运行 2 小时后报错OSError: [WinError 1450] 系统资源不足无法完成请求的服务原因pywinauto每次connect()或start()都创建新进程句柄但未显式app.kill()或app.disconnect()导致 Windows 句柄泄漏同时TdxHq_API连接未关闭占用 socket。解决在主程序退出前atexit.register()或try/finally执行if executor.app: executor.app.kill() # 强制结束进程 if data_cache.api: data_cache.api.disconnect() # 主动断开 TDX 连接在TDXExecutor.launch_or_connect()中每次connect()前先try: app.kill()避免残留进程答辩演示时务必展示“程序优雅退出”操作如按 CtrlC并检查任务管理器确认无残留TdxW.exe进程。提示这些坑90% 的课设代码里都存在。你在答辩 PPT 里专门放一页《我们如何应对真实环境挑战》逐条列出上述问题及解决方案老师会立刻意识到这不是玩具脚本而是认真考虑过生产约束的工程实践。4. 配置管理与策略参数化让同一套代码适配不同券商与策略硬编码是课设答辩的隐形杀手。broker_name XX证券、tdx_ip 127.0.0.1、macd_short 12这些值一旦写死老师问“换一家券商怎么改”你就得现场翻代码。本方案采用分层配置 策略参数注入所有可变项集中管理。4.1 配置文件结构config.yaml与环境隔离项目根目录下config.yaml定义全局配置按环境分组# config.yaml default: default tdx: host: 127.0.0.1 port: 7709 market_map: sh: 1 sz: 0 broker: name: XX证券 trade_window_title: .*XX证券.*|.*网上交易.* risk_control: max_daily_orders: 10 max_single_position: 50000 enable_auto_cancel: true development: : *default log_level: DEBUG strategy: macd_crossover strategy_params: short_period: 12 long_period: 26 production: : *default log_level: INFO strategy: rsi_overbought strategy_params: rsi_period: 14 overbought_threshold: 70 oversold_threshold: 30加载逻辑封装在config_loader.py# config_loader.py import yaml import os from pathlib import Path class ConfigLoader: def __init__(self, envdevelopment): self.env env self.config self._load_config() def _load_config(self): config_path Path(__file__).parent / config.yaml with open(config_path, r, encodingutf-8) as f: full_config yaml.safe_load(f) return full_config[self.env] def get(self, key, defaultNone): keys key.split(.) val self.config try: for k in keys: val val[k] return val except (KeyError, TypeError): return default # 使用示例 cfg ConfigLoader(envos.getenv(TRADE_ENV, development)) tdx_host cfg.get(tdx.host) rsi_period cfg.get(strategy_params.rsi_period, 14)注意os.getenv(TRADE_ENV)允许通过系统环境变量切换配置答辩时只需在命令行执行set TRADE_ENVproduction python main.py即可演示不同策略无需改代码。4.2 策略参数注入策略类构造函数接收配置策略类不再写死参数而是从配置中读取# strategy/rsi_overbought.py class RSIOverbought(BaseStrategy): def __init__(self, rsi_period14, overbought_threshold70, oversold_threshold30): self.rsi_period rsi_period self.overbought_threshold overbought_threshold self.oversold_threshold oversold_threshold # main.py 中动态实例化 from strategy import rsi_overbought cfg ConfigLoader() strategy_name cfg.get(strategy) strategy_params cfg.get(strategy_params, {}) if strategy_name rsi_overbought: strategy rsi_overbought.RSIOverbought(**strategy_params) elif strategy_name macd_crossover: strategy macd_crossover.MACDCrossOver(**strategy_params)4.3 券商适配配置交易窗口与控件定位规则外置不同券商交易窗口标题、按钮文本、输入框标签差异巨大。本方案将定位规则写入broker_configs/目录下的 YAML 文件# broker_configs/xx_securities.yaml trade_window_title: .*XX证券.*|.*网上交易V6.* code_input_label: .*证券代码.*|.*股票代码.* price_input_label: .*委托价格.*|.*买入价格.* volume_input_label: .*委托数量.*|.*卖出数量.* buy_button_text: .*买入.*|.*B.* sell_button_text: .*卖出.*|.*S.* success_message_pattern: .*已提交.*|.*成功.*执行层根据cfg.get(broker.name)动态加载对应配置实现“一套引擎多券商支持”。提示答辩时可准备两张截图一张是XX证券的交易窗口一张是YY证券的窗口然后指出配置文件中对应的title_re和label规则证明架构的可扩展性。这比讲一百遍“高内聚低耦合”都有力。5. 验收标准与答辩技巧如何让老师说“这确实是个完整系统”验收不是“程序跑起来就行”而是要证明它满足一套可度量、可演示、可复现的标准。我把这套标准拆成功能验收、鲁棒性验收、可维护性验收三张表每张表都对应答辩时的具体动作。你照着做老师挑不出毛病。5.1 功能验收五步现场演示清单10 分钟内完成步骤操作预期结果答辩话术要点1. 环境自检运行python health_check.py输出✅ TDX 连接正常、✅ 交易窗口可访问、✅ 配置文件加载成功、✅ SQLite 数据库可写入“老师这是我们的健康检查脚本确保所有依赖就绪。您看TDX 服务、交易软件、数据库权限全部通过。”2. 信号生成修改config.yaml中strategy_params如将short_period从 12 改为 20重启程序日志中出现INFO strategy.macd_crossover: MACD金叉价格突破MA5 for 000001“策略参数完全外部化修改后无需改代码重启即生效。这里我调高了短周期信号变少体现参数敏感性。”3. 模拟下单在通达信中手动挂一笔 100 股的测试单不成交观察程序日志日志中operation_log新增一条statussuccess记录order_id非空“注意看这条记录order_id是从通达信弹窗中正则提取的证明我们不只是点了按钮还拿到了系统返回的凭证。”4. 成交核对等待 2 分钟运行python check_deals.py --since 5m输出Found 1 deal for order 123456789: 10010.25且deal_check_log中新增记录“成交查询是异步的我们设置了 5 分钟窗口。您看这笔单子已确认成交价格和数量完全匹配。”5. 风控触发将max_daily_orders设为 1连续运行两次下单第二次日志中statusblockedmessage显示“今日订单已达上限”“风控不是摆设。当达到预设阈值系统主动拦截并留下审计日志符合金融系统基本安全要求。”血泪经验演示前务必用git clean -fdx清理所有.pyc和__pycache__再pip install -r requirements.txt重装依赖。曾有同学演示时因缓存旧版本pywinauto导致按钮找不到全场尴尬。5.2 鲁棒性验收三类必测异常场景答辩老师最爱问“如果……怎么办” 你得提前准备好答案并最好能现场演示。以下是三个必答场景附代码级应对方案场景一TDX 服务意外关闭操作演示中手动关闭通达信再触发下单。预期程序不崩溃日志输出ERROR data_loader: TDX connection error: ...并自动降级使用缓存行情若存在。话术“我们实现了连接降级。当 TDX 不可用策略仍能基于最后缓存的价格运行避免停摆。当然长期离线需告警这已在监控模块预留alert_on_cache_fallback配置项。”场景二交易窗口被微信遮挡操作演示前打开微信窗口盖住通达信再下单。预期place_order()中trade_dlg.set_focus()自动激活窗口操作继续成功。话术“UI 自动化最大的敌人是窗口焦点。我们每步操作前都强制聚焦确保控件可访问。这也是为什么我们不用坐标点击——它无法解决焦点问题。”场景三委托价格非 0.01 整数倍操作在策略中故意返回price10.255触发下单。预期执行层round(price, 2)自动修正为10.26日志记录WARN executor: Price rounded from 10.255 to 10.26。话术“A 股价格精度是硬性合规要求。我们在执行层做了兜底修正并记录警告确保不因策略计算误差导致下单失败。”5.3 可维护性验收答辩时必展示的三个文件老师判断你是不是真写了就看这三个文件是否真实存在、内容是否专业requirements.txt必须包含精确版本号如pywinauto0.6.10、pytdx1.97。不能写pywinauto0.6否则答辩环境版本不一致会翻车。README.md第一行必须是# pyautotrade_tdx接着是安装步骤含 TDX 设置说明、配置说明config.yaml字段解释、演示命令python main.py --env development。test/目录下的单元测试至少包含test_strategy.py验证 MACD 信号生成、test_executor.pyMock 通达信窗口测试下单逻辑、test_db.py验证日志写入。运行pytest test/应全部通过。从那以后我每次交课设都强制走一遍这三张表功能五步、鲁棒三问、可维护三件套。哪怕只有 30 分钟准备时间我也先确保这 11 件事全部就绪。因为答辩不是考你多懂而是考你多稳——稳就是把所有‘万一’都变成‘已处理’。希望帮到你。本文还有配套的精品资源点击获取