# 历史走势分析，及价值面中的历史行情部分（针对历史行情走势进行获取与分析）
|
# coding=utf-8
|
from __future__ import print_function, absolute_import, unicode_literals
|
|
import copy
|
# import logging
|
import os
|
import time
|
import datetime
|
# from datetime import datetime
|
import json
|
|
import constant
|
from log_module.log import logger_common
|
# 引入掘金API
|
# import utils.juejin_api
|
# import kpl_api
|
from strategy import data_cache
|
|
# 引入基础算法模块
|
from strategy import basic_methods
|
# 引入本地日志模块
|
from utils import hx_qc_value_util, tool
|
|
# 获取logger实例
|
logger = logger_common
|
|
start_time = time.time()
|
print(f"all_K_line开始运行--{start_time}")
|
|
|
def init_data():
|
pass
|
|
|
init_data()
|
|
|
# 90个交易日历史K线类对象
|
class KLineHistory:
|
all_stocks_base_K_line_dict = {}
|
|
# 1.类的构造方法，定义类的各个属性
|
def __init__(self):
|
pass
|
|
def init(self, today_date, Next_trading_day, symbols):
|
self.today_date = today_date # 定义属性today_date 当前交易日期
|
self.Next_trading_day = Next_trading_day # 定义属性Next_trading_day 下个交易日期
|
self.symbols = symbols
|
|
# 1.获取90个交易日期或者获取90个交易日期前的具体日期，为历史K线的start_time获取数据
|
|
def k_line_history_90day(self):
|
# 为了解决步长老是差一个，所以设定为92
|
start_time_data = basic_methods.pre_num_trading_day(self.today_date, 90)
|
# print(f"90个交易日前日期=========={start_time_data}")
|
# logger.info("获取历史K线===开始")
|
|
# 如果是16:00之前，截止日期为：前一天，如果在16:00之后截止日期为后一天
|
if int(tool.get_now_time_str().replace(":", "")) < int("160000"):
|
end_date = data_cache.DataCache().pre_trading_day
|
else:
|
end_date = data_cache.DataCache().today_date
|
|
all_symbol_K_line_history_90day_dict = {}
|
for symbol in self.symbols:
|
code = symbol[-6:]
|
k_bars = HistoryKDataManager().get_history_bars(code, end_date)
|
if not k_bars:
|
# 通过掘金接口迭代获取个股的90天基础K线，得到一个日期排列结果为増序的列表
|
k_bars = hx_qc_value_util.get_history_k_bars(symbol[-6:], 90,
|
end_date=end_date)
|
if k_bars:
|
# 将K线保存到文件
|
HistoryKDataManager().save_history_bars(symbol[-6:], k_bars[0]['bob'].strftime("%Y-%m-%d"),
|
k_bars)
|
# print(f"symbol==={symbol}")
|
# print(f"K_line_history_90day_data ====={K_line_history_90day_data} ")
|
# 将获取到的个股90天基础K线赋值给迭代个股的一个新建字典的symbol对应的value
|
all_symbol_K_line_history_90day_dict[symbol] = k_bars
|
# print(f"all_symbol_K_line_history_90day_dict==={len(all_symbol_K_line_history_90day_dict)}")
|
|
return all_symbol_K_line_history_90day_dict
|
|
# 3.获取90个交易日内标的股票的K线指标
|
@staticmethod
|
def get_property_limit_mark(it_K_line, symbol):
|
historical_high_price = 0.01 # 初始化历史最高价
|
historical_low_price = 0.01 # 初始化历史最低价
|
historical_low_price_index = 0 # 初始化历史最高价序号
|
historical_high_price_index = 0 # 初始化历史最低价序号
|
historical_average_price = 0.01 # 初始化历史均价
|
large_amplitude_day = [] # 初始化大幅震荡日期
|
current_count = 0 # 当前连续满足条件的天数
|
start = None
|
results = []
|
dates_list = []
|
# 查询交易标的基本信息并赋值
|
sec_name = data_cache.DataCache().code_name_dict.get(symbol)
|
# print(f"K_line_sec_name=={sec_name}")
|
# 将获取到的公司名称直接 生成字典放到 K线的第一个序列中
|
it_K_line[0]['sec_name'] = sec_name
|
for i in range(0, len(it_K_line) - 1):
|
previous_close = it_K_line[i + 1]['close'] # 昨日收盘价
|
previous_high = it_K_line[i + 1]['high'] # 昨日最高价
|
current_open = round(it_K_line[i]['open'], 2) # 当日开盘价
|
current_close = round(it_K_line[i]['close'], 2) # 当日收盘价
|
current_high = round(it_K_line[i]['high'], 2) # 当日最高价
|
current_low = round(it_K_line[i]['low'], 2) # 当日最低价
|
limit_up_price = basic_methods.limit_up_price(previous_close) # 计算出的当天涨停价
|
limit_up_price = float(limit_up_price) # 将涨停价转化为浮点
|
limit_down_price = basic_methods.limit_down_price(previous_close) # 计算出的当天跌停价
|
limit_down_price = float(limit_down_price) # 将跌停价转化为浮点
|
current_today_volume = it_K_line[i]['volume'] # 当日总交易量
|
current_yesterday_volume = it_K_line[i + 1]['volume'] # 昨日总交易量
|
# current_today_amount = it_K_line[i]['amount'] #当日总交易额
|
current_today_growth = basic_methods.intraday_growth(current_close, previous_close) # 计算出当日涨幅
|
it_K_line[i]['today_growth'] = current_today_growth # 将当日涨幅数据添加到字典中对应的日期中
|
current_today_amplitude = round((current_high - current_low) / previous_close, 2) # 计算出当日振幅
|
it_K_line[i]['today_amplitude'] = current_today_amplitude # 将当日振幅数据添加到字典中对应的日期中
|
historical_high_price = round(max(i['high'] for i in it_K_line), 2) # 计算出历史最高价
|
historical_high_price_index = max(range(len(it_K_line)), key=lambda ii: it_K_line[ii]['high']) # 找到最高价的索引
|
historical_low_price = round(min(i['low'] for i in it_K_line), 2) # 计算出历史最低价
|
historical_low_price_index = min(range(len(it_K_line)), key=lambda ii: it_K_line[ii]['low']) # 找到最低价的索引
|
historical_average_price = round((historical_high_price + historical_low_price) / 2, 2) # 计算出历史最均价
|
# 把最高价、最高价序号、最低价、最低价序号，添加到K线的昨日字典中
|
new_properties = {
|
'historical_high_price': historical_high_price,
|
'historical_high_price_index': historical_high_price_index,
|
'historical_low_price': historical_low_price,
|
'historical_low_price_index': historical_low_price_index,
|
'historical_average_price': historical_average_price
|
}
|
it_K_line[0].update(new_properties)
|
|
# 计算大振幅的日期并添加到大振幅日期列表中
|
if not (-4 < it_K_line[i]['today_growth'] < 4):
|
large_amplitude_day.append(it_K_line[i]['bob'])
|
|
# 判断K线图形风险位置 及 高低时间点
|
if historical_low_price_index > historical_high_price_index:
|
# 序列小于1是历史最高价 历史长度90，其实还可以根据更长来判断更显程度，历史长度很短的一般认为具有更高风险，而更长的之前有可能有更高的历史高位还比较远
|
# 最低价在前
|
it_K_line[0]['risk_position'] = 'low_price_ahead'
|
'''
|
以下部分条件分支由于都是对 it_K_line[0]['risk_position'] 数据进行修改，
|
条件判断逻辑有部分重叠，所以应按照逻辑的覆盖面进行先后判断，避免先把大部分过滤掉了。
|
所以当前条件分支语句的顺序不要轻易更换。
|
# [(historical_high_price / round(it_K_line[0]['close'], 2) < 1.05)的预设值取值范围选择其一：
|
# {"均价以上近前高的开盘价":1.05,"接近前高 的跌停价":1.11,"接近前高 的一个炸板加一个跌停的收盘价":1.17,"均价与最高价之间的涨停价":1.21,"均价与最高价之间的价":1.33}
|
'''
|
# 最高位据昨日较远 且 昨收价距历史均价差距较小 且 整体涨跌幅度较大 且 最高位 与 最低位为三倍涨幅的妖股【当前二度试探风险】
|
if (2 < historical_high_price_index < 15 and len(it_K_line) >= 90) and (
|
(historical_average_price / round(it_K_line[0]['close'], 2)) < 2) and (
|
historical_high_price > historical_low_price * 3):
|
it_K_line[0]['risk_position'] = 'recent_monster_second_break_risk'
|
# 最高位据昨日较远 且 昨收价距历史均价差距较小 且 整体涨跌幅度较大 【当前二度突破风险】
|
if (2 < historical_high_price_index < 15 and len(it_K_line) >= 90) and (
|
(historical_average_price / round(it_K_line[0]['close'], 2)) < 1.17) and (
|
historical_high_price > historical_low_price * 1.8):
|
it_K_line[0]['risk_position'] = 'recent_second_break_risk'
|
# 最高位据昨日较远 且 昨收价距历史最高价差距较小（小于5%，涨幅5%是左右是最容易把低吸骗进去的日内涨幅） 且 整体涨跌幅度较大 【当前二度突破近前高风险】
|
# 【双成药业 2024-11-08 这样的二次近前高，实际数值为1.4646...，为安全起见还是设置为1.5】
|
if (2 < historical_high_price_index < 15 and len(it_K_line) >= 90) and (
|
historical_high_price / round(it_K_line[0]['close'], 2) < 1.5) and (
|
historical_high_price > historical_low_price * 1.8):
|
it_K_line[0]['risk_position'] = 'recent_second_break_near_high_position_risk'
|
# 昨日就是最高位 且 整体涨跌幅度较大 【当前高位风险】
|
if (historical_high_price_index < 2 and len(it_K_line) >= 90) and (
|
historical_high_price > historical_low_price * 1.8):
|
it_K_line[0]['risk_position'] = 'recent_high_position_risk'
|
else:
|
# 最高价在前
|
it_K_line[0]['risk_position'] = 'high_price_ahead'
|
# 最低位据昨日较远 且 昨收价距历史均价差距较小 且 整体涨跌幅度较大 【当前二度崩溃风险】
|
if (2 < historical_low_price_index < 15 and len(it_K_line) >= 90) and (
|
(historical_average_price / round(it_K_line[0]['close'], 2)) > 1.1):
|
it_K_line[0]['risk_position'] = 'recent_second_crash_risk'
|
# 昨日就是最低位 且 整体涨跌幅度较大 【当前低位风险】
|
if (historical_low_price_index < 2 and len(it_K_line) >= 90) and (
|
historical_high_price > historical_low_price * 1.5):
|
it_K_line[0]['risk_position'] = 'recent_low_position_risk'
|
|
# 判断K线图形中连续低涨幅天数及具体序列号并把共识价格均价写入文件【阶段性价格共识位置】
|
if 0 < it_K_line[i]['today_amplitude'] < 0.025 and i < 60 and (
|
historical_high_price > historical_low_price * 1.5):
|
if current_count == 0:
|
start = i
|
# 如果当前涨幅小于阈值，增加连续天数计数器
|
current_count += 1
|
# 如果循环复合条件的连续天数达到5天，记录起始索引
|
if current_count > 5:
|
# 检查 results 是否为空或当前序列不与最后一个序列重叠
|
if not results or results[-1][1] < start:
|
if 0.98 < it_K_line[i]['high'] / it_K_line[start]['low'] < 1.02:
|
# 没有重叠，直接添加【其中start为循环到起始序号，i为循环到的结束序号。在实际的日期中要反过来】
|
results.append((start, i))
|
# 算出价格共识区间的均价
|
average_consensus_price = round((it_K_line[i]['high'] + it_K_line[start]['low']) / 2, 2)
|
dates_list.append((
|
it_K_line[start]['bob'].strftime('%Y-%m-%d %H:%M:%S'),
|
it_K_line[i]['bob'].strftime('%Y-%m-%d %H:%M:%S'),
|
average_consensus_price))
|
# if len(dates_list) > 1: # 如果只需要检测包含多段价格共识区
|
it_K_line[0]['rise_and_fall_is_small_start_and_end_date'] = dates_list
|
else:
|
# 重置计数器
|
current_count = 0
|
start = None
|
# 确保不会出现除以零的报错
|
if current_yesterday_volume == 0:
|
current_yesterday_volume = 1
|
if round(current_today_volume / current_yesterday_volume, 2) > 1.1:
|
# print(f"i=={i} {it_K_line[i]['bob']} {round(current_today_volume/current_yesterday_volume,2)} 【放量】")
|
if current_today_growth > 0:
|
it_K_line[i]['today_volume_shape'] = 'increases_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【放量上涨】")
|
elif current_today_growth < 0:
|
it_K_line[i]['today_volume_shape'] = 'increases_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【放量下跌】")
|
else:
|
it_K_line[i]['today_volume_shape'] = 'increases_balance'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【放量平收】")
|
|
elif round(current_today_volume / current_yesterday_volume, 2) < 0.9:
|
# print(f"i=={i} {it_K_line[i]['bob']} {round(current_today_volume/current_yesterday_volume,2)} 【缩量】")
|
if current_today_growth > 0:
|
it_K_line[i]['today_volume_shape'] = 'decreases_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【缩量上涨】")
|
elif current_today_growth < 0:
|
it_K_line[i]['today_volume_shape'] = 'decreases_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【缩量下跌】")
|
else:
|
it_K_line[i]['today_volume_shape'] = 'decreases_balance'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【缩量平收】")
|
else:
|
# print(f"i=={i} {it_K_line[i]['bob']} {round(current_today_volume/current_yesterday_volume,2)} 【平量】")
|
if current_today_growth > 0:
|
it_K_line[i]['today_volume_shape'] = 'remained_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【平量上涨】")
|
elif current_today_growth < 0:
|
it_K_line[i]['today_volume_shape'] = 'remained_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【平量下跌】")
|
else:
|
it_K_line[i]['today_volume_shape'] = 'remained_balance'
|
# print(f"i=={i} {it_K_line[i]['bob']} 【平量平收】")
|
|
if current_open - previous_close > 0:
|
# print(f"i=={i} {it_K_line[i]['bob']} 成交总量:{today_volume},,,成交总金额:{today_amount}")
|
# difference = current_close - previous_close
|
if current_close - current_open > 0:
|
it_K_line[i]['attribute'] = 'up_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 高开高走 【昨收价：{previous_close} 收盘价：{current_close}】")
|
else:
|
it_K_line[i]['attribute'] = 'up_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 高开低走 【昨收价：{previous_close} 收盘价：{current_close}】")
|
else:
|
if current_close - current_open > 0:
|
it_K_line[i]['attribute'] = 'down_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 低开高走 【昨收价：{previous_close} 收盘价：{current_close}】")
|
else:
|
it_K_line[i]['attribute'] = 'down_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 低开低走 【昨收价：{previous_close} 收盘价：{current_close}】")
|
|
if abs(current_high - limit_up_price) < 0.001 and abs(current_close - limit_up_price) < 0.001:
|
# print(f'整个列表更新了吗 ？ it_K_line==={it_K_line}')
|
if abs(current_open - limit_up_price) < 0.001 and abs(limit_up_price - current_low) < 0.01:
|
it_K_line[i]['attribute'] = 'one_line_limit_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 一字涨停板 ") # 打印出当前行的'bob'
|
elif abs(current_low - limit_down_price) < 0.01:
|
it_K_line[i]['attribute'] = 'limit_down_then_limit_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} [准]地天板") # 打印出当前行的'bob'
|
elif abs(current_open - limit_up_price) < 0.001 and abs(current_low - limit_down_price) < 0.01:
|
it_K_line[i]['attribute'] = 'limit_up_then_limit_down_then_limit_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 一字涨停天地天板") # 打印出当前行的'bob'
|
else:
|
it_K_line[i]['attribute'] = 'limit_up'
|
# print(f"i=={i} {it_K_line[i]['bob']} 涨停板") # 打印出当前行的'bob'
|
|
if abs(current_high - limit_up_price) < 0.001 <= abs(current_close - limit_up_price) and abs(
|
current_close - limit_down_price) >= 0.001:
|
it_K_line[i]['attribute'] = 'frying_plate'
|
# print(f'在炸板日的列表中更新了对应的字典的键对值吗 ？ it_K_line[i]==={it_K_line[i]}')
|
# print(f'整个列表更新了吗 ？ it_K_line==={it_K_line}')
|
if abs(previous_close - previous_high) >= 0.001:
|
it_K_line[i]['attribute'] = 'first_frying_plate'
|
# print(f"i=={i} {it_K_line[i]['bob']} 首板炸板") # 打印出当前行的'bob'
|
elif abs(current_open - limit_up_price) < 0.01:
|
it_K_line[i]['attribute'] = 'one_line_limit_up_then_frying_plate'
|
# print(f"i=={i} {it_K_line[i]['bob']} 一字天炸") # 打印出当前行的'bob'
|
elif abs(limit_down_price - limit_up_price) < 0.01 and abs(current_low - limit_down_price) < 0.01:
|
it_K_line[i]['attribute'] = 'one_line_limit_up_then_frying_plate_then_limit_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 一字天地") # 打印出当前行的'bob'
|
else:
|
it_K_line[i]['attribute'] = 'not_first_frying_plate'
|
# print(f"i=={i} {it_K_line[i]['bob']} 非首板炸板") # 打印出当前行的'bob'
|
|
if abs(current_low - limit_down_price) < 0.001:
|
if abs(current_close - limit_down_price) < 0.001:
|
if abs(current_high - limit_up_price) < 0.001:
|
it_K_line[i]['attribute'] = 'limit_up_then_limit_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} [准]天地板") # 打印出当前行的'bob'
|
elif abs(current_open - limit_down_price) < 0.001 and abs(current_high - limit_down_price) < 0.01:
|
it_K_line[i]['attribute'] = 'one_line_limit_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 一字跌停板") # 打印出当前行的'bob'
|
else:
|
it_K_line[i]['attribute'] = 'limit_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 跌停板") # 打印出当前行的'bob'
|
elif abs(current_close - limit_up_price) >= 0.001:
|
it_K_line[i]['attribute'] = 'touch_limit_down'
|
# print(f"i=={i} {it_K_line[i]['bob']} 触及跌停") # 打印出当前行的'bob'
|
|
# 将高涨跌幅日期打印出来
|
# it_K_line[0]['large_amplitude_day'] = large_amplitude_day
|
|
# 判断K线图形长期振幅 及 当前走势 【根据大振幅天数 反推算出 长期低幅度震荡持平、上涨、下跌的个股写入字典】
|
if len(large_amplitude_day) < 1:
|
# 长期低幅度震荡
|
it_K_line[0]['long_term_amplitude'] = 'low_amplitude_oscillation'
|
# it_K_line[0]['large_amplitude_day_len'] = len(large_amplitude_day)
|
if (historical_high_price / historical_low_price) < 1.15:
|
if 0.99 < (historical_average_price / round(it_K_line[0]['close'], 2)) < 1.09:
|
# 长期低幅度震荡持平
|
it_K_line[0]['long_term_amplitude'] = 'low_amplitude_oscillation_remains_stable'
|
else:
|
if (historical_low_price_index > historical_high_price_index) and (
|
historical_high_price_index < 2 and len(it_K_line) > 88):
|
# 长期低幅度震荡上涨
|
it_K_line[0]['long_term_amplitude'] = 'low_amplitude_oscillation_and_rise'
|
elif (historical_low_price_index < historical_high_price_index) and (
|
historical_low_price_index < 2 and len(it_K_line) > 88):
|
# 长期低幅度震荡下跌
|
it_K_line[0]['long_term_amplitude'] = 'low_amplitude_oscillation_and_decline'
|
|
|
# 实例化K线对象
|
k_line_history = KLineHistory()
|
# 实例化指数K线对象
|
main_index_k_line_history = KLineHistory()
|
|
|
# 在main.py中初始化函数里面就实例化上证A股和深证A股的历史K线方法【在本文件中调用且不写入本地文件时才需要在本文件内实例化】
|
# k_line_history.k_line_history_90day()
|
|
# 写入全目标标的股票90天K线
|
def all_stocks_all_k_line_dict_write():
|
all_stocks_base_K_line_dict = k_line_history.k_line_history_90day()
|
# 初始化所有个股的指标K线列表
|
all_stocks_all_K_line_property_dict = {}
|
for i in data_cache.DataCache().filtered_stocks:
|
# print(f"i==========={i}")
|
i_k_line = all_stocks_base_K_line_dict[i] # 获取i的K线
|
i_k_line_copy = copy.deepcopy(i_k_line) # 深拷贝i的K线
|
# it_K_line_reversed = list(reversed(i_k_line_copy)) # 开盘啦获取的数据需要反转i的K线
|
it_K_line_reversed = list(i_k_line_copy) # 小辉端的数据不需要反转i的K线
|
if not it_K_line_reversed:
|
continue
|
k_line_history.get_property_limit_mark(it_K_line_reversed, i) # 给标的的K线更新指标属性 把股票代码同时传给要调用的函数
|
symbol_K_line_property_dict = {i: it_K_line_reversed} # 添加 更新极限指标属性的K线 字典
|
# print(f"symbol_K_line_property_dict===={symbol_K_line_property_dict}")
|
all_stocks_all_K_line_property_dict.update(symbol_K_line_property_dict)
|
|
# 构造时间格式datetime转化为字符串，以便将K线属性指标转化为json格式写入本地文件
|
def convert_datetime(obj):
|
if isinstance(obj, datetime.datetime):
|
return obj.strftime('%Y-%m-%d %H:%M:%S') # 转换为字符串
|
elif isinstance(obj, dict):
|
return {k: convert_datetime(v) for k, v in obj.items()} # 递归处理字典
|
elif isinstance(obj, list):
|
return [convert_datetime(element) for element in obj] # 递归处理列表
|
# 可以添加其他类型的处理逻辑
|
else:
|
# 对于未知类型，你可以选择保留原样、跳过或引发异常
|
# 这里我们选择保留原样
|
return obj
|
|
try:
|
json_data = json.dumps(convert_datetime(all_stocks_all_K_line_property_dict), ensure_ascii=False, indent=4)
|
# 将转换后的JSON字符串写入文件
|
with open(constant.K_BARS_PATH, 'w', encoding='utf-8') as f:
|
f.write(json_data)
|
except Exception as error:
|
print(f"An error occurred while converting the data to JSON: {error}")
|
logger.info(f"标的个股历史k线写完了！{tool.get_now_time_str()}")
|
|
|
# 写入主要指数的90天K线
|
def main_index_k_line_dict_write():
|
main_index_base_K_line_dict = main_index_k_line_history.k_line_history_90day()
|
# 初始化所有个股的指标K线列表
|
main_index_k_line_property_dict = {}
|
for i in data_cache.DataCache().main_index_stocks:
|
# print(f"i==========={i}")
|
i_k_line = main_index_base_K_line_dict[i] # 获取i的K线
|
i_k_line_copy = copy.deepcopy(i_k_line) # 深拷贝i的K线
|
# it_K_line_reversed = list(reversed(i_k_line_copy)) # 开盘啦获取的数据需要反转i的K线
|
it_K_line_reversed = list(i_k_line_copy) # 小辉端的数据不需要反转i的K线
|
if not it_K_line_reversed:
|
continue
|
k_line_history.get_property_limit_mark(it_K_line_reversed, i) # 给标的的K线更新指标属性 把股票代码同时传给要调用的函数
|
index_k_line_property_dict = {i: it_K_line_reversed} # 添加 更新极限指标属性的K线 字典
|
# print(f"index_k_line_property_dict===={index_k_line_property_dict}")
|
main_index_base_K_line_dict.update(index_k_line_property_dict)
|
|
# 构造时间格式datetime转化为字符串，以便将K线属性指标转化为json格式写入本地文件
|
def convert_datetime(obj):
|
if isinstance(obj, datetime.datetime):
|
return obj.strftime('%Y-%m-%d %H:%M:%S') # 转换为字符串
|
elif isinstance(obj, dict):
|
return {k: convert_datetime(v) for k, v in obj.items()} # 递归处理字典
|
elif isinstance(obj, list):
|
return [convert_datetime(element) for element in obj] # 递归处理列表
|
# 可以添加其他类型的处理逻辑
|
else:
|
# 对于未知类型，你可以选择保留原样、跳过或引发异常
|
# 这里我们选择保留原样
|
return obj
|
|
try:
|
json_data = json.dumps(convert_datetime(main_index_k_line_property_dict), ensure_ascii=False, indent=4)
|
# 将转换后的JSON字符串写入文件
|
with open(constant.K_BARS_PATH, 'w', encoding='utf-8') as f:
|
f.write(json_data)
|
except Exception as error:
|
print(f"An error occurred while converting the data to JSON: {error}")
|
logger.info(f"主要指数的历史k线写完了！{tool.get_now_time_str()}")
|
|
|
|
# 用开盘啦数据检测昨日的K线中涨停属性是否有误（盘尾 集合竞价 炸开一个卖一档，但涨幅未变的）
|
def check_limit_up_attribute():
|
for key, values in data_cache.all_stocks_all_K_line_property_dict.items():
|
# print(f"data_cache.limit_up_code_list==={data_cache.limit_up_code_list}")
|
if 'attribute' in values[0]:
|
if key not in data_cache.yesterday_limit_up_code_list and values[0]['attribute'] in data_cache.limit_up_type:
|
logger.info(f"在K线中昨日涨停，但是在开盘啦中没有，key==={key} 公司名称==={values[0]['sec_name']} ")
|
data_cache.yesterday_frying_plate_last_minute_list.append(key)
|
else:
|
logger.info(f"K线中缺失昨日属性键值 {values[0]['sec_name']}:values[0]==={values[0]}")
|
|
|
# check_limit_up_attribute()
|
|
|
# 实时检测是否拉取K线线程的函数
|
def check_time_and_data_date():
|
while True:
|
try:
|
# 声明赋值实时时间
|
now_time = datetime.datetime.now().strftime("%H:%M:%S")
|
|
# print(f"now_time==={now_time}")
|
# if now_time > data_cache.after_closing_time and data_cache.execution is False:
|
# # if now_time > data_cache.after_closing_time:
|
# data_cache.execution = True
|
# # 整理当日涨停信息并写入本地管理好本地数据
|
# kpl_api.get_arrange_limit_up_info()
|
# logger.info(f"整理当日涨停信息 已经运行完成")
|
# # # 获取所有个股的板块概念并写入文件【耗时较长应该放在 核心主线程 和 仓位管理 后面】
|
# kpl_api.get_all_stocks_plate_dict(data_cache.min_stocks)
|
# # logger.info(f"获取所有个股的板块概念 已经运行完成")
|
|
# 构造一个循环检测K线子带你中所有个股代码下的日期是不是和某日一致的，如果不一致则返回False
|
def check_data_date(target_date):
|
# 再判断一下全局化缓存文件中的抽样日期是不是目标日期
|
if data_cache.all_stocks_all_K_line_property_dict['SZSE.000034'][0]['bob'][:10] == target_date:
|
data_date_and_target_date = True
|
return data_date_and_target_date
|
# for key,value in data_cache.all_stocks_all_K_line_property_dict.items():
|
# print(f"key===={key}")
|
# print(f"value[0]['bob'][:10]===={value[0]['bob'][:10]}")
|
# if value[0]['bob'][:10] != target_date and len(data_cache.all_stocks_all_K_line_property_dict) < 1 and key not in data_cache.no_k_line_stocks:
|
# data_date_and_target_date = False
|
# return data_date_and_target_date
|
|
# 盘中不进行时间与数据的判断(也不能早于服务器重启时间，因为次日凌晨拉取的K线会有错误)，一方面判断应该在开盘前完成，另一方面收盘后离最新拉取时间都还有很长时间
|
# 在这个时间段内运行 9:00--9:30 或 15:00--23:00
|
if data_cache.server_restart_time < now_time < data_cache.opening_time or data_cache.closing_time < now_time < data_cache.program_sleep_time:
|
# 然后判断一下，当K线字典中的任意一只个股的的第一个K线表中的第一个日期 等于 上一个交易日日期 或 今日日期 且 运行时时间未到 18:30 那么不需要新拉取直接读取已有的就行，
|
# 否者还是得调用K线对象方法拉取，并重新读取赋值全局化
|
# if hour < 18 or (hour == 18 and minute < 31):
|
if now_time < data_cache.update_data_time:
|
# if now_time < data_cache.closing_time:
|
check_pre_trading_day = check_data_date(data_cache.DataCache().pre_trading_day)
|
if check_pre_trading_day is True:
|
# if hour >= 17:
|
if now_time > data_cache.checking_data_time:
|
print(
|
f"未到【18:31】 但 既有数据 【是】 上个交易日数据 【不执行拉取K线】 ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε(/’-‘)/")
|
else:
|
print(
|
f"未到【18:31】 且 既有数据 【不是】 上个交易日数据 执行拉取K线 【既有数据日期：{data_cache.all_stocks_all_K_line_property_dict['SZSE.000034'][0]['bob'][:10]} 上个交易日：{data_cache.DataCache().pre_trading_day}】")
|
# 调用指标K线写入本地文件
|
all_stocks_all_k_line_dict_write()
|
# 读取已经获取到并存储在本地的所有票的指标K线字典，并赋值给data_cache全局缓存
|
if os.path.exists(constant.K_BARS_PATH):
|
with open(constant.K_BARS_PATH, 'r',
|
encoding='utf-8') as f:
|
data_cache.all_stocks_all_K_line_property_dict = json.load(f)
|
print(
|
f"data_cache.all_stocks_all_K_line_property_dict的个数{len(data_cache.all_stocks_all_K_line_property_dict)}")
|
|
else:
|
check_today_date = check_data_date(data_cache.DataCache().today_date)
|
if check_today_date is True:
|
print(f"到了【18:31】 但 既有数据 【是】 本交易日数据 【不执行拉取K线】 ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε=ε(/’-‘)/")
|
else:
|
print(
|
f"到了【18:31】 且 既有数据 【不是】 本交易日数据 执行拉取K线 【既有数据日期：{data_cache.all_stocks_all_K_line_property_dict['SZSE.000034'][0]['bob'][:10]} 本交易日：{data_cache.DataCache().today_date}】")
|
# 调用指标K线写入本地文件
|
all_stocks_all_k_line_dict_write()
|
# 读取已经获取到并存储在本地的所有票的指标K线字典，并赋值给data_cache全局缓存
|
if os.path.exists(constant.K_BARS_PATH):
|
with open(constant.K_BARS_PATH, 'r',
|
encoding='utf-8') as f:
|
data_cache.all_stocks_all_K_line_property_dict = json.load(f)
|
print(
|
f"data_cache.all_stocks_all_K_line_property_dict的个数{len(data_cache.all_stocks_all_K_line_property_dict)}")
|
except Exception as error:
|
logger.exception(error)
|
print(f"实时检测是否拉取K线线程报错An error occurred: {error}")
|
finally:
|
current_time = datetime.datetime.now()
|
# 使用strf time来格式化时间为字符串，只显示时分秒
|
formatted_time = current_time.strftime("%H:%M:%S")
|
# print(f"当前-时间：{formatted_time}")
|
time.sleep(1)
|
|
|
# 历史K线累计涨停天数函数
|
def count_limit_up_day(k_line_data):
|
limit_up_day = 0 # 初始化涨停天数
|
# 计算90天内涨停天数
|
for i in k_line_data:
|
if 'attribute' in i and (
|
i['attribute'] == 'one_line_limit_up' or i['attribute'] == 'limit_down_then_limit_up' or i[
|
'attribute'] == 'limit_up_then_limit_down_then_limit_up' or i['attribute'] == 'limit_up'):
|
limit_up_day += 1
|
# print(f"涨停提日期：{i['bob']}")
|
# print(f"涨停天数----limit_up_day======{limit_up_day}")
|
return limit_up_day
|
|
|
# 只在本文件内测试时开启调用
|
# all_stocks_all_k_line_dict_write()
|
|
|
# 原始K线管理器
|
class HistoryKDataManager:
|
__instance = None
|
__db = 0
|
__history_k_day_datas = {}
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(HistoryKDataManager, cls).__new__(cls, *args, **kwargs)
|
cls.__load_data()
|
return cls.__instance
|
|
@classmethod
|
def __load_data(cls):
|
pass
|
|
def __get_cache_dir(self):
|
"""
|
获取缓存路径
|
@return:
|
"""
|
dir_path = f"{constant.get_path_prefix()}/datas/k_bars"
|
if not os.path.exists(dir_path):
|
os.makedirs(dir_path)
|
return dir_path
|
|
def __del_outdate_datas(self, code):
|
dir_path = self.__get_cache_dir()
|
datas = []
|
for root, dirs, files in os.walk(dir_path):
|
for file in files:
|
# 输出文件的绝对路径
|
if file.find(code) < 0:
|
continue
|
path_ = os.path.join(root, file)
|
day = file.split("_")[0]
|
datas.append((day, path_))
|
# 保存最新的一条数据
|
if datas:
|
datas.sort(key=lambda x: int(x[0].replace("-", "")), reverse=True)
|
datas = datas[1:]
|
for d in datas:
|
os.remove(d[1])
|
|
def save_history_bars(self, code, day, datas, force=False):
|
"""
|
保存历史K线
|
@param code: 代码
|
@param day: K线最新的日期（取datas最新一条数据的日期）
|
@param datas: 数据
|
@return:
|
"""
|
cache_dir = self.__get_cache_dir()
|
file_name = f"{day}_{code}.txt"
|
path_str = f"{cache_dir}/{file_name}"
|
if os.path.exists(path_str) and not force:
|
return
|
if day not in self.__history_k_day_datas:
|
self.__history_k_day_datas[day] = {}
|
if datas:
|
self.__history_k_day_datas[day][code] = datas
|
# 将日期格式化
|
fdatas = []
|
for d in datas:
|
dd = copy.deepcopy(d)
|
for k in dd:
|
if type(dd[k]) == datetime.datetime:
|
dd[k] = dd[k].strftime("%Y-%m-%d %H:%M:%S")
|
fdatas.append(dd)
|
with open(path_str, encoding="utf-8", mode='w') as f:
|
f.write(f"{fdatas}")
|
self.__del_outdate_datas(code)
|
|
def get_history_bars(self, code, day):
|
"""
|
获取历史K线
|
@param code:
|
@param day:
|
@return:
|
"""
|
if day in self.__history_k_day_datas and code in self.__history_k_day_datas[day]:
|
return self.__history_k_day_datas[day][code]
|
cache_dir = self.__get_cache_dir()
|
file_name = f"{day}_{code}.txt"
|
path_str = f"{cache_dir}/{file_name}"
|
if not os.path.exists(path_str):
|
return None
|
with open(path_str, encoding="utf-8", mode='r') as f:
|
line = f.readline()
|
if line:
|
datas = eval(line)
|
# 将日期格式转为datetime
|
for d in datas:
|
for k in d:
|
if type(d[k]) == str and d[k].find("-") > 0 and d[k].find(":") > 0 and d[k].find(" ") > 0:
|
d[k] = datetime.datetime.strptime(d[k], "%Y-%m-%d %H:%M:%S")
|
return datas
|
return None
|
|
def get_pre_close(self, code, day):
|
"""
|
获取之前的收盘价
|
@param code:
|
@param day:
|
@return:
|
"""
|
if day in self.__history_k_day_datas and code in self.__history_k_day_datas[day]:
|
return self.__history_k_day_datas[day][code][0]["close"]
|
return None
|
|
def get_history_bars_codes(self, day):
|
"""
|
获取某一天的历史K线的代码数据
|
@param day:
|
@return: 代码集合
|
"""
|
|
dir_path = self.__get_cache_dir()
|
codes = set()
|
for root, dirs, files in os.walk(dir_path):
|
for file in files:
|
# 输出文件的绝对路径
|
if file.find(day) >= 0:
|
codes.add(file.split("_")[1][:6])
|
return codes
|
