自动读取通达信预警文件，并生成动态列表（含python源码）

admin

炒股的时候，最怕的不是没信号，而是信号太多，吵得你耳朵嗡嗡响，脑子嗡嗡响。

通达信那个预警窗口，你懂的。一到下午两点半，跟菜市场的大喇叭似的，叽叽喳喳全是股票代码。这边喊一个“放量突破”，那边吼一个“金叉买入”。哪个是主线？哪个是跟风杂鱼？根本分不清，看多了眼睛都花了。

那么，我们能不能给这个大嗓门的喇叭，配一个冷静、理智能快速整理信息的AI秘书？

我琢磨了一下，还真搞出来了这么个小东西。

给股票代码穿上“身份马甲”

这个秘书的工作，本质上就干一件事：给所有从通达信大喇叭里喊出来的股票，立刻穿上“身份马甲”，然后分门别类地站好队。

它的工作流程是这样的：

• 1. 监听“大喇叭”：秘书有个超灵敏的耳朵，时刻听着通达信导出的那个实时预警txt文件。只要这文件一动弹，有新股票喊出来了，它立马就知道了。
• 2. 翻“秘密花名册”：我给了它一本秘密花名册，这里面提前写好了各种“山头名册”（板块）的txt静态文件。比如，哪些是我的“亲兵卫队”（自选股），哪些是“近期起义的热门将领”（热门概念股），哪些是“国家队重点观察对象”（沪深300成分股）……
• 3. 实时贴标签、穿马甲：这是最精彩的一步。只要“大喇叭”里喊出一个股票代码，我这AI秘书就会一秒钟翻开花名册。
  
  • • 查无此人？那就是个陌生来客，在网页里就显示个普通样貌，不施粉黛。
  • • 查有此人？那可不的。它立刻判断这个股票属于哪个山头，然后“啪”地给它贴上一个电子标签，还刷上专属颜色。
    
    • • 比如，“自选股”来的，刷成大红色，这是我的嫡系部队，必须重点照顾。
    • • “热门概念”来的，刷成闪光的绿色，市场热点，值得关注。
    • • “权重股”来的，刷成稳重的蓝色， affecting the index, 注意大盘节奏。
      
• 4. 汇报“战况看板”：所有整理得漂漂亮亮的情报，最终都会呈现在一个自动刷新的HTML网页上。这个网页就像一个实时战况看板，一行行股票代码整整齐齐地列着，戴着不同颜色的标签。
  

效果如何？

效果就是：降噪、提纯。

原本是嘈杂无序的菜市场，现在变成了队列整齐的阅兵场。我一眼扫过去，就知道：

• • 今天我自选股里有几个动了。
• • 市场热点轮动到了哪个板块。
• • 有没有陌生人突然闯进来。
  

我不再需要费劲去一个一个搜代码，去回忆这个股票是干啥的。因为它“穿的马甲”已经告诉我一切了。

这东西解决的不是“如何发现信号”的问题，而是“如何高效处理你已经收到的信号”的问题。它帮你完成了从信息接收，到信息整合，再到信息呈现的最后，也是最重要的一公里。

通达信预警监控代码全解析

下面从代码结构、核心功能、运行逻辑三个维度，拆解这段代码，你可以把它理解为“一个带可视化界面的股票预警监控工具”：

一、代码整体结构（先认“骨架”）

这段代码分为4个核心部分：

• 1. 导入依赖：拿取实现功能需要的“工具包”；
• 2. 定义核心类StockMonitorApp：封装所有界面和监控逻辑（代码的核心）；
• 3. 定义启动函数main：创建窗口并启动工具；
• 4. 程序入口：触发main函数，让工具跑起来。
  

二、逐部分拆解（再看“血肉”）1. 导入依赖（代码第1-8行）import threading  # 多线程工具：让监控和界面操作同时进行（比如一边监控股票，一边能点按钮）
import time       # 时间工具：控制“多久查一次股票预警”
import os         # 文件路径工具：找文件、存文件的位置
import json       # 数据存储工具：保存/读取你的设置（比如选的文件、刷新间隔）
import tkinter as tk  # GUI界面工具：做窗口、按钮、表格
from tkinter import ttk, filedialog, messagebox  # GUI的扩展工具：更美观的控件、文件选择框、提示弹窗
from 通达信预警到html import StockMonitor  # 自定义工具：处理股票预警的核心逻辑（别人写好的“专用功能”）2. 核心类StockMonitorApp（代码第10行开始）

