from log_module import log_export, async_log_util
|
from log_module.log import logger_stock_of_markets_plate, logger_debug, logger_kpl_market_sift_plate, \
|
logger_stock_of_markets_plate_simple
|
from utils import tool
|
|
|
# 开盘啦板块强度日志管理
|
@tool.singleton
|
class KPLMarketsSiftPlateLogManager:
|
"""
|
精选流入板块与板块代码日志管理
|
"""
|
|
def __init__(self):
|
# 过滤后的日志数据
|
self.__filter_log_datas = []
|
|
def __load_data(self):
|
if tool.get_now_time_str() > '13:10:00':
|
logger_debug.info("KPLMarketsSiftPlateLogManager 开始加载数据")
|
datas = log_export.load_market_sift_plate()
|
for data in datas:
|
# (发生时间,[净流入板块], {"板块":(代码, 名称, 涨幅)})
|
self.__filter_log_datas.append(self.__filter_origin_data(data))
|
logger_debug.info("KPLMarketsSiftPlateLogManager 数据加载完成：{}", len(self.__filter_log_datas))
|
|
def load_data(self):
|
self.__load_data()
|
|
def __filter_origin_data(self, data):
|
"""
|
:param data: 格式：(时间,[数据,])
|
:return:
|
"""
|
return data
|
|
def add_log(self, in_plates_info, plate_codes_info):
|
oringin_data = (tool.get_now_time_str(), in_plates_info, plate_codes_info)
|
async_log_util.info(logger_kpl_market_sift_plate, f"{(oringin_data[1], oringin_data[2])}")
|
|
def get_filter_log_datas(self):
|
return self.__filter_log_datas
|
|
|
# 开盘啦个股强度日志管理
|
@tool.singleton
|
class KPLMarketStockHeatLogManager:
|
"""
|
精选流入板块与板块代码日志管理
|
"""
|
|
def __init__(self):
|
# 过滤后的日志数据
|
self.__filter_log_datas = []
|
|
def __load_data(self):
|
if tool.get_now_time_str() > '13:10:00':
|
logger_debug.info("KPLStockOfMarketsPlateLogManager 开始加载数据")
|
datas = log_export.load_kpl_market_stock_heat()
|
for data in datas:
|
# (发生时间,[净流入板块], {"板块":(代码, 名称, 涨幅)})
|
self.__filter_log_datas.append(self.__filter_origin_data(data))
|
logger_debug.info("KPLStockOfMarketsPlateLogManager 数据加载完成：{}", len(self.__filter_log_datas))
|
|
def load_data(self):
|
self.__load_data()
|
|
def __filter_origin_data(self, data):
|
return (data[0], [x[1] for x in data[1][0]],
|
{p: [xx for xx in data[1][1][p]] for p in data[1][1]})
|
|
def add_log(self, in_plates_info, plate_codes_info):
|
oringin_data = (tool.get_now_time_str(), in_plates_info, plate_codes_info)
|
async_log_util.info(logger_stock_of_markets_plate, f"{(oringin_data[1], oringin_data[2])}")
|
plate_codes_simple_info = {plate: [(s[1], s[0], s[6], s[9], s[4], s[2], s[40]) for s in oringin_data[2][plate]] for plate in oringin_data[2]}
|
async_log_util.info(logger_stock_of_markets_plate_simple, f"{(oringin_data[1], plate_codes_simple_info)}")
|
|
def get_filter_log_datas(self):
|
return self.__filter_log_datas
|
|
|
if __name__ == '__main__':
|
# KPLMarketsSiftPlateLogManager().load_data()
|
# KPLMarketStockHeatLogManager()
|
with open("D:/text.txt", encoding='utf-8', mode='r') as f:
|
line = f.readline()
|
res = eval(line)
|
KPLMarketStockHeatLogManager().add_log(res[0], res[1])
|
print(res)
|
