小白也量化：用Python实时抓取雪球热股榜TOP20（含源码及界面版）

admin

最近有同学问怎么找各平台的热门股票榜单，正好最近我在研究东财实盘跟单与雪球组合跟单。

后面我分几期介绍一下，今天先讲雪球。

行情波动时，我每天也会看雪球的热门榜。它是个不错的风向标，能看出市场在关注什么。

其实你可以用 Python 自己做一个实时更新的监控面板，操作很简单，新手也能搞定。

一、准备工具

需要2个 Python 库：

• • Requests：用来向雪球请求数据。
• • Pandas：用来处理数据。
  

安装命令：

pip install streamlit requests pandas

二、获取数据权限

直接爬雪球数据会被拦截，因为服务器能识别出是脚本访问。解决办法是模拟浏览器行为，主要靠两样东西：

• 1. 请求头：包含浏览器的基本信息。
• 2. Cookie：相当于登录凭证，最关键。
  

具体获取步骤：

• 1. 浏览器登录雪球账号。
• 2. 按 F12 打开开发者工具，切换到“网络”（Network）面板。
• 3. 刷新页面，找到对 xueqiu.com 的请求。
• 4. 在“请求头”里找到 Cookie 字段，复制其中 u= 和 xq_a_token= 后面的值备用。
  

三、代码实现

主要分三步：

第1步：请求数据
用 requests 库带上请求头和 Cookie，获取热门榜的原始数据。

第2步：处理数据
用 pandas 整理数据，提取股票名称、代码、价格、涨跌幅、热度等字段，并调整格式。

第3步：展示界面

完整代码已整理好，你可以直接复制使用。运行命令：

streamlit run xueqiu.py

这样就能在本地浏览器看到一个实时更新的雪球热股监控面板了。

下期再介绍其他平台的数据获取方法。

界面版及PY源码阅读原文或通过网盘获取：雪球热门榜单.zip

链接: https://pan.baidu.com/s/1gaIzc5_XZ8H6WzIcxL4Dvg?pwd=zj41 提取码: zj41

1. import os
   
2. import threading
   
3. import time
   
4. import tkinter as tk
   
5. from tkinter import ttk, messagebox
   
6. import webbrowser
   
7. import csv
   
8. from datetime import datetime
   
9. from typing import Any, Dict, List, Optional
   
10. 
    
11. from xueqiu_client import (
    
12.     XueqiuClient,
    
13.     XueqiuCookies,
    
14.     XueqiuAuthError,
    
15.     extract_required_cookies,
    
16.     load_config,
    
17.     save_config,
    
18. )
    
19. 
    
20. 
    
21. APP_TITLE = '雪球热门榜单--仓鼠量化'
    
22. CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'xueqiu_config.json')
    
23. 
    
24. HOT_STOCK_TYPE_ID = 20
    
25. 
    
26. 
    
27. COLUMNS = [
    
28.     ('name', '名称', 180),
    
29.     ('symbol', '代码', 110),
    
30.     ('current', '现价', 90),
    
31.     ('percent', '涨跌幅(%)', 90),
    
32.     ('chg', '涨跌额', 90),
    
33.     ('exchange', '交易所', 80),
    
34.     ('value', '热度值', 110),
    
35.     ('rank_change', '排名变化', 80),
    
36. ]
    
37. 
    
38. 
    
39. class XueqiuHotStockGUI:
    
40.     def __init__(self, root: tk.Tk):
    
41.         self.root = root
    
42.         self.root.title(APP_TITLE)
    
43.         self.root.geometry('1200x800')
    
44. 
    
45.         self._auto_refresh = False
    
46.         self._auto_after_id: Optional[str] = None
    
47.         self._fetch_lock = threading.Lock()
    
48.         self._last_items: List[Dict[str, Any]] = []
    
49. 
    
50.         self._build_ui()
    
51.         self._load_config_into_ui()
    
52. 
    