这个类是工具的“本体”，所有功能都在里面，我们按“工具的生命周期”拆解：

（1）初始化：工具刚启动时做什么（__init__方法）def __init__(self, root):
    self.root = root  # 把主窗口“绑定”到工具上
    self.root.title("通达信预警监控配置")  # 给窗口起名字

    # 初始化监控相关的“开关/变量”
    self.monitor = None# 监控核心实例（一开始为空）
    self.monitor_thread = None# 监控线程（一开始不干活）
    self.running = False# 监控运行状态（一开始是“关”）

    # 配置文件路径：和脚本存在同一个文件夹，文件名叫tdx_alert_gui_config.json
    self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tdx_alert_gui_config.json")

    self._build_ui()  # 第一步：搭建界面
    self._load_config()  # 第二步：读取上次保存的设置
    self._schedule_refresh()  # 第三步：设置表格自动刷新

核心作用：给工具“初始化参数”，并触发界面搭建、读取配置、表格刷新。

（2）搭建界面：做可视化窗口（_build_ui方法）

这部分是“画界面”，把窗口分成4个区域，对应你能看到的所有控件：

区域

功能

参数设置区

选“通达信预警文件”“静态标签文件”，设置“刷新间隔”（默认3秒）

控制区

启动/停止监控按钮，显示当前状态（未启动/运行中/已停止）

预警表格区

显示最近20条预警，列包括股票代码、名称、预警时间、价格等（带滚动条）

底部信息区

显示预警信息生成的HTML文件路径

核心作用：把“看不见的代码”变成“看得见的窗口”，让你能手动操作。

（3）配置保存/读取：记住你的设置（_load_config/_save_config）

• • _load_config：打开tdx_alert_gui_config.json文件，把上次选的文件、设的刷新间隔填回输入框（比如你上次设了5秒刷新，下次打开工具还能保留）；
• • _save_config：把当前输入框里的内容（选的文件、刷新间隔）存到JSON文件里（怕你关掉工具后忘了设置）。
  

核心作用：让工具“有记忆”，不用每次打开都重新设置。

（4）文件选择：找需要的文件（browse_dynamic/browse_static）

点击“浏览...”按钮时触发，弹出系统的文件选择框，选好文件后自动填到输入框里，同时保存配置。
核心作用：不用手动输文件路径，点点鼠标就能选文件。

（5）启动监控：让工具开始干活（start_monitor方法）

这是核心功能之一，步骤如下：

• 1. 先校验参数：没选预警文件/文件不存在/刷新间隔不是正整数 → 弹提示框报错；
• 2. 校验通过后，保存当前设置；
• 3. 创建StockMonitor实例（启动“股票预警处理核心”）；
• 4. 切换按钮状态：启动按钮变灰（不能点），停止按钮点亮（能点），状态改成“运行中”；
• 5. 开一个新线程（monitor_loop）：让工具每隔设置的秒数，调用StockMonitor查一次预警（线程的作用：查预警时不卡界面，你还能点按钮）。
  

核心作用：启动监控逻辑，让工具自动盯股票预警。

（6）停止监控：让工具歇着（stop_monitor方法）

• 1. 把“运行开关”（self.running）关掉，监控线程会自动停止；
• 2. 切换按钮状态：启动按钮点亮，停止按钮变灰，状态改成“已停止”；
• 3. 保存当前设置。
  核心作用：手动停止监控，避免工具一直跑。
  

（7）表格刷新：实时显示预警（_refresh_table/_schedule_refresh）

• • _refresh_table：清空表格旧数据，从StockMonitor里拿最近20条预警，填到表格里；
• • _schedule_refresh：用root.after(1000, ...)让表格每隔1秒刷新一次（1000毫秒=1秒）。
  核心作用：让你能实时看到最新的股票预警。
  

3. 启动函数main+程序入口def main():
    root = tk.Tk()  # 创建主窗口
    app = StockMonitorApp(root)  # 把工具绑定到主窗口
    root.mainloop()  # 让窗口一直显示（直到你关掉）

if __name__ == "__main__":
    main()  # 程序启动时，先执行main函数

核心作用：触发工具启动，显示窗口。

