""" 股票代码管理器 """ import json import time from db import redis_manager as redis_manager from db.redis_manager import RedisUtils from utils import tool import decimal __redisManager = redis_manager.RedisManager(0) class CodesNameManager: redisManager = redis_manager.RedisManager(0) __gp_list_names_first_cache = [] @classmethod def __get_redis(cls): return cls.redisManager.getRedis() @classmethod def list_code_name_dict(cls): dict_ = {} val = cls.list_first_code_name_dict() if val is not None: for k in val: dict_[k] = val[k] val = cls.list_second_code_name_dict() if val is not None: for k in val: dict_[k] = val[k] return dict_ @classmethod def list_first_code_name_dict(cls): val = RedisUtils.get(cls.__get_redis(), "gp_list_names_first") if val is not None: val = json.loads(val) return val return None @classmethod def list_first_code_name_dict_cache(cls): val = RedisUtils.get(cls.__get_redis(), "gp_list_names_first") if val is not None: val = json.loads(val) return val return None @classmethod def get_first_code_name(cls, code): val = RedisUtils.get(cls.__get_redis(), "gp_list_names_first") if not val: return None val = json.loads(val) for k in val: if val[k] == code: return k return None @classmethod def get_first_name_code(cls, name): val = RedisUtils.get(cls.__get_redis(), "gp_list_names_first") if not val: return None val = json.loads(val) return val.get(name) @classmethod def add_first_code_name(cls, code, name): val = RedisUtils.get(cls.__get_redis(), "gp_list_names_first") if not val: return None val = json.loads(val) val[name] = code cls.set_first_code_names(val) # 设置首板代码名称 @classmethod def set_first_code_names(cls, datas): RedisUtils.set(cls.__get_redis(), "gp_list_names_first", json.dumps(datas)) # 删除首板代码名称 @classmethod def clear_first_code_names(cls): RedisUtils.delete(cls.__get_redis(), "gp_list_names_first") @classmethod def list_second_code_name_dict(cls): val = RedisUtils.get(cls.__get_redis(), "gp_list_names") if val is not None: val = json.loads(val) return val return None @classmethod def get_second_code_name(cls, code): val = RedisUtils.get(cls.__get_redis(), "gp_list_names") if not val: return None val = json.loads(val) for k in val: if val[k] == code: return k @classmethod def get_second_name_code(cls, name): val = RedisUtils.get(cls.__get_redis(), "gp_list_names") if not val: return None val = json.loads(val) return val.get(name) # 设置二板代码名称 @classmethod def set_second_code_names(cls, datas): RedisUtils.set(cls.__get_redis(), "gp_list_names", json.dumps(datas)) # 设置二板代码名称 @classmethod def clear_second_code_names(cls): RedisUtils.delete(cls.__get_redis(), "gp_list_names") # 首板代码管理 class FirstCodeManager: __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(FirstCodeManager, cls).__new__(cls, *args, **kwargs) # 初始化设置 # 获取交易窗口的锁 cls.__instance.redisManager = redis_manager.RedisManager(0) cls.__instance.__first_code_record_cache = RedisUtils.smembers(cls.__instance.__get_redis(), "first_code_record") cls.__instance.__first_code_limited_up_record_cache = RedisUtils.smembers(cls.__instance.__get_redis(), "first_code_limited_up_record") return cls.__instance def __get_redis(self): return self.redisManager.getRedis() # 加入首板历史记录 def add_record(self, codes): hasChanged = False for code in codes: if code not in self.__first_code_record_cache: RedisUtils.sadd(self.__get_redis(), "first_code_record", code) hasChanged = True self.__first_code_record_cache.add(code) if hasChanged: RedisUtils.expire(self.__get_redis(), "first_code_record", tool.get_expire()) def is_in_first_record(self, code): if RedisUtils.sismember(self.__get_redis(), "first_code_record", code): return True else: return False def is_in_first_record_cache(self, code): return code in self.__first_code_record_cache # 加入首板涨停过代码集合 def add_limited_up_record(self, codes): hasChanged = False for code in codes: if code not in self.__first_code_limited_up_record_cache: RedisUtils.sadd(self.__get_redis(), "first_code_limited_up_record", code) hasChanged = True self.__first_code_limited_up_record_cache.add(code) if hasChanged: RedisUtils.expire(self.__get_redis(), "first_code_limited_up_record", tool.get_expire()) # 是否涨停过 def is_limited_up(self, code): if RedisUtils.sismember(self.__get_redis(), "first_code_limited_up_record", code): return True else: return False def is_limited_up_cache(self, code): return code in self.__first_code_limited_up_record_cache # 想要买的代码 class WantBuyCodesManager: __instance = None redisManager = redis_manager.RedisManager(0) __redis_key = "want_buy_codes" def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(WantBuyCodesManager, cls).__new__(cls, *args, **kwargs) # 初始化设置 # 获取交易窗口的锁 cls.__instance.__want_buy_codes_cache = RedisUtils.smembers(cls.__get_redis(), cls.__redis_key) return cls.__instance __want_buy_codes_cache = set() @classmethod def __get_redis(cls): return cls.redisManager.getRedis() def clear(self): RedisUtils.delete(self.__get_redis(), self.__redis_key) def add_code(self, code): self.__want_buy_codes_cache.add(code) RedisUtils.sadd(self.__get_redis(), self.__redis_key, code) RedisUtils.expire(self.__get_redis(), self.__redis_key, tool.get_expire()) def remove_code(self, code): self.__want_buy_codes_cache.discard(code) RedisUtils.srem(self.__get_redis(), self.__redis_key, code) def sync(self): codes = self.list_code() self.__want_buy_codes_cache.clear() if codes: self.__want_buy_codes_cache |= set(codes) def is_in(self, code): return RedisUtils.sismember(self.__get_redis(), self.__redis_key, code) def is_in_cache(self, code): return code in self.__want_buy_codes_cache def list_code(self): return RedisUtils.smembers(self.__get_redis(), self.__redis_key) def list_code_cache(self): return self.__want_buy_codes_cache # 暂停下单代码管理 # 与黑名单的区别是暂停交易代码只是不交易，不能移除L2监控位 class PauseBuyCodesManager: __instance = None redisManager = redis_manager.RedisManager(0) __redis_key = "pause_buy_codes" def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(PauseBuyCodesManager, cls).__new__(cls, *args, **kwargs) cls.__instance.__pause_buy_codes_cache = RedisUtils.smembers(cls.__get_redis(), cls.__redis_key) return cls.__instance @classmethod def __get_redis(cls): return cls.redisManager.getRedis() def clear(self): self.__pause_buy_codes_cache.clear() RedisUtils.delete(self.__get_redis(), self.__redis_key) def sync(self): data = RedisUtils.smembers(self.__get_redis(), self.__redis_key) self.__pause_buy_codes_cache.clear() if data: self.__pause_buy_codes_cache |= data def add_code(self, code): self.__pause_buy_codes_cache.add(code) RedisUtils.sadd(self.__get_redis(), self.__redis_key, code) RedisUtils.expire(self.__get_redis(), self.__redis_key, tool.get_expire()) def remove_code(self, code): self.__pause_buy_codes_cache.discard(code) RedisUtils.srem(self.__get_redis(), self.__redis_key, code) def is_in(self, code): return RedisUtils.sismember(self.__get_redis(), self.__redis_key, code) def is_in_cache(self, code): return code in self.__pause_buy_codes_cache def list_code(self): return RedisUtils.smembers(self.__get_redis(), self.__redis_key) def list_code_cache(self): return self.__pause_buy_codes_cache def __parse_codes_data(code_datas): codes = [] name_codes = {} for _data in code_datas: # 正常的股票 if _data["sec_type"] == 1 and _data["sec_level"] == 1: code = _data["symbol"].split(".")[1] if code.find("30") != 0 and code.find("68") != 0: name = _data["sec_name"] codes.append(code) # 保存代码对应的名称 name_codes[name] = code return codes, name_codes # -------------------------------二板代码管理--------------------------------- def set_gp_list(code_datas): codes, name_codes = __parse_codes_data(code_datas) redis_instance = __redisManager.getRedis() try: # 删除之前的 RedisUtils.delete(redis_instance, "gp_list", auto_free=False) CodesNameManager.clear_second_code_names() for d in codes: RedisUtils.sadd(redis_instance, "gp_list", d, auto_free=False) CodesNameManager.set_second_code_names(name_codes) finally: RedisUtils.realse(redis_instance) # 新增代码 def add_gp_list(code_datas): if len(code_datas) > 200: raise Exception("不能超过200个数据") redis_instance = __redisManager.getRedis() try: codes, name_codes = __parse_codes_data(code_datas) for d in codes: RedisUtils.sadd(redis_instance, "gp_list", d, auto_free=False) old_name_codes = CodesNameManager.list_second_code_name_dict() if old_name_codes is None: old_name_codes = name_codes else: for key in name_codes: old_name_codes[key] = name_codes[key] CodesNameManager.set_second_code_names(old_name_codes) finally: RedisUtils.realse(redis_instance) # -------------------------------首板代码管理------------------------------- class FirstGPCodesManager: __db = 0 __redisManager = redis_manager.RedisManager(0) __gp_list_first_codes_cache = set() __instance = None def __new__(cls, *args, **kwargs): if not cls.__instance: cls.__instance = super(FirstGPCodesManager, 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: cls.__gp_list_first_codes_cache = RedisUtils.smembers(__redis, "gp_list_first") finally: RedisUtils.realse(__redis) @classmethod def __parse_codes_data(cls, code_datas): codes = [] name_codes = {} for _data in code_datas: # 正常的股票 if _data["sec_type"] == 1 and _data["sec_level"] == 1: code = _data["symbol"].split(".")[1] if code.find("30") != 0 and code.find("68") != 0: name = _data["sec_name"] codes.append(code) # 保存代码对应的名称 name_codes[name] = code return codes, name_codes # 添加首板代码 # code_datas 掘金返回的数据 def set_first_gp_codes_with_data(self, code_datas): redis_instance = self.__get_redis() try: codes, name_codes = self.__parse_codes_data(code_datas) codes_set = set() for code in codes: codes_set.add(code) old_codes_set = self.__gp_list_first_codes_cache if old_codes_set is None: old_codes_set = set() del_set = old_codes_set - codes_set add_codes = codes_set - old_codes_set for code in add_codes: RedisUtils.sadd(redis_instance, "gp_list_first", code, auto_free=False) for code in del_set: RedisUtils.srem(redis_instance, "gp_list_first", code, auto_free=False) if add_codes or del_set: RedisUtils.expire(redis_instance, "gp_list_first", tool.get_expire(), auto_free=False) # 更新缓存 self.__gp_list_first_codes_cache.clear() self.__gp_list_first_codes_cache |= codes_set old_name_codes = CodesNameManager.list_first_code_name_dict() if old_name_codes is None: old_name_codes = name_codes else: for key in name_codes: old_name_codes[key] = name_codes[key] CodesNameManager.set_first_code_names(old_name_codes) finally: RedisUtils.realse(redis_instance) # 移除首板代码 def remove_first_gp_code(self, codes): redis_instance = self.__get_redis() try: for code in codes: self.__gp_list_first_codes_cache.discard(code) RedisUtils.srem(redis_instance, "gp_list_first", code, auto_free=False) finally: RedisUtils.realse(redis_instance) # 获取首板代码 def get_first_gp_codes(self): return RedisUtils.smembers(self.__get_redis(), "gp_list_first") def get_first_gp_codes_cache(self): return self.__gp_list_first_codes_cache # 是否在首板里面 def is_in_first_gp_codes(self, code): return RedisUtils.sismember(self.__get_redis(), "gp_list_first", code) # 是否在首板里面 def is_in_first_gp_codes_cache(self, code): return code in self.__gp_list_first_codes_cache # 获取名称对应的代码 def get_name_code(name): code = CodesNameManager.get_second_name_code(name) if code is not None: return code code = CodesNameManager.get_first_name_code(name) return code # 代码名字缓存 __code_name_dict = {} # 获取代码的名称 def get_code_name(code): if code in __code_name_dict: return __code_name_dict.get(code) name = CodesNameManager.get_second_code_name(code) if name is not None: __code_name_dict[code] = name return name name = CodesNameManager.get_first_code_name(code) if name: __code_name_dict[code] = name return name def get_name_codes(): return CodesNameManager.list_code_name_dict() # 涨停数据保存 def set_limit_up_list(gpset): if gpset is None: return # 获取基本信息 redis_instance = __redisManager.getRedis() try: # 删除之前的 RedisUtils.delete(redis_instance, "gp_limit_up_list", auto_free=False) for d in gpset: RedisUtils.sadd(redis_instance, "gp_limit_up_list", json.dumps(d), auto_free=False) RedisUtils.expire(redis_instance, "gp_limit_up_list", tool.get_expire(), auto_free=False) RedisUtils.setex(redis_instance, "gp_limit_up_list_update_time", tool.get_expire(), round(time.time() * 1000), auto_free=False) finally: RedisUtils.realse(redis_instance) # 获取涨停列表 def get_limit_up_list(): redis_instance = __redisManager.getRedis() try: return RedisUtils.get(redis_instance, "gp_limit_up_list_update_time", auto_free=False), RedisUtils.smembers(redis_instance, "gp_limit_up_list", auto_free=False) finally: RedisUtils.realse(redis_instance) def rm_gp(code): RedisUtils.srem(__redisManager.getRedis(), "gp_list", code) FirstGPCodesManager().remove_first_gp_code([code]) def is_in_gp_pool(code): return RedisUtils.sismember(__redisManager.getRedis(), "gp_list", code) or FirstGPCodesManager().is_in_first_gp_codes_cache(code) def get_gp_list(): codes = RedisUtils.smembers(__redisManager.getRedis(), "gp_list") first_codes = FirstGPCodesManager().get_first_gp_codes() return set.union(codes, first_codes) # 获取二板代码 def get_second_gp_list(): codes = RedisUtils.smembers(__redisManager.getRedis(), "gp_list") return codes def get_gp_list_with_prefix(data=None): if data is None: data = get_gp_list() list = [] for d in data: if d[0:2] == '00': list.append("SZSE.{}".format(d)) elif d[0:2] == '60': list.append("SHSE.{}".format(d)) return list class CodePrePriceManager: __price_pre_cache = {} __redisManager = redis_manager.RedisManager(0) # 获取收盘价 @classmethod def get_price_pre(cls, code): result = RedisUtils.get(cls.__redisManager.getRedis(), "price-pre-{}".format(code)) if result is not None: return float(result) return None # 获取缓存 @classmethod def get_price_pre_cache(cls, code): if code in cls.__price_pre_cache: return float(cls.__price_pre_cache[code]) val = cls.get_price_pre(code) if val: cls.__price_pre_cache[code] = val return val # 设置收盘价 @classmethod def set_price_pre(cls, code, price, force=False): codes = get_gp_list() if code not in codes and not FirstCodeManager().is_in_first_record_cache(code) and not force: return RedisUtils.setex(cls.__redisManager.getRedis(), "price-pre-{}".format(code), tool.get_expire(), str(price)) cls.__price_pre_cache[code] = float(price) __limit_up_price_dict = {} # 获取涨停价 def get_limit_up_price(code): # 读取内存中的值 if code in __limit_up_price_dict: return __limit_up_price_dict[code] price = CodePrePriceManager.get_price_pre_cache(code) if price is None: return None limit_up_price = tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal("1.1")) __limit_up_price_dict[code] = limit_up_price return limit_up_price def get_limit_up_price_by_preprice(price): if price is None: return None return tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal("1.1")) # 获取跌停价 def get_limit_down_price(code): price = CodePrePriceManager.get_price_pre_cache(code) if price is None: return None return tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal("0.9")) # 获取现价 def get_price(code): result = RedisUtils.get(__redisManager.getRedis(), "price-{}".format(code)) if result is not None: return float(result) return None __current_price_cache = {} # 设置现价 def set_price(code, price): if code in __current_price_cache and __current_price_cache[code] == price: return __current_price_cache[code] = price RedisUtils.setex(__redisManager.getRedis(), "price-{}".format(code), tool.get_expire(), price) # datas:[(code,price)] def set_prices(datas): pipe = __redisManager.getRedis().pipeline() for d in datas: code, price = d[0], d[1] if code in __current_price_cache and __current_price_cache[code] == price: continue __current_price_cache[code] = price RedisUtils.setex(pipe, "price-{}".format(code), tool.get_expire(), price) pipe.execute() # 获取正在监听的代码 def get_listen_codes(): redis_instance = __redisManager.getRedis() try: keys = RedisUtils.keys(redis_instance, "listen_code-*-*", auto_free=False) codes = set() for k in keys: code = RedisUtils.get(redis_instance, k, auto_free=False) if code is not None and len(code) > 0: codes.add(code) return codes finally: RedisUtils.realse(redis_instance) # 根据位置获取正在监听的代码 def get_listen_code_by_pos(client_id, pos): key = "listen_code-{}-{}".format(client_id, pos) value = RedisUtils.get(__redisManager.getRedis(), key) # print("redis:", key,value) return value # 设置位置的监听代码 def set_listen_code_by_pos(client_id, pos, code): RedisUtils.setex(__redisManager.getRedis(), "listen_code-{}-{}".format(client_id, pos), tool.get_expire(), code) # 同步监听的代码集合 __sync_listen_codes_pos() # 同步监听代码位置信息 def __sync_listen_codes_pos(): redis_instance = __redisManager.getRedis() try: # 获取已经正在监听的代码 keys = RedisUtils.keys(redis_instance, "code_listen_pos-*", auto_free=False) codes_set = set() for key in keys: codes_set.add(key.replace("code_listen_pos-", "")) keys = RedisUtils.keys(redis_instance, "listen_code-*-*", auto_free=False) for key in keys: result = RedisUtils.get(redis_instance, key, auto_free=False) if result: # 移除需要添加的代码 codes_set.discard(result) client_id_, pos_, code_ = int(key.split("-")[1]), int(key.split("-")[2]), result key_ = "code_listen_pos-{}".format(code_) val = RedisUtils.get(redis_instance, key_, auto_free=False) if val is None: RedisUtils.setex(redis_instance, key_, tool.get_expire(), json.dumps((client_id_, pos_)), auto_free=False) else: val = json.loads(val) if val[0] != client_id_ or val[1] != pos_: RedisUtils.setex(redis_instance, key_, tool.get_expire(), json.dumps((client_id_, pos_)), auto_free=False) # 移除没有监听的代码 for code_ in codes_set: RedisUtils.delete(redis_instance, code_, auto_free=False) finally: RedisUtils.realse(redis_instance) # 初始化位置 def init_listen_code_by_pos(client_id, pos): key = "listen_code-{}-{}".format(client_id, pos) RedisUtils.setnx(__redisManager.getRedis(), key, "") RedisUtils.expire(__redisManager.getRedis(), key, tool.get_expire()) # 清除所有监听代码 def clear_listen_codes(): redis_instance = __redisManager.getRedis() try: keys = RedisUtils.keys(redis_instance, "listen_code-*-*", auto_free=False) for key in keys: RedisUtils.setex(redis_instance, key, tool.get_expire(), "", auto_free=False) finally: RedisUtils.realse(redis_instance) def clear_first_codes(): redis_instance = __redisManager.getRedis() try: RedisUtils.delete(redis_instance, "gp_list_first", auto_free=False) RedisUtils.delete(redis_instance, "gp_list_names_first", auto_free=False) RedisUtils.delete(redis_instance, "first_code_record", auto_free=False) RedisUtils.delete(redis_instance, "first_code_limited_up_record", auto_free=False) finally: RedisUtils.realse(redis_instance) # 获取正在监听的代码的位置 def get_listen_code_pos(code): val = RedisUtils.get(__redisManager.getRedis(), "code_listen_pos-{}".format(code)) if val is None: return None, None val = json.loads(val) cid, pid = val[0], val[1] code_ = get_listen_code_by_pos(cid, pid) # 校验代码 if code_ == code: return cid, pid else: return None, None # 是否正在监听 def is_listen(code): val = RedisUtils.get(__redisManager.getRedis(), "code_listen_pos-{}".format(code)) if val is None: return False else: return True # codes = get_listen_codes() # return codes.__contains__(code) def is_listen_old(code): codes = get_listen_codes() return codes.__contains__(code) # 是否正在操作 def is_operate(code): return RedisUtils.get(__redisManager.getRedis(), "gp_operate-{}".format(code)) is not None # 设置正在操作的代码 def set_operate(code): RedisUtils.setex(__redisManager.getRedis(), "gp_operate-{}".format(code), 30, "1") # 批量设置正在操作的代码 def set_operates(codes): for code in codes: RedisUtils.setex(__redisManager.getRedis(), "gp_operate-{}".format(code), 30, "1") # 移除正在操作的代码 def rm_operate(code): RedisUtils.delete(__redisManager.getRedis(), "gp_operate-{}".format(code)) # 批量移除正在操作的代码 def rm_operates(codes): redis_instance = __redisManager.getRedis() try: for code in codes: RedisUtils.delete(redis_instance, "gp_operate-{}".format(code), auto_free=False) finally: RedisUtils.realse(redis_instance)