"""
|
新题材处理器
|
"""
|
import constant
|
from code_attribute import code_nature_analyse
|
from third_data import kpl_util, kpl_data_constant
|
from trade import l2_trade_util, trade_record_log_util
|
from trade.buy_radical.block_special_codes_manager import BlockSpecialCodesManager
|
from utils import tool, global_util
|
from utils.kpl_data_db_util import KPLLimitUpDataUtil
|
|
# 新题材的最少涨停个数
|
__MIN_LIMIT_UP_COUNT = 1
|
|
|
@tool.singleton
|
class BeforeBlocksComputer:
|
"""
|
往日题材计算器
|
"""
|
|
def __init__(self):
|
self.__before_blocks = set()
|
self.__load_data()
|
|
def __load_data(self):
|
# kpl_results: [( r.`_code`, r.`_day`, r.`_hot_block_name`, r.`_blocks`, r.`_open`)]
|
kpl_results = KPLLimitUpDataUtil.get_latest_block_infos()
|
days = set()
|
for r in kpl_results:
|
# 取60个交易日之内的题材
|
days.add(r[1])
|
days = list(days)
|
days.sort(key=lambda x: int(x.replace("-", "")), reverse=True)
|
days = days[:60]
|
# {"日期":{ "题材":{代码} }}
|
day_block_codes_dict = {}
|
for r in kpl_results:
|
if r[1] not in days:
|
continue
|
if r[4]:
|
# 不算炸板
|
continue
|
# 统计每一天的题材
|
day = r[1]
|
block = kpl_util.filter_block(r[2])
|
if block in constant.KPL_INVALID_BLOCKS:
|
continue
|
if day not in day_block_codes_dict:
|
day_block_codes_dict[day] = {}
|
if block not in day_block_codes_dict[day]:
|
day_block_codes_dict[day][block] = set()
|
day_block_codes_dict[day][block].add(r[0])
|
|
# 每一天走的板块
|
day_block_dict = {}
|
for day in day_block_codes_dict:
|
for b in day_block_codes_dict[day]:
|
if len(day_block_codes_dict[day][b]) >= 3:
|
if day not in day_block_dict:
|
day_block_dict[day] = set()
|
day_block_dict[day].add(b)
|
day_block_list = [(d, day_block_dict[d]) for d in day_block_dict]
|
day_block_list.sort(key=lambda x: x[0])
|
# 过去59天出现的题材
|
old_blocks = set()
|
for d in day_block_list:
|
if d[0] == days[0]:
|
continue
|
old_blocks |= d[1]
|
self.__before_blocks |= old_blocks
|
|
# 昨天出现的题材
|
# if days[0] == day_block_list[-1][0]:
|
# # 昨天出现的题材
|
# yesterday_blocks = day_block_list[-1][1]
|
|
def is_new_block(self, block):
|
return block not in self.__before_blocks
|
|
def get_old_blocks(self):
|
return self.__before_blocks
|
|
|
def screen_new_blocks_with_limit_up_datas(limit_up_datas):
|
"""
|
从涨停代码筛选出新题材
|
@param limit_up_datas:[(代码,涨停原因)]
|
@return:{"新题材":{新题材的代码}}
|
"""
|
block_codes_dict = {}
|
for d in limit_up_datas:
|
b = kpl_util.filter_block(d[1])
|
if b not in block_codes_dict:
|
block_codes_dict[b] = set()
|
block_codes_dict[b].add(d[0])
|
|
blocks = set()
|
for b in block_codes_dict:
|
if b in constant.KPL_INVALID_BLOCKS:
|
continue
|
if len(block_codes_dict[b]) >= __MIN_LIMIT_UP_COUNT:
|
# 涨停代码数大于2个的原因
|
blocks.add(b)
|
old_blocks = BeforeBlocksComputer().get_old_blocks()
|
# 新题材
|
new_blocks = blocks - old_blocks
|
return {x: block_codes_dict[x] for x in new_blocks}
|
|
|
def __is_can_add_new_block(code):
|
"""
|
是否可以添加到新板块
|
@param code:
|
@return:
|
"""
|
zyltgb = global_util.zyltgb_map.get(code)
|
if zyltgb and zyltgb < 10e8:
|
return False
|
k_format = code_nature_analyse.CodeNatureRecordManager().get_k_format_cache(code)
|
if not constant.TEST:
|
if not k_format or not k_format[1][0]:
|
return False
|
return True
|
|
|
def process_limit_up_list(code_block_dict):
|
# 删除变了的原因
|
for code in code_block_dict:
|
if kpl_data_constant.limit_up_code_new_block_dict.get(code, code_block_dict.get(code)) != code_block_dict.get(
|
code):
|
# 删除变之前的原因
|
if code in kpl_data_constant.limit_up_code_new_block_dict:
|
kpl_data_constant.limit_up_code_new_block_dict.pop(code)
|
__sync_data_to_special_codes(code)
|
|
|
def process_new_block_by_limit_up_list(code, block):
|
"""
|
根据涨停列表处理新题材
|
@param code_block_dict: 代码涨停原因字典
|
@param code:
|
@param block:
|
@return:
|
"""
|
# 不用删除炸板代码
|
if __is_can_add_new_block(code):
|
if kpl_data_constant.limit_up_code_new_block_dict.get(code, "") != block:
|
# 题材有变化
|
kpl_data_constant.limit_up_code_new_block_dict[code] = block
|
__sync_data_to_special_codes(code)
|
blocks = set()
|
for k in kpl_data_constant.limit_up_code_new_block_dict:
|
blocks.add(kpl_data_constant.limit_up_code_new_block_dict[k])
|
kpl_data_constant.new_blocks = blocks
|
|
|
def process_new_block_by_component_codes(block, codes, all_new_blocks):
|
"""
|
根据成分股处理新题材
|
@param block: 板块
|
@param codes: 成分股
|
@param all_new_blocks: 所有的新题材
|
@return:
|
"""
|
# 清除这个新题材下的代码
|
for c in kpl_data_constant.limit_up_component_code_new_blocks_dict:
|
if block in kpl_data_constant.limit_up_component_code_new_blocks_dict[c] and c not in codes:
|
# 原来的代码中包含该板块，而当前板块的辨识度不包含该代码
|
kpl_data_constant.limit_up_component_code_new_blocks_dict[c].discard(block)
|
|
# 清除不是新题材的代码
|
for c in kpl_data_constant.limit_up_component_code_new_blocks_dict:
|
# 原来的板块
|
diff = kpl_data_constant.limit_up_component_code_new_blocks_dict[c] - all_new_blocks
|
if diff:
|
for d in diff:
|
kpl_data_constant.limit_up_component_code_new_blocks_dict[c].discard(d)
|
|
# 在代码中加入新题材
|
for c in codes:
|
if __is_can_add_new_block(c):
|
if c not in kpl_data_constant.limit_up_component_code_new_blocks_dict:
|
kpl_data_constant.limit_up_component_code_new_blocks_dict[c] = set()
|
kpl_data_constant.limit_up_component_code_new_blocks_dict[c].add(block)
|
__sync_data_to_special_codes(c)
|
|
|
def __sync_data_to_special_codes(code):
|
"""
|
同步数据到辨识度
|
@param code:
|
@return:
|
"""
|
blocks = set()
|
if code in kpl_data_constant.limit_up_component_code_new_blocks_dict:
|
blocks |= kpl_data_constant.limit_up_component_code_new_blocks_dict[code]
|
if code in kpl_data_constant.limit_up_code_new_block_dict:
|
blocks.add(kpl_data_constant.limit_up_code_new_block_dict[code])
|
if blocks and l2_trade_util.is_in_forbidden_trade_codes(code):
|
l2_trade_util.remove_from_forbidden_trade_codes(code)
|
# 加想买单要从黑名单移除
|
trade_record_log_util.add_common_msg(code, "新题材移黑", f"{blocks}")
|
# 新板块
|
if constant.TEST:
|
print(code, kpl_data_constant.limit_up_code_new_block_dict.get(code),
|
kpl_data_constant.limit_up_component_code_new_blocks_dict.get(code))
|
BlockSpecialCodesManager().set_code_blocks_for_temp(code, blocks)
|
|
|
def is_can_forbidden(code):
|
"""
|
是否可以拉黑
|
@param code:
|
@return:
|
"""
|
if kpl_data_constant.has_new_block(code):
|
# 有新题材
|
k_format = code_nature_analyse.CodeNatureRecordManager().get_k_format_cache(code)
|
if k_format and k_format[1][0]:
|
# 突破形态
|
return False
|
return True
|