三、运行逻辑（工具的“工作流程”）

• 1. 你双击运行代码 → 执行main函数 → 弹出主窗口；
• 2. 窗口初始化：读取上次的设置（如果有），表格开始每秒刷新；
• 3. 你选好预警文件、设置刷新间隔，点击“启动监控”；
• 4. 工具校验参数 → 启动监控线程 → 每隔N秒查一次预警 → 表格实时显示；
• 5. 你点击“停止监控” → 监控线程停止 → 保存当前设置；
• 6. 你关掉窗口 → 程序结束。
  

四、关键细节（新手必懂）

• • 多线程：threading.Thread创建的监控线程是“后台线程”（daemon=True），关掉窗口时会自动结束，不会残留；
• • 容错处理：读/写配置、刷新表格时加了try/except，就算出错也不会直接崩溃；
• • 状态控制：self.running是“开关变量”，监控线程里的while self.running会一直循环，直到开关关掉。
  

打包见下：通过网盘分享的文件：通达信预警监控v1.0.zip链接: https://pan.baidu.com/s/1eIZ-IY3YYLiEUwFhZQXnEg?pwd=wx36 提取码: wx36 --来自百度网盘超级会员v7的分享
源码如下：

1. import threading
   
2. import time
   
3. import os
   
4. import json
   
5. import tkinter as tk
   
6. from tkinter import ttk, filedialog, messagebox
   
7. 
   
8. from 通达信预警到html import StockMonitor
   
9. 
   
10. 
    
11. class StockMonitorApp:
    
12.     def __init__(self, root):
    
13.         self.root = root
    
14.         self.root.title("通达信预警监控1.0--仓鼠量化")
    
15. 
    
16.         self.monitor = None
    
17.         self.monitor_thread = None
    
18.         self.running = False
    
19. 
    
20.         # 配置文件路径（与脚本同目录）
    
21.         self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tdx_alert_gui_config.json")
    
22. 
    
23.         self._build_ui()
    
24.         self._load_config()
    
25.         self._schedule_refresh()
    
26. 
    
27.     def _build_ui(self):
    
28.         # 参数区
    
29.         param_frame = ttk.LabelFrame(self.root, text="参数设置")
    
30.         param_frame.pack(fill=tk.X, padx=10, pady=5)
    
31. 
    
32.         # 通达信预警文件
    
33.         ttk.Label(param_frame, text="通达信预警文件:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=5)
    
34.         self.dynamic_var = tk.StringVar()
    
35.         self.dynamic_entry = ttk.Entry(param_frame, textvariable=self.dynamic_var, width=60)
    
36.         self.dynamic_entry.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W)
    
37.         ttk.Button(param_frame, text="浏览...", command=self.browse_dynamic).grid(row=0, column=2, padx=5, pady=5)
    
38. 
    
39.         # 静态标签文件
    
