| | |
|---|
| | | |
|---|
| | | # 实例化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线 |
|---|
| | | # 写入全目标标的股票90天K线 |
|---|
| | | def all_stocks_all_k_line_dict_write(): |
|---|
| | | all_stocks_base_K_line_dict = k_line_history.k_line_history_90day() |
|---|
| | | # 初始化所有个股的指标K线列表 |
|---|
| | |
|---|
| | | 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()}") |
|---|
| | | 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线中涨停属性是否有误（盘尾 集合竞价 炸开一个卖一档，但涨幅未变的） |
|---|
| | |
|---|
| | | # print(f"当前-时间：{formatted_time}") |
|---|
| | | time.sleep(1) |
|---|
| | | |
|---|
| | | |
|---|
| | | # 历史K线累计涨停天数函数 |
|---|
| | | def count_limit_up_day(k_line_data): |
|---|
| | | limit_up_day = 0 # 初始化涨停天数 |
|---|