53.         self._set_status('就绪')
    
54. 
    
55.     def _build_ui(self) -> None:
    
56.         top = tk.Frame(self.root)
    
57.         top.pack(fill=tk.X, padx=10, pady=10)
    
58. 
    
59.         title = tk.Label(top, text=APP_TITLE, font=('微软雅黑', 18, 'bold'))
    
60.         title.pack(side=tk.LEFT)
    
61. 
    
62.         self.status_var = tk.StringVar(value='')
    
63.         status = tk.Label(top, textvariable=self.status_var, anchor=tk.E)
    
64.         status.pack(side=tk.RIGHT, fill=tk.X, expand=True)
    
65. 
    
66.         body = tk.Frame(self.root)
    
67.         body.pack(fill=tk.BOTH, expand=True, padx=10, pady=(0, 10))
    
68. 
    
69.         self._build_left_panel(body)
    
70.         self._build_table(body)
    
71. 
    
72.     def _build_left_panel(self, parent: tk.Frame) -> None:
    
73.         left = tk.Frame(parent, width=380)
    
74.         left.pack(side=tk.LEFT, fill=tk.Y, padx=(0, 10))
    
75.         left.pack_propagate(False)
    
76. 
    
77.         cookie_frame = tk.LabelFrame(left, text='Cookie设置', padx=10, pady=10)
    
78.         cookie_frame.pack(fill=tk.X, pady=(0, 10))
    
79. 
    
80.         tk.Label(cookie_frame, text='Cookie整段(可选):').pack(anchor=tk.W)
    
81.         self.cookie_text = tk.Text(cookie_frame, height=5, width=40)
    
82.         self.cookie_text.pack(fill=tk.X, pady=(5, 10))
    
83. 
    
84.         row1 = tk.Frame(cookie_frame)
    
85.         row1.pack(fill=tk.X)
    
86.         tk.Label(row1, text='u:').pack(side=tk.LEFT)
    
87.         self.u_var = tk.StringVar(value='')
    
88.         tk.Entry(row1, textvariable=self.u_var).pack(side=tk.RIGHT, fill=tk.X, expand=True)
    
89. 
    
90.         row2 = tk.Frame(cookie_frame)
    
91.         row2.pack(fill=tk.X, pady=(6, 0))
    
92.         tk.Label(row2, text='xq_a_token:').pack(side=tk.LEFT)
    
93.         self.token_var = tk.StringVar(value='')
    
94.         tk.Entry(row2, textvariable=self.token_var).pack(side=tk.RIGHT, fill=tk.X, expand=True)
    
95. 
    
96.         btn_row = tk.Frame(cookie_frame)
    
97.         btn_row.pack(fill=tk.X, pady=(10, 0))
    
98. 
    
99.         tk.Button(btn_row, text='从整段Cookie解析', command=self.on_parse_cookie).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 6))
    
100.         tk.Button(btn_row, text='保存配置', command=self.on_save_config).pack(side=tk.RIGHT, fill=tk.X, expand=True, padx=(6, 0))
     
101. 
     
102.         action_frame = tk.LabelFrame(left, text='获取与刷新', padx=10, pady=10)
     
103.         action_frame.pack(fill=tk.X, pady=(0, 10))
     
104. 
     
105.         interval_row = tk.Frame(action_frame)
     
106.         interval_row.pack(fill=tk.X, pady=(10, 0))
     
107.         tk.Label(interval_row, text='自动刷新间隔(秒):').pack(side=tk.LEFT)
     
108.         self.interval_var = tk.StringVar(value='30')
     
109.         ttk.Spinbox(interval_row, from_=10, to=600, textvariable=self.interval_var, width=8).pack(side=tk.RIGHT)
     
110. 
     
111.         btn_row2 = tk.Frame(action_frame)
     
112.         btn_row2.pack(fill=tk.X, pady=(10, 0))
     
113. 
     
