"""
|
代码板块管理
|
"""
|
from local_api.log_module import async_log_util, log_export
|
from local_api.log_module.log import logger_kpl_block, printlog
|
from local_api.util import kpl_api, constant, kpl_util, tool
|
from local_api.util.juejin_util import JueJinApi
|
|
|
class CodeKPLBlockManager:
|
__instance = None
|
# 代码板块管理
|
__code_block_cache_dict = {}
|
# 上一个交易日的板块
|
__code_before_block_cache_dict = {}
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(CodeKPLBlockManager, cls).__new__(cls, *args, **kwargs)
|
cls.__load_datas()
|
return cls.__instance
|
|
@classmethod
|
def __load_datas(cls):
|
# 加载今日和上个交易日的数据
|
today = tool.get_now_date_str()
|
last_day = JueJinApi().get_previous_trading_date(today)
|
printlog("获取上个交易日", last_day)
|
block_dict = log_export.load_kpl_blocks(today)
|
for code in block_dict:
|
cls.__code_block_cache_dict[code] = block_dict[code]
|
|
block_dict = log_export.load_kpl_blocks(last_day)
|
for code in block_dict:
|
cls.__code_before_block_cache_dict[code] = block_dict[code]
|
|
def get_blocks(self, code):
|
"""
|
获取代码的板块，优先今日获取到的板块
|
:param code:
|
:return:
|
"""
|
blocks = self.__code_block_cache_dict.get(code)
|
if blocks:
|
return blocks
|
blocks = self.__code_before_block_cache_dict.get(code)
|
return blocks
|
|
|
def get_codes_count_info(self):
|
"""
|
获取代码数量信息
|
:return:（当日板块的代码数量,上个交易日的代码数量）
|
"""
|
return len(self.__code_block_cache_dict), len(self.__code_before_block_cache_dict)
|
|
def __request_block(self, code, force=False):
|
if code in self.__code_block_cache_dict and not force:
|
return
|
blocks = kpl_api.getCodeBlocks(code)
|
fblocks = set()
|
if blocks:
|
for b in blocks:
|
if b not in constant.KPL_INVALID_BLOCKS:
|
fblocks.add(kpl_util.filter_block(b))
|
self.__code_block_cache_dict[code] = fblocks
|
async_log_util.info(logger_kpl_block, f"{code}#{fblocks}")
|
|
def request_block(self, codes, force=False):
|
"""
|
请求板块
|
:param codes: 代码集合
|
:param force: 是否强制求情
|
:return:
|
"""
|
if type(codes) == str:
|
self.__request_block(codes, force)
|
else:
|
for code in codes:
|
self.__request_block(code, force)
|
