import datetime import json import logging import random import time as t import big_money_num_manager import code_data_util import constant import global_util import gpcode_manager import industry_codes_sort import l2_data_log import l2_data_manager import l2_data_util import l2_trade_factor import l2_trade_test import l2_trade_util import limit_up_time_manager import redis_manager import ths_industry_util import tool import trade_manager import trade_queue_manager import trade_data_manager from l2_data_manager import L2DataException, TradePointManager, local_today_datas, L2DataUtil, load_l2_data, \ local_today_num_operate_map from log import logger_l2_trade, logger_l2_trade_cancel, logger_l2_trade_buy, logger_l2_process, logger_buy_1_volumn, \ logger_l2_error # TODO l2数据管理 from trade_data_manager import CodeActualPriceProcessor class L2DataManager: # 格式化数据 def format_data(self, datas): format_datas = [] for data in datas: format_datas.append({"val": data, "re": 1}) return format_datas # 获取新增数据 def get_add_datas(self, format_datas): pass # 从数据库加载数据 def load_data(self, code=None, force=False): pass # 保存数据 def save_datas(self, add_datas, datas): pass # m值大单处理 class L2BigNumForMProcessor: def __init__(self): self._redis_manager = redis_manager.RedisManager(1) def __get_redis(self): return self._redis_manager.getRedis() # 保存计算开始位置 def set_begin_pos(self, code, index): if self.__get_begin_pos(code) is None: # 保存位置 key = "m_big_money_begin-{}".format(code) self.__get_redis().setex(key, tool.get_expire(), index) # 获取计算开始位置 def __get_begin_pos(self, code): key = "m_big_money_begin-{}".format(code) val = self.__get_redis().get(key) if val is None: return None return int(val) # 清除已经处理的数据 def clear_processed_end_index(self, code): key = "m_big_money_process_index-{}".format(code) self.__get_redis().delete(key) # 添加已经处理过的单 def __set_processed_end_index(self, code, index): key = "m_big_money_process_index-{}".format(code) self.__get_redis().setex(key, tool.get_expire(), index) # 是否已经处理过 def __get_processed_end_index(self, code): key = "m_big_money_process_index-{}".format(code) val = self.__get_redis().get(key) if val is None: return None return int(val) # 处理大单 def process(self, code, start_index, end_index, limit_up_price): begin_pos = self.__get_begin_pos(code) if begin_pos is None: # 没有获取到开始买入信号 return # 上次处理到的坐标 processed_index = self.__get_processed_end_index(code) if processed_index is None: processed_index = 0 if processed_index >= end_index: return start_time = round(t.time() * 1000) total_datas = local_today_datas[code] num_splites = [round(5000 / limit_up_price), round(10000 / limit_up_price), round(20000 / limit_up_price), round(30000 / limit_up_price)] total_num = 0 for i in range(max(start_index, processed_index), end_index + 1): data = total_datas[i] if not L2DataUtil.is_limit_up_price_buy_cancel(data["val"]) and not L2DataUtil.is_limit_up_price_buy( data["val"]): continue # 如果是涨停买撤信号需要看数据位置是否比开始处理时间早 if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]): # 获取买入信号 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(total_datas[i], local_today_num_operate_map.get(code)) if buy_index is not None and buy_index < begin_pos: continue # 计算成交金额 num = int(data["val"]["num"]) temp = 0 if num < num_splites[0]: pass elif num < num_splites[1]: temp = 1 elif num < num_splites[2]: temp = round(4 / 3, 3) elif num < num_splites[3]: temp = 2 else: temp = 4 count = int(temp * data["re"] * 1000) if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]): count = 0 - count total_num += count self.__set_processed_end_index(code, end_index) big_money_num_manager.add_num(code, total_num) print("m值大单计算范围：{}-{} 时间：{}".format(max(start_index, processed_index), end_index, round(t.time() * 1000) - start_time)) class L2TradeDataProcessor: unreal_buy_dict = {} random_key = {} l2BigNumForMProcessor = L2BigNumForMProcessor() __codeActualPriceProcessor = CodeActualPriceProcessor() buy1PriceManager = trade_queue_manager.Buy1PriceManager() __ths_l2_trade_queue_manager = trade_queue_manager.thsl2tradequeuemanager() @classmethod def debug(cls, code, content, *args): logger_l2_trade.debug(("thread-id={} code={} ".format(cls.random_key[code], code) + content).format(*args)) @classmethod def cancel_debug(cls, code, content, *args): logger_l2_trade_cancel.debug( ("thread-id={} code={} ".format(cls.random_key[code], code) + content).format(*args)) @classmethod def buy_debug(cls, code, content, *args): logger_l2_trade_buy.debug( ("thread-id={} code={} ".format(cls.random_key[code], code) + content).format(*args)) @classmethod # 数据处理入口 # datas: 本次截图数据 # capture_timestamp:截图时间戳 def process(cls, code, datas, capture_timestamp): cls.random_key[code] = random.randint(0, 100000) __start_time = round(t.time() * 1000) try: if len(datas) > 0: # 判断价格区间是否正确 if not code_data_util.is_same_code_with_price(code, float(datas[0]["val"]["price"])): raise L2DataException(L2DataException.CODE_PRICE_ERROR, "股价不匹配 code-{} price-{}".format(code, datas[0]["val"]["price"])) # 加载历史数据 l2_data_manager.load_l2_data(code) # 纠正数据 datas = l2_data_manager.L2DataUtil.correct_data(code, datas) _start_index = 0 if local_today_datas.get(code) is not None and len( local_today_datas[code]) > 0: _start_index = local_today_datas[code][-1]["index"] + 1 add_datas = l2_data_manager.L2DataUtil.get_add_data(code, datas, _start_index) # -------------数据增量处理------------ try: cls.process_add_datas(code, add_datas, capture_timestamp, __start_time) finally: # 保存数据 l2_data_manager.save_l2_data(code, datas, add_datas) __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "保存数据时间（{}）".format(len(add_datas))) finally: if code in cls.unreal_buy_dict: cls.unreal_buy_dict.pop(code) @classmethod def process_add_datas(cls, code, add_datas, capture_timestamp, __start_time): if len(add_datas) > 0: now_time_str = tool.get_now_time_str() # 拼接数据 local_today_datas[code].extend(add_datas) l2_data_util.load_num_operate_map(l2_data_manager.local_today_num_operate_map, code, add_datas) # ---------- 判断是否需要计算大单 ----------- try: average_need, buy_single_index, buy_exec_index = AverageBigNumComputer.is_need_compute_average( code, local_today_datas[code][-1]) # 计算平均大单 if average_need: end_index = local_today_datas[code][-1]["index"] if len(add_datas) > 0: end_index = add_datas[-1]["index"] AverageBigNumComputer.compute_average_big_num(code, buy_single_index, buy_single_index, end_index) except Exception as e: logging.exception(e) try: average_need, buy_single_index, buy_exec_index = SecondAverageBigNumComputer.is_need_compute_average( code, local_today_datas[code][-1]) # 计算平均大单 if average_need: end_index = local_today_datas[code][-1]["index"] if len(add_datas) > 0: end_index = add_datas[-1]["index"] SecondAverageBigNumComputer.compute_average_big_num(code, buy_single_index, buy_single_index, end_index) except Exception as e: logging.exception(e) # 第1条数据是否为09:30:00 if add_datas[0]["val"]["time"] == "09:30:00": if global_util.cuurent_prices.get(code): price_data = global_util.cuurent_prices.get(code) if price_data[1]: # 当前涨停价，设置涨停时间 logger_l2_process.info("开盘涨停：{}", code) # 保存涨停时间 limit_up_time_manager.save_limit_up_time(code, "09:30:00") total_datas = local_today_datas[code] __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "l2数据预处理时间") if len(add_datas) > 0: latest_time = add_datas[len(add_datas) - 1]["val"]["time"] # 时间差不能太大才能处理 if l2_data_manager.L2DataUtil.is_same_time(now_time_str, latest_time) and not l2_trade_util.is_in_forbidden_trade_codes( code): # 判断是否已经挂单 state = trade_manager.get_trade_state(code) start_index = len(total_datas) - len(add_datas) end_index = len(total_datas) - 1 if state == trade_manager.TRADE_STATE_BUY_DELEGATED or state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or state == trade_manager.TRADE_STATE_BUY_SUCCESS: # 已挂单 cls.__process_order(code, start_index, end_index, capture_timestamp) else: # 未挂单 cls.__process_not_order(code, start_index, end_index, capture_timestamp) logger_l2_process.info("code:{} 处理数据范围: {}-{} 处理时间:{} 截图时间戳：{}", code, add_datas[0]["index"], add_datas[-1]["index"], round(t.time() * 1000) - __start_time, capture_timestamp) __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "l2数据处理时间") # 处理未挂单 @classmethod def __process_not_order(cls, code, start_index, end_index, capture_time): __start_time = round(t.time() * 1000) # 获取阈值 threshold_money, msg = cls.__get_threshmoney(code) if round(t.time() * 1000) - __start_time > 10: __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "获取m值数据耗时") cls.__start_compute_buy(code, start_index, end_index, threshold_money, capture_time) # 测试专用 @classmethod def process_order(cls, code, start_index, end_index, capture_time, new_add=True): cls.__process_order(code, start_index, end_index, capture_time, new_add) # 处理已挂单 @classmethod def __process_order(cls, code, start_index, end_index, capture_time, new_add=True): if start_index < 0: start_index = 0 if end_index < start_index: return # 获取买入信号起始点 buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set = cls.__get_order_begin_pos( code) # 撤单计算,只看买1 cancel_data, cancel_msg = L2LimitUpMoneyStatisticUtil.process_data(code, start_index, end_index, buy_single_index, buy_exec_index) # 撤单计算，看秒级大单撤单 try: b_need_cancel, b_cancel_data = SecondAverageBigNumComputer.need_cancel(code, buy_single_index, buy_exec_index, start_index, end_index) if b_need_cancel and not cancel_data: cancel_data = b_cancel_data cancel_msg = "申报时间截至大单撤销比例触发阈值" except Exception as e: logging.exception(e) # 撤单计算，看分钟级大单撤单 try: b_need_cancel, b_cancel_data = AverageBigNumComputer.need_cancel(code, buy_single_index, buy_exec_index, start_index, end_index) if b_need_cancel and not cancel_data: cancel_data = b_cancel_data cancel_msg = "1分钟内大单撤销比例触发阈值" except Exception as e: logging.exception(e) if not cancel_data: # 统计板上卖 try: cancel_data, cancel_msg = L2LimitUpSellStatisticUtil.process(code, start_index, end_index, buy_exec_index) except Exception as e: logging.exception(e) # 计算m值大单 cls.l2BigNumForMProcessor.process(code, max(buy_single_index, start_index), end_index, gpcode_manager.get_limit_up_price(code)) if cancel_data: if cancel_data["index"] == 175: print("进入调试") cls.debug(code, "触发撤单，撤单位置：{} ，撤单原因：{}", cancel_data["index"], cancel_msg) # 撤单 if cls.cancel_buy(code, cancel_msg): # 撤单成功，继续计算下单 cls.__process_not_order(code, cancel_data["index"] + 1, end_index, capture_time) else: # 撤单尚未成功 pass else: # 如果有虚拟下单需要真实下单 unreal_buy_info = cls.unreal_buy_dict.get(code) if unreal_buy_info is not None: cls.debug(code, "有虚拟下单，无买撤信号，开始执行买入，执行位置：{},截图时间：{}", unreal_buy_info[0], capture_time) # unreal_buy_info 的内容格式为：(触法买操作下标,截图时间) # 真实下单 cls.__buy(code, unreal_buy_info[1], local_today_datas[code][unreal_buy_info[0]], unreal_buy_info[0]) # 判断是否需要计算长大单的信息 try: LongAverageBigNumComputer.compute_average_big_num(code, buy_single_index, buy_exec_index) except Exception as e: logging.exception(e) @classmethod def __buy(cls, code, capture_timestamp, last_data, last_data_index): can, reason = cls.__can_buy(code) # 删除虚拟下单 if code in cls.unreal_buy_dict: cls.unreal_buy_dict.pop(code) if not can: cls.debug(code, "不可以下单，原因：{}", reason) if not reason.startswith("买1价不为涨停价"): # 中断买入 trade_manager.break_buy(code, reason) return else: cls.debug(code, "可以下单，原因：{}", reason) try: cls.debug(code, "开始执行买入") trade_manager.start_buy(code, capture_timestamp, last_data, last_data_index) trade_data_manager.placeordercountmanager.place_order(code) # 获取买入位置信息 try: buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set = cls.__get_order_begin_pos( code) SecondAverageBigNumComputer.place_order_success(code, buy_single_index, buy_exec_index) AverageBigNumComputer.place_order_success(code, buy_single_index, buy_exec_index) LongAverageBigNumComputer.place_order_success(code, buy_single_index, buy_exec_index) except Exception as e: logging.exception(e) logger_l2_error.exception(e) l2_data_manager.TradePointManager.delete_buy_cancel_point(code) cls.debug(code, "执行买入成功") except Exception as e: cls.debug(code, "执行买入异常:{}", str(e)) pass finally: cls.debug(code, "m值影响因子：{}", l2_trade_factor.L2TradeFactorUtil.factors_to_string(code)) # 是否可以取消 @classmethod def __can_cancel(cls, code): if constant.TEST: return True, "" # 暂时注释掉 # 14点后如果是板块老大就不需要取消了 # now_time_str = tool.get_now_time_str() # if int(now_time_str.replace(":", "")) >= 140000: # industry, codes = ths_industry_util.get_same_industry_codes(code, gpcode_manager.get_gp_list()) # if industry is None: # return True, "没有获取到行业" # codes_index = industry_codes_sort.sort_codes(codes, code) # if codes_index is not None and codes_index.get(code) is not None: # # 同一板块中老二后面的不能买 # if codes_index.get(code) == 0: # return False, "14:00后老大不能撤单" # elif codes_index.get(code) == 1: # # 判断老大是否都是09：30：00涨停的 # # 同1板块老大是09:30:00涨停，老二14：00砸开的不撤 # first_count = 0 # for key in codes_index: # if codes_index[key] == 0: # first_count += 1 # if limit_up_time_manager.get_limit_up_time(key) == "09:30:00": # first_count -= 1 # if first_count == 0: # return False, "14:00后老大都开盘涨停，老二不能撤单" return True, "" # 是否可以买 @classmethod def __can_buy(cls, code): # 买1价格必须为涨停价才能买 # buy1_price = cls.buy1PriceManager.get_price(code) # if buy1_price is None: # return False, "买1价尚未获取到" # limit_up_price = gpcode_manager.get_limit_up_price(code) # if limit_up_price is None: # return False, "尚未获取到涨停价" # if abs(float(buy1_price) - float(limit_up_price)) >= 0.01: # return False, "买1价不为涨停价，买1价-{} 涨停价-{}".format(buy1_price, limit_up_price) # 从买入信号起始点到当前数据末尾的纯买手数与当前的卖1做比较，如果比卖1小则不能买入 total_datas = local_today_datas[code] try: sell1_time, sell1_price, sell1_volumn = cls.__ths_l2_trade_queue_manager.get_sell1_info(code) cls.buy_debug(code, "卖1信息为：({},{},{})", sell1_time, sell1_price, sell1_volumn) if sell1_time is not None and sell1_volumn > 0: # 获取执行位信息 buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set = cls.__get_order_begin_pos( code) buy_nums = num for i in range(buy_exec_index + 1, total_datas[-1]["index"] + 1): _val = total_datas[i]["val"] # 涨停买 if L2DataUtil.is_limit_up_price_buy(_val): # 涨停买 buy_nums += int(_val["num"]) * int(total_datas[i]["re"]) elif L2DataUtil.is_limit_up_price_buy_cancel(_val): buy_nums -= int(_val["num"]) * int(total_datas[i]["re"]) if buy_nums < sell1_volumn * 0.49: return False, "纯买量({})小于卖1量的49%{} 卖1时间：{}".format(buy_nums, sell1_volumn, sell1_time) except Exception as e: logging.exception(e) # 量比超过1.3的不能买 volumn_rate = l2_trade_factor.L2TradeFactorUtil.get_volumn_rate_by_code(code) if volumn_rate >= 1.3: return False, "最大量比超过1.3不能买" limit_up_time = limit_up_time_manager.get_limit_up_time(code) if limit_up_time is not None and l2_data_manager.L2DataUtil.get_time_as_second( limit_up_time) >= l2_data_manager.L2DataUtil.get_time_as_second( "14:30:00"): return False, "14:30后涨停的不能买，涨停时间为{}".format(limit_up_time) # 同一板块中老二后面的不能买 industry, codes = ths_industry_util.get_same_industry_codes(code, gpcode_manager.get_gp_list()) if industry is None: return True, "没有获取到行业" codes_index = industry_codes_sort.sort_codes(codes, code) if codes_index is not None and codes_index.get(code) is not None and codes_index.get(code) > 1: # 当老大老二当前没涨停 return False, "同一板块中老三,老四,...不能买" if cls.__codeActualPriceProcessor.is_under_water(code,total_datas[-1]["val"]["time"]): # 水下捞且板块中的票小于16不能买 # if global_util.industry_hot_num.get(industry) is not None and global_util.industry_hot_num.get( # industry) <= 16: # return False, "水下捞,板块中的票小于2只,为{}".format(global_util.industry_hot_num.get(industry)) # 水下捞自由流通市值大于老大的不要买 if codes_index.get(code) != 0: # 获取老大的市值 for c in codes_index: if codes_index.get(c) == 0 and global_util.zyltgb_map.get(code) > global_util.zyltgb_map.get(c): return False, "水下捞，不是老大，且自由流通市值大于老大" # 13:30后涨停，本板块中涨停票数<29不能买 # if limit_up_time is not None: # if int(limit_up_time.replace(":", "")) >= 133000 and global_util.industry_hot_num.get(industry) is not None: # if global_util.industry_hot_num.get(industry) < 16: # return False, "13:30后涨停，本板块中涨停票数<16不能买" if codes_index.get(code) is not None and codes_index.get(code) == 1: # 如果老大已经买成功了, 老二就不需要买了 first_codes = [] for key in codes_index: if codes_index.get(key) == 0: first_codes.append(key) # 暂时注释掉 # for key in first_codes: # state = trade_manager.get_trade_state(key) # if state == trade_manager.TRADE_STATE_BUY_SUCCESS: # # 老大已经买成功了 # return False, "老大{}已经买成功，老二无需购买".format(key) # # # 有9点半涨停的老大才能买老二，不然不能买 # # 获取老大的涨停时间 # for key in first_codes: # # 找到了老大 # time_ = limit_up_time_manager.get_limit_up_time(key) # if time_ == "09:30:00": # return True, "9:30涨停的老大，老二可以下单" # return False, "老大非9:30涨停，老二不能下单" # 过时 老二，本板块中涨停票数<29 不能买 # if codes_index.get(code) is not None and codes_index.get(code) == 1 and global_util.industry_hot_num.get( # industry) is not None: # if global_util.industry_hot_num.get(industry) < 29: # return False, "老二，本板块中涨停票数<29不能买" # 可以下单 return True, None @classmethod def __cancel_buy(cls, code): try: cls.debug(code, "开始执行撤单") trade_manager.start_cancel_buy(code) # 取消买入标识 l2_data_manager.TradePointManager.delete_buy_point(code) l2_data_manager.TradePointManager.delete_buy_cancel_point(code) l2_data_manager.TradePointManager.delete_compute_info_for_cancel_buy(code) l2_data_manager.TradePointManager.delete_count_info_for_cancel_buy(code) # 删除大群撤事件的大单 l2_data_manager.L2BetchCancelBigNumProcessor.del_recod(code) cls.debug(code, "执行撤单成功") except Exception as e: logging.exception(e) cls.debug(code, "执行撤单异常：{}", str(e)) @classmethod def cancel_buy(cls, code, msg=None, source="l2"): # 是否是交易队列触发 if source == "trade_queue": # 交易队列触发的需要下单后5s buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set = cls.__get_order_begin_pos( code) total_datas = local_today_datas[code] if buy_exec_index is not None and buy_exec_index > 0: now_time_str = tool.get_now_time_str() if tool.trade_time_sub(now_time_str, total_datas[buy_exec_index]["val"]["time"]) < 5: return False l2_data_manager.L2ContinueLimitUpCountManager.del_data(code) if code in cls.unreal_buy_dict: cls.unreal_buy_dict.pop(code) # 取消买入标识 l2_data_manager.TradePointManager.delete_buy_point(code) l2_data_manager.TradePointManager.delete_buy_cancel_point(code) l2_data_manager.TradePointManager.delete_compute_info_for_cancel_buy(code) l2_data_manager.TradePointManager.delete_count_info_for_cancel_buy(code) else: can_cancel, reason = cls.__can_cancel(code) if not can_cancel: # 不能取消 cls.cancel_debug(code, "撤单中断，原因：{}", reason) cls.debug(code, "撤单中断，原因：{}", reason) return False cls.__cancel_buy(code) l2_data_manager.L2BigNumProcessor.del_big_num_pos(code) cls.debug(code, "执行撤单成功，原因：{}", msg) return True # 虚拟下单 @classmethod def __virtual_buy(cls, code, buy_single_index, buy_exec_index, capture_time): cls.unreal_buy_dict[code] = (buy_exec_index, capture_time) SecondAverageBigNumComputer.place_order_success(code, buy_single_index, buy_exec_index) AverageBigNumComputer.place_order_success(code, buy_single_index, buy_exec_index) # 删除之前的板上卖信息 L2LimitUpSellStatisticUtil.delete(code) @classmethod def __start_compute_buy(cls, code, compute_start_index, compute_end_index, threshold_money, capture_time, new_add=True): if compute_end_index < compute_start_index: return _start_time = round(t.time() * 1000) total_datas = local_today_datas[code] # 获取买入信号计算起始位置 buy_single_index, buy_exec_index, buy_compute_index, num, count, max_num_set = cls.__get_order_begin_pos( code) # 是否为新获取到的位置 if buy_single_index is None: place_order_count = trade_data_manager.placeordercountmanager.get_place_order_count(code) continue_count = 3 # 前2次的信号连续笔数为3，后面为2 if place_order_count > 2: continue_count = 2 # 有买入信号 has_single, _index = cls.__compute_order_begin_pos(code, max( compute_start_index - 2 if new_add else compute_start_index, 0), continue_count, compute_end_index) buy_single_index = _index if has_single: num = 0 count = 0 cls.debug(code, "获取到买入信号起始点：{} ,计算范围：{}-{} ，数据：{}", buy_single_index, compute_start_index, compute_end_index, total_datas[buy_single_index]) # 如果是今天第一次有下单开始信号，需要设置大单起始点 cls.l2BigNumForMProcessor.set_begin_pos(code, buy_single_index) _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "下单信号计算时间") if buy_single_index is None: # 未获取到买入信号，终止程序 return None # 计算m值大单 cls.l2BigNumForMProcessor.process(code, max(buy_single_index, compute_start_index), compute_end_index, gpcode_manager.get_limit_up_price(code)) _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "计算m值大单") threshold_money, msg = cls.__get_threshmoney(code) # 买入纯买额统计 compute_index, buy_nums, buy_count, rebegin_buy_pos, max_num_set_new = cls.__sum_buy_num_for_order_3(code, max(buy_single_index,compute_start_index),compute_end_index,num,count,threshold_money,buy_single_index,max_num_set) _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "纯买额统计时间") cls.debug(code, "m值-{} m值因子-{}", threshold_money, msg) # 买入信号位与计算位置间隔2s及以上了 if rebegin_buy_pos is not None: # 需要重新计算纯买额 cls.__start_compute_buy(code, rebegin_buy_pos, compute_end_index, threshold_money, capture_time, False) return if compute_index is not None: cls.debug(code, "获取到买入执行位置：{} m值：{} 纯买手数：{} 纯买单数：{} 数据：{}", compute_index, threshold_money, buy_nums, buy_count, total_datas[compute_index]) # 记录买入信号位置 cls.__save_order_begin_data(code, buy_single_index, compute_index, compute_index, buy_nums, buy_count, max_num_set_new) # 如果是今天第一次有下单执行信号，涨停时间（买入执行位时间） limit_up_time_manager.save_limit_up_time(code, total_datas[compute_index]["val"]["time"]) # 虚拟下单 cls.__virtual_buy(code, buy_single_index, compute_index, capture_time) # 删除之前的所有撤单信号 l2_data_manager.TradePointManager.delete_buy_cancel_point(code) # 涨停封单额计算 L2LimitUpMoneyStatisticUtil.process_data(code, buy_single_index, compute_index, buy_single_index, buy_exec_index, False) _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "记录执行买入数据") # 数据是否处理完毕 if compute_index >= compute_end_index: need_cancel, cancel_data = SecondAverageBigNumComputer.need_cancel(code, buy_single_index, compute_index, buy_single_index, compute_index, True) # 分钟级大单计算 # need_cancel, cancel_data = AverageBigNumComputer.need_cancel(code, buy_single_index, compute_index, # buy_single_index, compute_index, True) cls.debug(code, "数据处理完毕，下单, 数据截图时间-{}", capture_time) # 数据已经处理完毕，如果还没撤单就实际下单 if need_cancel: if cls.cancel_buy(code, "S级大单撤销"): # 执行撤单成功 pass else: cls.__buy(code, capture_time, total_datas[compute_index], compute_index) else: # AverageBigNumComputer.need_cancel(code, buy_single_index, compute_index, # buy_single_index, compute_index, False) SecondAverageBigNumComputer.need_cancel(code, buy_single_index, compute_index, buy_single_index, compute_index, False) # 数据尚未处理完毕，进行下一步处理 cls.debug(code, "数据尚未处理完毕，进行下一步处理，处理进度：{}", compute_index) # 处理撤单步骤 cls.__process_order(code, compute_index + 1, compute_end_index, capture_time, False) else: # 未达到下单条件，保存纯买额，设置纯买额 # 记录买入信号位置 cls.__save_order_begin_data(code, buy_single_index, -1, compute_end_index, buy_nums, buy_count, max_num_set_new) print("保存大单时间", round((t.time() - _start_time) * 1000)) _start_time = t.time() pass # 获取下单起始信号 @classmethod def __get_order_begin_pos(cls, code): buy_single_index, buy_exec_index, compute_index, num, count, max_num_set = l2_data_manager.TradePointManager.get_buy_compute_start_data( code) return buy_single_index, buy_exec_index, compute_index, num, count, max_num_set # 保存下单起始信号 @classmethod def __save_order_begin_data(self, code, buy_single_index, buy_exec_index, compute_index, num, count, max_num_set): TradePointManager.set_buy_compute_start_data(code, buy_single_index, buy_exec_index, compute_index, num, count, max_num_set) # 计算下单起始信号 # compute_data_count 用于计算的l2数据数量 @classmethod def __compute_order_begin_pos(cls, code, start_index, continue_count, end_index): second_930 = 9 * 3600 + 30 * 60 + 0 # 倒数100条数据查询 datas = local_today_datas[code] if end_index - start_index + 1 < continue_count: return False, None __time = None last_index = None count = 0 start = None for i in range(start_index, end_index + 1): _val = datas[i]["val"] # 时间要>=09:30:00 if L2DataUtil.get_time_as_second(_val["time"]) < second_930: continue if L2DataUtil.is_limit_up_price_buy(_val): if last_index is None or (datas[last_index]["val"]["time"] == datas[i]["val"]["time"]): if start is None: start = i last_index = i count += datas[i]["re"] if count >= continue_count: return True, start else: # 本条数据作为起点 last_index = i count = datas[i]["re"] start = i elif not L2DataUtil.is_sell(_val) and not L2DataUtil.is_sell_cancel(_val): # 剔除卖与卖撤 last_index = None count = 0 start = None return False, None @classmethod def __get_threshmoney(cls, code): return l2_trade_factor.L2TradeFactorUtil.compute_m_value(code) # 是否为万手哥 @classmethod def __is_big_money(cls, limit_up_price, val): if int(val["num"]) >= constant.BIG_MONEY_NUM: return True if int(val["num"]) * limit_up_price >= constant.BIG_MONEY_AMOUNT: return True return False # 计算万手哥笔数 @classmethod def __compute_big_money_count(cls, total_datas, start_index, end_index): count = 0 for i in range(start_index, end_index + 1): if L2DataUtil.is_limit_up_price_buy(total_datas[i]["val"]): count += total_datas[i]["re"] elif L2DataUtil.is_limit_up_price_buy_cancel(total_datas[i]["val"]): count -= total_datas[i]["re"] return count # 统计买入净买量，不计算在买入信号之前的买撤单 @classmethod def __sum_buy_num_for_order_3(cls, code, compute_start_index, compute_end_index, origin_num, origin_count, threshold_money, buy_single_index, max_num_set): def get_threshold_count(): count = threshold_count - sub_threshold_count if count < 3: count = 3 count = round(count * buy1_factor) # 最高30笔，最低8笔 if count > 21: count = 21 if count < 8: count = 8 return count _start_time = t.time() total_datas = local_today_datas[code] # 计算从买入信号开始到计算开始位置的大单数量 sub_threshold_count = cls.__compute_big_money_count(total_datas, buy_single_index, compute_start_index - 1) if sub_threshold_count < 0: sub_threshold_count = 0 buy_nums = origin_num buy_count = origin_count limit_up_price = gpcode_manager.get_limit_up_price(code) buy1_price = cls.buy1PriceManager.get_price(code) if limit_up_price is None: raise Exception("涨停价无法获取") # 目标手数 threshold_num = threshold_money / (limit_up_price * 100) buy1_factor = 1 # 获取买1是否为涨停价 if buy1_price is None: buy1_factor = 1.3 elif limit_up_price is None: buy1_factor = 1.3 elif abs(float(buy1_price) - float(limit_up_price)) >= 0.01: print("买1价不为涨停价，买1价-{} 涨停价-{}".format(buy1_price, limit_up_price)) buy1_factor = 1.3 # 目标订单数量 threshold_count = l2_trade_factor.L2TradeFactorUtil.get_safe_buy_count(code) buy_single_time_seconds = L2DataUtil.get_time_as_second(total_datas[buy_single_index]["val"]["time"]) # 可以触发买，当有涨停买信号时才会触发买 trigger_buy = True place_order_count = trade_data_manager.placeordercountmanager.get_place_order_count(code) if place_order_count > 3: place_order_count = 3 # 间隔最大时间依次为：3,9,27,81 max_space_time = pow(3, place_order_count + 1) - 1 # 最大买量 max_buy_num = 0 max_buy_num_set = set(max_num_set) for i in range(compute_start_index, compute_end_index + 1): data = total_datas[i] _val = total_datas[i]["val"] trigger_buy = False # 必须为连续3秒内的数据 if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds > max_space_time: TradePointManager.delete_buy_point(code) if i == compute_end_index: # 数据处理完毕 return None, buy_nums, buy_count, None, max_buy_num_set else: # 计算买入信号，不能同一时间开始计算 for ii in range(buy_single_index + 1, compute_end_index + 1): if total_datas[buy_single_index]["val"]["time"] != total_datas[ii]["val"]["time"]: return None, buy_nums, buy_count, ii, max_buy_num_set # 涨停买 if L2DataUtil.is_limit_up_price_buy(_val): if cls.__is_big_money(limit_up_price, _val): sub_threshold_count += int(total_datas[i]["re"]) max_buy_num_set.add(i) if round(int(_val["num"]) * float(_val["price"])) >= 5900: trigger_buy = True # 只统计59万以上的金额 buy_nums += int(_val["num"]) * int(total_datas[i]["re"]) buy_count += int(total_datas[i]["re"]) if buy_nums >= threshold_num and buy_count >= get_threshold_count(): logger_l2_trade_buy.info("{}获取到买入执行点：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{} 目标纯买单数：{}, 大单数量：{}", code, i, buy_nums, threshold_num, buy_count, get_threshold_count(), sub_threshold_count) elif L2DataUtil.is_limit_up_price_buy_cancel(_val): if cls.__is_big_money(limit_up_price, _val): sub_threshold_count -= int(total_datas[i]["re"]) if round(int(_val["num"]) * float(_val["price"])) >= 5900: # 只统计59万以上的金额 # 涨停买撤 # 判断买入位置是否在买入信号之前 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(total_datas[i], local_today_num_operate_map.get( code)) if buy_index is not None: # 找到买撤数据的买入点 if buy_index >= buy_single_index: buy_nums -= int(_val["num"]) * int(data["re"]) buy_count -= int(data["re"]) cls.buy_debug(code, "{}数据在买入信号之后 撤买纯买手数：{} 目标手数：{}", i, buy_nums, threshold_num) else: cls.buy_debug(code, "{}数据在买入信号之前，买入位：{}", i, buy_index) if total_datas[buy_single_index]["val"]["time"] == buy_data["val"]["time"]: # 同一秒,当作买入信号之后处理 buy_nums -= int(_val["num"]) * int(data["re"]) buy_count -= int(data["re"]) cls.buy_debug(code, "{}数据买入位与预估买入位在同一秒", i) else: # 未找到买撤数据的买入点 cls.buy_debug(code, "未找到买撤数据的买入点: 位置-{} 数据-{}", i, data) buy_nums -= int(_val["num"]) * int(total_datas[i]["re"]) buy_count -= int(total_datas[i]["re"]) cls.buy_debug(code, "位置-{}，总手数：{}，目标手数：{}", i, buy_nums, threshold_num) # 有撤单信号，且小于阈值 if buy_nums >= threshold_num and buy_count >= get_threshold_count() and trigger_buy and len(max_buy_num_set)>1: return i, buy_nums, buy_count, None, max_buy_num_set cls.buy_debug(code, "尚未获取到买入执行点，起始计算位置：{} 统计纯买手数：{} 目标纯买手数：{} 统计纯买单数：{} 目标纯买单数：{} 大单数量：{}", compute_start_index, buy_nums, threshold_num, buy_count, get_threshold_count(), sub_threshold_count) return None, buy_nums, buy_count, None, max_buy_num_set @classmethod def test(cls): code = "002556" l2_trade_test.clear_trade_data(code) load_l2_data(code, True) _start = t.time() if True: state = trade_manager.get_trade_state(code) cls.random_key[code] = random.randint(0, 100000) capture_timestamp = 1999988888 try: if state == trade_manager.TRADE_STATE_BUY_DELEGATED or state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER: # 已挂单 cls.__process_order(code, 1552, 1641, capture_timestamp) else: # 未挂单 cls.__process_not_order(code, 1552, 1641, capture_timestamp) except Exception as e: logging.exception(e) print("处理时间", round((t.time() - _start) * 1000)) return # 按s批量化数据 total_datas = local_today_datas[code] start_time = total_datas[0]["val"]["time"] start_index = 0 for i in range(0, len(total_datas)): if total_datas[i]["val"]["time"] != start_time: cls.random_key[code] = random.randint(0, 100000) # 处理数据 start = start_index # if start != 201: # continue end = i - 1 print("处理进度：{},{}".format(start, end)) capture_timestamp = 1999999999 state = trade_manager.get_trade_state(code) try: if state == trade_manager.TRADE_STATE_BUY_DELEGATED or state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER: # 已挂单 cls.__process_order(code, start, end, capture_timestamp) else: # 未挂单 cls.__process_not_order(code, start, end, capture_timestamp) except Exception as e: logging.exception(e) # t.sleep(1) start_index = i start_time = total_datas[i]["val"]["time"] print("时间花费:", round((t.time() - _start) * 1000)) @classmethod def test1(cls): code = "002556" l2_trade_test.clear_trade_data(code) l2_data_manager.local_latest_datas[code] = [] load_l2_data(code, True) _start = t.time() capture_timestamp = 1999999999 cls.process(code, l2_data_manager.local_today_datas[code][1552:1641], capture_timestamp) print("时间花费:", round((t.time() - _start) * 1000)) pass @classmethod def test2(cls): code = "002864" load_l2_data(code) limit_up_time_manager.load_limit_up_time() limit_up_time = limit_up_time_manager.get_limit_up_time(code) if limit_up_time is not None and l2_data_manager.L2DataUtil.get_time_as_second( limit_up_time) >= l2_data_manager.L2DataUtil.get_time_as_second( "14:30:00"): return False, "14:30后涨停的不能买，涨停时间为{}".format(limit_up_time) # 同一板块中老二后面的不能买 industry, codes = ths_industry_util.get_same_industry_codes(code, gpcode_manager.get_gp_list()) if industry is None: return True, "没有获取到行业" codes_index = industry_codes_sort.sort_codes(codes, code) if codes_index is not None and codes_index.get(code) is not None and codes_index.get(code) > 1: return False, "同一板块中老三,老四,...不能买" if cls.__codeActualPriceProcessor.is_under_water(code): # 水下捞且板块中的票小于21不能买 if global_util.industry_hot_num.get(industry) is not None and global_util.industry_hot_num.get( industry) <= 16: return False, "水下捞,板块中的票小于2只,为{}".format(global_util.industry_hot_num.get(industry)) if codes_index.get(code) != 0: return False, "水下捞,不是老大，是老{}".format(codes_index.get(code)) # 13:30后涨停，本板块中涨停票数<29不能买 limit_up_time = limit_up_time_manager.get_limit_up_time(code) if limit_up_time is not None: if int(limit_up_time.replace(":", "")) >= 133000 and global_util.industry_hot_num.get(industry) is not None: if global_util.industry_hot_num.get(industry) < 16: return False, "13:30后涨停，本板块中涨停票数<16不能买" if codes_index.get(code) is not None and codes_index.get(code) == 1: # ----此条注释----- # 如果老大已经买成功了，老二就不需要买了 # first_codes = [] # for key in codes_index: # if codes_index.get(key) == 0: # first_codes.append(key) # # for key in first_codes: # state = trade_manager.get_trade_state(key) # if state == trade_manager.TRADE_STATE_BUY_SUCCESS: # # 老大已经买成功了 # return False, "老大{}已经买成功，老二无需购买".format(key) # ----此条注释----- # ----此条注释----- # 有9点半涨停的老大才能买老二，不然不能买 # 获取老大的涨停时间 # for key in first_codes: # # 找到了老大 # time_ = limit_up_time_manager.get_limit_up_time(key) # if time_ == "09:30:00": # return True, "9:30涨停的老大，老二可以下单" # return False, "老大非9:30涨停，老二不能下单" # ----此条注释----- return True, "老二可以下单" @classmethod def test3(cls): code = "002094" load_l2_data(code, True) cls.random_key[code] = random.randint(0, 100000) buy_single_begin_index, buy_exec_index = 426, 479 L2LimitUpMoneyStatisticUtil.process_data(code, 480, 519, buy_single_begin_index, buy_exec_index, False) L2LimitUpMoneyStatisticUtil.process_data(code, 480, 519, buy_single_begin_index, buy_exec_index, False) @classmethod def test_can_buy(cls): code = "002923" load_l2_data(code, True) limit_up_time_manager.load_limit_up_time() can, msg = cls.__can_buy(code) print(can, msg) # 涨停封单额统计 class L2LimitUpMoneyStatisticUtil: _redisManager = redis_manager.RedisManager(1) _thsBuy1VolumnManager = trade_queue_manager.THSBuy1VolumnManager() @classmethod def __get_redis(cls): return cls._redisManager.getRedis() # 设置l2的每一秒涨停封单额数据 @classmethod def __set_l2_second_money_record(cls, code, time, num, from_index, to_index): old_num, old_from, old_to = cls.__get_l2_second_money_record(code, time) if old_num is None: old_num = num old_from = from_index old_to = to_index else: old_num += num old_to = to_index key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", "")) cls.__get_redis().setex(key, tool.get_expire(), json.dumps((old_num, old_from, old_to))) @classmethod def __get_l2_second_money_record(cls, code, time): key = "l2_limit_up_second_money-{}-{}".format(code, time.replace(":", "")) val = cls.__get_redis().get(key) return cls.__format_second_money_record_val(val) @classmethod def __format_second_money_record_val(cls, val): if val is None: return None, None, None val = json.loads(val) return val[0], val[1], val[2] @classmethod def __get_l2_second_money_record_keys(cls, code, time_regex): key = "l2_limit_up_second_money-{}-{}".format(code, time_regex) keys = cls.__get_redis().keys(key) return keys # 设置l2最新的封单额数据 @classmethod def __set_l2_latest_money_record(cls, code, index, num): key = "l2_limit_up_money-{}".format(code) cls.__get_redis().setex(key, tool.get_expire(), json.dumps((num, index))) # 返回数量,索引 @classmethod def __get_l2_latest_money_record(cls, code): key = "l2_limit_up_money-{}".format(code) result = cls.__get_redis().get(key) if result: result = json.loads(result) return result[0], result[1] else: return 0, -1 # 矫正数据 # 矫正方法为取矫正时间两侧的秒分布数据，用于确定计算结束坐标 @classmethod def verify_num(cls, code, num, time_str): # 记录买1矫正日志 logger_buy_1_volumn.info("涨停封单量矫正：代码-{} 量-{} 时间-{}", code, num, time_str) time_ = time_str.replace(":", "") key = None for i in range(4, -2, -2): # 获取本(分钟/小时/天)内秒分布数据 time_regex = "{}*".format(time_[:i]) keys_ = cls.__get_l2_second_money_record_keys(code, time_regex) if keys_ and len(keys_) > 1: # 需要排序 keys = [] for k in keys_: keys.append(k) keys.sort(key=lambda tup: int(tup.split("-")[-1])) # 有2个元素 for index in range(0, len(keys) - 1): time_1 = keys[index].split("-")[-1] time_2 = keys[index + 1].split("-")[-1] if int(time_1) <= int(time_) <= int(time_2): # 在此时间范围内 if time_ == time_2: key = keys[index + 1] else: key = keys[index] break if key: val = cls.__get_redis().get(key) old_num, old_from, old_to = cls.__format_second_money_record_val(val) end_index = old_to # 保存最近的数据 cls.__set_l2_latest_money_record(code, end_index, num) logger_buy_1_volumn.info("涨停封单量矫正结果：代码-{} 位置-{} 量-{}", code, end_index, num) break # 计算量，用于涨停封单量的计算 @classmethod def __compute_num(cls, code, data, buy_single_data): if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]) or L2DataUtil.is_sell(data["val"]): # 涨停买撤与卖 return 0 - int(data["val"]["num"]) * data["re"] else: # 卖撤 if L2DataUtil.is_sell_cancel(data["val"]): # 卖撤的买数据是否在买入信号之前，如果在之前就不计算，不在之前就计算 if l2_data_util.is_sell_index_before_target(data, buy_single_data, local_today_num_operate_map.get(code)): return 0 return int(data["val"]["num"]) * data["re"] @classmethod def clear(cls, code): key = "l2_limit_up_money-{}".format(code) cls.__get_redis().delete(key) # 返回取消的标志数据 # with_cancel 是否需要判断是否撤销 @classmethod def process_data(cls, code, start_index, end_index, buy_single_begin_index, buy_exec_index, with_cancel=True): if buy_single_begin_index is None or buy_exec_index is None: return None, None start_time = round(t.time() * 1000) total_datas = local_today_datas[code] time_dict_num = {} # 记录计算的坐标 time_dict_num_index = {} # 坐标-量的map num_dict = {} # 统计时间分布 time_dict = {} for i in range(start_index, end_index + 1): data = total_datas[i] val = data["val"] time_ = val["time"] if time_ not in time_dict: time_dict[time_] = i for i in range(start_index, end_index + 1): data = total_datas[i] val = data["val"] time_ = val["time"] if time_ not in time_dict_num: time_dict_num[time_] = 0 time_dict_num_index[time_] = {"s": i, "e": i} time_dict_num_index[time_]["e"] = i num = cls.__compute_num(code, data, total_datas[buy_single_begin_index]) num_dict[i] = num time_dict_num[time_] = time_dict_num[time_] + num for t_ in time_dict_num: cls.__set_l2_second_money_record(code, t_, time_dict_num[t_], time_dict_num_index[t_]["s"], time_dict_num_index[t_]["e"]) print("保存涨停封单额时间：", round(t.time() * 1000) - start_time) # 累计最新的金额 total_num, index = cls.__get_l2_latest_money_record(code) record_msg = f"同花顺买1信息 {total_num},{index}" if index == -1: # 没有获取到最新的矫正封单额，需要从买入信号开始点计算 index = buy_single_begin_index - 1 total_num = 0 cancel_index = None cancel_msg = None # 待计算量 limit_up_price = gpcode_manager.get_limit_up_price(code) min_volumn = round(10000000 / (limit_up_price * 100)) min_volumn_big = min_volumn * 5 # 不同时间的数据开始坐标 time_start_index_dict = {} # 数据时间分布 time_list = [] # 到当前时间累积的买1量 time_total_num_dict = {} # 大单撤销笔数 cancel_big_num_count = 0 buy_exec_time = tool.get_time_as_second(total_datas[buy_exec_index]["val"]["time"]) # 获取最大封单额 max_buy1_volume = cls._thsBuy1VolumnManager.get_max_buy1_volume(code) # 从同花顺买1矫正过后的位置开始计算，到end_index结束 for i in range(index + 1, end_index + 1): data = total_datas[i] # 统计撤销数量 try: if big_money_num_manager.is_big_num(data["val"]): if L2DataUtil.is_limit_up_price_buy_cancel(data["val"]): cancel_big_num_count += int(data["re"]) # TODO 大量重复的工作需要处理，可以暂存在内存中，从而减少计算 # 获取是否在买入执行信号周围2s buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and buy_data is not None: # 相差1s buy_time = buy_data["val"]["time"] if abs(buy_exec_time - tool.get_time_as_second(buy_time)) < 2: cancel_big_num_count += int(data["re"]) elif L2DataUtil.is_limit_up_price_buy(data["val"]): cancel_big_num_count -= int(data["re"]) except Exception as e: logging.exception(e) threshold_rate = 0.5 if cancel_big_num_count >= 0: if cancel_big_num_count < 10: threshold_rate = threshold_rate - cancel_big_num_count * 0.01 else: threshold_rate = threshold_rate - 10 * 0.01 time_ = data["val"]["time"] if time_ not in time_start_index_dict: # 记录每一秒的开始位置 time_start_index_dict[time_] = i # 记录时间分布 time_list.append(time_) # 上一段时间的总数 time_total_num_dict[time_] = total_num exec_time_offset = tool.trade_time_sub(data["val"]["time"], total_datas[buy_exec_index]["val"]["time"]) val = num_dict.get(i) if val is None: val = cls.__compute_num(code, data, total_datas[buy_single_begin_index]) total_num += val # 在处理数据的范围内，就需要判断是否要撤单了 if start_index <= i <= end_index: # 如果是减小项 if val < 0: # 当前量小于最大量的24%则需要取消 if exec_time_offset >= 30: if total_num <= min_volumn_big and max_buy1_volume * 0.24 > total_num: cancel_index = i cancel_msg = "封板额小于最高封板额的24% {}/{}".format(total_num, max_buy1_volume) break # 累计封单金额小于1000万 if total_num < min_volumn: # 与执行位相隔>=5s时规则生效 if exec_time_offset >= 5: cancel_index = i cancel_msg = "封单金额小于1000万,为{}".format(total_num) break # 相邻2s内的数据减小50% # 上1s的总数 last_second_total_volumn = time_total_num_dict.get(time_list[-1]) if last_second_total_volumn > 0 and ( last_second_total_volumn - total_num) / last_second_total_volumn >= threshold_rate: # 与执行位相隔>=5s时规则生效 if exec_time_offset >= 5: # 相邻2s内的数据减小50% cancel_index = i cancel_msg = "相邻2s({})内的封单量减小50%({}->{})".format(time_, last_second_total_volumn, total_num) break # 记录中有上2个数据 if len(time_list) >= 2: # 倒数第2个数据 last_2_second_total_volumn = time_total_num_dict.get(time_list[-2]) if last_2_second_total_volumn > 0: if last_2_second_total_volumn > last_second_total_volumn > total_num: dif = last_2_second_total_volumn - total_num if dif / last_2_second_total_volumn >= threshold_rate: # 与执行位相隔>=5s时规则生效 if exec_time_offset >= 5: cancel_index = i cancel_msg = "相邻3s({})内的封单量(第3秒 与 第1的 减小比例)减小50%({}->{}->{})".format(time_, last_2_second_total_volumn, last_second_total_volumn, total_num) break # ------大单撤处理------- # if total_num < min_volumn_big: if exec_time_offset < 31: pass # try: # b_need_cancel, b_cancel_index = AverageBigNumComputer.need_cancel(code, i, i) # if b_need_cancel: # cancel_index = b_cancel_index # cancel_msg = "1分钟内大单撤销比例触发阈值" # break # except Exception as e: # logging.exception(e) # 30s外才执行 elif 31 <= exec_time_offset: try: b_need_cancel, b_cancel_index = LongAverageBigNumComputer.need_cancel(code, buy_exec_index, i, i) if b_need_cancel: cancel_index = b_cancel_index cancel_msg = "30s后内大单撤销比例触发阈值" break except Exception as e: logging.exception(e) # ------大单撤处理结束------- if not with_cancel: cancel_index = None print("封单额计算时间：", round(t.time() * 1000) - start_time) process_end_index = end_index if cancel_index: process_end_index = cancel_index # 保存最新累计金额 # cls.__set_l2_latest_money_record(code, process_end_index, total_num) l2_data_log.l2_time(code, round(t.time() * 1000) - start_time, "l2数据封单额计算时间", False) if cancel_index: L2TradeDataProcessor.cancel_debug(code, "数据处理位置：{}-{}，{}，最终买1为：{}", start_index, end_index, record_msg, total_num) return total_datas[cancel_index], cancel_msg return None, None # 涨停卖统计 class L2LimitUpSellStatisticUtil: _redisManager = redis_manager.RedisManager(0) @classmethod def __get_redis(cls): return cls._redisManager.getRedis() # 新增卖数据 @classmethod def __incre_sell_data(cls, code, num): key = "limit_up_sell_num-{}".format(code) cls.__get_redis().incrby(key, num) @classmethod def __get_sell_data(cls, code): key = "limit_up_sell_num-{}".format(code) val = cls.__get_redis().get(key) if val is None: return 0 return int(val) @classmethod def __save_process_index(cls, code, index): key = "limit_up_sell_index-{}".format(code) cls.__get_redis().setex(key, tool.get_expire(), index) @classmethod def __get_process_index(cls, code): key = "limit_up_sell_index-{}".format(code) val = cls.__get_redis().get(key) if val is None: return -1 return int(val) # 清除数据,当取消成功与买入之前需要清除数据 @classmethod def delete(cls, code): key = "limit_up_sell_num-{}".format(code) cls.__get_redis().delete(key) key = "limit_up_sell_index-{}".format(code) cls.__get_redis().delete(key) @classmethod def clear(cls): keys = cls.__get_redis().keys("limit_up_sell_num-*") for k in keys: cls.__get_redis().delete(k) # 处理数据，返回是否需要撤单 # 处理范围：买入执行位-当前最新位置 @classmethod def process(cls, code, start_index, end_index, buy_exec_index): # 获取涨停卖的阈值 limit_up_price = gpcode_manager.get_limit_up_price(code) zyltgb = l2_trade_factor.L2TradeFactorUtil.get_zyltgb(code) # 大于自由流通市值的4.8% threshold_num = int(zyltgb * 0.048) // (limit_up_price * 100) total_num = cls.__get_sell_data(code) cancel_index = None process_index = cls.__get_process_index(code) total_datas = local_today_datas.get(code) for i in range(start_index, end_index + 1): if i < buy_exec_index: continue if i <= process_index: continue if L2DataUtil.is_limit_up_price_sell(total_datas[i]["val"]) or L2DataUtil.is_sell(total_datas[i]["val"]): num = int(total_datas[i]["val"]["num"]) cls.__incre_sell_data(code, num) total_num += num if total_num > threshold_num: cancel_index = i break if cancel_index is not None: process_index = cancel_index else: process_index = end_index L2TradeDataProcessor.cancel_debug(code, "板上卖信息：计算位置：{}-{} 板上卖数据{}/{}", start_index, end_index, total_num, threshold_num) cls.__save_process_index(code, process_index) if cancel_index is not None: return total_datas[cancel_index], "板上卖的手数{} 超过{}".format(total_num, threshold_num) return None, "" @classmethod def test(cls): code = "003005" load_l2_data(code) L2TradeDataProcessor.random_key[code] = 123123 cls.process(code, 126, 171, 126) # s级平均大单计算 # 计算范围到申报时间的那一秒 class SecondAverageBigNumComputer: __redis_manager = redis_manager.RedisManager(0) __place_order_time_dict = {} @classmethod def __getRedis(cls): return cls.__redis_manager.getRedis() @classmethod def __save_average_data(cls, code, average_num, average_up_count, start_index, end_index): key = "s_average_big_num-{}".format(code) cls.__getRedis().setex(key, 2000, json.dumps((average_num, average_up_count, start_index, end_index))) L2TradeDataProcessor.cancel_debug(code, "保存秒级大单位置信息：平均手数-{} 大单数量-{} 计算开始范围-{}:{}".format(average_num, average_up_count, start_index, end_index)) @classmethod def __get_average_data(cls, code): key = "s_average_big_num-{}".format(code) val = cls.__getRedis().get(key) if val is None: return None, None, None, None val = json.loads(val) return val[0], val[1], val[2], val[3] # 保存买撤数据 @classmethod def __save_cancel_data(cls, code, cancel_index): key = "s_average_big_num_comput_info-{}".format(code) cls.__getRedis().sadd(key, cancel_index) # 获取买撤的数据 @classmethod def __get_cancel_datas(cls, code): key = "s_average_big_num_comput_info-{}".format(code) val = cls.__getRedis().smembers(key) return val # 保存买撤数据 @classmethod def __save_apply_time(cls, code, time_str): key = "s_average_big_num_apply_time-{}".format(code) cls.__getRedis().setex(key, tool.get_expire(), time_str) # 获取买撤的数据 @classmethod def __get_apply_time(cls, code): key = "s_average_big_num_apply_time-{}".format(code) val = cls.__getRedis().get(key) return val # 保存结束位置 @classmethod def __save_end_index(cls, code, end_index): key = "s_average_big_num_end_index_set-{}".format(code) cls.__getRedis().sadd(key, end_index) @classmethod def __list_end_indexs(cls, code): key = "s_average_big_num_end_index_set-{}".format(code) vals = cls.__getRedis().smembers(key) if vals is None: return None results = [] for val in vals: results.append(int(val)) results.sort() return results @classmethod def __clear_data(cls, code): ks = ["s_average_big_num_comput_info-{}".format(code), "s_average_big_num-{}".format(code), "s_average_big_num_end_index_set-{}".format(code)] for key in ks: cls.__getRedis().delete(key) @classmethod def clear_data(cls): ks = ["s_average_big_num_comput_info-*", "s_average_big_num-*", "s_average_big_num_end_index_set-*"] for key in ks: keys = cls.__getRedis().keys(key) for k in keys: cls.__getRedis().delete(k) # 计算平均手数 # 计算范围：买入信号起始点到买入执行位的下一张图结束点数据为止 @classmethod def compute_average_big_num(cls, code, buy_single_index, start_index, end_index): cls.__save_end_index(code, end_index) # 保存结束位置 end_indexs = cls.__list_end_indexs(code) print("compute_average_big_num", code, buy_single_index, start_index, end_index) L2TradeDataProcessor.cancel_debug(code, "开始计算短大单位置") total_data = local_today_datas[code] num = 0 count = 0 apply_time = cls.get_apply_time(code) apply_time_second = int(apply_time.replace(":", "")) for ei in end_indexs: if int(total_data[ei]["val"]["time"].replace(":", "")) >= apply_time_second: end_index = ei break for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] # if int(val["time"].replace(":", "")) > apply_time_second: # # 重新设置计算结束位置 # end_index = i - 1 # break if L2DataUtil.is_limit_up_price_buy(val): # and float(val["price"]) * int(val["num"]) > 7500: # 75万以上的才参与计算平均大单 count += data["re"] num += int(val["num"]) # 如果没有找到75万以上的单就不添加75w的筛选条件 if count == 0: for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val): if int(val["time"].replace(":", "")) > apply_time_second: break # 75万以上的才参与计算平均大单 count += data["re"] num += int(val["num"]) average_num = num // count average_num = min(constant.BIG_MONEY_NUM, round(constant.BIG_MONEY_AMOUNT / gpcode_manager.get_limit_up_price(code))) average_up_count = 0 for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val): if int(val["time"].replace(":", "")) > apply_time_second: break if int(val["num"]) >= average_num: average_up_count += data["re"] print("平均手数：", average_num, "大单总数：", average_up_count) # 保存数据 cls.__save_average_data(code, average_num, average_up_count, start_index, end_index) # 是否需要撤单 @classmethod def need_cancel(cls, code, buy_single_index, buy_exec_index, start_index, end_index, need_cancel=True): average_num, average_up_count, a_start_index, a_end_index = cls.__get_average_data(code) L2TradeDataProcessor.cancel_debug(code, "s级是否需要撤单，数据范围：{}-{} 平均大单信息-({},{},{},{})", start_index, end_index, average_num, average_up_count, a_start_index, a_end_index) if average_num is None: return False, None total_data = local_today_datas[code] # 只守护30s if tool.trade_time_sub(total_data[start_index]["val"]["time"], total_data[buy_exec_index]["val"]["time"]) > 30: return False, None # 如果start_index与buy_single_index相同，即是下单后的第一次计算 # 需要查询买入信号之前的同1s是否有涨停撤的数据 if buy_single_index == start_index: for i in range(buy_single_index - 1, 0, -1): data = total_data[i] val = data["val"] if val["time"] != total_data[buy_single_index]["val"]["time"]: break if L2DataUtil.is_limit_up_price_buy_cancel(val) and int(val["cancelTime"]) == 0: # 涨停买撤销且撤销的间隔时间为0 # 查询买入信号，如果无法查询到或者是买入位置比买入信号小就不算 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and a_start_index <= buy_index <= a_end_index: # 在买入信号之后 cls.__save_cancel_data(code, i) for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] # print("处理进度", i) if L2DataUtil.is_limit_up_price_buy_cancel(val): # 查询买入位置 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and a_start_index <= buy_index <= a_end_index: cls.__save_cancel_data(code, i) else: # 有部分撤销从而导致的无法溯源，这时就需要判断预估买入时间是否在a_start_index到a_end_index的时间区间 min_space, max_space = l2_data_util.compute_time_space_as_second(val["cancelTime"], val["cancelTimeUnit"]) # 只判断S级撤销,只有s级撤销才有可能相等 if max_space - min_space <= 1: buy_time = tool.trade_time_add_second(val["time"], 0 - min_space) if int(total_data[a_start_index]["val"]["time"].replace(":", "")) <= int( buy_time.replace(":", "")) <= int( total_data[a_end_index]["val"]["time"].replace(":", "")): cls.__save_cancel_data(code, i) if need_cancel: # 计算买撤大单暂比 cancel_datas = cls.__get_cancel_datas(code) if cancel_datas is not None and len(cancel_datas) > 0: L2TradeDataProcessor.cancel_debug(code, "s级大单 取消数量:{}", len(cancel_datas)) cancel_rate_threshold = 0.49 place_order_count = trade_data_manager.placeordercountmanager.get_place_order_count(code) if place_order_count <= 1: cancel_rate_threshold = 0.49 elif place_order_count <= 2: cancel_rate_threshold = 0.59 else: cancel_rate_threshold = 0.69 cancel_indexs = [] for index in cancel_datas: cancel_indexs.append(int(index)) cancel_indexs.sort() # print("取消的数据", cancel_indexs) cancel_count = 0 for index in cancel_indexs: data = total_data[index] if int(data["val"]["num"]) >= average_num: cancel_count += data["re"] if cancel_count / average_up_count > cancel_rate_threshold: return True, total_data[index] return False, None # 是否需要计算 @classmethod def is_need_compute_average(cls, code, latest_data): total_datas = local_today_datas[code] data = cls.__place_order_time_dict.get(code) if data is None: return False, None, None elif tool.trade_time_sub(latest_data["val"]["time"], cls.get_apply_time(code)) < 5: # 有5s时间上传申报时间 return True, data[1], data[2] else: cls.__place_order_time_dict.pop(code) return False, None, None # 设置申报时间 @classmethod def set_apply_time(cls, code, time_str, force=False): old_time_str = cls.get_apply_time(code) if not force: if old_time_str is not None: sub_time = tool.trade_time_sub(time_str, old_time_str) if sub_time <= 0 or sub_time > 4: # 申报时间与下单时间不能操过4s return cls.__save_apply_time(code, time_str) @classmethod def get_apply_time(cls, code): return cls.__get_apply_time(code) # 下单成功 @classmethod def place_order_success(cls, code, buy_single_index, buy_exec_index): cls.__clear_data(code) cls.__place_order_time_dict[code] = (t.time(), buy_single_index, buy_exec_index) # 以防万一，先保存下单信息 total_data = local_today_datas[code] cls.set_apply_time(code, total_data[buy_exec_index]["val"]["time"], True) cls.compute_average_big_num(code, buy_single_index, buy_single_index, total_data[-1]["index"]) @classmethod def __test(cls, datas): code = datas[0] load_l2_data(code) L2TradeDataProcessor.random_key[code] = 123123 # 先执行下单 buy_single_index = datas[1] buy_exec_index = datas[2] local_today_datas[code] = local_today_datas[code][0:datas[4]] cls.place_order_success(code, buy_single_index, buy_exec_index) # 执行是否需要计算average cls.compute_average_big_num(code, buy_single_index, buy_single_index, datas[3]) cancel, cancel_data = cls.need_cancel(code, buy_single_index, buy_exec_index, buy_single_index, buy_exec_index, False) for i in range(buy_exec_index + 1, datas[4]): cancel, cancel_data = cls.need_cancel(code, buy_single_index, buy_exec_index, i, i) if cancel: print("需要撤单", cancel, cancel_data["index"]) break @classmethod def test(cls): # cls.__test(("000909", 607, 646, 646, 694)) # 代码 买入信号起始点 买入信息执行位置 计算末位 最远计算位置 # cls.__test(("002793", 292, 308, 314, 410)) cls.__save_end_index("000333", 200) cls.__save_end_index("000333", 101) cls.__save_end_index("000333", 99) cls.__save_end_index("000333", 120) cls.__save_end_index("000333", 126) cls.__save_end_index("000333", 126) print(cls.__list_end_indexs("000333")) # 执行是否需要撤销 # 平均大单计算 class AverageBigNumComputer: __redis_manager = redis_manager.RedisManager(0) __place_order_time_dict = {} @classmethod def __getRedis(cls): return cls.__redis_manager.getRedis() @classmethod def __save_average_data(cls, code, average_num, average_up_count, start_index, end_index): key = "average_big_num-{}".format(code) cls.__getRedis().setex(key, 2000, json.dumps((average_num, average_up_count, start_index, end_index))) L2TradeDataProcessor.cancel_debug(code, "保存短大单位置信息：平均手数-{} 大单数量-{} 计算开始范围-{}:{}".format(average_num, average_up_count, start_index, end_index)) @classmethod def __get_average_data(cls, code): key = "average_big_num-{}".format(code) val = cls.__getRedis().get(key) if val is None: return None, None, None, None val = json.loads(val) return val[0], val[1], val[2], val[3] # 保存买撤数据 @classmethod def __save_cancel_data(cls, code, cancel_index): key = "average_big_num_comput_info-{}".format(code) cls.__getRedis().sadd(key, cancel_index) # 获取买撤的数据 @classmethod def __get_cancel_datas(cls, code): key = "average_big_num_comput_info-{}".format(code) val = cls.__getRedis().smembers(key) return val @classmethod def __clear_data(cls, code): key = "average_big_num_comput_info-{}".format(code) cls.__getRedis().delete(key) key = "average_big_num-{}".format(code) cls.__getRedis().delete(key) @classmethod def clear_data(cls): key = "average_big_num_comput_info-*" keys = cls.__getRedis().keys(key) for k in keys: cls.__getRedis().delete(k) key = "average_big_num-*" keys = cls.__getRedis().keys(key) for k in keys: cls.__getRedis().delete(k) # 计算平均手数 # 计算范围：买入信号起始点到买入执行位的下一张图结束点数据为止 @classmethod def compute_average_big_num(cls, code, buy_single_index, start_index, end_index): print("compute_average_big_num", code, buy_single_index, start_index, end_index) L2TradeDataProcessor.cancel_debug(code, "开始计算短大单位置") total_data = local_today_datas[code] num = 0 count = 0 for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val) and float(val["price"]) * int(val["num"]) >= 5000: # 75万以上的才参与计算平均大单 count += data["re"] num += int(val["num"]) # 如果没有找到75万以上的单就不添加75w的筛选条件 if count == 0: for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val): # 75万以上的才参与计算平均大单 count += data["re"] num += int(val["num"]) average_num = num // count # average_num = 0 average_num = min(constant.BIG_MONEY_NUM, round(constant.BIG_MONEY_AMOUNT / gpcode_manager.get_limit_up_price(code))) average_up_count = 0 for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val): if int(val["num"]) >= average_num: average_up_count += data["re"] print("平均手数：", average_num, "大单总数：", average_up_count) # 保存数据 cls.__save_average_data(code, average_num, average_up_count, start_index, end_index) # 是否需要撤单 @classmethod def need_cancel(cls, code, buy_single_index, buy_exec_index, start_index, end_index, need_cancel=True): # 暂时取消此撤单条件 return False, None average_num, average_up_count, a_start_index, a_end_index = cls.__get_average_data(code) if average_num is None: return False, None total_data = local_today_datas[code] # 如果start_index与buy_single_index相同，即是下单后的第一次计算 # 需要查询买入信号之前的同1s是否有涨停撤的数据 if buy_single_index == start_index: for i in range(buy_single_index - 1, 0, -1): data = total_data[i] val = data["val"] if val["time"] != total_data[buy_single_index]["val"]["time"]: break if L2DataUtil.is_limit_up_price_buy_cancel(val) and int(val["cancelTime"]) == 0: # 涨停买撤销且撤销的间隔时间为0 # 查询买入信号，如果无法查询到或者是买入位置比买入信号小就不算 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and a_start_index <= buy_index <= a_end_index: # 在买入信号之后 cls.__save_cancel_data(code, i) for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] # print("处理进度", i) if L2DataUtil.is_limit_up_price_buy_cancel(val): # 查询买入位置 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and a_start_index <= buy_index <= a_end_index: cls.__save_cancel_data(code, i) if need_cancel: # 计算买撤大单暂比 cancel_datas = cls.__get_cancel_datas(code) if cancel_datas is not None and len(cancel_datas) > 0: cancel_rate_threshold = 0.49 place_order_count = trade_data_manager.placeordercountmanager.get_place_order_count(code) if place_order_count <= 1: cancel_rate_threshold = 0.49 elif place_order_count <= 2: cancel_rate_threshold = 0.549 else: cancel_rate_threshold = 0.59 cancel_indexs = [] for index in cancel_datas: cancel_indexs.append(int(index)) cancel_indexs.sort() # print("取消的数据", cancel_indexs) cancel_count = 0 for index in cancel_indexs: data = total_data[index] if int(data["val"]["num"]) >= average_num: cancel_count += data["re"] if cancel_count / average_up_count > cancel_rate_threshold: return True, total_data[index] return False, None # 是否需要计算 @classmethod def is_need_compute_average(cls, code, latest_data): total_datas = local_today_datas[code] data = cls.__place_order_time_dict.get(code) if data is None: return False, None, None elif tool.trade_time_sub(latest_data["val"]["time"], total_datas[data[2]]["val"]["time"]) < 3: # 3s内的数据才需要计算average return True, data[1], data[2] else: cls.__place_order_time_dict.pop(code) return False, None, None # 下单成功 @classmethod def place_order_success(cls, code, buy_single_index, buy_exec_index): cls.__clear_data(code) cls.__place_order_time_dict[code] = (t.time(), buy_single_index, buy_exec_index) # 以防万一，先保存下单信息 total_data = local_today_datas[code] cls.compute_average_big_num(code, buy_single_index, buy_single_index, total_data[-1]["index"]) @classmethod def __test(cls, datas): code = datas[0] load_l2_data(code) L2TradeDataProcessor.random_key[code] = 123123 # 先执行下单 buy_single_index = datas[1] buy_exec_index = datas[2] local_today_datas[code] = local_today_datas[code][0:datas[4]] cls.place_order_success(code, buy_single_index, buy_exec_index) # 执行是否需要计算average cls.compute_average_big_num(code, buy_single_index, buy_single_index, datas[3]) for i in range(buy_single_index, datas[4]): cancel, cancel_data = cls.need_cancel(code, i, i) if cancel: print("需要撤单", cancel, cancel_data["index"]) break @classmethod def test(cls): cls.__test(("000716", 410, 420, 461, 536)) # 代码 买入信号起始点 买入信息执行位置 计算末位 最远计算位置 # cls.__test(("002793", 292, 308, 314, 410)) # 执行是否需要撤销 # 平均大单计算 class LongAverageBigNumComputer: __redis_manager = redis_manager.RedisManager(0) @classmethod def __getRedis(cls): return cls.__redis_manager.getRedis() @classmethod def __save_average_data(cls, code, average_num, average_up_count, total_count, start_index, end_index): L2TradeDataProcessor.cancel_debug(code, "获取到长大单位置信息：平均手数-{} 大单数量-{} 样本数量-{} 计算开始范围-{}:{}".format(average_num, average_up_count, total_count, start_index, end_index)) key = "l_average_big_num-{}".format(code) cls.__getRedis().setex(key, tool.get_expire(), json.dumps((average_num, average_up_count, total_count, start_index, end_index))) @classmethod def __get_average_data(cls, code): key = "l_average_big_num-{}".format(code) val = cls.__getRedis().get(key) if val is None: return None, None, None, None, None val = json.loads(val) return val[0], val[1], val[2], val[3], val[4] @classmethod def __save_compute_info(cls, code, cancel_count, process_index): key = "l_average_big_num_comput_info-{}".format(code) cls.__getRedis().setex(key, tool.get_expire(), json.dumps((cancel_count, process_index))) @classmethod def __get_compute_info(cls, code): key = "l_average_big_num_comput_info-{}".format(code) val = cls.__getRedis().get(key) if val is None: return None, None val = json.loads(val) return val[0], val[1] @classmethod def __clear_data(cls, code): key = "l_average_big_num_comput_info-{}".format(code) cls.__getRedis().delete(key) key = "l_average_big_num-{}".format(code) cls.__getRedis().delete(key) # 计算平均手数 # 计算范围：买入信号起始点到买入执行位的下一张图结束点数据为止 @classmethod def compute_average_big_num(cls, code, buy_single_index, buy_exec_index): total_data = local_today_datas[code] latest_index = total_data[-1]["index"] end_index = total_data[-1]["index"] if end_index >= 434: print("测试") start_index = buy_exec_index if tool.trade_time_sub(total_data[end_index]["val"]["time"], total_data[buy_exec_index]["val"]["time"]) < 3: return exec_time = total_data[buy_exec_index]["val"]["time"] o_average_num, o_average_up_count, o_count, o_start_index, o_start_index = cls.__get_average_data(code) if o_average_num is not None and o_count >= constant.H_CANCEL_BUY_COUNT: return # 获取买入执行位后2s的数据末位 for i in range(end_index, buy_exec_index, - 1): time_ = total_data[i]["val"]["time"] if tool.trade_time_sub(time_, exec_time) <= 2: end_index = i break num = 0 count = 0 for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val) and int(val["num"]) * float(val["price"]) >= 5900: count += data["re"] num += int(val["num"]) * data["re"] # 如果小于30笔，需要再往后计算 if count < constant.H_CANCEL_BUY_COUNT: for i in range(end_index + 1, latest_index + 1, 1): data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy(val) and int(val["num"]) * float(val["price"]) >= 5900: count += data["re"] num += int(val["num"]) * data["re"] if count >= constant.H_CANCEL_BUY_COUNT: end_index = i break # 获取大单数量 average_up_count = 0 average_num = round(num / count) for i in range(start_index, end_index + 1): data = total_data[i] val = data["val"] if int(val["num"]) >= average_num: average_up_count += data["re"] # 保存数据 cls.__save_average_data(code, average_num, average_up_count, count, start_index, end_index) cls.__save_compute_info(code, 0, buy_exec_index) # 是否需要撤单 @classmethod def need_cancel(cls, code, buy_exec_index, start_index, end_index): average_num, average_up_count, total_count, a_start_index, a_end_index = cls.__get_average_data(code) if average_num is None: return False, None cancel_count, process_index = cls.__get_compute_info(code) total_data = local_today_datas[code] # 14:30过后不再守护 if int(total_data[end_index]["val"]["time"].replace(":", "")) > int("143000"): return False, None try: for i in range(min(start_index, buy_exec_index + 1), end_index + 1): if i <= buy_exec_index: continue if process_index >= i: continue data = total_data[i] val = data["val"] if L2DataUtil.is_limit_up_price_buy_cancel(val) and int(val["num"]) >= average_num: # 查询买入位置 buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(data, local_today_num_operate_map.get( code)) if buy_index is not None and a_start_index <= buy_index <= a_end_index: # 买入位置要在平均值计算范围内 cancel_count += data["re"] process_index = i sj = 0 # 5 * tool.trade_time_sub(val["time"],total_data[buy_exec_index]["val"]["time"]) print("h平均大单计算结果:", "取消数量", cancel_count, "大单总数", average_up_count, sj) if cancel_count / (average_up_count - sj) >= 0.75: return True, i finally: cls.__save_compute_info(code, cancel_count, process_index) return False, None # 下单成功 @classmethod def place_order_success(cls, code, buy_single_index, buy_exec_index): cls.__clear_data(code) @classmethod def __test(cls, datas): code = datas[0] load_l2_data(code) L2TradeDataProcessor.random_key[code] = random.randint(0, 100000) # 先执行下单 buy_single_index = datas[1] buy_exec_index = datas[2] cls.__clear_data(code) cls.place_order_success(code, buy_single_index, buy_exec_index) # 执行是否需要计算average cls.compute_average_big_num(code, buy_single_index, buy_exec_index) for i in range(buy_exec_index + 1, datas[4]): cancel, index = cls.need_cancel(code, buy_exec_index, i, i) if cancel: print("需要撤单", cancel, index) break @classmethod def test(cls): # 代码 买入信号起始点 买入信息执行位置 计算末位 最远计算位置 cls.__test(("002793", 292, 308, 332, 410)) # 执行是否需要撤销 if __name__ == "__main__": # trade_manager.start_cancel_buy("000637") # t.sleep(10) # AverageBigNumComputer.test() # LongAverageBigNumComputer.test() # L2TradeDataProcessor.test() SecondAverageBigNumComputer.test() # load_l2_data("600213") # # buy_index, buy_data = l2_data_util.get_buy_data_with_cancel_data(local_today_datas["600213"][84], # local_today_num_operate_map.get( # "600213")) # print(buy_index, buy_data)