471 lines
19 KiB
Python
471 lines
19 KiB
Python
import os
|
||
import sys
|
||
import subprocess
|
||
import tkinter as tk
|
||
from tkinter import ttk, messagebox, filedialog
|
||
import tushare as ts
|
||
import threading
|
||
import time
|
||
import datetime # 新增导入
|
||
import re
|
||
|
||
class StockMonitor:
|
||
def __init__(self, master):
|
||
self.master = master
|
||
self.master.title("股票价格监控系统")
|
||
ts.set_token('9343e641869058684afeadfcfe7fd6684160852e52e85332a7734c8d')
|
||
self.pro = ts.pro_api()
|
||
self.time_interval = 10 # 监控间隔时间（秒）
|
||
self.auto_push_var = tk.IntVar(value=0) # 自动下单开关
|
||
self.create_widgets()
|
||
self.monitor_active = True
|
||
self.start_monitor()
|
||
self.trigger_count = 0
|
||
self.previous_trading_status = None
|
||
self.start_status_update()
|
||
|
||
|
||
# 构建界面
|
||
def create_widgets(self):
|
||
# 创建Treeview容器框架
|
||
tree_frame = ttk.Frame(self.master)
|
||
tree_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)
|
||
# tree_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True, side=tk.LEFT)
|
||
# 设置字体和样式
|
||
style = ttk.Style()
|
||
style.configure('Larger.TButton', font=('微软雅黑', 12))
|
||
# style.configure('Log.TText', font=('微软雅黑', 12))
|
||
|
||
# # 创建日志输出框
|
||
log_frame = ttk.Frame(self.master)
|
||
log_frame.pack(side=tk.BOTTOM, fill=tk.BOTH, padx=5, pady=5)
|
||
|
||
# self.log_label = ttk.Label(log_frame, text="日志输出:", font=('微软雅黑', 12))
|
||
self.log_text = tk.Text(log_frame, wrap=tk.WORD, height=10, state=tk.DISABLED, font=('微软雅黑', 11), fg='blue')
|
||
scrollbar = ttk.Scrollbar(log_frame, command=self.log_text.yview)
|
||
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
|
||
self.log_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
||
self.log_text.config(yscrollcommand=scrollbar.set)
|
||
|
||
# 创建Treeview
|
||
self.tree = ttk.Treeview(
|
||
tree_frame,
|
||
columns=('code', 'name', 'open_zf', 'lt_pan', 'price', 'target', 'status', 'alert_time'),
|
||
show='headings',
|
||
height=20 # 设置默认显示行数
|
||
)
|
||
# 文字新增样式配置
|
||
self.tree.tag_configure('triggered', foreground='red', font=('Verdana', 12, 'bold')) # 修改样式配置
|
||
self.tree.tag_configure('wrong_buy', foreground='green', font=('Verdana', 11, 'bold')) # 修改样式配置
|
||
|
||
# 配置列属性
|
||
columns = [
|
||
('code', '代码', 80),
|
||
('name', '名称', 80),
|
||
('open_zf', '开盘涨幅', 80),
|
||
('lt_pan', '流通盘', 80),
|
||
('price', '当前价', 80),
|
||
('target', '目标价', 80),
|
||
('status', '状态', 60),
|
||
('alert_time', '预警时间', 200)
|
||
]
|
||
|
||
for col_id, col_text, col_width in columns:
|
||
self.tree.heading(col_id, text=col_text)
|
||
self.tree.column(col_id, width=col_width, anchor=tk.CENTER)
|
||
|
||
# 创建滚动条
|
||
vsb = ttk.Scrollbar(
|
||
tree_frame,
|
||
orient="vertical",
|
||
command=self.tree.yview
|
||
)
|
||
self.tree.configure(yscrollcommand=vsb.set)
|
||
|
||
# 布局组件
|
||
self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
||
vsb.pack(side=tk.RIGHT, fill=tk.Y)
|
||
|
||
# 创建控制按钮框架
|
||
control_frame = ttk.Frame(self.master)
|
||
control_frame.pack(pady=20)
|
||
|
||
# 文件选择按钮（原有代码保持不变）
|
||
self.btn_load = ttk.Button(
|
||
control_frame,
|
||
text="选择监控文件",
|
||
command=self.select_file,
|
||
style='Larger.TButton', # 应用样式
|
||
width=18 # 新增宽度设置
|
||
)
|
||
self.btn_load.pack(side=tk.LEFT, padx=(5, 15)) # 显式指定侧边排列
|
||
|
||
# 文件选择按钮（原有代码保持不变）
|
||
self.btn_moniter = ttk.Button(
|
||
control_frame,
|
||
text="开始监控",
|
||
command=self.select_file,
|
||
style='Larger.TButton', # 应用样式
|
||
width=15, # 新增宽度设置,
|
||
state=tk.DISABLED # 初始禁用
|
||
)
|
||
self.btn_moniter.pack(side=tk.LEFT, padx=(10, 15)) # 显式指定侧边排列
|
||
|
||
# 复选框-自动下单
|
||
self.auto_push_cb = ttk.Checkbutton(
|
||
control_frame,
|
||
text="自动下单",
|
||
variable=self.auto_push_var,
|
||
state=tk.DISABLED
|
||
)
|
||
self.auto_push_cb.pack(side=tk.LEFT, padx=5)
|
||
|
||
# 显示触发股票数
|
||
status_frame = ttk.Frame(self.master)
|
||
status_frame.pack(pady=5, fill=tk.X)
|
||
# 创建数量显示Label
|
||
self.trigger_count_label = ttk.Label(
|
||
status_frame,
|
||
text="已触发股票数量：0",
|
||
font=('微软雅黑', 14),
|
||
foreground='red'
|
||
)
|
||
self.trigger_count_label.pack(padx=10, pady=5, side=tk.LEFT, anchor=tk.W)
|
||
|
||
# 开盘时间显示
|
||
self.trading_status_label = ttk.Label(
|
||
status_frame,
|
||
text="当前状态：非开盘时间",
|
||
font=('微软雅黑', 14),
|
||
foreground='red'
|
||
)
|
||
self.trading_status_label.pack(padx=10, pady=5, side=tk.RIGHT, anchor=tk.E)
|
||
|
||
# 在control_frame内添加定时监控组件
|
||
timing_frame = ttk.Frame(control_frame)
|
||
timing_frame.pack(side=tk.LEFT, padx=(5, 15))
|
||
|
||
# 时间输入框
|
||
timing_label = ttk.Label(timing_frame, text="定时时间:")
|
||
timing_label.pack(side=tk.LEFT)
|
||
self.timing_time = tk.StringVar()
|
||
self.timing_entry = ttk.Entry(timing_frame, textvariable=self.timing_time, width=8)
|
||
self.timing_entry.pack(side=tk.LEFT)
|
||
|
||
# 定时监控复选框
|
||
self.timing_enabled = tk.BooleanVar()
|
||
self.timing_checkbox = ttk.Checkbutton(
|
||
control_frame,
|
||
text="定时监控",
|
||
variable=self.timing_enabled
|
||
)
|
||
self.timing_checkbox.pack(side=tk.LEFT, padx=5)
|
||
|
||
def append_log(self, message):
|
||
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 格式化到秒
|
||
self.log_text.config(state=tk.NORMAL)
|
||
self.log_text.insert(tk.END, f"[{current_time}] {message}\n")
|
||
self.log_text.see(tk.END)
|
||
self.log_text.config(state=tk.DISABLED)
|
||
|
||
# 选中文件读取
|
||
def select_file(self):
|
||
filepath = filedialog.askopenfilename(
|
||
title="选择监控列表文件",
|
||
filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")]
|
||
)
|
||
if filepath:
|
||
# 验证文件名是否为纯日期格式
|
||
filename = os.path.basename(filepath)
|
||
if not re.match(r'^\d{8}\.txt$', filename):
|
||
messagebox.showerror("错误", "文件名必须为8位数字日期格式（如：20250401.txt）")
|
||
return
|
||
|
||
trade_date = filename.split('.')[0]
|
||
self.load_stocks(filepath, trade_date)
|
||
|
||
def load_stocks(self, filepath, trade_date):
|
||
try:
|
||
self.tree.delete(*self.tree.get_children())
|
||
with open(filepath, 'r') as f:
|
||
# 读取标题行并验证列
|
||
headers = next(f).strip().split(',')
|
||
required_cols = ['条件选股', '代码']
|
||
for col in required_cols:
|
||
if col not in headers:
|
||
raise KeyError(f"缺少必要列：{col}")
|
||
# 获取列索引
|
||
name_col = headers.index('条件选股')
|
||
code_col = headers.index('代码')
|
||
# price_col = headers.index('成交')
|
||
|
||
# 批量获取历史数据
|
||
codes = []
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
parts = line.split(',')
|
||
if len(parts) <= max(name_col, code_col):
|
||
continue
|
||
code = parts[code_col].strip()
|
||
code = f"{code:0>6}" # 标准化为6位代码
|
||
codes.append(self.format_code(code))
|
||
|
||
# 获取历史收盘价
|
||
df = self.pro.daily(
|
||
ts_code=','.join(codes),
|
||
trade_date=trade_date
|
||
)
|
||
if df.empty:
|
||
raise ValueError("未找到历史数据，请确认：\n1.日期是否为交易日\n2.股票代码是否正确")
|
||
|
||
# 创建收盘价映射表
|
||
close_prices = {row['ts_code']: row['close'] for _, row in df.iterrows()}
|
||
|
||
# 回到文件开头并处理数据
|
||
f.seek(0)
|
||
next(f) # 跳过标题行
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
parts = line.split(',')
|
||
if len(parts) <= max(name_col, code_col):
|
||
raise ValueError(f"数据行字段不足：{line}")
|
||
name = parts[name_col].strip()
|
||
code = parts[code_col].strip()
|
||
# 确保代码保持字符串格式，保留前导零
|
||
code = f"{code:0>6}" # 确保代码是6位，不足前面补零
|
||
f_code = self.format_code(code)
|
||
|
||
# 获取收盘价
|
||
if f_code not in close_prices:
|
||
print(f"跳过 {code}，未找到收盘价数据")
|
||
continue
|
||
|
||
close_price = close_prices[f_code]
|
||
# 计算目标价
|
||
target = round(float(close_price) * 1.049, 2)
|
||
self.add_stock(code, name, target)
|
||
|
||
# 在加载成功后更新标题
|
||
formatted_date = f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:]}" # 格式化为YYYY-MM-DD
|
||
self.master.title(f"股票价格监控系统 - {formatted_date}") # 新增标题设置
|
||
self.append_log(f"加载文件成功：{filepath}")
|
||
except Exception as e:
|
||
self.append_log(f"加载文件失败：{str(e)}")
|
||
# messagebox.showerror("错误", f"加载文件失败：{str(e)}")
|
||
|
||
def add_stock(self, code, name, target):
|
||
"""添加股票到监控列表"""
|
||
codes = self.format_code(str(code))
|
||
self.tree.insert('', 'end', values=(codes, name, '-', '-', '-', target, '监控中', ''))
|
||
|
||
# 检查是否为交易日
|
||
def is_trading_time(self):
|
||
now = datetime.datetime.now()
|
||
current_time = now.time()
|
||
weekday = now.weekday() # 0-4是周一到周五
|
||
return (
|
||
weekday < 5 and # 仅工作日
|
||
(
|
||
(datetime.time(9, 30) <= current_time <= datetime.time(11, 30)) or
|
||
(datetime.time(13, 0) <= current_time <= datetime.time(15, 0))
|
||
)
|
||
)
|
||
|
||
def update_trading_status(self):
|
||
current_status = "开盘中" if self.is_trading_time() else "非开盘时间"
|
||
color = "green" if current_status == "开盘中" else "red"
|
||
# 状态变更时才记录日志
|
||
if current_status != self.previous_trading_status:
|
||
self.append_log(f"系统状态更新：{current_status}")
|
||
self.previous_trading_status = current_status
|
||
# 变更UI
|
||
self.master.after(0, lambda s=current_status, c=color:
|
||
self.trading_status_label.config(text=f"当前状态：{s}", foreground=c))
|
||
|
||
# 监控状态
|
||
def start_status_update(self):
|
||
def _loop():
|
||
while self.monitor_active:
|
||
self.update_trading_status()
|
||
time.sleep(1) # 每秒检查一次
|
||
|
||
threading.Thread(target=_loop, daemon=True).start()
|
||
|
||
def validate_time(self, time_str):
|
||
try:
|
||
datetime.datetime.strptime(time_str, "%H:%M")
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
def on_timing_change(self, *args):
|
||
if self.timing_enabled.get():
|
||
time_str = self.timing_time.get().strip()
|
||
if not self.validate_time(time_str):
|
||
messagebox.showerror("错误", "时间格式应为 HH:MM（如：09:30）")
|
||
self.timing_enabled.set(False)
|
||
return
|
||
self.start_monitor_at_time(time_str)
|
||
else:
|
||
self.monitor_active = True
|
||
self.start_monitor()
|
||
|
||
def start_monitor_at_time(self, target_time):
|
||
now = datetime.datetime.now()
|
||
target = datetime.datetime.strptime(target_time, "%H:%M")
|
||
target = now.replace(
|
||
hour=target.hour, minute=target.minute, second=0, microsecond=0
|
||
)
|
||
if target < now:
|
||
target += datetime.timedelta(days=1)
|
||
delta = target - now
|
||
threading.Timer(delta.total_seconds(), self.start_monitor).start()
|
||
|
||
# 格式化代码
|
||
def format_code(self, code):
|
||
# 为股票代码添加后缀
|
||
code = f"{code:0>6}" # 确保代码是6位，不足前面补零
|
||
if code.startswith(("6", "9")):
|
||
f_code = f"{code}.SH"
|
||
elif code.startswith(("0", "2", "3")):
|
||
f_code = f"{code}.SZ"
|
||
else:
|
||
print(f"未知的股票代码格式: {code}")
|
||
return None
|
||
return f_code
|
||
|
||
def play_audio(self, file_path):
|
||
try:
|
||
if not os.path.exists(file_path):
|
||
raise FileNotFoundError(f"文件路径不存在: {file_path}")
|
||
|
||
if sys.platform == "win32":
|
||
os.startfile(file_path)
|
||
elif sys.platform == "darwin":
|
||
subprocess.run(["open", file_path], check=True)
|
||
else:
|
||
subprocess.run(["xdg-open", file_path], check=True)
|
||
except FileNotFoundError as e:
|
||
print(f"错误: {e}")
|
||
except subprocess.CalledProcessError as e:
|
||
print(f"播放失败: {e}")
|
||
except Exception as e:
|
||
print(f"未知错误: {e}")
|
||
|
||
def update_prices(self):
|
||
codes = [self.tree.item(item)['values'][0] for item in self.tree.get_children()]
|
||
codes = [self.format_code(str(code)) for code in codes]
|
||
required_columns = ['TS_CODE', 'LOW', 'PRICE']
|
||
|
||
# 判断是否开盘时间
|
||
if not self.is_trading_time():
|
||
return
|
||
|
||
try:
|
||
batch_size = 40
|
||
for i in range(0, len(codes), batch_size):
|
||
batch = codes[i:i+batch_size]
|
||
self.append_log("行情实时数据更新...")
|
||
# print(f"更新批次：{batch}")
|
||
df = ts.realtime_quote(ts_code=','.join(batch))
|
||
for _, row in df.iterrows():
|
||
code = row['TS_CODE']
|
||
current_price = row['PRICE']
|
||
# 计算开盘涨幅 / (OPEN-PRE_CLOSE)/PRE_CLOSE*100
|
||
open_price = row['OPEN']
|
||
pre_close = row['PRE_CLOSE']
|
||
high_price = row['HIGH']
|
||
open_zf = round((float(open_price) - float(pre_close)) / float(pre_close) * 100, 2)
|
||
|
||
# 计算流通盘
|
||
lt_pan = row['VOLUME'] / 100000000 # 转换为亿
|
||
|
||
item = self.find_tree_item(code)
|
||
if item:
|
||
self.tree.set(item, 'price', f"{current_price:.2f}")
|
||
target = float(self.tree.item(item)['values'][5])
|
||
# 设置开盘涨幅/ (OPEN-PRE_CLOSE)*
|
||
self.tree.set(item, 'open_zf', f"{open_zf:.2f}%")
|
||
self.tree.set(item, 'lt_pan', f"{lt_pan:.2f}亿")
|
||
|
||
if current_price >= target and self.tree.item(item)['values'][6] != '已触发':
|
||
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
self.master.after(0, self.tree.set, item, 'alert_time', now)
|
||
|
||
# 自动下单
|
||
if self.auto_push_var.get():
|
||
self.autopush(item)
|
||
|
||
self.tree.set(item, 'status', '已触发')
|
||
self.tree.item(item, tags=('triggered',))
|
||
self.append_log(f"{code} 已触发！当前价：{current_price}")
|
||
|
||
# 使用系统命令播放音频（Windows）
|
||
self.play_audio("example.mp3")
|
||
|
||
# 错买的判断
|
||
if current_price < target and high_price >= target and self.tree.item(item)['values'][6] != '已触发':
|
||
self.tree.set(item, 'status', '错买')
|
||
self.tree.item(item, tags=('wrong_buy',))
|
||
|
||
except Exception as e:
|
||
print(f"更新失败：{str(e)}")
|
||
# 在遍历完所有股票后统计数量
|
||
triggered_count = 0
|
||
for item in self.tree.get_children():
|
||
if self.tree.item(item)['values'][6] == '已触发' or self.tree.item(item)['values'][6] == '错买':
|
||
triggered_count += 1
|
||
|
||
# 正确写法：
|
||
self.master.after(0, lambda: self.trigger_count_label.config(text=f"已触发股票数量：{triggered_count}"))
|
||
|
||
def find_tree_item(self, code):
|
||
for item in self.tree.get_children():
|
||
if self.tree.item(item)['values'][0] == code:
|
||
return item
|
||
return None
|
||
|
||
def start_monitor(self):
|
||
if self.timing_enabled.get():
|
||
self.monitor_active = True
|
||
else:
|
||
self.monitor_active = True # 非定时模式直接启动
|
||
|
||
def monitor_loop():
|
||
while self.monitor_active:
|
||
self.update_prices()
|
||
time.sleep(self.time_interval)
|
||
threading.Thread(target=monitor_loop, daemon=True).start()
|
||
|
||
def on_closing(self):
|
||
self.monitor_active = False
|
||
self.master.destroy()
|
||
|
||
# 扩展自动下单----------------------------------------------------------
|
||
def autopush(self, item):
|
||
code = self.tree.item(item)['values'][0]
|
||
current_price = float(self.tree.item(item)['values'][2])
|
||
print(code)
|
||
confirm = messagebox.askyesno(
|
||
"确认下单",
|
||
f"确认为 {code} 在当前价 {current_price} 自动买入吗？"
|
||
)
|
||
if confirm:
|
||
# 调用真实交易接口示例：
|
||
# result = trade_api.sell(code, current_price)
|
||
# ... 处理下单结果 ...
|
||
messagebox.showinfo("下单成功", "已执行自动买入操作")
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
root = tk.Tk()
|
||
app = StockMonitor(root)
|
||
root.protocol("WM_DELETE_WINDOW", app.on_closing)
|
||
root.mainloop()
|
||
|