添加托管交易通道测试

Administrator

2023-08-22 a933bf8e8e7f0c5a5131a95ed5d0f0b58bf9ac34

 third_data/code_plate_key_manager.py

@@ -21,22 +21,39 @@
class KPLCodeJXBlockManager:
    __redisManager = redis_manager.RedisManager(3)
    __code_blocks = {}
    # 备用
    __code_by_blocks = {}

    def __get_redis(self):
        return self.__redisManager.getRedis()

    def save_jx_blocks(self, code, blocks):
    def save_jx_blocks(self, code, blocks, by=False):
        if blocks is None:
            return
        if len(blocks) >2:
            blocks = blocks[:2]

        # 保存前2条数据
        if by:
            RedisUtils.setex(self.__get_redis(), f"kpl_jx_blocks_by-{code}", tool.get_expire(), json.dumps(blocks))
            self.__code_by_blocks[code] = blocks
        else:
        RedisUtils.setex(self.__get_redis(), f"kpl_jx_blocks-{code}", tool.get_expire(), json.dumps(blocks))
        self.__code_blocks[code] = blocks

    # 获取精选板块
    def get_jx_blocks(self, code):
    def get_jx_blocks(self, code, by=False):
        if by:
            if code in self.__code_by_blocks:
                return self.__code_by_blocks[code]
            val = RedisUtils.get(self.__get_redis(), f"kpl_jx_blocks_by-{code}")
            if val is None:
                return None
            else:
                val = json.loads(val)
                self.__code_by_blocks[code] = val
            return self.__code_by_blocks[code]
        else:
        if code in self.__code_blocks:
            return self.__code_blocks[code]
        val = RedisUtils.get(self.__get_redis(), f"kpl_jx_blocks-{code}")
@@ -47,7 +64,10 @@
            self.__code_blocks[code] = val
        return self.__code_blocks[code]

    def get_jx_blocks_cache(self, code):
    def get_jx_blocks_cache(self, code, by=False):
        if by:
            return self.__code_by_blocks.get(code)
        else:
        return self.__code_blocks.get(code)


@@ -367,6 +387,8 @@

        k4 = set()
        jingxuan_blocks = self.__KPLCodeJXBlockManager.get_jx_blocks_cache(code)
        if not jingxuan_blocks:
            jingxuan_blocks = self.__KPLCodeJXBlockManager.get_jx_blocks_cache(code, by=True)
        if jingxuan_blocks:
            jingxuan_blocks = jingxuan_blocks[:2]
            k4 |= set([x[1] for x in jingxuan_blocks])