40.         ttk.Label(param_frame, text="静态标签文件:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=5)
    
41.         self.static_var = tk.StringVar(value="static_codes.txt")
    
42.         self.static_entry = ttk.Entry(param_frame, textvariable=self.static_var, width=60)
    
43.         self.static_entry.grid(row=1, column=1, padx=5, pady=5, sticky=tk.W)
    
44.         ttk.Button(param_frame, text="浏览...", command=self.browse_static).grid(row=1, column=2, padx=5, pady=5)
    
45. 
    
46.         # 刷新间隔
    
47.         ttk.Label(param_frame, text="刷新间隔(秒):").grid(row=2, column=0, sticky=tk.W, padx=5, pady=5)
    
48.         self.interval_var = tk.StringVar(value="3")
    
49.         self.interval_entry = ttk.Entry(param_frame, textvariable=self.interval_var, width=10)
    
50.         self.interval_entry.grid(row=2, column=1, padx=5, pady=5, sticky=tk.W)
    
51. 
    
52.         # 控制区
    
53.         control_frame = ttk.LabelFrame(self.root, text="控制")
    
54.         control_frame.pack(fill=tk.X, padx=10, pady=5)
    
55. 
    
56.         self.start_button = ttk.Button(control_frame, text="启动监控", command=self.start_monitor)
    
57.         self.start_button.grid(row=0, column=0, padx=5, pady=5)
    
58. 
    
59.         self.stop_button = ttk.Button(control_frame, text="停止监控", command=self.stop_monitor, state=tk.DISABLED)
    
60.         self.stop_button.grid(row=0, column=1, padx=5, pady=5)
    
61. 
    
62.         self.status_var = tk.StringVar(value="状态: 未启动")
    
63.         ttk.Label(control_frame, textvariable=self.status_var).grid(row=0, column=2, padx=5, pady=5, sticky=tk.W)
    
64. 
    
65.         # 最近预警表格
    
66.         table_frame = ttk.LabelFrame(self.root, text="最近预警 (最多显示最近20条)")
    
67.         table_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
    
68. 
    
69.         columns = ("code", "name", "time", "price", "inc", "tag", "reason")
    
70.         self.tree = ttk.Treeview(table_frame, columns=columns, show="headings", height=12)
    
71.         self.tree.heading("code", text="股票代码")
    
72.         self.tree.heading("name", text="股票名称")
    
73.         self.tree.heading("time", text="预警时间")
    
74.         self.tree.heading("price", text="预警价格")
    
75.         self.tree.heading("inc", text="预警涨幅")
    
76.         self.tree.heading("tag", text="静态标签")
    
77.         self.tree.heading("reason", text="预警原因")
    
78. 
    
79.         self.tree.column("code", width=80, anchor=tk.CENTER)
    
80.         self.tree.column("name", width=100, anchor=tk.W)
    
81.         self.tree.column("time", width=80, anchor=tk.CENTER)
    
82.         self.tree.column("price", width=80, anchor=tk.E)
    
83.         self.tree.column("inc", width=80, anchor=tk.E)
    
84.         self.tree.column("tag", width=100, anchor=tk.W)
    
85.         self.tree.column("reason", width=200, anchor=tk.W)
    
86. 
    
87.         self.tree.pack(fill=tk.BOTH, expand=True, side=tk.LEFT)
    
88. 
    
89.         scrollbar = ttk.Scrollbar(table_frame, orient=tk.VERTICAL, command=self.tree.yview)
    
90.         self.tree.configure(yscroll=scrollbar.set)
    
91.         scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
    
92. 
    
93.         # 底部信息
    
94.         bottom_frame = ttk.Frame(self.root)
    
95.         bottom_frame.pack(fill=tk.X, padx=10, pady=5)
    
96. 
    
97.         html_path = os.path.abspath("stock_alerts_table.html")
    
98.         self.html_info_var = tk.StringVar(value=f"HTML 输出: {html_path}")
    
99.         ttk.Label(bottom_frame, textvariable=self.html_info_var).pack(side=tk.LEFT)
    
100. 
     
101.     def _load_config(self):
     
102.         """从本地JSON配置文件加载上次使用的参数"""
     
103.         if not os.path.exists(self.config_path):
     
104.             return
     
105.         try:
     
106.             with open(self.config_path, "r", encoding="utf-8") as f:
     
107.                 cfg = json.load(f)
     
108.             dynamic_file = cfg.get("dynamic_file", "")
     
109.             static_file = cfg.get("static_file", "static_codes.txt")
     
110.             interval = str(cfg.get("interval", 3))
     
111. 
     
112.             if dynamic_file:
     
113.                 self.dynamic_var.set(dynamic_file)
     
114.             if static_file:
     
115.                 self.static_var.set(static_file)
     
116.             if interval:
     
117.                 self.interval_var.set(interval)
     
118.         except Exception:
     
119.             # 配置损坏时忽略，使用默认值
     
120.             pass
     
121. 
     
122.     def _save_config(self):
     
123.         """将当前参数保存到本地JSON配置文件"""
     
124.         cfg = {
     
125.             "dynamic_file": self.dynamic_var.get().strip(),
     
126.             "static_file": self.static_var.get().strip(),
     
127.             "interval": int(self.interval_var.get().strip()) if self.interval_var.get().strip().isdigit() else 3,
     
128.         }
     
129.         try:
     
130.             with open(self.config_path, "w", encoding="utf-8") as f:
     
131.                 json.dump(cfg, f, ensure_ascii=False, indent=2)
     
132.         except Exception:
     
133.             # 保存失败时静默忽略，不影响程序运行
     
134.             pass
     
135. 
     
136.     def browse_dynamic(self):
     
137.         path = filedialog.askopenfilename(title="选择通达信预警文件", filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")])
     
138.         if path:
     
139.             self.dynamic_var.set(path)
     
140.             self._save_config()
     
141. 
     
142.     def browse_static(self):
     
143.         path = filedialog.askopenfilename(title="选择静态标签文件", filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")])
     
144.         if path:
     
145.             self.static_var.set(path)
     
146.             self._save_config()
     
147. 
     
148.     def start_monitor(self):
     
149.         if self.running:
     
150.             return
     
151. 
     
152.         dynamic_file = self.dynamic_var.get().strip()
     
153.         static_file = self.static_var.get().strip()
     
154.         interval_str = self.interval_var.get().strip()
     
155. 
     
156.         if not dynamic_file:
     
157.             messagebox.showwarning("提示", "请先选择通达信预警文件")
     
158.             return
     
159.         if not os.path.exists(dynamic_file):
     
160.             messagebox.showerror("错误", f"通达信预警文件不存在:\n{dynamic_file}")
     
161.             return
     
162. 
     
163.         if not interval_str.isdigit() or int(interval_str) <= 0:
     
164.             messagebox.showerror("错误", "刷新间隔必须是正整数")
     
165.             return
     
166. 
     
167.         update_interval = int(interval_str)
     
168. 
     
169.         # 参数校验通过后立即保存配置
     
170.         self._save_config()
     
171. 
     
172.         try:
     
173.             self.monitor = StockMonitor(dynamic_file, static_file)
     
174.         except Exception as e:
     
175.             messagebox.showerror("错误", f"创建监控实例失败:\n{e}")
     
176.             self.monitor = None
     
177.             return
     
178. 
     
179.         self.running = True
     
180.         self.start_button.configure(state=tk.DISABLED)
     
181.         self.stop_button.configure(state=tk.NORMAL)
     
182.         self.status_var.set("状态: 运行中")
     
183. 
     
184.         def monitor_loop():
     
185.             while self.running:
     
186.                 try:
     
187.                     if self.monitor is not None:
     
188.                         self.monitor.process_once()
     
189.                 except Exception as e:
     
190.                     # 在后台线程中不能直接弹窗，只更新状态
     
191.                     self.status_var.set(f"状态: 运行中(发生错误: {e})")
     
192.                 time.sleep(update_interval)
     
193. 
     
194.         self.monitor_thread = threading.Thread(target=monitor_loop, daemon=True)
     
195.         self.monitor_thread.start()
     
196. 
     
197.     def stop_monitor(self):
     
198.         if not self.running:
     
199.             return
     
200.         self.running = False
     
201.         self.start_button.configure(state=tk.NORMAL)
     
202.         self.stop_button.configure(state=tk.DISABLED)
     
203.         self.status_var.set("状态: 已停止")
     
204.         # 停止时也保存一次当前参数
     
205.         self._save_config()
     
206. 
     
207.     def _refresh_table(self):
     
208.         if self.monitor is not None:
     
209.             try:
     
210.                 df = self.monitor.get_recent_alerts(20)
     
211.                 # 清空表格
     
212.                 for item in self.tree.get_children():
     
213.                     self.tree.delete(item)
     
214.                 # 插入新的行
     
215.                 if not df.empty:
     
216.                     for _, row in df.iterrows():
     
217.                         self.tree.insert("", tk.END, values=(
     
218.                             row.get("股票代码", ""),
     
219.                             row.get("股票名称", ""),
     
220.                             row.get("预警时间", ""),
     
221.                             row.get("预警价格", ""),
     
222.                             f"{row.get('预警涨幅', '')}%",
     
223.                             row.get("静态标签", ""),
     
224.                             row.get("预警原因", ""),
     
225.                         ))
     
226.             except Exception:
     
227.                 # 刷新失败时静默忽略，避免打断界面
     
228.                 pass
     
229. 
     
230.     def _schedule_refresh(self):
     
231.         self._refresh_table()
     
232.         self.root.after(1000, self._schedule_refresh)
     
233. 
     
234. 
     
235. def main():
     
236.     root = tk.Tk()
     
237.     app = StockMonitorApp(root)
     
238.     root.mainloop()
     
239. 
     
240. 
     
241. if __name__ == "__main__":
     
242.     main()
     

复制代码

pengcheng4989

感谢分享