"""
|
策略分析
|
"""
|
import datetime
|
|
from db.mysql_data_delegate import Mysqldb
|
from third_data import history_k_data_util
|
from utils import tool
|
|
|
def __load_limit_up_records(day):
|
mysql = Mysqldb()
|
datas = mysql.select_all(f"select * from kpl_limit_up_record where _day = '{day}'")
|
return datas
|
|
|
if __name__ == '__main__':
|
dates = history_k_data_util.JueJinApi.get_trading_dates("2024-05-01", "2024-08-19")
|
datas_dict = {}
|
for day in dates:
|
datas = __load_limit_up_records(day)
|
# 将数据根据涨停原因聚合
|
temp_dict = {}
|
for d in datas:
|
if d[2] not in temp_dict:
|
temp_dict[d[2]] = []
|
temp_dict[d[2]].append(d)
|
datas_dict[day] = temp_dict
|
# 统计前面7天的所有板块
|
datas_list = [(x, datas_dict[x]) for x in datas_dict]
|
SD = 7
|
for i in range(SD, len(datas_list)):
|
fblocks = set()
|
for j in range(1, SD + 1):
|
blocks = set(datas_list[i - j][1].keys())
|
fblocks |= blocks
|
c_blocks = datas_list[i][1].keys()
|
first_limit_up_fblocks = c_blocks - fblocks
|
print(f"-------------[{datas_list[i][0]}]--------------")
|
for b in first_limit_up_fblocks:
|
codes_list = [(x[3], x[4], x[12], datetime.datetime.fromtimestamp(int(x[5])).strftime('%H:%M:%S')) for x in
|
datas_list[i][1][b]]
|
if len(codes_list) >= 2:
|
print(b, codes_list)
|
|
print(len(datas_dict))
|
