""" 成交量管理 """ # 设置历史量 # max60 60天最大量 # yesterday 昨天的量 import json from code_attribute import gpcode_manager from db.redis_manager_delegate import RedisUtils from log_module import async_log_util from utils import global_util, tool from db import redis_manager_delegate as redis_manager from log_module.log import logger_day_volumn class CodeVolumeManager: __db = 0 __redis_manager = redis_manager.RedisManager(0) __instance = None __today_volumn_cache = {} __reference_volume_as_money_y_dict = {} __max_volume_in_5days = {} def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(CodeVolumeManager, cls).__new__(cls, *args, **kwargs) cls.__load_data() return cls.__instance # 将量从数据库加入内存 @classmethod def __load_data(cls): redis = cls.__redis_manager.getRedis() try: keys = RedisUtils.keys(redis, "volumn_max60-*", auto_free=False) if keys is not None: for k in keys: code = k.split("-")[1] max60_volumn = RedisUtils.get(redis, k, auto_free=False) if max60_volumn: max60_volumn = json.loads(max60_volumn) global_util.max60_volumn[code] = max60_volumn keys = RedisUtils.keys(redis, "volumn_yes-*", auto_free=False) if keys is not None: for k in keys: code = k.split("-")[1] global_util.yesterday_volumn[code] = RedisUtils.get(redis, k, auto_free=False) keys = RedisUtils.keys(redis, "volumn_max_5days-*", auto_free=False) if keys is not None: for k in keys: code = k.split("-")[1] val = RedisUtils.get(redis, k, auto_free=False) cls.__max_volume_in_5days[code] = int(val) finally: RedisUtils.realse(redis) # 设置历史量 def set_histry_volumn(self, code, max60, yesterday, max60_day, max60_day_count, max5_volume): redis = self.__redis_manager.getRedis() global_util.max60_volumn[code] = (max60, max60_day, max60_day_count) global_util.yesterday_volumn[code] = yesterday self.__save_max_volume_in_5days(code, max5_volume) try: RedisUtils.setex_async(self.__db, "volumn_max60-{}".format(code), tool.get_expire(), json.dumps((max60, max60_day, max60_day_count)), auto_free=False) RedisUtils.setex_async(self.__db, "volumn_yes-{}".format(code), tool.get_expire(), yesterday, auto_free=False) finally: RedisUtils.realse(redis) def __save_max_volume_in_5days(self, code, volume): self.__max_volume_in_5days[code] = volume RedisUtils.setex_async(self.__db, "volumn_max_5days-{}".format(code), tool.get_expire(), volume, auto_free=False) def get_max_volume_in_5days(self, code): return self.__max_volume_in_5days.get(code) # 获取历史量 def get_histry_volumn(self, code): max60 = global_util.max60_volumn.get(code) yesterday = global_util.yesterday_volumn.get(code) redis = self.__redis_manager.getRedis() try: if max60 is None: max60 = RedisUtils.get(redis, "volumn_max60-{}".format(code), auto_free=False) if max60: max60 = json.loads(max60) if yesterday is None: yesterday = RedisUtils.get(redis, "volumn_yes-{}".format(code), auto_free=False) return max60, yesterday finally: RedisUtils.realse(redis) # 量的变化大保存 # 设置今日量 def set_today_volumn(self, code, volumn): async_log_util.info(logger_day_volumn, "code:{} volumn:{}".format(code, volumn)) global_util.today_volumn[code] = volumn # 有1000手的变化才保存 if code in self.__today_volumn_cache and volumn - self.__today_volumn_cache[code] < 100000: return self.__today_volumn_cache[code] = volumn RedisUtils.setex(self.__redis_manager.getRedis(), "volumn_today-{}".format(code), tool.get_expire(), volumn) # datas:[(code, volumn)] def set_today_volumns(self, datas): for d in datas: code, volumn = d async_log_util.info(logger_day_volumn, "code:{} volumn:{}".format(code, volumn)) global_util.today_volumn[code] = volumn # 有1000手的变化才保存 if code in self.__today_volumn_cache and volumn - self.__today_volumn_cache[code] < 100000: continue self.__today_volumn_cache[code] = volumn RedisUtils.setex_async(self.__db, "volumn_today-{}".format(code), tool.get_expire(), volumn) # 获取今日量 def get_today_volumn(self, code): _volumn = global_util.today_volumn.get(code) if _volumn is None: _volumn = RedisUtils.get(self.__redis_manager.getRedis(), "volumn_today-{}".format(code)) return _volumn # 获取今日量 def get_today_volumn_cache(self, code): return global_util.today_volumn.get(code) # 获取量比（今日量/max(60天最大量,昨日量)） # 将总卖量计算在内 def get_volume_rate(self, code, total_sell_volume=0, with_info=False): today = self.get_today_volumn(code) max60, yesterday = self.get_histry_volumn(code) if today is None: today = 0 if max60 is None or yesterday is None: max60 = [today, ''] yesterday = today if max60[0] < 1: max60[0] = 1 rate = round((int(today) + total_sell_volume) / max(int(max60[0]), int(yesterday)), 2) if not with_info: return rate return rate, (today, max(int(max60[0]), int(yesterday))) # 获取量参考日期 # 返回(参考量日期,距今的交易日个数) def get_volume_refer_date(self, code): max60, yesterday = self.get_histry_volumn(code) if max60 is None or yesterday is None: raise Exception("获取失败") if int(max60[0]) >= int(yesterday): return max60[1], max60[2] else: return "上个交易日", 0 # 获取量比索引 def get_volume_rate_index(self, volume_rate): rates = [0.2, 0.4, 0.6, 0.8, 1, 1.2, 1.4, 1.6] for index in range(0, len(rates)): if volume_rate <= rates[index]: return index return len(rates) - 1 # 保存今日量 def save_today_volumn(self, code, volumn, volumnUnit): _volumn = None if volumnUnit == 0: _volumn = round(float(volumn) * 100000000) elif volumnUnit == 1: _volumn = round(float(volumn) * 10000) elif volumnUnit == 2: _volumn = int(volumn) if _volumn is not None: self.set_today_volumn(code, _volumn * 100) def get_reference_volume_as_money_y(self, code): """ 返回参考量今日对应的金额（单位为亿） @param code: @return: """ if code in self.__reference_volume_as_money_y_dict: return self.__reference_volume_as_money_y_dict.get(code) max60, yesterday = self.get_histry_volumn(code) if max60: num = max60[0] limit_up_price = gpcode_manager.get_limit_up_price(code) if limit_up_price: money_y = round((num * float(limit_up_price)) / 1e8, 1) self.__reference_volume_as_money_y_dict[code] = money_y return money_y # 默认为5亿 return 5 if __name__ == "__main__": print(CodeVolumeManager().get_volume_rate("000059"))