114.         tk.Button(btn_row2, text='测试Cookie', command=self.on_test_cookie).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 6))
     
115.         tk.Button(btn_row2, text='手动刷新', command=self.on_refresh).pack(side=tk.RIGHT, fill=tk.X, expand=True, padx=(6, 0))
     
116. 
     
117.         btn_row3 = tk.Frame(action_frame)
     
118.         btn_row3.pack(fill=tk.X, pady=(10, 0))
     
119.         self.auto_btn = tk.Button(btn_row3, text='开始自动刷新', command=self.on_toggle_auto)
     
120.         self.auto_btn.pack(fill=tk.X)
     
121. 
     
122.         export_frame = tk.LabelFrame(left, text='导出', padx=10, pady=10)
     
123.         export_frame.pack(fill=tk.X)
     
124.         tk.Button(export_frame, text='导出CSV', command=self.on_export_csv).pack(fill=tk.X)
     
125. 
     
126.     def _build_table(self, parent: tk.Frame) -> None:
     
127.         right = tk.Frame(parent)
     
128.         right.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)
     
129. 
     
130.         table_frame = tk.Frame(right)
     
131.         table_frame.pack(fill=tk.BOTH, expand=True)
     
132. 
     
133.         self.tree = ttk.Treeview(table_frame, columns=[c[0] for c in COLUMNS], show='headings')
     
134.         vsb = ttk.Scrollbar(table_frame, orient='vertical', command=self.tree.yview)
     
135.         hsb = ttk.Scrollbar(table_frame, orient='horizontal', command=self.tree.xview)
     
136.         self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
     
137. 
     
138.         self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
     
139.         vsb.pack(side=tk.RIGHT, fill=tk.Y)
     
140.         hsb.pack(side=tk.BOTTOM, fill=tk.X)
     
141. 
     
142.         for key, title, width in COLUMNS:
     
143.             self.tree.heading(key, text=title)
     
144.             self.tree.column(key, width=width, anchor=tk.W)
     
145. 
     
146.         self.tree.bind('<Double-1>', self.on_open_selected)
     
147. 
     
148.         menu = tk.Menu(self.root, tearoff=0)
     
149.         menu.add_command(label='打开雪球链接', command=self.on_open_selected)
     
150.         menu.add_command(label='复制代码', command=self.on_copy_symbol)
     
151.         self._context_menu = menu
     
152. 
     
153.         self.tree.bind('<Button-3>', self._show_context_menu)
     
154. 
     
155.     def _show_context_menu(self, event) -> None:
     
156.         item_id = self.tree.identify_row(event.y)
     
157.         if item_id:
     
158.             self.tree.selection_set(item_id)
     
159.             self._context_menu.tk_popup(event.x_root, event.y_root)
     
160. 
     
161.     def _set_status(self, msg: str) -> None:
     
162.         ts = datetime.now().strftime('%H:%M:%S')
     
163.         self.status_var.set(f'{ts} - {msg}')
     
164. 
     
165.     def _get_client(self) -> XueqiuClient:
     
166.         u = (self.u_var.get() or '').strip()
     
167.         token = (self.token_var.get() or '').strip()
     
168.         if not u or not token:
     
169.             raise ValueError('请先填写 u 和 xq_a_token，或粘贴整段Cookie后解析')
     
170.         return XueqiuClient(XueqiuCookies(u=u, xq_a_token=token))
     
171. 
     
172.     def on_parse_cookie(self) -> None:
     
173.         cookie_string = self.cookie_text.get('1.0', tk.END).strip()
     
174.         u, token = extract_required_cookies(cookie_string)
     
175.         if u:
     
176.             self.u_var.set(u)
     
177.         if token:
     
178.             self.token_var.set(token)
     
179. 
     
180.         if u and token:
     
181.             self._set_status('已从整段Cookie解析成功')
     
182.         else:
     
