import time import pyautogui import win32gui import win32con import tkinter as tk from tkinter import filedialog from logger_utils import setup_logger # 导入日志配置函数 import datetime from logger_utils import log_info, log_warning, log_error, log_trigger from order_executor import OrderExecutor # 导入 OrderExecutor 类 下单类 import os from threading import Thread import pygetwindow as gw class Autotrading: def __init__(self, master): self.master = master master.title("监控自动交易工具") # 注册日志回调 self.order_executor = OrderExecutor() self.default_buy_price = 0.00 self.default_sell_price = 0.00 self.default_buy_volume = 100 self.default_sell_volume = 100 self.stock_code = "" self.monitoring = False self.monitor_thread = None self.last_file_size = 0 self.monitored_file = "" self.monitoring = False # 添加监控状态标志 self.hwnd = None # 交易窗口的句柄 self.target_window_title = "东方财富终端" # 交易窗口的 # 创建界面组件 self.label = tk.Label(master, text="选择类型:", anchor="w") self.label.pack(anchor="w") # 添加单选按钮组 self.mode_var = tk.StringVar(value="方式1") # 默认选择方式1 self.mode1 = tk.Radiobutton(master, text="监控文件", variable=self.mode_var, font=('微软雅黑', 11), value="方式1") self.mode2 = tk.Radiobutton(master, text="自动推送", variable=self.mode_var, font=('微软雅黑', 11), value="方式2") self.mode1.pack(anchor="w") self.mode2.pack(anchor="w") # 添加ST排除复选框 self.exclude_st_var = tk.BooleanVar(value=True) self.exclude_st_check = tk.Checkbutton( master, text="排除ST个股", variable=self.exclude_st_var, font=('微软雅黑', 11) ) self.exclude_st_check.pack(anchor="w") # 在 __init__ 中添加 self.use_executor_var = tk.BooleanVar(value=True) # 默认启用 executor self.executor_check = tk.Checkbutton( master, text="使用专业下单模块 (OrderExecutor)", variable=self.use_executor_var, font=('微软雅黑', 11) ) self.executor_check.pack(anchor="w") # 新增编辑框 self.entry_label = tk.Label(master, text="输入策略名:", anchor="w") self.entry_label.pack(anchor="w") self.stock_entry = tk.Entry(master, width=25) self.stock_entry.insert(0, "凤随☆") # 设置默认内容 self.stock_entry.pack(anchor="w", padx=5) # 添加文件监控按钮 self.monitor_button = tk.Button( master, text="监控文件", command=self.choose_model, width=15, font=('微软雅黑', 12), # 设置字体和大小 fg='red' # 字体颜色 ) self.monitor_button.pack() # 文件选择按钮 self.file_button = tk.Button( master, text="选择文件", command=self.select_file, font=('微软雅黑', 12), # 设置字体和大小 width=12 ) self.file_button.pack() # 文件显示标签 self.file_label = tk.Label(master, text="未选择文件", font=('微软雅黑', 11)) self.file_label.pack() # 新增窗口状态标签 self.window_status = tk.Label(master, text="窗口状态: 未检测到", fg="red", font=('微软雅黑', 11)) self.window_status.pack() # 添加日志窗口 self.log_frame = tk.Frame(master) self.log_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 日志文本框，设置字体大小 self.log_text = tk.Text( self.log_frame, height=20, state='disabled', font=('微软雅黑', 12) # 设置字体和大小，可按需调整 ) self.log_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) # 滚动条 self.log_scroll = tk.Scrollbar(self.log_frame, command=self.log_text.yview) self.log_scroll.pack(side=tk.RIGHT, fill=tk.Y) self.log_text.config(yscrollcommand=self.log_scroll.set) # === 设置日志颜色样式（放在这里）=== self.log_text.tag_configure('error', foreground='red') self.log_text.tag_configure('loading', foreground='orange') self.log_text.tag_configure('trigger', foreground='green') self.log_text.tag_configure('default', foreground='blue') self.log_text.tag_configure('info', foreground='blue') self.log_text.tag_configure('warning', foreground='orange') # 注册日志回调 setup_logger(self.gui_log) def gui_log(self, message, level='default'): """日志代理函数，用于将日志信息更新到 GUI""" current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") full_message = f" {message}\n" self.log_text.config(state=tk.NORMAL) self.log_text.insert(tk.END, full_message, level) app.log_text.see(tk.END) app.log_text.config(state=tk.DISABLED) def get_entry_content(self): """获取输入框内容""" content = self.stock_entry.get() if not content.strip(): # 如果内容为空或只有空格 log_error("警告：策略名不能为空！") return None return content.strip() def choose_model(self): """根据单选按钮选择功能""" selected_mode = self.mode_var.get() if selected_mode == "方式1": self.start_file_monitoring() elif selected_mode == "方式2": self.start_auto_push() # 获取窗口句柄 def get_window_handle(self, window_title): """获取指定标题的窗口句柄""" hwnd = win32gui.FindWindow(None, window_title) if not hwnd: self.window_status.config(text=f"窗口状态: {window_title} 未打开", fg="red") raise Exception(f"未找到标题为 '{window_title}' 的窗口") self.window_status.config(text=f"窗口状态: {window_title}", fg="green") return hwnd def get_window_content(self, hwnd): """获取窗口可视内容（截图）""" # 将窗口置顶 win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(hwnd) # 获取窗口位置和尺寸 left, top, right, bottom = win32gui.GetWindowRect(hwnd) width = right - left height = bottom - top # 截图并保存 screenshot = pyautogui.screenshot(region=(left, top, width, height)) return screenshot def capture_window(self): """截图功能""" try: hwnd = self.get_window_handle(self.target_window_title) content = self.get_window_content(hwnd) content.save(f"{self.target_window_title}.png") self.file_label.config(text=f"截图已保存为 {self.target_window_title}.png") except Exception as e: self.file_label.config(text=f"错误: {str(e)}") def start_auto_push(self): """开始自动推送功能""" self.place_order("002183", "买入", 100, 2.00, ) def start_file_monitoring(self): if not self.monitored_file: log_warning("请先选择要监控的文件") return if self.monitoring: self.monitoring = False self.monitor_button.config(text="监控文件") log_info(f"已停止监控文件: {self.monitored_file}") else: try: self.get_window_handle(self.target_window_title) self.monitoring = True self.monitor_button.config(text="停止监控") self.last_file_size = os.path.getsize(self.monitored_file) self.monitor_thread = Thread(target=self.monitor_file_changes) self.monitor_thread.daemon = True self.monitor_thread.start() log_info(f"开始监控文件: {self.monitored_file}") except Exception as e: log_error(f"未找到 {self.target_window_title} 窗口，请先打开该窗口再开始监控。错误信息: {str(e)}") def monitor_file_changes(self): """监控文件变化的后台线程，处理ANSI格式文件""" last_position = os.path.getsize(self.monitored_file) if os.path.exists(self.monitored_file) else 0 last_warning_type = "" # 记录上一条预警类型 # 获取策略名作为预警名 warning_name = self.get_entry_content() log_error(f"当前策略：{warning_name}") while self.monitoring: try: current_size = os.path.getsize(self.monitored_file) if current_size > last_position: # 比较当前文件大小和上次记录的位置 with open(self.monitored_file, 'r', encoding='mbcs') as f: f.seek(last_position) new_lines = f.readlines() last_position = f.tell() if new_lines: for line in new_lines: # 解析每行数据 parts = line.strip().split('\t') if len(parts) >= 6: code = parts[0] # 代码 name = parts[1] # 个股名称 t_time = parts[2] # 预警时间 price = parts[3] # 预警价格 increase = parts[4] # 预警涨幅 code_num = parts[5] # 编码 warning_type = parts[6] if len(parts) > 6 else "" # 预警类型 output = f"预警类型: {warning_type}\n代码: {code}\n个股名称: {name}\n预警时间: {t_time}\n预警价格: {price}\n预警涨幅: {increase}\n编码: {code_num}\n" # 检查是否排除ST个股 if self.exclude_st_var.get() and ('ST' in name or '*ST' in name): log_error(f"排除ST个股: {name}({code})") continue # 判断是否匹配当前策略 if warning_type == warning_name: log_trigger(f"策略触发----{warning_name}\n{output}") # 调用下单函数（示例） # self.place_order(code, name, price, self.default_buy_volume, 'buy') else: log_error(f"非指定策略-{warning_type}") log_info("--------------------------------") except Exception as e: log_error(f"监控出错: {str(e)}") time.sleep(1) time.sleep(0.5) def select_file(self): """文件选择功能""" if self.monitoring: log_warning("请先停止监控再选择文件") return file_path = filedialog.askopenfilename() if file_path: self.file_label.config(text=f"已选择文件: {file_path}") self.monitored_file = file_path log_info(f"已选择监控文件: {file_path}") def get_min_interval(self): """检测系统最小可操作间隔""" # 测试几次点击操作的最小间隔 test_times = 3 start_time = time.time() for _ in range(test_times): pyautogui.click(100, 100) # 在屏幕角落测试点击 end_time = time.time() # 计算平均间隔时间，并乘以安全系数(1.5) return (end_time - start_time) / test_times * 1.5 def place_order(self, code, code_name, price, volume, order_type): """ 下单函数，优先使用 OrderExecutor 下单 :param code: 股票代码 :param code_name: 股票名称 :param price: 下单价格 :param volume: 下单数量 :param order_type: 下单类型，如 'buy' 或 'sell' """ try: pure_code = code[-6:] # 假设输入为 "SH600000" 或 "SZ000001" 等格式 auto_push = 1 # 表示是否自动下单，这里默认开启 log_trigger(f"开始 {order_type} 下单，代码: {code}，{code_name}，价格: {price}，数量: {volume}") self.order_executor.place_order(pure_code, auto_push) log_trigger(f"{order_type} 下单成功，代码: {code}，{code_name}，价格: {price}，数量: {volume}") except Exception as e: log_error(f"{order_type} 下单失败，代码: {code}，错误信息: {str(e)}") if __name__ == "__main__": root = tk.Tk() app = Autotrading(root) root.mainloop()