import datetime import json import logging import random import time as t import big_money_num_manager import code_data_util import global_util import gpcode_manager import l2_data_log import l2_data_manager import l2_data_util import l2_trade_factor import l2_trade_test import limit_up_time_manager import redis_manager import ths_industry_util import tool import trade_data_manager import trade_manager import trade_queue_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 # 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() @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) now_time_str = datetime.datetime.now().strftime("%H:%M:%S") __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) if len(add_datas) > 0: # 拼接数据 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) # 第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"] # 时间差不能太大才能处理 # TODO 暂时关闭处理 if l2_data_manager.L2DataUtil.is_same_time(now_time_str, latest_time): # 判断是否已经挂单 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: # 已挂单 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) __start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - __start_time, "l2数据处理时间") # 保存数据 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_not_order(cls, code, start_index, end_index, capture_time): __start_time = t.time() # 获取阈值 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): if start_index < 0: start_index = 0 if end_index < start_index: return # 获取买入信号起始点 buy_single_index, buy_exec_index, buy_compute_index, num, count = cls.__get_order_begin_pos(code) # 撤单计算,只看买1 cancel_data, cancel_msg = L2LimitUpMoneyStatisticUtil.process_data(code, start_index, end_index, buy_single_index) # 计算m值大单 cls.l2BigNumForMProcessor.process(code, max(buy_single_index, start_index), end_index, gpcode_manager.get_limit_up_price(code)) if cancel_data: cls.debug(code, "触发撤单，撤单位置：{} ，撤单原因：{}", cancel_data["index"], cancel_msg) # 撤单 cls.cancel_buy(code, cancel_msg) # 继续计算下单 cls.__process_not_order(code, cancel_data["index"] + 1, end_index, capture_time) 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]) @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) 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): # 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 = limit_up_time_manager.sort_code_by_limit_time(codes) 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.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 = limit_up_time_manager.sort_code_by_limit_time(codes) 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涨停，老二不能下单" # 过时 老二，本板块中涨停票数<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: cls.debug(code, "执行撤单异常：{}", str(e)) @classmethod def cancel_buy(cls, code, msg=None): can_cancel, reason = cls.__can_cancel(code) if not can_cancel: # 不能取消 cls.cancel_debug(code, "撤单中断，原因：{}", reason) return 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: cls.__cancel_buy(code) l2_data_manager.L2BigNumProcessor.del_big_num_pos(code) cls.debug(code, "执行撤单成功，原因：{}", msg) @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 = t.time() total_datas = local_today_datas[code] # 获取买入信号计算起始位置 buy_single_index, buy_exec_index, buy_compute_index, num, count = cls.__get_order_begin_pos(code) # 是否为新获取到的位置 if buy_single_index is None: # 有买入信号 has_single, _index = cls.__compute_order_begin_pos(code, max( compute_start_index - 2 if new_add else compute_start_index, 0), 3, compute_end_index) buy_single_index = _index if has_single: num = 0 count = 0 cls.debug(code, "获取到买入信号起始点：{} 数据：{}", buy_single_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 = 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, capture_time) _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) # 如果是今天第一次有下单执行信号，涨停时间（买入执行位时间） limit_up_time_manager.save_limit_up_time(code, total_datas[compute_index]["val"]["time"]) # 虚拟下单 cls.unreal_buy_dict[code] = (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, False) _start_time = l2_data_log.l2_time(code, round(t.time() * 1000) - _start_time, "记录执行买入数据") # 数据是否处理完毕 if compute_index >= compute_end_index: cls.debug(code, "数据处理完毕，下单, 数据截图时间-{}", capture_time) # 数据已经处理完毕，如果还没撤单就实际下单 cls.__buy(code, capture_time, total_datas[compute_index], compute_index) else: # 数据尚未处理完毕，进行下一步处理 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) 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 = l2_data_manager.TradePointManager.get_buy_compute_start_data( code) return buy_single_index, buy_exec_index, compute_index, num, count # 保存下单起始信号 @classmethod def __save_order_begin_data(self, code, buy_single_index, buy_exec_index, compute_index, num, count): TradePointManager.set_buy_compute_start_data(code, buy_single_index, buy_exec_index, compute_index, num, count) # 计算下单起始信号 # 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"]) >= 7888: return True if int(val["num"]) * limit_up_price >= 29900: 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, capture_time): def get_threshold_count(): count = threshold_count - sub_threshold_count if count < 3: count = 3 return round(count*buy1_factor) _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"]) for i in range(compute_start_index, compute_end_index + 1): data = total_datas[i] _val = total_datas[i]["val"] # 必须为连续3秒内的数据 if L2DataUtil.get_time_as_second(_val["time"]) - buy_single_time_seconds > 2: TradePointManager.delete_buy_point(code) if i == compute_end_index: # 数据处理完毕 return None, buy_nums, buy_count, None 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 # 涨停买 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"]) # 涨停买 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"]) # 涨停买撤 # 判断买入位置是否在买入信号之前 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(): return i, buy_nums, buy_count, None 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 @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 = limit_up_time_manager.sort_code_by_limit_time(codes) 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涨停，老二不能下单" # 涨停封单额统计 class L2LimitUpMoneyStatisticUtil: _redisManager = redis_manager.RedisManager(1) @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, with_cancel=True): start_time = round(t.time() * 1000) total_datas = local_today_datas[code] time_dict_num = {} # 记录计算的坐标 time_dict_num_index = {} 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)) # 不同时间的数据开始坐标 time_start_index_dict = {} # 数据时间分布 time_list = [] # 到当前时间累积的买1量 time_total_num_dict = {} for i in range(index + 1, end_index + 1): data = total_datas[i] 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 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 val < 0 and start_index <= i <= end_index: # 累计封单金额小于1000万 if total_num < min_volumn: cancel_index = i cancel_msg = "封单金额小于1000万" 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 >= 0.5: # 相邻2s内的数据减小50% cancel_index = i cancel_msg = "相邻2s({})内的封单量减小50%({}->{})".format(time_, last_second_total_volumn, total_num) break 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 if __name__ == "__main__": L2TradeDataProcessor.test2() print("----------------------") # L2TradeDataProcessor.test()