183.             messagebox.showwarning('提示', '未能解析到 u 或 xq_a_token，请确认Cookie内容')
     
184. 
     
185.     def _load_config_into_ui(self) -> None:
     
186.         cfg = load_config(CONFIG_FILE)
     
187.         if not isinstance(cfg, dict):
     
188.             return
     
189. 
     
190.         self.u_var.set(str(cfg.get('u', '') or ''))
     
191.         self.token_var.set(str(cfg.get('xq_a_token', '') or ''))
     
192.         self.interval_var.set(str(cfg.get('interval_seconds', '30') or '30'))
     
193. 
     
194.     def on_save_config(self) -> None:
     
195.         data = {
     
196.             'u': (self.u_var.get() or '').strip(),
     
197.             'xq_a_token': (self.token_var.get() or '').strip(),
     
198.             'interval_seconds': int((self.interval_var.get() or '30').strip() or '30'),
     
199.         }
     
200.         save_config(CONFIG_FILE, data)
     
201.         self._set_status('配置已保存')
     
202. 
     
203.     def on_test_cookie(self) -> None:
     
204.         def worker():
     
205.             try:
     
206.                 client = self._get_client()
     
207.                 start = time.time()
     
208.                 items = client.get_hot_stocks(type_id=HOT_STOCK_TYPE_ID, page=1, size=10)
     
209.                 cost = int((time.time() - start) * 1000)
     
210.                 self.root.after(0, lambda: self._set_status(f'Cookie有效，获取到 {len(items)} 条 (耗时 {cost}ms)'))
     
211.             except XueqiuAuthError as e:
     
212.                 self.root.after(0, lambda: self._on_cookie_expired(str(e)))
     
213.             except Exception as e:
     
214.                 self.root.after(0, lambda: messagebox.showerror('错误', str(e)))
     
215. 
     
216.         threading.Thread(target=worker, daemon=True).start()
     
217. 
     
218.     def _on_cookie_expired(self, msg: str) -> None:
     
219.         self._set_status('Cookie疑似失效')
     
220.         messagebox.showwarning('Cookie可能已失效', msg)
     
221. 
     
222.     def on_refresh(self) -> None:
     
223.         if not self._fetch_lock.acquire(blocking=False):
     
224.             return
     
225. 
     
226.         self._set_status('正在获取数据...')
     
227. 
     
228.         def worker():
     
229.             try:
     
230.                 client = self._get_client()
     
231.                 start = time.time()
     
232.                 items = client.get_hot_stocks(type_id=HOT_STOCK_TYPE_ID, page=1, size=100)
     
233.                 cost = int((time.time() - start) * 1000)
     
234.                 self.root.after(0, lambda: self._update_table(items, cost))
     
235.             except XueqiuAuthError as e:
     
236.                 self.root.after(0, lambda: self._on_cookie_expired(str(e)))
     
237.             except Exception as e:
     
238.                 self.root.after(0, lambda: messagebox.showerror('错误', str(e)))
     
239.                 self.root.after(0, lambda: self._set_status('获取失败'))
     
240.             finally:
     
241.                 self._fetch_lock.release()
     
242. 
     
243.         threading.Thread(target=worker, daemon=True).start()
     
244. 
     
245.     def _update_table(self, items: List[Dict[str, Any]], cost_ms: int) -> None:
     
246.         self._last_items = items
     
247.         for iid in self.tree.get_children():
     
248.             self.tree.delete(iid)
     
249. 
     
250.         for item in items:
     
251.             values = []
     
252.             for key, _title, _w in COLUMNS:
     
253.                 v = item.get(key)
     
254.                 if v is None:
     
255.                     values.append('')
     
256.                 else:
     
257.                     values.append(str(v))
     
258.             self.tree.insert('', tk.END, values=values)
     
259. 
     
260.         self._set_status(f'已更新 {len(items)} 条 (耗时 {cost_ms}ms)')
     
261. 
     
