# 悬浮框 import json import logging import os import sys import threading import time import requests import schedule import wx from utils import tool import kpl_api def show_warning(content, click): toastone = wx.MessageDialog(None, content, "提示", wx.YES_DEFAULT | wx.ICON_WARNING) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 if click is not None: click() toastone.Destroy() def show_info(content, click): toastone = wx.MessageDialog(None, content, "提示", wx.YES_DEFAULT | wx.ICON_INFORMATION) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 click() toastone.Destroy() class MyLogHandler(logging.Handler): def __init__(self, obj): logging.Handler.__init__(self) self.setLevel(logging.DEBUG) self.Object = obj def emit(self, record): if record.levelno: if len(self.Object.GetValue()) > 1024: self.Object.SetValue('') tstr = time.strftime('%Y-%m-%d_%H:%M:%S') self.Object.AppendText("[%s][%s] %s\n" % (tstr, record.levelname, record.getMessage())) class KPLDataCaptureFrame(wx.Frame): def __create_item(self, name, checked=True, trade_time=True, circulate=True): btn = wx.Button(self, label=f"{'开启' if circulate else ''}{name}", size=(120, 28)) check = wx.CheckBox(self, size=(-1, -1)) if checked: check.SetValue(True) else: check.SetValue(False) label = wx.StaticText(self, label=f"{'是否交易时间采集' if trade_time else ''}") bs1 = wx.BoxSizer(wx.HORIZONTAL) bs1.Add(label) bs1.Add(check, 0, wx.LEFT, 10) bs2 = wx.BoxSizer(wx.VERTICAL) bs2.Add(bs1) bs2.Add(btn) if not trade_time: check.Hide() return bs2, btn, check # 初始化采集线程 def __init_capture_threads(self): def create_thread(target): t1 = threading.Thread(target=target) t1.setDaemon(True) t1.start() def create_circulate_task(target): while True: try: target() except Exception as e: print(str(e)) time.sleep(3) create_thread(lambda: create_circulate_task(self.__exec_limit_up)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_open_limit_up)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_limit_down)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_ever_limit_down)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_fengkou)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_fengkou_best)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_fengxiang)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_industry_rank)) time.sleep(0.5) create_thread(lambda: create_circulate_task(self.__exec_jingxuan_rank)) def __init__(self): self.__init_data() wx.Frame.__init__(self, None, -1, "开盘啦数据采集客户端", style=wx.CAPTION ^ wx.MINIMIZE_BOX ^ wx.CLOSE_BOX ^ wx.STAY_ON_TOP, size=(800, 400)) self.SetBackgroundColour(wx.Colour(224, 224, 224)) self.Bind(wx.EVT_CLOSE, self.OnExit) boxsier = wx.FlexGridSizer(5, 2, 10, 20) self.item_bidding = self.__create_item("竞价采集", trade_time=False, circulate=False) boxsier.Add(self.item_bidding[0], 0, wx.TOP, 5) self.items = [] names = ["涨停采集", "炸板采集", "跌停采集", "曾跌停采集", "市场风口采集", "最强风口采集", "风向标采集", "行业涨幅", "精选流入"] for name in names: self.items.append(self.__create_item(name)) boxsier.Add(self.items[-1][0], 0, wx.TOP, 5) # 绑定事件 self.item_bidding[1].Bind(wx.EVT_BUTTON, self.__capture_bidding) for i in range(0, len(names)): self.items[i][1].Bind(wx.EVT_BUTTON, lambda e, x=i: self.__capture_btn_click(x, e)) self.items[i][2].Bind(wx.EVT_CHECKBOX, lambda e, x=i: self.__capture_btn_check(x, e)) root_boxsier = wx.BoxSizer(wx.HORIZONTAL) left_boxsier = wx.BoxSizer(wx.VERTICAL) left_boxsier.Add(boxsier) whole_boxsier = wx.BoxSizer(wx.HORIZONTAL) self.start_all_btn = wx.Button(self, label="开启全部任务", size=(120, 28)) self.start_all_btn.SetForegroundColour("#008040") self.start_all_btn.Bind(wx.EVT_BUTTON, self.__start_all_task) self.stop_all_btn = wx.Button(self, label="关闭全部任务", size=(120, 28)) self.stop_all_btn.SetForegroundColour("#FF0000") self.stop_all_btn.Bind(wx.EVT_BUTTON, self.__stop_all_task) whole_boxsier.Add(self.stop_all_btn) whole_boxsier.Add(self.start_all_btn, 1, wx.LEFT, 25) left_boxsier.Add(whole_boxsier, 1, wx.TOP, 30) root_boxsier.Add(left_boxsier, 1, wx.LEFT, 10) self.text_log = wx.TextCtrl(parent=self, size=(500, self.Size[1]), style=wx.TE_MULTILINE) self.text_log.SetEditable(False) self.text_log.SetBackgroundColour("#EEE") self.text_log.SetForegroundColour("#333") handler = MyLogHandler(self.text_log) self.logger = logging.getLogger(__name__) self.logger.addHandler(handler) self.logger.setLevel(logging.DEBUG) root_boxsier.Add(self.text_log, 1, wx.LEFT, 10) self.SetSizer(root_boxsier) self.timer = wx.Timer(self) # 创建定时器 # self.Bind(wx.EVT_TIMER, self.clear_msg, self.timer) self.logger.info("初始化成功") self.__init_capture_threads() # 开启定时器，在9点26时读取竞价数据 def schedule_fun(): self.__capture_bidding(None) schedule.every().day.at("09:26:00").do(schedule_fun) #lambda: self.__capture_bidding(None) t1 = threading.Thread(target=self.run_schedule) t1.setDaemon(True) t1.start() # 自动开启采集 self.__start_all_task(None) def run_schedule(self): while True: schedule.run_pending() def __init_data(self): self.capture_status = {} self.keys = ["limit_up", "open_limit_up", "limit_down", "ever_limit_down", "feng_kou", "best_feng_kou", "feng_xiang", "industry_rank", "jingxuan_rank"] for key in self.keys: self.capture_status[key] = [False, True] def __set_btn_text(self, controls): text = controls[1].GetLabelText() if text.find("开启") == 0: text = text.replace("开启", "关闭") else: text = text.replace("关闭", "开启") controls[1].SetLabelText(text) def __capture_bidding(self, event): results = kpl_api.daBanList(kpl_api.DABAN_TYPE_BIDDING) result = json.loads(results) self.logger.info(f"获取到竞价数据：{len(result['list'])}") self.__upload_data("biddings", result) # show_info("采集成功", None) def __capture_btn_click(self, key_index, event): self.__set_btn_text(self.items[key_index]) self.capture_status[self.keys[key_index]][0] = not self.capture_status[self.keys[key_index]][0] def __capture_btn_check(self, key_index, event): self.capture_status[self.keys[key_index]][1] = not self.capture_status[self.keys[key_index]][1] def __can_capture(self, key_index): if not self.capture_status[self.keys[key_index]][0]: return False if self.capture_status[self.keys[key_index]][1] and not tool.is_trade_time(): return False return True def __exec_limit_up(self): if not self.__can_capture(0): return try: results = kpl_api.daBanList(kpl_api.DABAN_TYPE_LIMIT_UP) result = json.loads(results) self.logger.info(f"涨停代码数量：{len(result['list'])}") self.__upload_data(self.keys[0], result) except: pass def __exec_open_limit_up(self): if not self.__can_capture(1): return try: results = kpl_api.daBanList(kpl_api.DABAN_TYPE_OPEN_LIMIT_UP) result = json.loads(results) self.logger.info(f"炸板代码数量：{len(result['list'])}") self.__upload_data(self.keys[1], result) except: pass def __exec_limit_down(self): if not self.__can_capture(2): return try: results = kpl_api.daBanList(kpl_api.DABAN_TYPE_LIMIT_DOWN) result = json.loads(results) self.logger.info(f"跌停代码数量：{len(result['list'])}") self.__upload_data(self.keys[2], result) except: pass def __exec_ever_limit_down(self): if not self.__can_capture(3): return try: results = kpl_api.daBanList(kpl_api.DABAN_TYPE_EVER_LIMIT_DOWN) result = json.loads(results) self.logger.info(f"曾经跌停代码数量：{len(result['list'])}") self.__upload_data(self.keys[3], result) except: pass def __exec_fengkou(self): if not self.__can_capture(4): return try: results = kpl_api.getFengKouList(tool.get_now_date_str()) result = json.loads(results) self.logger.info(f"市场风口代码数量：{len(result['List'])}") self.__upload_data(self.keys[4], result) except: pass def __exec_fengkou_best(self): try: if not self.__can_capture(5): return if not os.path.exists("D:/kpl/GetFengKListBest_0_0.log"): return with open("D:/kpl/GetFengKListBest_0_0.log", mode='r', encoding="utf-16") as f: lines = f.readlines() if lines: data = json.loads(lines[0]) self.logger.info(f"最强风口代码数量：{data['Count']}") self.__upload_data(self.keys[5], data) except: pass def __exec_fengxiang(self): if not self.__can_capture(6): return self.logger.info(f"开始风向标采集") try: results = kpl_api.getFengXiangBiao() result = json.loads(results) self.logger.info(f"风向标代码数量：{len(result['list'])}") self.__upload_data(self.keys[6], result) except Exception as e: self.logger.error(str(e)) def __exec_industry_rank(self): if not self.__can_capture(7): return try: results = kpl_api.getMarketIndustryRealRankingInfo() result = json.loads(results) self.logger.info(f"行业涨幅排行代码数量：{len(result['list'])}") self.__upload_data(self.keys[7], result) except: pass def __exec_jingxuan_rank(self): if not self.__can_capture(8): return try: results = kpl_api.getMarketJingXuanRealRankingInfo() result = json.loads(results) self.logger.info(f"精选流入排行代码数量：{len(result['list'])}") self.__upload_data(self.keys[8], result) except: pass # 开始所有的任务 def __start_all_task(self, event): for i in range(0, len(self.items)): # 是否已经开启 if not self.capture_status[self.keys[i]][0]: self.__capture_btn_click(i, None) # 开始所有的任务 def __stop_all_task(self, event): for i in range(0, len(self.items)): # 是否已经开启 if self.capture_status[self.keys[i]][0]: self.__capture_btn_click(i, None) def __upload_data(self, type, datas): root_data = { "type": type, "data": datas } requests.post("http://192.168.3.252:9004/upload_kpl_data", json.dumps(root_data)) def OnExit(self, e): sys.exit(0) class mainApp(wx.App): def OnInit(self): self.SetAppName("开盘啦") # 传递代码设置成功的回调 self.kpl = KPLDataCaptureFrame() self.kpl.Show() return True # cd D:\workspace\GP\trade_desk # D:\workspace\GP\trade_desk\dist\env\pk_env\Scripts\pyinstaller.exe kpl/gui.spec # if __name__ == "__main__": app = mainApp(redirect=True, filename="logs/output.log") app.MainLoop()