import json
|
import os
|
|
import constant
|
from strategy import data_cache
|
from log_module.log import logger_common, logger_debug
|
from utils import tool
|
|
|
@tool.singleton
|
class KBarsManager:
|
def __init__(self):
|
self.all_stocks_all_K_line_property_dict = {}
|
self.all_index_k_line_property_dict = {}
|
self.day = ''
|
self.now_time = tool.get_now_time_str()
|
self.stock_path = ''
|
self.index_path = ''
|
self.load_data()
|
|
def get_now_day(self):
|
"""
|
获取现在K线的日期
|
:return:
|
"""
|
if tool.get_now_time_str() > '16:00:00':
|
return tool.get_now_date_str()
|
else:
|
# 获取上一个交易日
|
return data_cache.DataCache().pre_trading_day
|
|
def load_data(self):
|
# 判断是否为16:00之后
|
if tool.get_now_time_str() > '16:00:00':
|
self.day = tool.get_now_date_str()
|
else:
|
# 获取上一个交易日
|
self.day = data_cache.DataCache().pre_trading_day
|
self.stock_path = os.path.join(constant.DATA_DIR_PATH, f"all_stocks_all_K_line_property_dict.{self.day}.json")
|
self.index_path = os.path.join(constant.DATA_DIR_PATH, f"all_index_k_line_property_dict.{self.day}.json")
|
self.__load_data()
|
|
def __load_data(self):
|
if os.path.exists(self.stock_path):
|
|
with open(self.stock_path, 'r', encoding='utf-8') as f:
|
self.all_stocks_all_K_line_property_dict = json.load(f)
|
logger_debug.info(f"历史K线数据数量（{self.stock_path}）：{len(self.all_stocks_all_K_line_property_dict)}")
|
else:
|
logger_debug.info(f"历史K线数据不存在：{self.stock_path}")
|
self.all_stocks_all_K_line_property_dict.clear()
|
|
if os.path.exists(self.index_path):
|
with open(self.index_path, 'r', encoding='utf-8') as f:
|
data_cache.all_index_k_line_property_dict = json.load(f)
|
self.all_index_k_line_property_dict = data_cache.all_index_k_line_property_dict
|
|
def set_stock_k_bars_data(self, datas, force=False):
|
"""
|
设置股票K线数据
|
:param force: 是否强制更新
|
:param datas:
|
:return:
|
"""
|
if not force and os.path.exists(self.stock_path):
|
return
|
self.all_stocks_all_K_line_property_dict = datas
|
with open(self.stock_path, mode='w', encoding='utf-8') as f:
|
f.write(json.dumps(datas))
|
|
def set_index_k_bars_data(self, datas, force=False):
|
"""
|
设置股票K线数据
|
:param force: 是否强制更新
|
:param datas:
|
:return:
|
"""
|
if not force and os.path.exists(self.index_path):
|
return
|
self.all_index_k_line_property_dict = datas
|
with open(self.index_path, mode='w', encoding='utf-8') as f:
|
f.write(json.dumps(datas))
|
|
|
# 读取本地的所有带属性K线数据（所有个股K线及指数线）
|
def read_local_K_line_data():
|
KBarsManager().load_data()
|
data_cache.all_stocks_all_K_line_property_dict = KBarsManager().all_stocks_all_K_line_property_dict
|
data_cache.all_index_k_line_property_dict = KBarsManager().all_index_k_line_property_dict
|
return data_cache.all_index_k_line_property_dict
|
|
|
# 读取本地的个股所属概念板块数据
|
def read_local_all_stocks_plate_data():
|
# 读取已经获取到并存储在本地的目标范围的个股的板块概念
|
# 读取JSON文件并解析为字典
|
if os.path.exists(constant.ALL_STOCKS_PLATE_PATH):
|
with open(constant.ALL_STOCKS_PLATE_PATH, 'r',
|
encoding='utf-8') as f:
|
json_data = f.read()
|
else:
|
json_data = "{}"
|
data_cache.all_stocks_plate_dict = json.loads(json_data)
|
logger_common.info(f"all_stocks_plate_dict的数量={len(data_cache.all_stocks_plate_dict)}")
|
|
|
if __name__ == '__main__':
|
all_index_k_line_property_dict = read_local_K_line_data()
|
print(f"all_index_k_line_property_dict=={all_index_k_line_property_dict}")
|
if all_index_k_line_property_dict['SHSE.000001'][1]['today_volume_shape'] == 'remained_down':
|
print(f"昨日平量下跌")
|