262.     def on_toggle_auto(self) -> None:
     
263.         if not self._auto_refresh:
     
264.             self._auto_refresh = True
     
265.             self.auto_btn.config(text='停止自动刷新')
     
266.             self._schedule_next_refresh(0)
     
267.             self._set_status('自动刷新已启动')
     
268.         else:
     
269.             self._auto_refresh = False
     
270.             self.auto_btn.config(text='开始自动刷新')
     
271.             if self._auto_after_id:
     
272.                 self.root.after_cancel(self._auto_after_id)
     
273.                 self._auto_after_id = None
     
274.             self._set_status('自动刷新已停止')
     
275. 
     
276.     def _schedule_next_refresh(self, delay_ms: int) -> None:
     
277.         if not self._auto_refresh:
     
278.             return
     
279.         self._auto_after_id = self.root.after(delay_ms, self._auto_tick)
     
280. 
     
281.     def _auto_tick(self) -> None:
     
282.         if not self._auto_refresh:
     
283.             return
     
284. 
     
285.         try:
     
286.             interval = int((self.interval_var.get() or '30').strip() or '30')
     
287.         except Exception:
     
288.             interval = 30
     
289. 
     
290.         self.on_refresh()
     
291.         self._schedule_next_refresh(interval * 1000)
     
292. 
     
293.     def _get_selected_symbol(self) -> Optional[str]:
     
294.         sel = self.tree.selection()
     
295.         if not sel:
     
296.             return None
     
297.         values = self.tree.item(sel[0], 'values')
     
298.         if not values:
     
299.             return None
     
300.         symbol_idx = next((i for i, c in enumerate(COLUMNS) if c[0] == 'symbol'), None)
     
301.         if symbol_idx is None:
     
302.             return None
     
303.         if symbol_idx >= len(values):
     
304.             return None
     
305.         return str(values[symbol_idx])
     
306. 
     
307.     def on_open_selected(self, _event=None) -> None:
     
308.         symbol = self._get_selected_symbol()
     
309.         if not symbol:
     
310.             return
     
311.         url = f'https://xueqiu.com/S/{symbol}'
     
312.         webbrowser.open(url)
     
313. 
     
314.     def on_copy_symbol(self) -> None:
     
315.         symbol = self._get_selected_symbol()
     
316.         if not symbol:
     
317.             return
     
318.         self.root.clipboard_clear()
     
319.         self.root.clipboard_append(symbol)
     
320.         self.root.update()
     
321.         self._set_status('已复制代码到剪贴板')
     
322. 
     
323.     def on_export_csv(self) -> None:
     
324.         if not self._last_items:
     
325.             messagebox.showinfo('提示', '没有数据可导出，请先获取数据')
     
326.             return
     
327. 
     
328.         from tkinter import filedialog
     
329. 
     
330.         filename = filedialog.asksaveasfilename(defaultextension='.csv', filetypes=[('CSV文件', '*.csv')])
     
331.         if not filename:
     
332.             return
     
333. 
     
334.         try:
     
335.             with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
     
336.                 writer = csv.writer(f)
     
337.                 writer.writerow([c[1] for c in COLUMNS] + ['链接'])
     
338.                 for item in self._last_items:
     
339.                     row = [item.get(c[0], '') for c in COLUMNS]
     
340.                     symbol = item.get('symbol', '')
     
341.                     row.append(f'https://xueqiu.com/S/{symbol}' if symbol else '')
     
342.                     writer.writerow(row)
     
343.             self._set_status(f'已导出: {filename}')
     
344.         except Exception as e:
     
345.             messagebox.showerror('错误', f'导出失败: {e}')
     
346. 
     
347. 
     
348. def main() -> None:
     
349.     root = tk.Tk()
     
350.     app = XueqiuHotStockGUI(root)
     
351.     root.mainloop()
     
352. 
     
353. 
     
354. if __name__ == '__main__':
     
355.     main()
     

复制代码