"""
|
板块辨识度票管理
|
"""
|
import difflib
|
|
import constant
|
from db import mysql_data_delegate as mysql_data
|
from log_module.log import logger_plate_special_codes
|
from third_data import kpl_util
|
from third_data.history_k_data_util import HistoryKDatasUtils
|
from third_data.kpl_data_constant import LimitUpCodesBlockRecordManager
|
from utils import tool
|
|
|
@tool.singleton
|
class BlockSpecialCodesManager:
|
__block_codes_dict = {}
|
__code_blocks_dict = {}
|
|
__temp_block_codes_dict = {}
|
__temp_code_blocks_dict = {}
|
|
def __init__(self):
|
self.mysql = mysql_data.Mysqldb()
|
self.__load_data()
|
|
def __load_data(self):
|
results = self.mysql.select_all("select _block,_code from block_special_codes")
|
temp_dict = {}
|
for row in results:
|
block, code = row[0], row[1]
|
if block not in temp_dict:
|
temp_dict[block] = set()
|
if code not in self.__code_blocks_dict:
|
self.__code_blocks_dict[code] = set()
|
self.__code_blocks_dict[code].add(block)
|
temp_dict[block].add(code)
|
self.__block_codes_dict = temp_dict
|
# 保存辨识度的日志
|
try:
|
logger_plate_special_codes.info(f"{temp_dict}")
|
except:
|
pass
|
|
def get_block_codes(self, block):
|
return self.__block_codes_dict.get(block, set()) | self.__temp_block_codes_dict.get(block, set())
|
|
def get_code_blocks(self, code):
|
return self.__code_blocks_dict.get(code, set()) | self.__temp_code_blocks_dict.get(code, set())
|
|
def get_code_blocks_dict(self):
|
fdict = {}
|
for code in self.__code_blocks_dict:
|
fdict[code] = set(self.__code_blocks_dict.get(code))
|
|
for code in self.__temp_code_blocks_dict:
|
if code not in fdict:
|
fdict[code] = set()
|
fdict[code] |= self.__temp_code_blocks_dict.get(code)
|
|
return fdict
|
|
def get_temp_code_blocks_dict(self):
|
fdict = {}
|
for code in self.__temp_code_blocks_dict:
|
if code not in fdict:
|
fdict[code] = set()
|
fdict[code] |= self.__temp_code_blocks_dict.get(code)
|
return fdict
|
|
def set_code_blocks_for_temp(self, code, blocks):
|
"""
|
临时添加代码的辨识度板块
|
@param code: 带啊吗
|
@param blocks: 板块
|
@return:
|
"""
|
|
old_blocks = self.__temp_code_blocks_dict.get(code, set())
|
self.__temp_code_blocks_dict[code] = set(blocks)
|
|
add_blocks = blocks - old_blocks
|
del_blocks = old_blocks - blocks
|
if add_blocks:
|
for b in add_blocks:
|
if b not in self.__temp_block_codes_dict:
|
self.__temp_block_codes_dict[b] = set()
|
self.__temp_block_codes_dict[b].add(code)
|
if del_blocks:
|
for b in del_blocks:
|
if b not in self.__temp_block_codes_dict:
|
self.__temp_block_codes_dict[b] = set()
|
self.__temp_block_codes_dict[b].discard(code)
|
|
if add_blocks or del_blocks:
|
trade_record_log_util.add_temp_special_codes(code, f"新题材辨识度设置：{blocks}")
|
|
def set_block_codes_list(self, datas):
|
"""
|
设置数据
|
@param datas:[(板块,代码,代码名称,涨停次数,自由市值)]
|
@return:
|
"""
|
self.mysql.execute("delete from block_special_codes")
|
for d in datas:
|
sql = f"insert into block_special_codes(_block,_code,_code_name,_limit_up_count,_zyltgb,_create_time) values('{d[0]}', '{d[2]}', '{d[1]}', {d[3]}, {d[4]}, now())"
|
self.mysql.execute(sql)
|
|
|
class AnalysisBlockSpecialCodesManager:
|
__mysql = mysql_data.Mysqldb()
|
|
def __get_code_blocks(self, min_day, max_day):
|
results = self.__mysql.select_all(
|
f"SELECT r.`_hot_block_name`, r.`_code`, COUNT(*) FROM kpl_limit_up_record r WHERE r.`_day`>'{min_day}' and r.`_day`<'{max_day}' and r._code not like '68%' group by r.`_hot_block_name`, r.`_code`")
|
return results
|
|
def __list_code_blocks(self, min_day, max_day):
|
results = self.__mysql.select_all(
|
f"SELECT r.`_hot_block_name`, r.`_code`, r.`_day`, r.`_open` FROM kpl_limit_up_record r WHERE r.`_day`>'{min_day}' and r.`_day`<'{max_day}' and r._code not like '68%'")
|
return results
|
|
def __get_limit_up_info(self, min_day):
|
sql = f"SELECT r.`_code`, COUNT(r.`_code`),r.`_code_name`,IF( r.`_zylt_val` is null, 1, r.`_zylt_val`/100000000 ) FROM (SELECT * FROM kpl_limit_up_record r ORDER BY r.`_create_time` DESC) r WHERE r.`_day`>'{min_day}' GROUP BY r.`_code`"
|
results = self.__mysql.select_all(sql)
|
# {"代码":(涨停次数, 名称, 自由流通市值)}
|
return {x[0]: (x[1], x[2], x[3]) for x in results}
|
|
def __get_top(self, block, min_day, max_day):
|
sql = f"SELECT r.`_code`,r.`_code_name`, COUNT(*), IF( r.`_zylt_val` is null, 1, r.`_zylt_val`/100000000 ) FROM (SELECT * FROM kpl_limit_up_record r ORDER BY r.`_create_time` DESC) r WHERE r.`_hot_block_name`='{block}' AND r.`_day`>'{min_day}' and r.`_day`<'{max_day}' and r._code not like '68%' GROUP BY r.`_code`"
|
results = self.__mysql.select_all(sql)
|
results = list(results)
|
info_map = {x[1]: x for x in results}
|
max_count = min(10, len(results) // 2)
|
results.sort(key=lambda x: x[2], reverse=True)
|
names1 = set([x[0] for x in results[:max_count]])
|
results.sort(key=lambda x: x[3], reverse=True)
|
names2 = set([x[0] for x in results[:max_count]])
|
fnames = names1 & names2
|
return fnames
|
|
def __get_block_map(self):
|
"""
|
获取板块裂变
|
@return:
|
"""
|
constant.get_path_prefix()
|
with open(f"{constant.get_path_prefix()}/板块对应.txt", encoding="utf-8") as f:
|
lines = f.readlines()
|
block_map = {}
|
for line in lines:
|
line = line.strip()
|
if line:
|
line = line.replace("，", ",").replace("丨", "|")
|
parent_blocks = set()
|
if line.find("|") >= 0:
|
sts = line.split("|")
|
parent_blocks |= set(sts[0].split(","))
|
line = sts[1]
|
blocks = line.split(",")
|
if not blocks:
|
continue
|
if not parent_blocks:
|
parent_blocks.add(blocks[0])
|
for b in blocks:
|
b = b.strip()
|
if b not in block_map:
|
block_map[b] = set()
|
block_map[b] |= parent_blocks
|
if len(parent_blocks) > 1:
|
# parent加入
|
for b in parent_blocks:
|
if b not in block_map:
|
block_map[b] = set()
|
block_map[b].add(b)
|
return block_map
|
|
def test_block_map(self):
|
print(self.__get_block_map())
|
|
def get_block_special_codes(self):
|
"""
|
获取板块有辨识度的代码
|
@return:
|
"""
|
|
def statistic_continue_target_codes():
|
"""
|
统计连续涨停的辨识度代码代码
|
@return:
|
"""
|
dates = HistoryKDatasUtils.get_latest_trading_date_cache(120)
|
code_days_map = {}
|
for d in code_block_infos:
|
if d[3]:
|
# 不看炸板
|
continue
|
if d[1] not in code_days_map:
|
code_days_map[d[1]] = set()
|
|
code_days_map[d[1]].add(d[2])
|
target_codes_info = set()
|
for code in code_days_map:
|
# if code != '002400':
|
# continue
|
# 且有3天属于连续涨停
|
day_list = list(code_days_map[code])
|
day_list.sort(reverse=True)
|
step = 4
|
matcher = difflib.SequenceMatcher(None, dates, day_list)
|
match = matcher.find_longest_match(0, len(dates), 0, len(day_list))
|
if match.size < step:
|
continue
|
target_codes_info.add((code, match.size))
|
print(f"大于4板：{target_codes_info}")
|
|
continue_count_dict = {x[0]: x[1] for x in target_codes_info}
|
|
# 统计板块的代码
|
block_codes_2 = {}
|
for code_info in target_codes_info:
|
code = code_info[0]
|
bs = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code)
|
if bs:
|
for bb in bs:
|
if bb not in block_codes_2:
|
block_codes_2[bb] = set()
|
block_codes_2[bb].add(code)
|
for k in block_codes_2:
|
block_codes_2[k] = list(block_codes_2[k])
|
block_codes_2[k].sort(key=lambda x: continue_count_dict[x], reverse=True)
|
block_codes_2[k] = block_codes_2[k][:3]
|
|
return block_codes_2
|
|
trading_dates = HistoryKDatasUtils.get_latest_trading_date(9)
|
max_day = trading_dates[0]
|
min_day = tool.date_sub(max_day, 120)
|
|
block_map = self.__get_block_map()
|
# 统计最近120天涨停数据
|
# [(板块名称,代码, 日期, 是否炸板)]
|
code_block_infos = self.__list_code_blocks(min_day, max_day)
|
|
# 统计代码的涨停次数
|
limit_up_count_dict = {}
|
for d in code_block_infos:
|
if d[1] not in limit_up_count_dict:
|
limit_up_count_dict[d[1]] = set()
|
limit_up_count_dict[d[1]].add(d[2])
|
limit_up_count_dict = {c: len(limit_up_count_dict[c]) for c in limit_up_count_dict}
|
|
# 拥有连板高度的辨识度：{"板块":[代码1,代码2]}
|
continue_target_block_codes = statistic_continue_target_codes()
|
|
# 统计最近8天的涨停数据
|
min_day = trading_dates[-1]
|
code_block_infos_8 = self.__get_code_blocks(min_day, max_day)
|
# 统计最近8天涨停次数大于3次的涨停代码
|
# 格式：{"代码":涨停次数}
|
count_dict = {}
|
filter_codes = set()
|
for d in code_block_infos_8:
|
if d[1] not in count_dict:
|
count_dict[d[1]] = 0
|
count_dict[d[1]] += d[2]
|
if count_dict[d[1]] > 3:
|
filter_codes.add(d[1])
|
count_dict.clear()
|
code_block_infos_8.clear()
|
|
# 删除数据
|
temp_data_dict = {}
|
for d in code_block_infos:
|
if d[1] in filter_codes and int(d[2].replace("-", "")) > int(min_day.replace("-", "")):
|
# 需要过滤且最近8天涨停
|
continue
|
block, code = d[0], d[1]
|
k = f"{block}#{code}"
|
if k not in temp_data_dict:
|
temp_data_dict[k] = [block, code, 0]
|
temp_data_dict[k][2] += 1
|
code_block_infos = [temp_data_dict[k] for k in temp_data_dict]
|
|
min_day = tool.date_sub(max_day, 120)
|
|
code_block_dict = {} # {"代码":{"板块": 涨停次数}}
|
for b in code_block_infos:
|
if b[1] not in code_block_dict:
|
code_block_dict[b[1]] = {}
|
bs = block_map.get(b[0])
|
if not bs:
|
bs = {b[0]}
|
for bb in bs:
|
if bb not in code_block_dict[b[1]]:
|
code_block_dict[b[1]][bb] = 0
|
# if tool.is_ge_code(b[1]):
|
# # 创业板1次涨停算2次涨停
|
# code_block_dict[b[1]][bb] += b[2] * 2
|
# else:
|
code_block_dict[b[1]][bb] += b[2]
|
block_codes_dict = {} # {"板块":[（代码,涨停次数）]}
|
for code in code_block_dict:
|
for b in code_block_dict[code]:
|
if b not in block_codes_dict:
|
block_codes_dict[b] = []
|
block_codes_dict[b].append((code, code_block_dict[code][b]))
|
|
# 统计最近120天的涨停数据 {"代码": (涨停次数, 名称, 自由流通市值)}
|
limit_up_info_map = self.__get_limit_up_info(min_day)
|
fdatas = []
|
for b in block_codes_dict:
|
# if b != '机器人':
|
# continue
|
|
if b in constant.KPL_INVALID_BLOCKS:
|
continue
|
|
code_info_list = block_codes_dict[b]
|
# 取自由市值的前1/3 和涨停次数最多的2个代码
|
code_info_list.sort(key=lambda x: float(limit_up_info_map[x[0]][2]), reverse=True)
|
max_count = len(code_info_list) // 3
|
zylt_list = code_info_list[:max_count]
|
# 取涨停次数最多的
|
temp_codes = [x[0] for x in zylt_list]
|
code_info_list.sort(key=lambda x: x[1], reverse=True)
|
limit_up_count_max = code_info_list[:2]
|
for i in range(2, len(code_info_list)):
|
if code_info_list[i][1] == limit_up_count_max[-1][1]:
|
limit_up_count_max.append(code_info_list[i])
|
else:
|
break
|
rank2_codes = set()
|
for d in limit_up_count_max:
|
if d[1] >= 6:
|
# 涨停次数大于6次
|
if d[0] not in temp_codes:
|
zylt_list.append(d)
|
rank2_codes.add(d[0])
|
|
# 添加连板辨识度
|
continue_codes = continue_target_block_codes.get(b)
|
if continue_codes:
|
for c in continue_codes:
|
if int(float(limit_up_info_map[c][2])) < 50:
|
continue
|
zylt_list.append((c, limit_up_count_dict.get(c)))
|
|
# 按照股票涨停次数排序
|
zylt_list.sort(key=lambda x: x[1], reverse=True)
|
zylt_list = [x[0] for x in zylt_list]
|
index = 0
|
|
# 获取股价，是否是ST
|
if not zylt_list:
|
continue
|
juejin_results = HistoryKDatasUtils.get_gp_latest_info(zylt_list, fields="sec_id,sec_level,upper_limit")
|
if juejin_results is None:
|
continue
|
|
juejin_result_dict = {x['sec_id']: (x['sec_id'], x['sec_level'], x['upper_limit']) for x in juejin_results}
|
codes = set()
|
for code in zylt_list:
|
if code in codes:
|
continue
|
|
if limit_up_count_dict[code] <= 3:
|
# 累计涨停次数小于3次
|
continue
|
|
if code not in juejin_result_dict:
|
# 查询不到信息
|
continue
|
|
if juejin_result_dict[code][1] != 1:
|
# 非正常票
|
continue
|
|
if juejin_result_dict[code][2] < 3 or juejin_result_dict[code][2] > 50:
|
# 小于3块/大于50块
|
continue
|
|
index += 1
|
if int(float(limit_up_info_map[code][2])) < 50 and code not in rank2_codes:
|
continue
|
# [(板块, 代码名称, 代码, 涨停次数, 自由市值)]
|
fdatas.append(
|
(kpl_util.filter_block(b), limit_up_info_map[code][1], code, limit_up_count_dict[code],
|
int(float(limit_up_info_map[code][2])), limit_up_count_dict.get(code)))
|
codes.add(code)
|
if index >= 10:
|
break
|
# BlockSpecialCodesManager().set_block_codes_list(fdatas)
|
return fdatas
|
|
|
def update_block_special_codes():
|
datas = AnalysisBlockSpecialCodesManager().get_block_special_codes()
|
for d in datas:
|
print(d)
|
BlockSpecialCodesManager().set_block_codes_list(datas)
|
return len(datas)
|
|
|
if __name__ == "__main__":
|
# print(datas)
|
datas = AnalysisBlockSpecialCodesManager().get_block_special_codes()
|
# print(datas)
|
# for d in datas:
|
# print(d)
|
# BlockSpecialCodesManager().set_block_codes_list(datas)
|
# print(BlockSpecialCodesManager().get_code_blocks("002582"))
|
