""" 交易数据股那里器 用于对交易临时数据（交易状态，代码状态等）进行管理 """ import datetime import json import time # 交易撤销数据管理器 import constant import global_util import l2_data_util import redis_manager import tool from log import logger_trade class TradeCancelDataManager: capture_time_dict = {} # 保存截图时间 @classmethod def save_l2_capture_time(cls, client_id, pos, code, capture_time): cls.capture_time_dict["{}-{}-{}".format(client_id, pos, code)] = {"create_time": round(time.time() * 1000), "capture_time": capture_time} # 获取最近一次的截图时间 @classmethod def get_latest_l2_capture_time(cls, client_id, pos, code): val = cls.capture_time_dict.get("{}-{}-{}".format(client_id, pos, code)) if val is None: return -1 # 间隔时间不能大于1s if round(time.time() * 1000) - val["create_time"] > 1000: return -1 return val["capture_time"] # 获取l2数据的增长速度 @classmethod def get_l2_data_grow_speed(cls, client_id, pos, code, add_datas, capture_time): count = 0 for data in add_datas: count += data["re"] lastest_capture_time = cls.get_latest_l2_capture_time(client_id, pos, code) if lastest_capture_time < 0: raise Exception("获取上次l2数据截图时间出错") return count / (capture_time - lastest_capture_time) # 获取买入确认点的位置 @classmethod def get_buy_sure_position(cls, index, speed, trade_time): return index + round(speed * trade_time) class TradeBuyDataManager: redisManager = redis_manager.RedisManager(0) buy_sure_position_dict = {} # 设置买入点的信息 # trade_time: 买入点截图时间与下单提交时间差值 # capture_time: 买入点截图时间 # last_data: 买入点最后一条数据 @classmethod def set_buy_position_info(cls, code, capture_time, trade_time, last_data, last_data_index): redis = cls.redisManager.getRedis() redis.setex("buy_position_info-{}".format(code), tool.get_expire(), json.dumps((capture_time, trade_time, last_data, last_data_index))) # 获取买入点信息 @classmethod def get_buy_position_info(cls, code): redis = cls.redisManager.getRedis() val_str = redis.get("buy_position_info-{}".format(code)) if val_str is None: return None, None, None, None else: val = json.loads(val_str) return val[0], val[1], val[2], val[3] # 删除买入点信息 @classmethod def remove_buy_position_info(cls, code): redis = cls.redisManager.getRedis() redis.delete("buy_position_info-{}".format(code)) # 设置买入确认点信息 @classmethod def __set_buy_sure_position(cls, code, index, data): logger_trade.debug("买入确认点信息： code:{} index:{} data:{}", code, index, data) redis = cls.redisManager.getRedis() key = "buy_sure_position-{}".format(code) redis.setex(key, tool.get_expire(), json.dumps((index, data))) cls.buy_sure_position_dict[code] = (index, data) # 移除下单信号的详细信息 cls.remove_buy_position_info(code) # 清除买入确认点信息 @classmethod def __clear_buy_sure_position(cls, code): redis = cls.redisManager.getRedis() key = "buy_sure_position-{}".format(code) redis.delete(key) if code in cls.buy_sure_position_dict: cls.buy_sure_position_dict.pop(code) # 获取买入确认点信息 @classmethod def get_buy_sure_position(cls, code): temp = cls.buy_sure_position_dict.get(code) if temp is not None: return temp[0], temp[1] redis = cls.redisManager.getRedis() key = "buy_sure_position-{}".format(code) val = redis.get(key) if val is None: return None, None else: val = json.loads(val) cls.buy_sure_position_dict[code] = (val[0], val[1]) return val[0], val[1] # 处理买入确认点信息 @classmethod def process_buy_sure_position_info(cls, code, capture_time, l2_today_datas, l2_latest_data, l2_add_datas): buy_capture_time, trade_time, l2_data, l2_data_index = cls.get_buy_position_info(code) if buy_capture_time is None: # 没有购买者信息 return None if capture_time - buy_capture_time < trade_time: # 时间未等待足够 return None # 时间差是否相差2s及以上 old_time = l2_data["val"]["time"] new_time = l2_latest_data["val"]["time"] old_time_int = l2_data_util.get_time_as_seconds(old_time) new_time_int = l2_data_util.get_time_as_seconds(new_time) if new_time_int - old_time_int >= 2: # 间隔2s及其以上表示数据异常 # 间隔2s以上的就以下单时间下一秒末尾作为确认点 start_index = l2_data_index if len(l2_today_datas) - 1 > start_index: for i in range(start_index + 1, len(l2_today_datas)): _time = l2_today_datas[i]["val"]["time"] if l2_data_util.get_time_as_seconds(_time) - old_time_int >= 2: index = i - 1 data = l2_today_datas[index] cls.__set_buy_sure_position(code, index, data) break else: cls.__set_buy_sure_position(code, l2_data_index, l2_data) elif new_time_int - old_time_int >= 0: # 间隔2s内表示数据正常,将其位置设置为新增数据的中间位置 index = len(l2_today_datas) - 1 - (len(l2_add_datas)) // 2 data = l2_today_datas[index] cls.__set_buy_sure_position(code, index, data) else: # 间隔时间小于0 ，一般产生原因是数据回溯产生，故不做处理 logger_trade.warning("预估委托位置错误：数据间隔时间小于0 code-{}", code) pass # 代码实时价格管理器 class CodeActualPriceProcessor: __redisManager = redis_manager.RedisManager(0) def __get_redis(self): return self.__redisManager.getRedis() # 保存跌价的时间 def __save_down_price_time(self, code, time_str): key = "under_water_last_time-{}".format(code) self.__get_redis().setex(key, tool.get_expire(), time_str) def __remove_down_price_time(self, code): key = "under_water_last_time-{}".format(code) self.__get_redis().delete(key) def __get_last_down_price_time(self, code): key = "under_water_last_time-{}".format(code) return self.__get_redis().get(key) def __increment_down_price_time(self, code, seconds): key = "under_water_seconds-{}".format(code) self.__get_redis().incrby(key, seconds) # 设置个失效时间 self.__get_redis().expire(key, tool.get_expire()) def __get_down_price_time_as_seconds(self, code): key = "under_water_seconds-{}".format(code) val = self.__get_redis().get(key) if val is None: return None else: return int(val) # 清除所有的水下捞数据 def clear_under_water_data(self): key = "under_water_*" keys = self.__get_redis().keys(key) for k in keys: self.__get_redis().delete(k) def __save_current_price_codes_count(self, count): key = "current_price_codes_count" self.__get_redis().setex(key, 10, count) def __get_current_price_codes_count(self): key = "current_price_codes_count" count = self.__get_redis().get(key) return 0 if count is None else count # 保存当前涨幅 def __save_current_rate(self, code, rate): key = "code_current_rate-{}".format(code) self.__get_redis().setex(key, tool.get_expire(), rate) # 获取当前涨幅 def __get_current_rate(self, code): key = "code_current_rate-{}".format(code) rate = self.__get_redis().get(key) if rate is not None: return float(rate) return None def process_rate(self, code, rate, time_str): # 保存目前的代码涨幅 self.__save_current_rate(code, rate) # 9点半之前的数据不处理 if int(time_str.replace(":", "")) < int("093000"): return # now_str = datetime.datetime.now().strftime("%H:%M:%S") if rate >= 0: down_start_time = self.__get_last_down_price_time(code) if down_start_time is None: return else: # 累计增加时间 time_second = tool.trade_time_sub(time_str, down_start_time) self.__increment_down_price_time(code, time_second) # 删除起始时间 self.__remove_down_price_time(code) else: # 记录开始值 if self.__get_last_down_price_time(code) is None: self.__save_down_price_time(code, time_str) # 保存现价 def save_current_price(self, code, price, is_limit_up): global_util.cuurent_prices[code] = (price, is_limit_up, round(time.time())) pass # 获取现价 def get_current_price(self, code): return global_util.cuurent_prices.get(code) # 现价代码数量 def save_current_price_codes_count(self, count): self.__save_current_price_codes_count(count) def get_current_price_codes_count(self): return self.__get_current_price_codes_count() # 是否为水下捞 def is_under_water(self, code): time_seconds = self.__get_down_price_time_as_seconds(code) if time_seconds is None: return False else: return time_seconds >= constant.UNDER_WATER_PRICE_TIME_AS_SECONDS # 当前代码是否涨停 def current_is_limit_up(self, code): data = self.get_current_price(code) if data is None: return None return data[1] # 获取涨幅前几的代码 def get_top_rate_codes(self, top_n): keys = "code_current_rate-*" keys = self.__get_redis().keys(keys) infos = [] for k in keys: code = k.split("-")[1] rate = self.__get_current_rate(code) infos.append((code, rate)) # 排序信息 sorted_infos = sorted(infos, key=lambda tup: tup[1], reverse=True) sorted_infos = sorted_infos[:top_n] codes = [] for data in sorted_infos: codes.append(data[0]) return codes # 涨停次数管理 class placeordercountmanager: __redisManager = redis_manager.RedisManager(0) @classmethod def __get_redis(cls): return cls.__redisManager.getRedis() @classmethod def __incre_place_order_count(cls, code): key = "place_order_count-{}".format(code) cls.__get_redis().incrby(key, 1) cls.__get_redis().expire(key, tool.get_expire()) @classmethod def __get_place_order_count(cls, code): key = "place_order_count-{}".format(code) count = cls.__get_redis().get(key) if count is not None: return int(count) return 0 @classmethod def place_order(cls, code): cls.__incre_place_order_count(code) @classmethod def get_place_order_count(cls, code): return cls.__get_place_order_count(code) if __name__ == "__main__": processor = CodeActualPriceProcessor() print(processor.get_top_rate_codes(30))