""" 交易数据股那里器 用于对交易临时数据（交易状态，代码状态等）进行管理 """ import json import time # 交易撤销数据管理器 import constant from db.redis_manager_delegate import RedisUtils from utils import global_util, tool import l2_data_util from db import redis_manager_delegate as redis_manager from log_module.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: __db = 0 redisManager = redis_manager.RedisManager(0) buy_sure_position_dict = {} __buy_position_info_cache = {} __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(TradeBuyDataManager, cls).__new__(cls, *args, **kwargs) cls.__load_datas() return cls.__instance @classmethod def __get_redis(cls): return cls.redisManager.getRedis() @classmethod def __load_datas(cls): __redis = cls.__get_redis() try: keys = RedisUtils.keys(__redis, "buy_position_info-*") for k in keys: code = k.split("-")[-1] val = RedisUtils.get(__redis, k) val = json.loads(val) tool.CodeDataCacheUtil.set_cache(cls.__buy_position_info_cache, code, val) finally: RedisUtils.realse(__redis) # 设置买入点的信息 # trade_time: 买入点截图时间与下单提交时间差值 # capture_time: 买入点截图时间 # last_data: 买入点最后一条数据 def set_buy_position_info(self, code, capture_time, trade_time, last_data, last_data_index): val = (capture_time, trade_time, last_data, last_data_index) tool.CodeDataCacheUtil.set_cache(self.__buy_position_info_cache, code, val) RedisUtils.setex_async(self.__db, "buy_position_info-{}".format(code), tool.get_expire(), json.dumps(val)) # 获取买入点信息 def get_buy_position_info(self, code): val_str = RedisUtils.get(self.redisManager.getRedis(), "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] def get_buy_position_info_cache(self, code): cache_result = tool.CodeDataCacheUtil.get_cache(self.__buy_position_info_cache, code) if cache_result[0]: return cache_result[1] return None, None, None, None # 删除买入点信息 def remove_buy_position_info(self, code): tool.CodeDataCacheUtil.clear_cache(self.__buy_position_info_cache, code) RedisUtils.delete_async(self.__db, "buy_position_info-{}".format(code)) # 设置买入确认点信息 def __set_buy_sure_position(self, code, index, data): logger_trade.debug("买入确认点信息： code:{} index:{} data:{}", code, index, data) key = "buy_sure_position-{}".format(code) RedisUtils.setex(self.redisManager.getRedis(), key, tool.get_expire(), json.dumps((index, data))) self.buy_sure_position_dict[code] = (index, data) # 移除下单信号的详细信息 self.remove_buy_position_info(code) # 清除买入确认点信息 def __clear_buy_sure_position(self, code): key = "buy_sure_position-{}".format(code) RedisUtils.delete(self.redisManager.getRedis(), key) if code in self.buy_sure_position_dict: self.buy_sure_position_dict.pop(code) # 获取买入确认点信息 def get_buy_sure_position(self, code): temp = self.buy_sure_position_dict.get(code) if temp is not None: return temp[0], temp[1] key = "buy_sure_position-{}".format(code) val = RedisUtils.get(self.redisManager.getRedis(), key) if val is None: return None, None else: val = json.loads(val) self.buy_sure_position_dict[code] = (val[0], val[1]) return val[0], val[1] # 处理买入确认点信息 def process_buy_sure_position_info(self, code, capture_time, l2_today_datas, l2_latest_data, l2_add_datas): buy_capture_time, trade_time, l2_data, l2_data_index = self.get_buy_position_info_cache(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] self.__set_buy_sure_position(code, index, data) break else: self.__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] self.__set_buy_sure_position(code, index, data) else: # 间隔时间小于0 ，一般产生原因是数据回溯产生，故不做处理 logger_trade.warning("预估委托位置错误：数据间隔时间小于0 code-{}", code) pass # 代码实时价格管理器 class CodeActualPriceProcessor: __under_water_last_time_cache = {} __code_current_rate_cache = {} __code_current_rate_latest = {} __db = 0 __redisManager = redis_manager.RedisManager(0) __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(CodeActualPriceProcessor, cls).__new__(cls, *args, **kwargs) cls.__load_datas() return cls.__instance @classmethod def __get_redis(cls): return cls.__redisManager.getRedis() @classmethod def __load_datas(cls): __redis = cls.__get_redis() try: keys = RedisUtils.keys(__redis, "under_water_last_time-*") for k in keys: code = k.split("-")[-1] val = RedisUtils.get(__redis, k) tool.CodeDataCacheUtil.set_cache(cls.__under_water_last_time_cache, code, val) keys = RedisUtils.keys(__redis, "code_current_rate-*") for k in keys: code = k.split("-")[-1] val = RedisUtils.get(__redis, k) tool.CodeDataCacheUtil.set_cache(cls.__code_current_rate_cache, code, float(val)) except Exception as e: pass finally: RedisUtils.realse(__redis) # 保存跌价的时间 def __save_down_price_time(self, code, time_str): key = "under_water_last_time-{}".format(code) tool.CodeDataCacheUtil.set_cache(self.__under_water_last_time_cache, code, time_str) RedisUtils.setex(self.__get_redis(), key, tool.get_expire(), time_str) def __remove_down_price_time(self, code): key = "under_water_last_time-{}".format(code) tool.CodeDataCacheUtil.clear_cache(self.__under_water_last_time_cache, code) RedisUtils.delete(self.__get_redis(), key) def __get_last_down_price_time(self, code): key = "under_water_last_time-{}".format(code) return RedisUtils.get(self.__get_redis(), key) def __get_last_down_price_time_cache(self, code): cache_result = tool.CodeDataCacheUtil.get_cache(self.__under_water_last_time_cache, code) if cache_result[0]: return cache_result[1] return None def __increment_down_price_time(self, code, seconds): key = "under_water_seconds-{}".format(code) RedisUtils.incrby( self.__get_redis(), key, seconds) # 设置个失效时间 RedisUtils.expire(self.__get_redis(), key, tool.get_expire()) def __get_down_price_time_as_seconds(self, code): key = "under_water_seconds-{}".format(code) val = RedisUtils.get(self.__get_redis(), key) if val is None: return None else: return int(val) # 清除所有的水下捞数据 def clear_under_water_data(self): key = "under_water_*" keys = RedisUtils.keys(self.__get_redis(), key) for k in keys: RedisUtils.delete(self.__get_redis(), k) def __save_current_price_codes_count(self, count): key = "current_price_codes_count" RedisUtils.setex(self.__get_redis(), key, 10, count) def __get_current_price_codes_count(self): key = "current_price_codes_count" count = RedisUtils.get(self.__get_redis(), key) return 0 if count is None else count # 保存当前涨幅 def __save_current_rate(self, code, rate): # 变化之后才会持久化 if self.__code_current_rate_latest.get(code) == rate: return self.__code_current_rate_latest[code] = rate tool.CodeDataCacheUtil.set_cache(self.__code_current_rate_cache, code, rate) key = "code_current_rate-{}".format(code) RedisUtils.setex_async(self.__db, key, tool.get_expire(), rate) # 批量保存 def __save_current_rates(self, datas): # 变化之后才会持久化 for d in datas: if self.__code_current_rate_latest.get(d[0]) == d[1]: continue self.__code_current_rate_latest[d[0]] = d[1] tool.CodeDataCacheUtil.set_cache(self.__code_current_rate_cache, d[0], d[1]) key = "code_current_rate-{}".format(d[0]) RedisUtils.setex_async(self.__db, key, tool.get_expire(), d[1]) # 获取当前涨幅 def __get_current_rate(self, code): key = "code_current_rate-{}".format(code) rate = RedisUtils.get(self.__get_redis(), key) if rate is not None: return float(rate) return None def get_current_rate(self, code): cache_result = tool.CodeDataCacheUtil.get_cache(self.__code_current_rate_cache, code) if cache_result[0]: return cache_result[1] 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 = tool.get_now_time_str() if rate >= 0: down_start_time = self.__get_last_down_price_time_cache(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_cache(code) is None: self.__save_down_price_time(code, time_str) # datas:[(代码,比例)] def process_rates(self, datas, time_str): # 9点半之前的数据不处理 if int(time_str.replace(":", "")) < int("093000"): return # 保存目前的代码涨幅 self.__save_current_rates(datas) # now_str = tool.get_now_time_str() for d in datas: code, rate = d[0], d[1] if rate >= 0: down_start_time = self.__get_last_down_price_time_cache(code) if down_start_time is None: continue 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_cache(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, now_time=None): time_seconds = self.__get_down_price_time_as_seconds(code) if time_seconds is None: return False else: if time_seconds >= constant.UNDER_WATER_PRICE_TIME_AS_SECONDS: if now_time is None: now_time = tool.get_now_time_str() space = tool.trade_time_sub(now_time, "09:30:00") if space > 0 and time_seconds / space >= 0.2: return True return False # 当前代码是否涨停 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 = RedisUtils.keys(self.__get_redis(), 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: __db = 0 __redisManager = redis_manager.RedisManager(0) __place_order_count_cache = {} __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(PlaceOrderCountManager, cls).__new__(cls, *args, **kwargs) cls.__load_datas() return cls.__instance @classmethod def __get_redis(cls): return cls.__redisManager.getRedis() @classmethod def __load_datas(cls): redis_ = cls.__get_redis() try: keys = RedisUtils.keys(redis_, "place_order_count-*") for k in keys: code = k.split("-")[-1] count = RedisUtils.get(redis_, k) cls.__place_order_count_cache[code] = int(count) finally: RedisUtils.realse(redis_) def __incre_place_order_count(self, code): if code not in self.__place_order_count_cache: self.__place_order_count_cache[code] = 0 self.__place_order_count_cache[code] += 1 key = "place_order_count-{}".format(code) RedisUtils.incrby_async(self.__db, key, 1) RedisUtils.expire_async(self.__db, key, tool.get_expire()) def __get_place_order_count(self, code): key = "place_order_count-{}".format(code) count = RedisUtils.get(self.__get_redis(), key) if count is not None: return int(count) return 0 def __get_place_order_count_cache(self, code): cache_result = tool.CodeDataCacheUtil.get_cache(self.__place_order_count_cache, code) if cache_result[0]: return cache_result[1] return 0 def place_order(self, code): self.__incre_place_order_count(code) def get_place_order_count(self, code): return self.__get_place_order_count_cache(code) def clear_place_order_count(self, code): self.__place_order_count_cache[code] = 0 key = "place_order_count-{}".format(code) RedisUtils.delete_async(self.__db, key) def clear(self): self.__place_order_count_cache.clear() keys = RedisUtils.keys(self.__get_redis(), "place_order_count-*") for k in keys: RedisUtils.delete(self.__get_redis(), k) if __name__ == "__main__": processor = CodeActualPriceProcessor() print(processor.get_top_rate_codes(30))