简介这是一套面向Python开发者与量化交易初学者的股票自动化实战源码聚焦通达信数据对接、北向/南向资金分析、早盘异动监控及可转债联动选股等高频交易场景解决手动盯盘效率低、策略执行滞后等实际痛点。资源共77个文件含49个Python脚本覆盖数据采集、策略执行、UI逻辑、3个Qt Designer设计的.ui界面文件、2个HTML前端页面、6个XML配置文件及SQL数据库脚本辅以日志文本、图标图片与Jupyter分析笔记压缩包大小为74.09MB。已有822人学习下载内容结构清晰核心模块如‘龙头盯盘’‘操盘神器ui’‘北向资金分析工具’均提供完整可运行代码配套requirements.txt与ini配置说明支持快速部署同时包含通达信早盘数据入库、微信通知、语音文字互转等增强型实用功能兼顾策略开发与工程落地需求。1. 项目概述1.1 核心需求解析先说结论这个项目做的是一件很多人想做但没坚持下来的事——把盯盘、看信号、手动下单这一整套流程用 Python 和 HTML 串成一个自动化的闭环。项目名字里的 TDXPystock可以理解成一套围绕通达信体系行情数据、基于 Python 封装的股票分析组件库后端负责数据抓取、指标计算、信号生成和模拟执行前端用 HTML 搭建实时监控面板让整个交易决策过程变成一个可视化、可回测、可追溯的工程系统。动手做之前我把需求拆成了四个层面数据层自动拉取行情数据包括日线、分钟线和实时快照覆盖自选股池。策略层把均线、MACD、RSI 这类指标写成可复用函数生成买卖信号。执行层模拟交易引擎按照信号执行虚拟下单、持仓管理和盈亏统计。展示层HTML 网页实时展示行情、信号、持仓曲线浏览器打开即可查看。这个项目最适合两类人一类是有 Python 基础、想接触量化交易但不知道从哪下手的开发者另一类是长期用通达信看盘、对指标公式有概念、想用代码把人工规则固化的交易者。前者能学会怎么把零散的行情数据变成工程化系统后者能把我感觉要涨了变成信号触发阈值 0.86连续 3 根 K 线确认。1.2 为什么选 Python HTML 这套组合选型的时候纠结过几个方案用 PyQt 写桌面端、用 Django 搭完整 Web 应用、或者纯 Python 脚本跑完在终端打印结果。但最终定下来 Python 做逻辑、HTML 做界面原因很实际Python 在行情处理和策略回测方面生态太成熟了pandas 处理 K 线数据、numpy 做数值计算几行代码就能完成复杂的指标运算。HTML 页面不需要安装客户端手机和电脑都能通过浏览器访问策略跑在服务器上看盘设备只要能打开网页就行跨平台成本几乎是零。用 Django 或 Flask 渲染整个后台对交易工具来说太重了。交易面板的核心是信息密度高、刷新及时、一眼看懂纯 HTML 加少量 Ajax 就够用没必要为了架构好看而堆框架。架构上我划分了四个独立模块模块之间通过标准化的数据接口通信模块职责技术选型数据采集模块获取历史 K 线与实时行情Python requests pandas策略计算模块指标计算与信号生成Python numpy核心指标封装成类模拟交易模块虚拟持仓管理与成交撮合Python 自研引擎独立数据库存储可视化面板行情与持仓监控HTML ECharts 原生 JavaScript模块之间解耦的好处是如果将来想换数据源只需要重写数据采集模块策略层和展示层完全不用动。这个设计在后面实际开发中帮我省了无数次事。2. 核心模块设计与实现要点2.1 行情数据模块兼容通达信体系的 K 线处理数据是整个系统的基础数据不对后面全部白算。我处理数据时参考了通达信体系常见的本地数据结构思路但实际落地时改成了更通用的文件格式存储每只股票一个 CSV 文件字段包含 date、open、high、low、close、volume、amount 这几列。之所以不用数据库是因为 CSV 文件用 pandas 读取速度快、调试方便、出了问题可以直接用文本编辑器打开检查。数据更新采用增量拉取机制。首次运行时全量下载近三年的日线数据之后每次启动只拉取最近五个交易日的数据做增量补齐。这里有个关键细节停牌股票当天没有数据如果直接按当天缺失就跳过会把除权除息的日子弄错。我的处理方式是把最近拉到的数据与本地已有数据做日期对比只追加新日期保证了 K 线的连续性。数据校验方面我也是踩了坑才补上的逻辑。有一回策略跑出来的信号特别密集排查半天发现是某天的 high 数值比 low 还小属于数据源脏数据。后来加了过滤规则high 必须大于等于 low、volume 必须大于零、日期必须是递增的不合法的直接丢弃该记录并打日志告警。对于拿到 TDX 格式本地数据的场景还需要额外处理复权因子前复权和后复权在回测中的结果差异非常大这在后面回测部分会细说。2.2 策略信号模块从指标函数到信号生成策略模块是系统的大脑。我先把常见指标都拆成独立的纯函数输入是 DataFrame输出是计算结果列。比如简单移动平均的实现def sma(data: pd.DataFrame, window: int) - pd.Series: return data[close].rolling(windowwindow, min_periods1).mean()MACD 的计算稍微复杂一点需要计算快线、慢线和柱状值def macd(data: pd.DataFrame, fast: int 12, slow: int 26, signal: int 9): ema_fast data[close].ewm(spanfast, adjustFalse).mean() ema_slow data[close].ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() histogram (dif - dea) * 2 return dif, dea, histogram指标函数写好之后信号生成是另一层逻辑。以双均线策略为例金叉只是辅助判断真正触发交易还需要成交量配合和价格位置确认这样能过滤掉大量无效信号。信号生成器的输出统一成字典格式包含方向buy/sell、强度、触发原因和当前指标快照方便后续在 HTML 面板上展示。一个容易被忽略的细节是信号去重。比如均线金叉出现后如果价格一直在均线上方震荡策略可能会在每一根 K 线都产生一次买入信号。所以我加了状态机只有当当前状态是空仓时才响应买入信号持仓状态下只响应卖出信号。这个逻辑极大减少了日志里的噪音也让模拟交易的持仓记录变得干净。2.3 可视化面板模块HTML ECharts 实时监控HTML 面板是整个项目最有成就感的部分。页面加载之后左侧是自选股列表中间是 K 线图和成交量图右侧是实时信号记录和当前持仓信息。整体布局就是一张大屏打开就能看。K 线图用的是 ECharts 的 candlestick 组件数据由 Python 后端以 JSON 格式通过接口返回。前端每五秒轮询一次最新行情接口拿到增量数据后调用 setOption 更新图表。这里有个性能方面的心得每次轮询不要刷新整个图表只更新最后一个点的数据否则鼠标悬停查看历史数据时会经常被强制刷新打断体验很差。信号展示区域我单独做了一个滚动列表每条信号记录包含时间、股票代码、信号方向、触发原因和当时的价格。为了直观买入信号用红底白字卖出信号用绿底白字国内习惯红涨绿跌这个细节还挺多朋友问的。前端逻辑全部用原生 JavaScript 实现不引框架就引了一个 ECharts 的 CDN。原因很简单页面功能只有轮询数据 更新图表 渲染列表引入 Vue 或 React 反而增加了打包和部署成本没必要。2.4 模拟交易模块风险控制与持仓管理模拟交易模块严格来说是一个简化版的撮合引擎。收到买入信号后按当前价加一个滑点比例计算成交价更新账户的持仓和可用资金收到卖出信号则按当前价减滑点计算成交价落袋盈亏。每次成交都会写入交易记录表包含时间、方向、价格、数量、手续费和备注。资金管理上我设置了最大单票仓位 20%、单日最大回撤 5% 自动停止新开仓、单笔止损 8% 强制平仓三条风控规则。有一段时间我发现模拟账户的收益曲线波动特别剧烈复盘后发现是因为单票仓位太重一只票的涨跌就能带动整个账户大起大落。限仓之后曲线平滑了很多虽然单次收益变小了但回撤也显著降低。这个模块还顺带解决了一个需求每天开盘前自动加载前一日所有持仓股票的公告和停复牌信息如果发现持仓中有停牌股会在面板上打上醒目标记提醒手动处理。3. 实操过程与关键代码解析3.1 环境准备与项目初始化实际搭建这个项目我建议直接用 Anaconda 管理 Python 环境避开不少 Windows 上安装包的坑。Python 版本用的 3.10依赖库版本也锁定了一个稳定的组合pip install pandas numpy requests flask apscheduler后端接口我用 Flask 起了一个轻量服务就两个路由一个返回全部 K 线数据一个返回最新行情和信号记录。不选 FastAPI 的原因是我对异步没有硬需求Flask 本身就是够了体积也小。项目目录结构建议这样组织tdxpystock/ ├── data/ # 行情数据 CSV 存储目录 ├── strategies/ # 策略指标代码 │ ├── indicators.py │ └── signals.py ├── trader/ # 模拟交易与风控 │ ├── portfolio.py │ └── risk.py ├── web/ # HTML 前端文件 │ └── index.html ├── server.py # Flask 后端入口 └── main.py # 程序主流程3.2 行情数据采集实操数据采集这块我单独封装了一个类统一管理行情源的接入。核心方法是获取日 K 线数据和获取实时快照class MarketData: def __init__(self, data_dir: str data): self.data_dir data_dir os.makedirs(data_dir, exist_okTrue) def get_kline(self, symbol: str, days: int 750) - pd.DataFrame: file_path os.path.join(self.data_dir, f{symbol}.csv) if os.path.exists(file_path): df pd.read_csv(file_path, parse_dates[date]) else: df self.download_kline(symbol, days) self.save_to_csv(symbol, df) return df[(df[date] (pd.Timestamp.now() - pd.Timedelta(daysdays)))] def update_realtime(self, symbol: str) - dict: # 拉取实时行情返回最新价、成交量等 pass我写的 download_kline 里面用的是第三方行情接口通过 requests 调用返回 JSON 后解析成 DataFrame。这里有个实际经验接口返回的数据量较大时一定做好异常重试和超时处理。我给 requests 设置了 timeout10连续三次失败就直接跳过本只股票的本次更新避免因为一只票拉不到数据导致整个循环卡死。另一个细节是 pandas 的列名处理。各家接口返回的字段名千奇百怪有返回大写 Open、High 的也有返回小写 open、high 的甚至有的时间字段叫 timestamp。我统一在 download_kline 里做了一层字段映射保证落盘的数据永远是统一的 schema这样策略层永远不会因为列名问题报 KeyError。3.3 策略信号生成实操策略层我写了一个 SignalEngine 类把指标计算和信号判定分成两步class SignalEngine: def __init__(self, indicators: dict): self.indicators indicators def compute(self, data: pd.DataFrame) - pd.DataFrame: result data.copy() for name, params in self.indicators.items(): if name sma: result[fsma_{params[window]}] sma(result, params[window]) elif name macd: result[dif], result[dea], result[hist] macd(result) elif name rsi: result[frsi_{params[window]}] rsi(result, params[window]) return result def generate_signals(self, data: pd.DataFrame) - list: # 基于计算指标生成信号列表 pass双均线策略的信号判定逻辑是核心。我的流程是这样的计算 5 日均线和 20 日均线。判断是否发生金叉昨天的 5 日均线小于等于昨天的 20 日均线今天的 5 日均线大于今天的 20 日均线。金叉确认后查看当日成交量是否大于过去 5 日均量的 1.2 倍。如果量能配合再检查当前价格是否站在 20 日均线上方。四重条件全部满足才生成买入信号。这样写出来的信号数量虽然比只看金叉少但胜率明显更好模拟交易跑了三个月胜率大概在 58% 左右盈亏比在 1.6 附近。需要说明的是这个结果只基于特定行情区间不要觉得 58% 就能直接拿到实盘去用。信号生成器还有一个冷却时间参数默认买入信号触发后 120 分钟内不再重复触发同一股票的买入。这个机制是用来防止信号闪烁问题的也就是价格在临界点来回波动导致频繁开仓。3.4 可视化面板实操HTML 面板的页面结构是这样的!DOCTYPE html html langzh-cn head meta charsetutf-8 titleTDXPystock 交易监控面板/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script style /* 页面布局样式 */ /style /head body div idsidebar h3自选股/h3 ul idsymbol-list/ul /div div idmain-chart h3 idcurrent-symbol--/h3 div idkline-chart/div /div div idsignal-panel h3最新信号/h3 ul idsignal-list/ul /div script srcapp.js/script /body /html前端的关键逻辑是数据轮询和图表更新。我写了一个 updateData 函数每次请求后端的 /api/market_data 接口拿到最新的 K 线 JSON 后直接更新图表。K 线数据的数据结构是{ dates: [2025-01-02, 2025-01-03], kline: [[3.65, 3.72, 3.58, 3.70, 112233], ...], volumes: [112233, ...] }ECharts 的 candlestick 组件接收的数据顺序是 [open, close, lowest, highest]这个顺序很容易搞错我第一次开发时就是因为顺序写反了K 线图看着完全变形。后来我干脆在 Python 端就生成好 ECharts 需要的顺序前端不再做任何数据处理减少出错的可能。轮询频率我设置在 http://192.168.1.100:80805 秒一次白天跑一整天也不会有明显资源占用。如果策略逻辑再重一些可以把轮询改成 WebSocket 推送模式但现在的实现已经能满足需求没必要过度设计。3.5 主流程整合与定时调度主流程通过 APScheduler 做定时调度设置了三个任务每个交易日的 09:25 拉取集合竞价结束后的行情快照更新当日开盘价。盘中每 5 秒拉取一次实时行情并交给策略引擎判断是否有信号。每天 15:10 收盘后做一次全量数据更新然后执行一次全持仓对账。调度任务的代码如下from apscheduler.schedulers.blocking import BlockingScheduler def create_scheduler(engine, market_data, portfolio): scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.scheduled_job(cron, day_of_weekmon-fri, hour9,10,11,13,14, minute*, second*/5) def realtime_job(): for symbol in portfolio.watchlist: quote market_data.update_realtime(symbol) signals engine.generate_signals(quote) portfolio.execute_signals(signals) scheduler.scheduled_job(cron, day_of_weekmon-fri, hour15, minute10) def daily_close(): market_data.update_all() portfolio.sync_positions() return scheduler需要注意的是交易日历的问题。我一开始没有处理节假日结果国庆节后第一天系统 9:25 触发任务发现所有股票的实时接口都返回空数据。后来引入了交易日历库判断当天是不是交易日不是交易日就跳过所有盘中任务是交易日才正常执行。这一个改动让系统在节假日和周末跑得特别安静日志里也不会刷出一堆无效拉取记录。4. 常见问题与排查技巧实录4.1 数据源返回的数据出现空值或异常值现象某只股票的 K 线数据里出现 NaN 值策略计算时直接报错或者指标结果异常偏大。排查过程第一次遇到时我以为是数据源的问题后来加了日志才发现是股票停牌日返回了空成交量而我在合并数据时没有做填充。还有一个更隐蔽的情况是股票当天不交易但数据源返回了前一天的旧数据如果不判断日期会把这根假 K 线当成当天数据参与计算。解决方案在数据入库前增加空值检测close 为空、volume 为空、日期不为交易日的记录直接过滤。对分钟线数据非交易时段自动跳过。设置数据质量监控脚本每天收盘后检查 CSV 文件的最后一行日期是否为当天如果不是就发送告警日志。4.2 模拟交易持仓与手动操作不一致现象模拟账户显示持仓 1000 股但实际上行情接口返回的持仓文件里是 0 股两边对不上。排查过程这个问题我是折腾了很久才找到原因的。原来模拟交易引擎在收到卖出信号后先从持仓里扣减了数量但没有立即落盘而是等到收盘统一写文件。如果中途程序崩溃内存里的持仓数据就丢掉了落盘的文件还停留在卖出之前的状态。解决方案每次成交立即同步持仓文件不等到收盘。增加交易流水表每次交易都记录前后持仓变化方便回溯。每日启动时先做一次全量对账发现差异就打印告警并自动按流水重建持仓。4.3 HTML 页面无法打开或图表不显示现象浏览器打开页面背景正常但 K 线图区域是空的控制台报错。排查过程这类问题九成出在数据格式上。有一次后端接口返回的日期字段是2025/01/02格式前端 JS 直接 new Date 解析后变成了 Invalid Date导致 ECharts 的 xAxis 数据全是 null图表就空了。解决方案后端统一输出 ISO 格式的日期字符串比如2025-01-02前端直接使用。在 JS 里增加数据校验拿到数据先判断数组长度长度为零时显示占位提示而不是让图表白屏。把 ECharts 初始化代码包裹在 window.onload 里避免 DOM 还没渲染完就执行脚本。4.4 策略回测表现好但模拟交易表现差现象同一套策略历史回测年化收益 25%放到模拟盘跑一个月只有 5%回撤还比回测大了不少。排查过程回测和模拟盘之间最大的差异因素有三个——滑点、手续费、信号延迟。回测时我用的是收盘价直接成交完全忽略了真实交易中的滑点和冲击成本模拟盘我加了千分之一的滑点后很多本来就贴着均线波动的信号就很容易被触发后又反向打掉。解决方案回测中加入固定滑点 0.2% 和双边手续费 0.1%让自己的预期更接近真实。加一个信号过滤条件如果当前价格距离均线小于 0.5%放弃本次信号避免在震荡行情里反复开仓。用模拟盘跑满 30 个交易日再和回测数据做对比如果收益差距超过 10 个百分点首先检查策略的参数是不是已经过拟合了。5. 系统功能扩展方向5.1 增加多策略并行目前信号引擎一次只能跑一种策略组合如果想把双均线策略和 MACD 背离策略同时跑就需要重构信号引擎支持多策略结果合并。我的思路是把每个策略封装成一个独立的 Generator 类输出同样的信号结构汇总层再做信号融合。多个策略同步给出买入信号的票优先执行单一策略信号只作为观察对象不直接下单。这样能减少试错成本胜率也会更稳定。5.2 接入消息通知HTML 面板需要主动打开才能看到信号对于移动场景还是不够方便。可以加一个消息推送扩展当信号产生或持仓达到止损线时通过 Server 酱或企业微信机器人推送到手机。我在实际扩展中发现这类推送在信号产生时作用最大当你有持仓时一个已触发止损的通知比盯一整天盘更高效。实现上就是多一个通知模块在 portfolio.execute_signals 里调用 notify 方法几十行代码的事。5.3 参数寻优与稳健性检查策略参数比如均线窗口、RSI 阈值目前是手工设定的可以加一个简单的参数寻优模块用网格搜索在历史数据上跑一遍找出一组在近三年表现最优的参数。但这里要特别提醒网格搜索出来的最优参数大概率是过拟合的不经过样本外测试就直接实盘会死得很惨。稳妥的做法是留出最近 6 个月的数据做验证集参数在训练集上寻优然后在验证集上确认收益和回撤两项都达标才纳入正式配置。6. 我的实操体会跑这套系统半年多我最大的感受是交易自动化最难的不是写代码而是把模糊的交易规则翻译成精确的代码逻辑。比如感觉要启动了这句话在代码里必须变成5 日均线上穿 20 日均线成交量放大 20%收盘价站上 20 日均线才有意义。另外一点很重要的是自动化系统需要容忍它犯错。我的模拟账户也有阶段性亏损但我不会因为亏损就临时改参数那样做系统就永远处于不稳定状态。每一次参数调整都要做完整的回归测试比较新旧参数在所有历史区间的表现差异再决定是否启用。最后给打算实践这个项目的朋友一个建议第一步不要追求完整的功能体系先做通一个最简单的小闭环——拉取一只股票的数据、算一个均线指标、在 HTML 页面上画出来。当这个最小闭环跑通之后你自然知道下一步该加什么。很多人做量化项目半途而废都是因为从一开始就想做一个完美的系统结果卡在某个环节整整一个周末然后就再也没打开过那个文件夹了。本文还有配套的精品资源点击获取 基于Python和HTML的股票量化交易监控系统开发实践
简介这是一套面向Python开发者与量化交易初学者的股票自动化实战源码聚焦通达信数据对接、北向/南向资金分析、早盘异动监控及可转债联动选股等高频交易场景解决手动盯盘效率低、策略执行滞后等实际痛点。资源共77个文件含49个Python脚本覆盖数据采集、策略执行、UI逻辑、3个Qt Designer设计的.ui界面文件、2个HTML前端页面、6个XML配置文件及SQL数据库脚本辅以日志文本、图标图片与Jupyter分析笔记压缩包大小为74.09MB。已有822人学习下载内容结构清晰核心模块如‘龙头盯盘’‘操盘神器ui’‘北向资金分析工具’均提供完整可运行代码配套requirements.txt与ini配置说明支持快速部署同时包含通达信早盘数据入库、微信通知、语音文字互转等增强型实用功能兼顾策略开发与工程落地需求。1. 项目概述1.1 核心需求解析先说结论这个项目做的是一件很多人想做但没坚持下来的事——把盯盘、看信号、手动下单这一整套流程用 Python 和 HTML 串成一个自动化的闭环。项目名字里的 TDXPystock可以理解成一套围绕通达信体系行情数据、基于 Python 封装的股票分析组件库后端负责数据抓取、指标计算、信号生成和模拟执行前端用 HTML 搭建实时监控面板让整个交易决策过程变成一个可视化、可回测、可追溯的工程系统。动手做之前我把需求拆成了四个层面数据层自动拉取行情数据包括日线、分钟线和实时快照覆盖自选股池。策略层把均线、MACD、RSI 这类指标写成可复用函数生成买卖信号。执行层模拟交易引擎按照信号执行虚拟下单、持仓管理和盈亏统计。展示层HTML 网页实时展示行情、信号、持仓曲线浏览器打开即可查看。这个项目最适合两类人一类是有 Python 基础、想接触量化交易但不知道从哪下手的开发者另一类是长期用通达信看盘、对指标公式有概念、想用代码把人工规则固化的交易者。前者能学会怎么把零散的行情数据变成工程化系统后者能把我感觉要涨了变成信号触发阈值 0.86连续 3 根 K 线确认。1.2 为什么选 Python HTML 这套组合选型的时候纠结过几个方案用 PyQt 写桌面端、用 Django 搭完整 Web 应用、或者纯 Python 脚本跑完在终端打印结果。但最终定下来 Python 做逻辑、HTML 做界面原因很实际Python 在行情处理和策略回测方面生态太成熟了pandas 处理 K 线数据、numpy 做数值计算几行代码就能完成复杂的指标运算。HTML 页面不需要安装客户端手机和电脑都能通过浏览器访问策略跑在服务器上看盘设备只要能打开网页就行跨平台成本几乎是零。用 Django 或 Flask 渲染整个后台对交易工具来说太重了。交易面板的核心是信息密度高、刷新及时、一眼看懂纯 HTML 加少量 Ajax 就够用没必要为了架构好看而堆框架。架构上我划分了四个独立模块模块之间通过标准化的数据接口通信模块职责技术选型数据采集模块获取历史 K 线与实时行情Python requests pandas策略计算模块指标计算与信号生成Python numpy核心指标封装成类模拟交易模块虚拟持仓管理与成交撮合Python 自研引擎独立数据库存储可视化面板行情与持仓监控HTML ECharts 原生 JavaScript模块之间解耦的好处是如果将来想换数据源只需要重写数据采集模块策略层和展示层完全不用动。这个设计在后面实际开发中帮我省了无数次事。2. 核心模块设计与实现要点2.1 行情数据模块兼容通达信体系的 K 线处理数据是整个系统的基础数据不对后面全部白算。我处理数据时参考了通达信体系常见的本地数据结构思路但实际落地时改成了更通用的文件格式存储每只股票一个 CSV 文件字段包含 date、open、high、low、close、volume、amount 这几列。之所以不用数据库是因为 CSV 文件用 pandas 读取速度快、调试方便、出了问题可以直接用文本编辑器打开检查。数据更新采用增量拉取机制。首次运行时全量下载近三年的日线数据之后每次启动只拉取最近五个交易日的数据做增量补齐。这里有个关键细节停牌股票当天没有数据如果直接按当天缺失就跳过会把除权除息的日子弄错。我的处理方式是把最近拉到的数据与本地已有数据做日期对比只追加新日期保证了 K 线的连续性。数据校验方面我也是踩了坑才补上的逻辑。有一回策略跑出来的信号特别密集排查半天发现是某天的 high 数值比 low 还小属于数据源脏数据。后来加了过滤规则high 必须大于等于 low、volume 必须大于零、日期必须是递增的不合法的直接丢弃该记录并打日志告警。对于拿到 TDX 格式本地数据的场景还需要额外处理复权因子前复权和后复权在回测中的结果差异非常大这在后面回测部分会细说。2.2 策略信号模块从指标函数到信号生成策略模块是系统的大脑。我先把常见指标都拆成独立的纯函数输入是 DataFrame输出是计算结果列。比如简单移动平均的实现def sma(data: pd.DataFrame, window: int) - pd.Series: return data[close].rolling(windowwindow, min_periods1).mean()MACD 的计算稍微复杂一点需要计算快线、慢线和柱状值def macd(data: pd.DataFrame, fast: int 12, slow: int 26, signal: int 9): ema_fast data[close].ewm(spanfast, adjustFalse).mean() ema_slow data[close].ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() histogram (dif - dea) * 2 return dif, dea, histogram指标函数写好之后信号生成是另一层逻辑。以双均线策略为例金叉只是辅助判断真正触发交易还需要成交量配合和价格位置确认这样能过滤掉大量无效信号。信号生成器的输出统一成字典格式包含方向buy/sell、强度、触发原因和当前指标快照方便后续在 HTML 面板上展示。一个容易被忽略的细节是信号去重。比如均线金叉出现后如果价格一直在均线上方震荡策略可能会在每一根 K 线都产生一次买入信号。所以我加了状态机只有当当前状态是空仓时才响应买入信号持仓状态下只响应卖出信号。这个逻辑极大减少了日志里的噪音也让模拟交易的持仓记录变得干净。2.3 可视化面板模块HTML ECharts 实时监控HTML 面板是整个项目最有成就感的部分。页面加载之后左侧是自选股列表中间是 K 线图和成交量图右侧是实时信号记录和当前持仓信息。整体布局就是一张大屏打开就能看。K 线图用的是 ECharts 的 candlestick 组件数据由 Python 后端以 JSON 格式通过接口返回。前端每五秒轮询一次最新行情接口拿到增量数据后调用 setOption 更新图表。这里有个性能方面的心得每次轮询不要刷新整个图表只更新最后一个点的数据否则鼠标悬停查看历史数据时会经常被强制刷新打断体验很差。信号展示区域我单独做了一个滚动列表每条信号记录包含时间、股票代码、信号方向、触发原因和当时的价格。为了直观买入信号用红底白字卖出信号用绿底白字国内习惯红涨绿跌这个细节还挺多朋友问的。前端逻辑全部用原生 JavaScript 实现不引框架就引了一个 ECharts 的 CDN。原因很简单页面功能只有轮询数据 更新图表 渲染列表引入 Vue 或 React 反而增加了打包和部署成本没必要。2.4 模拟交易模块风险控制与持仓管理模拟交易模块严格来说是一个简化版的撮合引擎。收到买入信号后按当前价加一个滑点比例计算成交价更新账户的持仓和可用资金收到卖出信号则按当前价减滑点计算成交价落袋盈亏。每次成交都会写入交易记录表包含时间、方向、价格、数量、手续费和备注。资金管理上我设置了最大单票仓位 20%、单日最大回撤 5% 自动停止新开仓、单笔止损 8% 强制平仓三条风控规则。有一段时间我发现模拟账户的收益曲线波动特别剧烈复盘后发现是因为单票仓位太重一只票的涨跌就能带动整个账户大起大落。限仓之后曲线平滑了很多虽然单次收益变小了但回撤也显著降低。这个模块还顺带解决了一个需求每天开盘前自动加载前一日所有持仓股票的公告和停复牌信息如果发现持仓中有停牌股会在面板上打上醒目标记提醒手动处理。3. 实操过程与关键代码解析3.1 环境准备与项目初始化实际搭建这个项目我建议直接用 Anaconda 管理 Python 环境避开不少 Windows 上安装包的坑。Python 版本用的 3.10依赖库版本也锁定了一个稳定的组合pip install pandas numpy requests flask apscheduler后端接口我用 Flask 起了一个轻量服务就两个路由一个返回全部 K 线数据一个返回最新行情和信号记录。不选 FastAPI 的原因是我对异步没有硬需求Flask 本身就是够了体积也小。项目目录结构建议这样组织tdxpystock/ ├── data/ # 行情数据 CSV 存储目录 ├── strategies/ # 策略指标代码 │ ├── indicators.py │ └── signals.py ├── trader/ # 模拟交易与风控 │ ├── portfolio.py │ └── risk.py ├── web/ # HTML 前端文件 │ └── index.html ├── server.py # Flask 后端入口 └── main.py # 程序主流程3.2 行情数据采集实操数据采集这块我单独封装了一个类统一管理行情源的接入。核心方法是获取日 K 线数据和获取实时快照class MarketData: def __init__(self, data_dir: str data): self.data_dir data_dir os.makedirs(data_dir, exist_okTrue) def get_kline(self, symbol: str, days: int 750) - pd.DataFrame: file_path os.path.join(self.data_dir, f{symbol}.csv) if os.path.exists(file_path): df pd.read_csv(file_path, parse_dates[date]) else: df self.download_kline(symbol, days) self.save_to_csv(symbol, df) return df[(df[date] (pd.Timestamp.now() - pd.Timedelta(daysdays)))] def update_realtime(self, symbol: str) - dict: # 拉取实时行情返回最新价、成交量等 pass我写的 download_kline 里面用的是第三方行情接口通过 requests 调用返回 JSON 后解析成 DataFrame。这里有个实际经验接口返回的数据量较大时一定做好异常重试和超时处理。我给 requests 设置了 timeout10连续三次失败就直接跳过本只股票的本次更新避免因为一只票拉不到数据导致整个循环卡死。另一个细节是 pandas 的列名处理。各家接口返回的字段名千奇百怪有返回大写 Open、High 的也有返回小写 open、high 的甚至有的时间字段叫 timestamp。我统一在 download_kline 里做了一层字段映射保证落盘的数据永远是统一的 schema这样策略层永远不会因为列名问题报 KeyError。3.3 策略信号生成实操策略层我写了一个 SignalEngine 类把指标计算和信号判定分成两步class SignalEngine: def __init__(self, indicators: dict): self.indicators indicators def compute(self, data: pd.DataFrame) - pd.DataFrame: result data.copy() for name, params in self.indicators.items(): if name sma: result[fsma_{params[window]}] sma(result, params[window]) elif name macd: result[dif], result[dea], result[hist] macd(result) elif name rsi: result[frsi_{params[window]}] rsi(result, params[window]) return result def generate_signals(self, data: pd.DataFrame) - list: # 基于计算指标生成信号列表 pass双均线策略的信号判定逻辑是核心。我的流程是这样的计算 5 日均线和 20 日均线。判断是否发生金叉昨天的 5 日均线小于等于昨天的 20 日均线今天的 5 日均线大于今天的 20 日均线。金叉确认后查看当日成交量是否大于过去 5 日均量的 1.2 倍。如果量能配合再检查当前价格是否站在 20 日均线上方。四重条件全部满足才生成买入信号。这样写出来的信号数量虽然比只看金叉少但胜率明显更好模拟交易跑了三个月胜率大概在 58% 左右盈亏比在 1.6 附近。需要说明的是这个结果只基于特定行情区间不要觉得 58% 就能直接拿到实盘去用。信号生成器还有一个冷却时间参数默认买入信号触发后 120 分钟内不再重复触发同一股票的买入。这个机制是用来防止信号闪烁问题的也就是价格在临界点来回波动导致频繁开仓。3.4 可视化面板实操HTML 面板的页面结构是这样的!DOCTYPE html html langzh-cn head meta charsetutf-8 titleTDXPystock 交易监控面板/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script style /* 页面布局样式 */ /style /head body div idsidebar h3自选股/h3 ul idsymbol-list/ul /div div idmain-chart h3 idcurrent-symbol--/h3 div idkline-chart/div /div div idsignal-panel h3最新信号/h3 ul idsignal-list/ul /div script srcapp.js/script /body /html前端的关键逻辑是数据轮询和图表更新。我写了一个 updateData 函数每次请求后端的 /api/market_data 接口拿到最新的 K 线 JSON 后直接更新图表。K 线数据的数据结构是{ dates: [2025-01-02, 2025-01-03], kline: [[3.65, 3.72, 3.58, 3.70, 112233], ...], volumes: [112233, ...] }ECharts 的 candlestick 组件接收的数据顺序是 [open, close, lowest, highest]这个顺序很容易搞错我第一次开发时就是因为顺序写反了K 线图看着完全变形。后来我干脆在 Python 端就生成好 ECharts 需要的顺序前端不再做任何数据处理减少出错的可能。轮询频率我设置在 http://192.168.1.100:80805 秒一次白天跑一整天也不会有明显资源占用。如果策略逻辑再重一些可以把轮询改成 WebSocket 推送模式但现在的实现已经能满足需求没必要过度设计。3.5 主流程整合与定时调度主流程通过 APScheduler 做定时调度设置了三个任务每个交易日的 09:25 拉取集合竞价结束后的行情快照更新当日开盘价。盘中每 5 秒拉取一次实时行情并交给策略引擎判断是否有信号。每天 15:10 收盘后做一次全量数据更新然后执行一次全持仓对账。调度任务的代码如下from apscheduler.schedulers.blocking import BlockingScheduler def create_scheduler(engine, market_data, portfolio): scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.scheduled_job(cron, day_of_weekmon-fri, hour9,10,11,13,14, minute*, second*/5) def realtime_job(): for symbol in portfolio.watchlist: quote market_data.update_realtime(symbol) signals engine.generate_signals(quote) portfolio.execute_signals(signals) scheduler.scheduled_job(cron, day_of_weekmon-fri, hour15, minute10) def daily_close(): market_data.update_all() portfolio.sync_positions() return scheduler需要注意的是交易日历的问题。我一开始没有处理节假日结果国庆节后第一天系统 9:25 触发任务发现所有股票的实时接口都返回空数据。后来引入了交易日历库判断当天是不是交易日不是交易日就跳过所有盘中任务是交易日才正常执行。这一个改动让系统在节假日和周末跑得特别安静日志里也不会刷出一堆无效拉取记录。4. 常见问题与排查技巧实录4.1 数据源返回的数据出现空值或异常值现象某只股票的 K 线数据里出现 NaN 值策略计算时直接报错或者指标结果异常偏大。排查过程第一次遇到时我以为是数据源的问题后来加了日志才发现是股票停牌日返回了空成交量而我在合并数据时没有做填充。还有一个更隐蔽的情况是股票当天不交易但数据源返回了前一天的旧数据如果不判断日期会把这根假 K 线当成当天数据参与计算。解决方案在数据入库前增加空值检测close 为空、volume 为空、日期不为交易日的记录直接过滤。对分钟线数据非交易时段自动跳过。设置数据质量监控脚本每天收盘后检查 CSV 文件的最后一行日期是否为当天如果不是就发送告警日志。4.2 模拟交易持仓与手动操作不一致现象模拟账户显示持仓 1000 股但实际上行情接口返回的持仓文件里是 0 股两边对不上。排查过程这个问题我是折腾了很久才找到原因的。原来模拟交易引擎在收到卖出信号后先从持仓里扣减了数量但没有立即落盘而是等到收盘统一写文件。如果中途程序崩溃内存里的持仓数据就丢掉了落盘的文件还停留在卖出之前的状态。解决方案每次成交立即同步持仓文件不等到收盘。增加交易流水表每次交易都记录前后持仓变化方便回溯。每日启动时先做一次全量对账发现差异就打印告警并自动按流水重建持仓。4.3 HTML 页面无法打开或图表不显示现象浏览器打开页面背景正常但 K 线图区域是空的控制台报错。排查过程这类问题九成出在数据格式上。有一次后端接口返回的日期字段是2025/01/02格式前端 JS 直接 new Date 解析后变成了 Invalid Date导致 ECharts 的 xAxis 数据全是 null图表就空了。解决方案后端统一输出 ISO 格式的日期字符串比如2025-01-02前端直接使用。在 JS 里增加数据校验拿到数据先判断数组长度长度为零时显示占位提示而不是让图表白屏。把 ECharts 初始化代码包裹在 window.onload 里避免 DOM 还没渲染完就执行脚本。4.4 策略回测表现好但模拟交易表现差现象同一套策略历史回测年化收益 25%放到模拟盘跑一个月只有 5%回撤还比回测大了不少。排查过程回测和模拟盘之间最大的差异因素有三个——滑点、手续费、信号延迟。回测时我用的是收盘价直接成交完全忽略了真实交易中的滑点和冲击成本模拟盘我加了千分之一的滑点后很多本来就贴着均线波动的信号就很容易被触发后又反向打掉。解决方案回测中加入固定滑点 0.2% 和双边手续费 0.1%让自己的预期更接近真实。加一个信号过滤条件如果当前价格距离均线小于 0.5%放弃本次信号避免在震荡行情里反复开仓。用模拟盘跑满 30 个交易日再和回测数据做对比如果收益差距超过 10 个百分点首先检查策略的参数是不是已经过拟合了。5. 系统功能扩展方向5.1 增加多策略并行目前信号引擎一次只能跑一种策略组合如果想把双均线策略和 MACD 背离策略同时跑就需要重构信号引擎支持多策略结果合并。我的思路是把每个策略封装成一个独立的 Generator 类输出同样的信号结构汇总层再做信号融合。多个策略同步给出买入信号的票优先执行单一策略信号只作为观察对象不直接下单。这样能减少试错成本胜率也会更稳定。5.2 接入消息通知HTML 面板需要主动打开才能看到信号对于移动场景还是不够方便。可以加一个消息推送扩展当信号产生或持仓达到止损线时通过 Server 酱或企业微信机器人推送到手机。我在实际扩展中发现这类推送在信号产生时作用最大当你有持仓时一个已触发止损的通知比盯一整天盘更高效。实现上就是多一个通知模块在 portfolio.execute_signals 里调用 notify 方法几十行代码的事。5.3 参数寻优与稳健性检查策略参数比如均线窗口、RSI 阈值目前是手工设定的可以加一个简单的参数寻优模块用网格搜索在历史数据上跑一遍找出一组在近三年表现最优的参数。但这里要特别提醒网格搜索出来的最优参数大概率是过拟合的不经过样本外测试就直接实盘会死得很惨。稳妥的做法是留出最近 6 个月的数据做验证集参数在训练集上寻优然后在验证集上确认收益和回撤两项都达标才纳入正式配置。6. 我的实操体会跑这套系统半年多我最大的感受是交易自动化最难的不是写代码而是把模糊的交易规则翻译成精确的代码逻辑。比如感觉要启动了这句话在代码里必须变成5 日均线上穿 20 日均线成交量放大 20%收盘价站上 20 日均线才有意义。另外一点很重要的是自动化系统需要容忍它犯错。我的模拟账户也有阶段性亏损但我不会因为亏损就临时改参数那样做系统就永远处于不稳定状态。每一次参数调整都要做完整的回归测试比较新旧参数在所有历史区间的表现差异再决定是否启用。最后给打算实践这个项目的朋友一个建议第一步不要追求完整的功能体系先做通一个最简单的小闭环——拉取一只股票的数据、算一个均线指标、在 HTML 页面上画出来。当这个最小闭环跑通之后你自然知道下一步该加什么。很多人做量化项目半途而废都是因为从一开始就想做一个完美的系统结果卡在某个环节整整一个周末然后就再也没打开过那个文件夹了。本文还有配套的精品资源点击获取