使用Python簡單編寫一個股票監(jiān)控系統(tǒng)
更新時間：2024年12月19日 10:31:16 作者：PieroPc
這篇文章主要為大家詳細介紹了如何使用Python簡單編寫一個股票監(jiān)控系統(tǒng),文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下
圖樣
最小化時
上代碼
import json
import logging
import threading
from typing import Dict, List
import efinance as ef
import time
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import sys
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import queue
import requests
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('stock_monitor.log', encoding='utf-8'),
logging.StreamHandler()
]
)
class StockMonitorGUI:
def __init__(self, root):
self.root = root
self.root.title("股票監(jiān)控系統(tǒng)")
self.root.geometry("1024x600")
# 添加最小化事件綁定
self.root.protocol("WM_DELETE_WINDOW", self.on_closing)
self.root.bind("<Unmap>", self.on_minimize)
self.float_window = None # 懸浮窗口
self.is_minimized = False
self.monitor = StockMonitor()
self.monitor.set_log_callback(self.add_log)
self.running = False
self.monitor_thread = None
self.log_queue = queue.Queue()
self.setup_gui()
self.update_log()
self.update_stock_table(self.get_initial_stock_data())
def setup_gui(self):
# 創(chuàng)建主框架
main_frame = ttk.Frame(self.root, padding="10")
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 股票信息表格
table_frame = ttk.LabelFrame(main_frame, text="股票監(jiān)控列表", padding="5")
table_frame.grid(row=0, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S))
self.tree = ttk.Treeview(table_frame, columns=('代碼', '名稱', '當前價格', '買入價', '賣出價', '狀態(tài)'), show='headings', height=8) # 設置固定高度
self.tree.heading('代碼', text='代碼')
self.tree.heading('名稱', text='名稱')
self.tree.heading('當前價格', text='當前價格')
self.tree.heading('買入價', text='買入價')
self.tree.heading('賣出價', text='賣出價')
self.tree.heading('狀態(tài)', text='狀態(tài)')
# 設置列寬
self.tree.column('代碼', width=80)
self.tree.column('名稱', width=100)
self.tree.column('當前價格', width=80)
self.tree.column('買入價', width=80)
self.tree.column('賣出價', width=80)
self.tree.column('狀態(tài)', width=100)
self.tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 添加滾動條
scrollbar = ttk.Scrollbar(table_frame, orient=tk.VERTICAL, command=self.tree.yview)
self.tree.configure(yscrollcommand=scrollbar.set)
scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S))
# 日志顯示區(qū)域
log_frame = ttk.LabelFrame(main_frame, text="運行日志", padding="5")
log_frame.grid(row=1, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S))
self.log_text = scrolledtext.ScrolledText(log_frame, height=8) # 減小日志區(qū)域高度
self.log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 控制按鈕
button_frame = ttk.Frame(main_frame, padding="5")
button_frame.grid(row=2, column=0, columnspan=3, sticky=(tk.W, tk.E))
# 使用更緊湊的按鈕布局
self.start_button = ttk.Button(button_frame, text="開始監(jiān)控", command=self.start_monitoring, width=10)
self.start_button.grid(row=0, column=0, padx=3)
self.stop_button = ttk.Button(button_frame, text="停止監(jiān)控", command=self.stop_monitoring, state=tk.DISABLED, width=10)
self.stop_button.grid(row=0, column=1, padx=3)
self.add_button = ttk.Button(button_frame, text="添加股票", command=self.show_add_dialog, width=10)
self.add_button.grid(row=0, column=2, padx=3)
self.edit_button = ttk.Button(button_frame, text="修改股票", command=self.show_edit_dialog, width=10)
self.edit_button.grid(row=0, column=3, padx=3)
self.delete_button = ttk.Button(button_frame, text="刪除股票", command=self.delete_stock, width=10)
self.delete_button.grid(row=0, column=4, padx=3)
self.email_settings_button = ttk.Button(button_frame, text="郵件設置", command=self.show_email_settings, width=10)
self.email_settings_button.grid(row=0, column=5, padx=3)
self.qq_settings_button = ttk.Button(button_frame, text="QQ設置", command=self.show_qq_settings, width=10)
self.qq_settings_button.grid(row=0, column=6, padx=3)
self.weixin_settings_button = ttk.Button(button_frame, text="微信設置", command=self.show_weixin_settings, width=10)
self.weixin_settings_button.grid(row=0, column=7, padx=3)
# 調(diào)整窗口大小
self.root.geometry("800x500") # 減小窗口高度
# 配置grid權重
self.root.grid_rowconfigure(0, weight=1)
self.root.grid_columnconfigure(0, weight=1)
main_frame.grid_rowconfigure(1, weight=1) # 讓日志區(qū)域可以擴展
main_frame.grid_columnconfigure(0, weight=1)
def update_log(self):
while True:
try:
log_message = self.log_queue.get_nowait()
self.log_text.insert(tk.END, log_message + '\n')
self.log_text.see(tk.END)
except queue.Empty:
break
self.root.after(100, self.update_log)
def update_stock_table(self, stock_data):
# 清空現(xiàn)有數(shù)據(jù)
for item in self.tree.get_children():
self.tree.delete(item)
# 插入新數(shù)據(jù)
for stock in stock_data:
self.tree.insert('', tk.END, values=stock)
# 更新懸浮窗口
self.update_float_window(stock_data)
def start_monitoring(self):
self.add_log("啟動股票監(jiān)控系統(tǒng)...")
self.running = True
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
self.monitor_thread = threading.Thread(target=self.monitoring_task)
self.monitor_thread.daemon = True
self.monitor_thread.start()
def stop_monitoring(self):
self.running = False
self.start_button.config(state=tk.NORMAL)
self.stop_button.config(state=tk.DISABLED)
self.add_log("正在停止股票監(jiān)控系統(tǒng)...")
def monitoring_task(self):
self.add_log("開始監(jiān)控股票...")
while self.running:
try:
stock_data = self.monitor.get_stock_data()
self.root.after(0, self.update_stock_table, stock_data)
time.sleep(self.monitor.check_interval)
except Exception as e:
self.add_log(f"錯誤: {str(e)}")
time.sleep(self.monitor.check_interval)
self.add_log("停止監(jiān)控股票...")
def show_add_dialog(self):
dialog = tk.Toplevel(self.root)
dialog.title("添加股票")
dialog.geometry("300x200")
dialog.transient(self.root)
ttk.Label(dialog, text="股票代碼:").grid(row=0, column=0, padx=5, pady=5)
code_entry = ttk.Entry(dialog)
code_entry.grid(row=0, column=1, padx=5, pady=5)
ttk.Label(dialog, text="股票名稱:").grid(row=1, column=0, padx=5, pady=5)
name_entry = ttk.Entry(dialog)
name_entry.grid(row=1, column=1, padx=5, pady=5)
ttk.Label(dialog, text="買入價:").grid(row=2, column=0, padx=5, pady=5)
buy_entry = ttk.Entry(dialog)
buy_entry.grid(row=2, column=1, padx=5, pady=5)
ttk.Label(dialog, text="賣出價:").grid(row=3, column=0, padx=5, pady=5)
sell_entry = ttk.Entry(dialog)
sell_entry.grid(row=3, column=1, padx=5, pady=5)
def save_stock():
try:
new_stock = {
"code": code_entry.get(),
"name": name_entry.get(),
"buy_price": float(buy_entry.get()),
"sell_price": float(sell_entry.get())
}
if not new_stock["code"] or not new_stock["name"]:
messagebox.showerror("錯誤", "股票代碼和名稱不能為空！")
return
self.monitor.add_stock(new_stock)
dialog.destroy()
self.add_log(f"添加股票成功：{new_stock['name']}")
self.update_stock_table(self.get_initial_stock_data())
except ValueError:
messagebox.showerror("錯誤", "請輸入有效的價格！")
ttk.Button(dialog, text="保存", command=save_stock).grid(row=4, column=0, columnspan=2, pady=20)
def show_edit_dialog(self):
selected = self.tree.selection()
if not selected:
messagebox.showwarning("提示", "請先選擇要修改的股票！")
return
item = self.tree.item(selected[0])
values = item['values']
dialog = tk.Toplevel(self.root)
dialog.title("修改股票")
dialog.geometry("300x200")
dialog.transient(self.root)
ttk.Label(dialog, text="股票代碼:").grid(row=0, column=0, padx=5, pady=5)
code_entry = ttk.Entry(dialog)
code_entry.insert(0, values[0])
code_entry.config(state='readonly')
code_entry.grid(row=0, column=1, padx=5, pady=5)
ttk.Label(dialog, text="股票名稱:").grid(row=1, column=0, padx=5, pady=5)
name_entry = ttk.Entry(dialog)
name_entry.insert(0, values[1])
name_entry.grid(row=1, column=1, padx=5, pady=5)
ttk.Label(dialog, text="買入價:").grid(row=2, column=0, padx=5, pady=5)
buy_entry = ttk.Entry(dialog)
buy_entry.insert(0, values[3])
buy_entry.grid(row=2, column=1, padx=5, pady=5)
ttk.Label(dialog, text="賣出價:").grid(row=3, column=0, padx=5, pady=5)
sell_entry = ttk.Entry(dialog)
sell_entry.insert(0, values[4])
sell_entry.grid(row=3, column=1, padx=5, pady=5)
def save_changes():
try:
updated_stock = {
"code": values[0],
"name": name_entry.get(),
"buy_price": float(buy_entry.get()),
"sell_price": float(sell_entry.get())
}
if not updated_stock["name"]:
messagebox.showerror("錯誤", "股票名稱不能為空！")
return
self.monitor.update_stock(updated_stock)
dialog.destroy()
self.add_log(f"修改股票成功：{updated_stock['name']}")
self.update_stock_table(self.get_initial_stock_data())
except ValueError:
messagebox.showerror("錯誤", "請輸入有效的價格！")
ttk.Button(dialog, text="保存", command=save_changes).grid(row=4, column=0, columnspan=2, pady=20)
def delete_stock(self):
selected = self.tree.selection()
if not selected:
messagebox.showwarning("提示", "請先選擇要刪除的股票！")
return
item = self.tree.item(selected[0])
stock_code = item['values'][0]
stock_name = item['values'][1]
if messagebox.askyesno("確認", f"確定要刪除股票 {stock_name} 嗎？"):
self.monitor.delete_stock(stock_code)
self.add_log(f"刪除股票成功：{stock_name}")
self.update_stock_table(self.get_initial_stock_data())
def show_email_settings(self):
dialog = tk.Toplevel(self.root)
dialog.title("郵件服務設置")
dialog.geometry("400x350") # 增加一點高度
dialog.transient(self.root)
frame = ttk.Frame(dialog, padding="10")
frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 添加啟用郵件通知選項
enabled_var = tk.BooleanVar(value=self.monitor.email_config.get('enabled', False))
ttk.Checkbutton(frame, text="啟用郵件通知", variable=enabled_var).grid(row=0, column=0, columnspan=2, pady=5)
# SMTP服務器???置
ttk.Label(frame, text="SMTP服務器:").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W)
host_entry = ttk.Entry(frame, width=30)
host_entry.insert(0, self.monitor.email_config['host'])
host_entry.grid(row=1, column=1, padx=5, pady=5)
# 端口設置
ttk.Label(frame, text="端口:").grid(row=2, column=0, padx=5, pady=5, sticky=tk.W)
port_entry = ttk.Entry(frame, width=30)
port_entry.insert(0, "465") # 默認端口
port_entry.grid(row=2, column=1, padx=5, pady=5)
# 用戶名設置
ttk.Label(frame, text="郵箱賬號:").grid(row=3, column=0, padx=5, pady=5, sticky=tk.W)
user_entry = ttk.Entry(frame, width=30)
user_entry.insert(0, self.monitor.email_config['user'])
user_entry.grid(row=3, column=1, padx=5, pady=5)
# 密碼設置
ttk.Label(frame, text="授權密碼:").grid(row=4, column=0, padx=5, pady=5, sticky=tk.W)
pass_entry = ttk.Entry(frame, width=30, show="*")
pass_entry.insert(0, self.monitor.email_config['password'])
pass_entry.grid(row=4, column=1, padx=5, pady=5)
# 發(fā)件人設置
ttk.Label(frame, text="發(fā)件人:").grid(row=5, column=0, padx=5, pady=5, sticky=tk.W)
sender_entry = ttk.Entry(frame, width=30)
sender_entry.insert(0, self.monitor.email_config['sender'])
sender_entry.grid(row=5, column=1, padx=5, pady=5)
# 收件人設置
ttk.Label(frame, text="收件人:").grid(row=6, column=0, padx=5, pady=5, sticky=tk.W)
receivers_entry = ttk.Entry(frame, width=30)
receivers_entry.insert(0, ",".join(self.monitor.email_config['receivers']))
receivers_entry.grid(row=6, column=1, padx=5, pady=5)
# 郵件主題設置
ttk.Label(frame, text="郵件主題:").grid(row=7, column=0, padx=5, pady=5, sticky=tk.W)
title_entry = ttk.Entry(frame, width=30)
title_entry.insert(0, self.monitor.email_config['title'])
title_entry.grid(row=7, column=1, padx=5, pady=5)
def test_email():
if not enabled_var.get():
messagebox.showwarning("提示", "請先啟用郵件通知！")
return
# 臨時保存當前設置
temp_config = {
'enabled': enabled_var.get(),
'host': host_entry.get(),
'user': user_entry.get(),
'password': pass_entry.get(),
'sender': sender_entry.get(),
'receivers': [r.strip() for r in receivers_entry.get().split(',')],
'title': title_entry.get()
}
try:
with smtplib.SMTP_SSL(temp_config['host'], 465) as smtp:
smtp.login(temp_config['user'], temp_config['password'])
message = MIMEText("這是一封測試郵件，如果您收到這封郵件，說明郵件服務設置正確。", 'plain', 'utf-8')
message['From'] = temp_config['sender']
message['To'] = ",".join(temp_config['receivers'])
message['Subject'] = "測試郵件"
smtp.sendmail(
temp_config['sender'],
temp_config['receivers'],
message.as_string()
)
messagebox.showinfo("成功", "測試郵件發(fā)送成功！")
except Exception as e:
messagebox.showerror("錯誤", f"測試郵件發(fā)送失?。簕str(e)}")
def save_settings():
try:
new_config = {
'enabled': enabled_var.get(),
'host': host_entry.get(),
'user': user_entry.get(),
'password': pass_entry.get(),
'sender': sender_entry.get(),
'receivers': [r.strip() for r in receivers_entry.get().split(',')],
'title': title_entry.get()
}
# 驗證必填字段
if new_config['enabled'] and not all([new_config['host'], new_config['user'],
new_config['password'], new_config['sender'],
new_config['receivers'], new_config['title']]):
messagebox.showerror("錯誤", "啟用郵件通知時所有字段都必須填寫！")
return
# 更新配置
self.monitor.update_email_config(new_config)
messagebox.showinfo("成功", "郵件設置已保存！")
dialog.destroy()
except Exception as e:
messagebox.showerror("錯誤", f"保存設置失?。簕str(e)}")
# 按鈕框架
button_frame = ttk.Frame(frame)
button_frame.grid(row=8, column=0, columnspan=2, pady=20)
test_button = ttk.Button(button_frame, text="測試", command=test_email)
test_button.grid(row=0, column=0, padx=5)
save_button = ttk.Button(button_frame, text="保存", command=save_settings)
save_button.grid(row=0, column=1, padx=5)
def show_qq_settings(self):
dialog = tk.Toplevel(self.root)
dialog.title("QQ機器人設置")
dialog.geometry("400x250")
dialog.transient(self.root)
frame = ttk.Frame(dialog, padding="10")
frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 啟用QQ通知
enabled_var = tk.BooleanVar(value=self.monitor.qq_config.get('enabled', False))
ttk.Checkbutton(frame, text="啟用QQ通知", variable=enabled_var).grid(row=0, column=0, columnspan=2, pady=5)
# API地址
ttk.Label(frame, text="API地址:").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W)
api_entry = ttk.Entry(frame, width=30)
api_entry.insert(0, self.monitor.qq_config.get('api_url', 'http://127.0.0.1:5700'))
api_entry.grid(row=1, column=1, padx=5, pady=5)
# QQ號
ttk.Label(frame, text="接收QQ:").grid(row=2, column=0, padx=5, pady=5, sticky=tk.W)
qq_entry = ttk.Entry(frame, width=30)
qq_entry.insert(0, self.monitor.qq_config.get('qq_id', ''))
qq_entry.grid(row=2, column=1, padx=5, pady=5)
# 訪問令牌
ttk.Label(frame, text="訪問令牌:").grid(row=3, column=0, padx=5, pady=5, sticky=tk.W)
token_entry = ttk.Entry(frame, width=30, show="*")
token_entry.insert(0, self.monitor.qq_config.get('access_token', ''))
token_entry.grid(row=3, column=1, padx=5, pady=5)
def test_qq():
config = {
'enabled': enabled_var.get(),
'api_url': api_entry.get(),
'qq_id': qq_entry.get(),
'access_token': token_entry.get()
}
try:
url = f"{config['api_url']}/send_private_msg"
params = {
'user_id': config['qq_id'],
'message': "這是一條測試消息，如果您收到這消息，說明QQ機器人設置正確。"
}
headers = {
'Authorization': f"Bearer {config['access_token']}"
}
response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
messagebox.showinfo("成功", "測試消息發(fā)送成功！")
else:
messagebox.showerror("錯誤", f"測試消息發(fā)送失敗：{response.text}")
except Exception as e:
messagebox.showerror("錯誤", f"測試消息發(fā)送失?。簕str(e)}")
def save_settings():
try:
new_config = {
'enabled': enabled_var.get(),
'api_url': api_entry.get(),
'qq_id': qq_entry.get(),
'access_token': token_entry.get()
}
if new_config['enabled'] and not all([new_config['api_url'], new_config['qq_id']]):
messagebox.showerror("錯誤", "啟用QQ通知時必須填寫API地址和接收QQ！")
return
self.monitor.qq_config = new_config
self.monitor.config['qq'] = new_config
self.monitor._save_config()
messagebox.showinfo("成功", "QQ設置已保存！")
dialog.destroy()
except Exception as e:
messagebox.showerror("錯誤", f"保存設置失敗：{str(e)}")
# 按鈕框架
button_frame = ttk.Frame(frame)
button_frame.grid(row=4, column=0, columnspan=2, pady=20)
test_button = ttk.Button(button_frame, text="測試", command=test_qq)
test_button.grid(row=0, column=0, padx=5)
save_button = ttk.Button(button_frame, text="保存", command=save_settings)
save_button.grid(row=0, column=1, padx=5)
def show_weixin_settings(self):
dialog = tk.Toplevel(self.root)
dialog.title("企業(yè)微信設置")
dialog.geometry("400x300")
dialog.transient(self.root)
frame = ttk.Frame(dialog, padding="10")
frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
# 啟用微信通知
enabled_var = tk.BooleanVar(value=self.monitor.weixin_config.get('enabled', False))
ttk.Checkbutton(frame, text="啟用企業(yè)微信通知", variable=enabled_var).grid(row=0, column=0, columnspan=2, pady=5)
# 企業(yè)ID
ttk.Label(frame, text="企業(yè)ID:").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W)
corp_id_entry = ttk.Entry(frame, width=30)
corp_id_entry.insert(0, self.monitor.weixin_config.get('corp_id', ''))
corp_id_entry.grid(row=1, column=1, padx=5, pady=5)
# 應用ID
ttk.Label(frame, text="應用ID:").grid(row=2, column=0, padx=5, pady=5, sticky=tk.W)
agent_id_entry = ttk.Entry(frame, width=30)
agent_id_entry.insert(0, self.monitor.weixin_config.get('agent_id', ''))
agent_id_entry.grid(row=2, column=1, padx=5, pady=5)
# 應用密鑰
ttk.Label(frame, text="應用密鑰:").grid(row=3, column=0, padx=5, pady=5, sticky=tk.W)
secret_entry = ttk.Entry(frame, width=30, show="*")
secret_entry.insert(0, self.monitor.weixin_config.get('corp_secret', ''))
secret_entry.grid(row=3, column=1, padx=5, pady=5)
# 接收用戶
ttk.Label(frame, text="接收用戶:").grid(row=4, column=0, padx=5, pady=5, sticky=tk.W)
to_user_entry = ttk.Entry(frame, width=30)
to_user_entry.insert(0, self.monitor.weixin_config.get('to_user', '@all'))
to_user_entry.grid(row=4, column=1, padx=5, pady=5)
def test_weixin():
if not enabled_var.get():
messagebox.showwarning("提示", "請先啟用企業(yè)微信通知！")
return
config = {
'enabled': enabled_var.get(),
'corp_id': corp_id_entry.get(),
'agent_id': agent_id_entry.get(),
'corp_secret': secret_entry.get(),
'to_user': to_user_entry.get()
}
# 創(chuàng)建臨時的 StockMonitor 實例來測試
temp_monitor = StockMonitor()
temp_monitor.weixin_config = config
temp_monitor.set_log_callback(self.add_log)
temp_monitor.send_weixin_message("這是一條測試消息，如果您收到這條消息，說明企業(yè)微信設置正確。")
def save_settings():
try:
new_config = {
'enabled': enabled_var.get(),
'corp_id': corp_id_entry.get(),
'agent_id': agent_id_entry.get(),
'corp_secret': secret_entry.get(),
'to_user': to_user_entry.get()
}
if new_config['enabled'] and not all([new_config['corp_id'],
new_config['agent_id'],
new_config['corp_secret']]):
messagebox.showerror("錯誤", "啟用企業(yè)微信通知時必須填寫企業(yè)ID、應用ID和應用密鑰！")
return
self.monitor.weixin_config = new_config
self.monitor.config['weixin'] = new_config
self.monitor._save_config()
messagebox.showinfo("成功", "企業(yè)微信設置已保存！")
dialog.destroy()
except Exception as e:
messagebox.showerror("錯誤", f"保存設置失?。簕str(e)}")
# 按鈕框架
button_frame = ttk.Frame(frame)
button_frame.grid(row=5, column=0, columnspan=2, pady=20)
test_button = ttk.Button(button_frame, text="測試", command=test_weixin)
test_button.grid(row=0, column=0, padx=5)
save_button = ttk.Button(button_frame, text="保存", command=save_settings)
save_button.grid(row=0, column=1, padx=5)
def add_log(self, message: str):
"""添加日志到界面"""
self.log_text.insert(tk.END, message + '\n')
self.log_text.see(tk.END) # 滾動到最新的日志
def get_initial_stock_data(self):
"""獲取初始股票數(shù)據(jù)用于顯示"""
result = []
for stock in self.monitor.stocks:
result.append((
stock['code'],
stock['name'],
'等待更新', # 初始顯示時還沒有實時價格
stock['buy_price'],
stock['sell_price'],
'等待監(jiān)控' # 初始狀態(tài)
))
return result
def create_float_window(self):
"""創(chuàng)建懸浮窗口"""
self.float_window = tk.Toplevel()
self.float_window.title("股票監(jiān)控")
# 設置窗口屬性
self.float_window.attributes('-topmost', True) # 始終置頂
self.float_window.overrideredirect(True) # 無邊框
self.float_window.attributes('-alpha', 0.9) # 設置透明度
# 獲取屏幕尺寸
screen_width = self.root.winfo_screenwidth()
screen_height = self.root.winfo_screenheight()
# 設置窗口位置（右下角）
window_width = 200
window_height = 300
x = screen_width - window_width - 10
y = screen_height - window_height - 50
self.float_window.geometry(f"{window_width}x{window_height}+{x}+{y}")
# 創(chuàng)建標題欄
title_frame = ttk.Frame(self.float_window)
title_frame.pack(fill=tk.X, padx=2, pady=2)
ttk.Label(title_frame, text="股票監(jiān)控").pack(side=tk.LEFT, padx=5)
# 添加關閉和還原按鈕
ttk.Button(title_frame, text="□", width=3, command=self.restore_window).pack(side=tk.RIGHT, padx=1)
ttk.Button(title_frame, text="×", width=3, command=self.on_closing).pack(side=tk.RIGHT, padx=1)
# 添加拖動功能
title_frame.bind("<Button-1>", self.start_move)
title_frame.bind("<B1-Motion>", self.on_move)
# 創(chuàng)建股票信息顯示區(qū)域
self.float_tree = ttk.Treeview(self.float_window, columns=('名稱', '價格', '狀態(tài)'), show='headings', height=10)
self.float_tree.heading('名稱', text='名稱')
self.float_tree.heading('價格', text='價格')
self.float_tree.heading('狀態(tài)', text='狀態(tài)')
# 設置列寬
self.float_tree.column('名稱', width=60)
self.float_tree.column('價格', width=60)
self.float_tree.column('狀態(tài)', width=60)
self.float_tree.pack(fill=tk.BOTH, expand=True, padx=2, pady=2)
def update_float_window(self, stock_data):
"""更新懸浮窗口的股票信息"""
if self.float_window and self.is_minimized:
for item in self.float_tree.get_children():
self.float_tree.delete(item)
for stock in stock_data:
self.float_tree.insert('', tk.END, values=(stock[1], stock[2], stock[5]))
def start_move(self, event):
"""開始拖動窗口"""
self.x = event.x
self.y = event.y
def on_move(self, event):
"""拖動窗口"""
deltax = event.x - self.x
deltay = event.y - self.y
x = self.float_window.winfo_x() + deltax
y = self.float_window.winfo_y() + deltay
self.float_window.geometry(f"+{x}+{y}")
def on_minimize(self, event):
"""最小化主窗口時顯示懸浮窗"""
if not self.float_window:
self.create_float_window()
self.is_minimized = True
self.float_window.deiconify()
def restore_window(self):
"""還原主窗口"""
self.root.deiconify()
self.is_minimized = False
if self.float_window:
self.float_window.withdraw()
def on_closing(self):
"""關閉程序"""
if messagebox.askokcancel("退出", "確定要退出程序嗎？"):
if self.float_window:
self.float_window.destroy()
self.root.destroy()
class StockMonitor:
def __init__(self, config_file: str = 'stock_config.json'):
self.config = self._load_config(config_file)
self.email_config = self.config['email']
self.qq_config = self.config.get('qq', {'enabled': False}) # 添加QQ配置
self.weixin_config = self.config.get('weixin', {'enabled': False}) # 添加微信配置
self.stocks = self.config['stocks']
self.check_interval = self.config['check_interval']
self.log_callback = None # 添加日志回調(diào)函數(shù)
def set_log_callback(self, callback):
"""設置日志回調(diào)函數(shù)"""
self.log_callback = callback
def log(self, message: str, level: str = 'info'):
"""統(tǒng)一的日志處理方法"""
if level == 'error':
logging.error(message)
else:
logging.info(message)
if self.log_callback:
self.log_callback(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {message}")
def _load_config(self, config_file: str) -> Dict:
try:
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
logging.error(f"配置文件 {config_file} 不存在")
sys.exit(1)
except json.JSONDecodeError:
logging.error(f"配置文件 {config_file} 格式錯誤")
sys.exit(1)
def send_email(self, content: str) -> None:
"""發(fā)送郵件"""
if not self.email_config.get('enabled', False): # 檢查是否啟用
return
message = MIMEText(content, 'plain', 'utf-8')
message['From'] = self.email_config['sender']
message['To'] = ",".join(self.email_config['receivers'])
message['Subject'] = self.email_config['title']
try:
with smtplib.SMTP_SSL(self.email_config['host'], 465) as smtp:
smtp.login(self.email_config['user'], self.email_config['password'])
smtp.sendmail(
self.email_config['sender'],
self.email_config['receivers'],
message.as_string()
)
self.log("郵件發(fā)送成功！")
except Exception as e:
self.log(f"郵件發(fā)送失敗: {str(e)}", 'error')
def get_stock_status(self, current_price: float, buy_price: float, sell_price: float) -> str:
if current_price <= buy_price:
return "建議買入"
elif current_price >= sell_price:
return "建議賣出"
return "觀察中"
def send_qq_message(self, message: str) -> None:
"""發(fā)送QQ消息"""
if not self.qq_config.get('enabled', False):
return
try:
url = f"{self.qq_config['api_url']}/send_private_msg"
params = {
'user_id': self.qq_config['qq_id'],
'message': message
}
headers = {
'Authorization': f"Bearer {self.qq_config.get('access_token', '')}"
}
response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
self.log("QQ消息發(fā)送成功！")
else:
self.log(f"QQ消息發(fā)送失敗: {response.text}", 'error')
except Exception as e:
self.log(f"QQ消息發(fā)送失敗: {str(e)}", 'error')
def send_weixin_message(self, message: str) -> None:
"""發(fā)送企業(yè)微信消息"""
if not self.weixin_config.get('enabled', False):
return
try:
# 獲取訪問令牌
token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
token_params = {
'corpid': self.weixin_config['corp_id'],
'corpsecret': self.weixin_config['corp_secret']
}
token_response = requests.get(token_url, params=token_params)
token_data = token_response.json()
if token_data['errcode'] != 0:
self.log(f"獲取微信訪問令牌失敗: {token_data['errmsg']}", 'error')
return
access_token = token_data['access_token']
# 發(fā)送消息
send_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
send_data = {
'touser': self.weixin_config['to_user'],
'msgtype': 'text',
'agentid': self.weixin_config['agent_id'],
'text': {
'content': message
}
}
response = requests.post(send_url, json=send_data)
result = response.json()
if result['errcode'] == 0:
self.log("微信消息發(fā)送成功！")
else:
self.log(f"微信消息發(fā)送失敗: {result['errmsg']}", 'error')
except Exception as e:
self.log(f"微信消息發(fā)送失敗: {str(e)}", 'error')
def get_stock_data(self) -> List:
try:
df = ef.stock.get_realtime_quotes()
stock_list = df.values.tolist()
result = []
for stock_info in self.stocks:
stock_data = next(
(item for item in stock_list if item[0] == stock_info['code']),
None
)
if stock_data:
current_price = float(stock_data[3])
status = self.get_stock_status(
current_price,
stock_info['buy_price'],
stock_info['sell_price']
)
result.append((
stock_info['code'],
stock_info['name'],
current_price,
stock_info['buy_price'],
stock_info['sell_price'],
status
))
# 檢查是否需要發(fā)送提醒
if status in ["建議買入", "建議賣出"]:
content = f'當前{stock_info["name"]}的價格是: {current_price}, {status}'
self.send_email(content)
self.send_qq_message(content)
self.send_weixin_message(content) # 添加微信消息提醒
self.log(content)
else:
self.log(f'當前{stock_info["name"]}的價格是: {current_price} 耐心觀察中...')
else:
self.log(f"未找到股票 {stock_info['code']} 的數(shù)據(jù)", 'error')
return result
except Exception as e:
self.log(f"獲取股票數(shù)據(jù)失敗: {str(e)}", 'error')
return []
def add_stock(self, stock: Dict) -> None:
self.stocks.append(stock)
self._save_config()
def update_stock(self, updated_stock: Dict) -> None:
for i, stock in enumerate(self.stocks):
if stock['code'] == updated_stock['code']:
self.stocks[i] = updated_stock
break
self._save_config()
def delete_stock(self, stock_code: str) -> None:
self.stocks = [s for s in self.stocks if s['code'] != stock_code]
self._save_config()
def _save_config(self) -> None:
try:
self.config['stocks'] = self.stocks
with open('stock_config.json', 'w', encoding='utf-8') as f:
json.dump(self.config, f, indent=4, ensure_ascii=False)
except Exception as e:
logging.error(f"保存配置文件失敗: {str(e)}")
def update_email_config(self, new_config: Dict) -> None:
self.email_config = new_config
self.config['email'] = new_config
self._save_config()
def main():
root = tk.Tk()
app = StockMonitorGUI(root)
root.mainloop()
if __name__ == "__main__":
main()上配置文件：stock_config.json
{
"email": {
"enabled": false,
"host": "smtp.163.com",
"user": "i238@163.com",
"password": "JIMRV",
"sender": "i25@163.com",
"receivers": [
"123@qq.com"
],
"title": "股票監(jiān)控買賣提示"
},
"qq": {
"enabled": false,
"api_url": "http://127.0.0.1:5700",
"qq_id": "123456789",
"access_token": "your_access_token"
},
"weixin": {
"enabled": false,
"corp_id": "your_corp_id",
"agent_id": "your_agent_id",
"corp_secret": "your_corp_secret",
"to_user": "@all"
},
"stocks": [
{
"code": "000776",
"name": "廣發(fā)證券",
"buy_price": 11.3,
"sell_price": 12.6
},
{
"code": "002945",
"name": "華林證券",
"buy_price": 12.0,
"sell_price": 16.0
}
],
"check_interval": 60
}上文檔說明--- 股票監(jiān)控系統(tǒng)使用說明
1. 系統(tǒng)簡介
這是一個基于Python開發(fā)的股票監(jiān)控系統(tǒng)，可以實時監(jiān)控多支股票的價格變動，并通過多種方式（郵件、QQ、企業(yè)微信）發(fā)送買賣提醒。
主要功能：
- 實時監(jiān)控多支股票價格
- 自定義買入賣出價格
- 多種提醒方式（郵件/QQ/企業(yè)微信）
- 懸浮窗口顯示實時行情
- 完整的日志記錄
2. 系統(tǒng)要求
Python 3.6 或更高版本
需要安裝的依賴包：
pip install efinance requests
3. 配置文件說明
配置文件為 stock_config.json，包含以下主要配置項：
son
{
"email": {
"enabled": false, // 是否啟用郵件通知
"host": "smtp.163.com", // SMTP服務器
"user": "xxx@163.com", // 郵箱賬號
"password": "xxx", // 郵箱授權碼
"sender": "xxx@163.com", // 發(fā)件人
"receivers": ["xxx@qq.com"],// 收件人列表
"title": "股票監(jiān)控提示" // 郵件主題
},
"qq": {
"enabled": false, // 是否啟用QQ通知
"api_url": "http://127.0.0.1:5700", // go-cqhttp服務地址
"qq_id": "123456789", // 接收消息的QQ號
"access_token": "xxx" // 訪問令牌
},
"weixin": {
"enabled": false, // 是否啟用企業(yè)微信通知
"corp_id": "xxx", // 企業(yè)ID
"agent_id": "xxx", // 應用ID
"corp_secret": "xxx", // 應用密鑰
"to_user": "@all" // 接收用戶
},
"stocks": [ // 監(jiān)控的股票列表
{
"code": "000776", // 股票代碼
"name": "廣發(fā)證券", // 股票名稱
"buy_price": 11.3, // 買入價
"sell_price": 12.6 // 賣出價
}
],
"check_interval": 60 // 檢查間隔（秒）
}
4. 使用說明
4.1 啟動程序
運行 股票監(jiān)聽器.py 文件啟動程序。
4.2 股票管理
- 添加股票：點擊"添加股票"按鈕，輸入股票代碼、名稱、買入價和賣出價
- 修改股票：選中要修改的股票，點擊"修改股票"按鈕
- 刪除股票：選中要刪除的股票，點擊"刪除股票"按鈕
4.3 通知設置
郵件設置：
- 點擊"郵件設置"按鈕
- 勾選"啟用郵件通知"
- 填寫SMTP服務器、郵箱賬號等信息
- 可點擊"測試"按鈕測試設置
QQ設置：
- 需要先配置并運行 go-cqhttp
- 點擊"QQ設置"按鈕
- 勾選"啟用QQ通知"
- 填寫API地址和接收QQ號
- 可點擊"測試"按鈕測試設置
企業(yè)微信設置：
- 需要有企業(yè)微信管理員權限
- 點擊"微信設置"按鈕
- 勾選"啟用企業(yè)微信通知"
- 填寫企業(yè)ID、應用ID等信息
- 可點擊"測試"按鈕測試設置
4.4 監(jiān)控操作
開始監(jiān)控：點擊"開始監(jiān)控"按鈕
停止監(jiān)控：點擊"停止監(jiān)控"按鈕
最小化：
- 程序最小化后會在屏幕右下角顯示懸浮窗
- 懸浮窗可以拖動位置
- 點擊"□"按鈕可以還原主窗口
- 點擊"×"按鈕可以關閉程序
5. 提醒規(guī)則
當股票價格低于或等于設定的買入價時，發(fā)送"建議買入"提醒
當股票價格高于或等于設定的賣出價時，發(fā)送"建議賣出"提醒
提醒會同時通過所有已啟用的通知方式發(fā)送
6. 注意事項
郵件通知需要使用郵箱的授權碼，而不是登錄密碼
QQ通知需要正確配置并運行 go-cqhttp
企業(yè)微信通知需要管理員在企業(yè)微信后臺創(chuàng)建應用
所有密碼和密鑰信息都保存在本地配置文件中，請注意信息安全
程序會自動保存所有設置到配置文件
7. 常見問題
郵件發(fā)送失敗：
- 檢查郵箱賬號和授權碼是否正確
- 確認SMTP服務器地址是否正確
QQ消息發(fā)送失敗：
- 確認go-cqhttp是否正常運行
- 檢查API地址是否可以訪問
企業(yè)微信消息發(fā)送失?。?/p>
- 確認企業(yè)ID和應用密鑰是否正確
- 檢查應用是否有發(fā)送消息權限
股票數(shù)據(jù)獲取失敗：
- 檢查網(wǎng)絡連接
- 確認股票代碼是否正確
到此這篇關于使用Python簡單編寫一個股票監(jiān)控系統(tǒng)的文章就介紹到這了,更多相關Python股票監(jiān)控系統(tǒng)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
本文主要介紹了numpy 產(chǎn)生隨機數(shù)的幾種方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-02-02
文章描述了在部署Python項目到生產(chǎn)環(huán)境時遇到的一個錯誤,即端口號無法正確轉(zhuǎn)換為整數(shù)值,該錯誤在測試環(huán)境和本地調(diào)試中沒有出現(xiàn),但在生產(chǎn)環(huán)境中才出現(xiàn),通過分析錯誤信息和代碼,作者發(fā)現(xiàn)問題出在URL解析過程中,特別是在處理包含特殊字符（如#號）的URL時2024-12-12
用python 獲取本機ip地址的多種方法，需要的朋友可以參考下2013-02-02
這篇文章主要介紹了python?中Mixin混入類的使用方法詳解，Mixin?混入也可以說是編程模式，并不是什么新的語法，用好混入類可以使自己的代碼結構清晰，功能明了，所以以后在設計類時要多考慮使用Mixin混入類的實現(xiàn)方式2022-07-07
這篇文章主要介紹了python爬蟲入門教程之點點美女圖片爬蟲代碼分享,本文以采集抓取點點網(wǎng)美女圖片為例,需要的朋友可以參考下2014-09-09
文章介紹了Python的pip包管理工具更換鏡像源的方法,包括清華大學、阿里云、中國科技大學和豆瓣等常用鏡像源,文章詳細說明了臨時使用鏡像源和永久配置鏡像源的方法,并附上了針對Conda的鏡像配置示例,最后,作者推薦使用國內(nèi)鏡像源以提高下載速度2025-02-02
基于Python中isfile函數(shù)和isdir函數(shù)使用詳解
今天小編就為大家分享一篇基于Python中isfile函數(shù)和isdir函數(shù)使用詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
這篇文章主要介紹了Python中捕獲鍵盤的方式，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-03-03
Pandas排序和分組排名(sort和rank)的實現(xiàn)
Pandas是Python中廣泛使用的數(shù)據(jù)處理庫,提供了豐富的功能來處理和分析數(shù)據(jù),本文主要介紹了Pandas排序和分組排名(sort和rank)的實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2024-07-07
set是一個無序且不重復的元素集合。這篇文章主要介紹了Python set常用操作函數(shù)集錦,需要的朋友可以參考下2017-11-11