611 lines
28 KiB
Python
611 lines
28 KiB
Python
import re
|
||
from datetime import date, timedelta
|
||
from datetime import datetime
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
from openpyxl import load_workbook
|
||
from openpyxl.styles import PatternFill
|
||
|
||
from data_manager import DataReader
|
||
from llm_manager import get_llm_manager
|
||
from logger_manager import get_logger
|
||
|
||
import os
|
||
|
||
logger = get_logger()
|
||
|
||
|
||
def create_concept_analysis_excel(force_update=False):
|
||
"""创建或更新板块分析Excel文件，根据需要从20250101或现有表格最后日期到今天
|
||
参数:
|
||
force_update: 布尔值，如果为True，则强制从20250101开始重新获取所有数据
|
||
"""
|
||
# 检查文件是否存在
|
||
excel_path = 'concept_analysis.xlsx'
|
||
today = date.today()
|
||
# 如果强制更新，直接设置起始日期为20250101
|
||
if force_update:
|
||
start_date = '20250101'
|
||
logger.info(f"强制更新模式：将从 {start_date} 获取所有数据")
|
||
existing_z_t_df = None
|
||
min_date = pd.Timestamp('2025-01-01')
|
||
elif os.path.exists(excel_path):
|
||
logger.info(f"找到现有文件 {excel_path}，准备更新...")
|
||
# 读取现有Excel文件
|
||
try:
|
||
existing_z_t_df = pd.read_excel(excel_path, sheet_name='涨停板', index_col=0)
|
||
# 将索引转换为日期时间类型
|
||
existing_z_t_df.index = pd.to_datetime(existing_z_t_df.index)
|
||
# 找出有实际数据(非空)的最后一个日期
|
||
z_t_last_date = None
|
||
# 检查z_t_df中有实际数据的最后一天
|
||
for idx in sorted(existing_z_t_df.index, reverse=True):
|
||
row = existing_z_t_df.loc[idx]
|
||
if not row.isnull().all() and not (row == '').all():
|
||
z_t_last_date = idx
|
||
break
|
||
# 如果没有找到有效的最后日期(可能是空表)，则从头开始
|
||
if z_t_last_date is None:
|
||
start_date = '20250101'
|
||
min_date = pd.Timestamp('2025-01-01')
|
||
logger.info(f"文件中没有有效数据，将从 {start_date} 开始获取数据")
|
||
else:
|
||
# 找到最后的有效日期
|
||
last_date = z_t_last_date
|
||
start_date = (last_date + timedelta(days=1)).strftime('%Y%m%d')
|
||
min_date = min(existing_z_t_df.index.min(), pd.Timestamp('2025-01-01'))
|
||
logger.info(f"现有数据最后日期为 {last_date.strftime('%Y%m%d')}，将更新从 {start_date} 到今天的数据")
|
||
except Exception as e:
|
||
logger.info(f"读取现有文件时出错: {e}")
|
||
logger.info("将创建新文件，数据范围从20250101到今天")
|
||
existing_z_t_df = None
|
||
start_date = '20250101'
|
||
min_date = pd.Timestamp('2025-01-01')
|
||
else:
|
||
logger.info(f"未找到文件 {excel_path}，将创建新文件，数据范围从20250101到今天")
|
||
existing_z_t_df = None
|
||
start_date = '20250101'
|
||
min_date = pd.Timestamp('2025-01-01')
|
||
|
||
# 检查是否需要获取新数据
|
||
need_data_update = pd.to_datetime(start_date, format='%Y%m%d') <= pd.Timestamp(today)
|
||
# 定义空的新数据框架
|
||
new_z_t_df = pd.DataFrame()
|
||
# 只有在需要更新数据且今天或之前有日期需要获取时才从数据库获取数据
|
||
if need_data_update:
|
||
# 获取新的板块数据
|
||
logger.info(f"获取从 {start_date} 到今天的板块数据...")
|
||
kpl_concept = DataReader.get_table_data_by_date(
|
||
table_name='kpl_concept',
|
||
start_date=start_date,
|
||
end_date=None,
|
||
filter_main_board=False
|
||
)
|
||
# 如果获取到了新数据，处理这些数据
|
||
if not kpl_concept.empty:
|
||
logger.info(f"获取到 {len(kpl_concept)} 条新数据")
|
||
# 确保z_t_num和up_num列是数值类型
|
||
kpl_concept['z_t_num'] = pd.to_numeric(kpl_concept['z_t_num'], errors='coerce')
|
||
# 将交易日期转换为日期类型
|
||
kpl_concept['trade_date'] = pd.to_datetime(kpl_concept['trade_date'], format='%Y%m%d')
|
||
# 创建新数据的透视表
|
||
new_z_t_df = kpl_concept.set_index(['trade_date', 'name'])['z_t_num'].unstack(level='name')
|
||
new_z_t_df = new_z_t_df.replace(0, np.nan) # 将0替换为NaN而不是空字符串
|
||
else:
|
||
logger.info("没有新的交易数据，但仍将更新表格至今天")
|
||
else:
|
||
logger.info(f"今天 {today.strftime('%Y%m%d')} 已经是最新数据，无需从数据库获取")
|
||
|
||
# 合并数据或创建新数据框
|
||
if existing_z_t_df is not None:
|
||
# 合并现有数据和新数据
|
||
if not new_z_t_df.empty:
|
||
z_t_df = pd.concat([existing_z_t_df, new_z_t_df])
|
||
# 处理可能的重复行
|
||
z_t_df = z_t_df[~z_t_df.index.duplicated(keep='last')]
|
||
logger.info(f"合并了现有数据和新数据，共 {len(z_t_df)} 行")
|
||
else:
|
||
# 如果没有新数据，仍使用现有数据
|
||
z_t_df = existing_z_t_df.copy()
|
||
logger.info(f"没有新数据，保留原有数据 {len(z_t_df)} 行，仅添加新日期的空行")
|
||
else:
|
||
# 如果是第一次创建表格且有新数据
|
||
if not new_z_t_df.empty:
|
||
z_t_df = new_z_t_df
|
||
else:
|
||
# 如果是第一次创建表格但没有数据，创建空表格
|
||
z_t_df = pd.DataFrame()
|
||
|
||
# 获取所有列名 - 确保即使是空表格也有适当的列
|
||
all_columns = set()
|
||
if not z_t_df.empty:
|
||
all_columns.update(z_t_df.columns)
|
||
# 如果没有任何列但需要创建表格，尝试从数据库获取列名
|
||
if not all_columns and existing_z_t_df is None:
|
||
logger.info("尝试从数据库获取板块名称以创建空表格...")
|
||
try:
|
||
sample_data = DataReader.get_table_data_by_date(
|
||
table_name='kpl_concept',
|
||
start_date='20250101',
|
||
end_date='20250101',
|
||
filter_main_board=False
|
||
)
|
||
if not sample_data.empty:
|
||
sector_names = sample_data['name'].unique()
|
||
all_columns = set(sector_names)
|
||
logger.info(f"获取到 {len(all_columns)} 个板块名称")
|
||
except Exception as e:
|
||
logger.error(f"获取板块名称时出错: {e}")
|
||
# 如果无法获取列名，创建带有默认列的空表格
|
||
all_columns = {"未知板块"}
|
||
|
||
# 确保有列名
|
||
if not all_columns:
|
||
all_columns = {"未知板块"}
|
||
|
||
# 确保最晚日期为今天
|
||
max_date = pd.Timestamp(today)
|
||
logger.info(f"创建从 {min_date.strftime('%Y%m%d')} 到 {max_date.strftime('%Y%m%d')} 的日期范围")
|
||
|
||
# 创建一个包含所有日期的连续序列（包括非交易日）
|
||
all_dates = pd.date_range(start=min_date, end=max_date)
|
||
|
||
# 如果z_t_df是空的，用适当的列创建它
|
||
if z_t_df.empty:
|
||
z_t_df = pd.DataFrame(columns=list(all_columns), index=[])
|
||
|
||
# 确保DataFrame有所有可能的列
|
||
for col in all_columns:
|
||
if col not in z_t_df.columns:
|
||
z_t_df[col] = np.nan
|
||
|
||
# 记录处理前的行数
|
||
original_row_count = len(z_t_df)
|
||
|
||
# 确保透视表包含所有日期
|
||
# 首先获取唯一的交易日
|
||
unique_trade_dates = set(z_t_df.index) if not z_t_df.empty else set()
|
||
|
||
# 找出缺失的日期
|
||
missing_dates = [date for date in all_dates if date not in unique_trade_dates]
|
||
|
||
# 为缺失的日期创建空行 - 使用NaN而不是空字符串
|
||
for missing_date in missing_dates:
|
||
# 添加空行到DataFrame中
|
||
z_t_df.loc[missing_date] = np.nan
|
||
|
||
# 重新排序索引，确保日期按顺序排列
|
||
z_t_df = z_t_df.sort_index()
|
||
|
||
# 记录处理后的行数
|
||
new_row_count = len(z_t_df)
|
||
logger.info(
|
||
f"原有数据行数: {original_row_count}, 添加空行后总行数: {new_row_count}, 新增空行: {new_row_count - original_row_count}")
|
||
|
||
# 保存Excel文件
|
||
sheet_name = '涨停板'
|
||
|
||
# 如果文件存在且非强制更新模式，使用openpyxl保留格式
|
||
if os.path.exists(excel_path) and not force_update:
|
||
import openpyxl
|
||
from openpyxl.utils.dataframe import dataframe_to_rows
|
||
|
||
# 加载现有Excel文件
|
||
logger.info("使用openpyxl加载Excel文件以保留格式...")
|
||
wb = openpyxl.load_workbook(excel_path)
|
||
|
||
# 检查是否存在目标工作表，如不存在则创建
|
||
if sheet_name in wb.sheetnames:
|
||
ws = wb[sheet_name]
|
||
# 清除工作表内容但保留格式
|
||
for row in ws.iter_rows(min_row=2): # 跳过标题行
|
||
for cell in row:
|
||
cell.value = None
|
||
else:
|
||
ws = wb.create_sheet(sheet_name)
|
||
|
||
# 将DataFrame写入工作表
|
||
# 首先写入索引名称到A1单元格
|
||
ws.cell(row=1, column=1, value="日期")
|
||
|
||
# 写入列名（从第2列开始）
|
||
for col_idx, col_name in enumerate(z_t_df.columns, start=2):
|
||
ws.cell(row=1, column=col_idx, value=col_name)
|
||
|
||
# 写入索引和数据（从第2行开始）
|
||
for row_idx, (idx, row_data) in enumerate(z_t_df.iterrows(), start=2):
|
||
# 写入索引（日期）到第1列
|
||
ws.cell(row=row_idx, column=1, value=idx)
|
||
|
||
# 写入数据（从第2列开始）
|
||
for col_idx, value in enumerate(row_data, start=2):
|
||
if pd.isna(value):
|
||
ws.cell(row=row_idx, column=col_idx, value=None)
|
||
else:
|
||
ws.cell(row=row_idx, column=col_idx, value=value)
|
||
|
||
# 保存工作簿
|
||
wb.save(excel_path)
|
||
logger.info(f"Excel文件已更新并保留原格式：{excel_path}")
|
||
else:
|
||
# 对于新文件或强制更新，使用pandas的to_excel
|
||
with pd.ExcelWriter(excel_path) as writer:
|
||
z_t_df.to_excel(writer, sheet_name=sheet_name)
|
||
logger.info(f"Excel文件已{'更新' if os.path.exists(excel_path) else '创建'}：{excel_path}")
|
||
|
||
logger.info(f"已包含 {len(missing_dates)} 个非交易日或未来日期")
|
||
|
||
|
||
def analyze_sectors_from_news(force_update=False):
|
||
"""分析宏观新闻并评估对各板块的影响
|
||
参数:
|
||
force_update: 布尔值，如果为True，则强制从头开始分析所有新闻
|
||
"""
|
||
|
||
# 1. 首先获取所有板块名称，作为LLM分析的参考
|
||
logger.info("正在读取板块信息...")
|
||
|
||
# 检查Excel文件是否存在
|
||
excel_exists = os.path.exists('concept_analysis.xlsx')
|
||
|
||
if excel_exists:
|
||
# 从已有的Excel文件中读取板块名称
|
||
try:
|
||
z_t_df = pd.read_excel('concept_analysis.xlsx', sheet_name='涨停板', index_col=0)
|
||
sector_names = list(z_t_df.columns)
|
||
logger.info(f"从Excel中读取到 {len(sector_names)} 个板块")
|
||
|
||
# 如果不是强制更新，查找Excel中已有分析结果的最新日期
|
||
if not force_update:
|
||
try:
|
||
# 尝试识别Excel中已经分析过(有染色)的最新日期
|
||
logger.info("查找已有分析结果的最新日期...")
|
||
wb = load_workbook('concept_analysis.xlsx')
|
||
sheet = wb['涨停板'] # 使用新的表名
|
||
|
||
last_analyzed_date = None
|
||
# 遍历行，查找带有背景色的单元格(表示已分析过)
|
||
for row in range(2, sheet.max_row + 1): # 跳过标题行
|
||
date_cell = sheet.cell(row=row, column=1)
|
||
# 检查该行是否有任何单元格带有背景色
|
||
has_color = False
|
||
for col in range(2, sheet.max_column + 1):
|
||
cell = sheet.cell(row=row, column=col)
|
||
if cell.fill.start_color.index != '00000000': # 非默认背景色
|
||
has_color = True
|
||
break
|
||
|
||
if has_color:
|
||
# 更新最新的分析日期
|
||
if date_cell.value:
|
||
current_date = pd.to_datetime(date_cell.value).strftime('%Y%m%d')
|
||
if last_analyzed_date is None or current_date > last_analyzed_date:
|
||
last_analyzed_date = current_date
|
||
|
||
if last_analyzed_date:
|
||
# 设置分析起始日期为最后分析日期后一天
|
||
last_date = datetime.strptime(last_analyzed_date, '%Y%m%d')
|
||
news_start_date = (last_date + timedelta(days=1)).strftime('%Y-%m-%d 00:00:00')
|
||
logger.info(f"找到已分析到 {last_analyzed_date}，将继续分析从 {news_start_date} 开始的新闻")
|
||
else:
|
||
logger.info("未找到已分析的数据，将从默认起始日期开始分析")
|
||
news_start_date = '2025-04-01 00:00:00'
|
||
except Exception as e:
|
||
logger.info(f"读取已分析日期时出错: {e}")
|
||
logger.info("将使用默认起始日期")
|
||
news_start_date = '2025-04-01 00:00:00'
|
||
else:
|
||
# 强制更新模式
|
||
logger.info("强制更新模式：将从默认起始日期开始重新分析所有新闻")
|
||
news_start_date = '2025-04-01 00:00:00'
|
||
except Exception as e:
|
||
logger.info(f"读取Excel文件时出错: {e}")
|
||
logger.info("将从数据库获取板块信息")
|
||
excel_exists = False # 重置文件存在标志，使用数据库获取信息
|
||
|
||
# 如果Excel不存在或读取失败，从数据库获取板块名称
|
||
if not excel_exists or force_update:
|
||
logger.info("从数据库获取板块信息...")
|
||
kpl_concept = DataReader.get_table_data_by_date(
|
||
table_name='kpl_concept',
|
||
start_date='20250101',
|
||
end_date=None,
|
||
filter_main_board=False
|
||
)
|
||
if not kpl_concept.empty:
|
||
sector_names = kpl_concept['name'].unique().tolist()
|
||
logger.info(f"从数据库中读取到 {len(sector_names)} 个板块")
|
||
else:
|
||
# 如果数据库也没有返回数据，使用一个默认的空列表
|
||
sector_names = []
|
||
logger.info("警告: 数据库中没有获取到板块信息，将使用空板块列表")
|
||
|
||
# 强制更新或表格不存在时，使用默认起始日期
|
||
news_start_date = '2025-04-01 00:00:00'
|
||
|
||
# 如果表格不存在或强制更新，需要创建新表格
|
||
logger.info("将创建新的分析表格")
|
||
# 创建一个新的空表格，以便后续分析结果可以写入
|
||
if not os.path.exists('concept_analysis.xlsx'):
|
||
# 使用create_concept_analysis_excel创建基本表格
|
||
# 创建一个包含所有板块名称的空数据框
|
||
df = pd.DataFrame(columns=sector_names)
|
||
# 添加今天的日期作为索引
|
||
df.loc[pd.Timestamp.today()] = np.nan
|
||
# 保存到Excel
|
||
with pd.ExcelWriter('concept_analysis.xlsx') as writer:
|
||
df.to_excel(writer, sheet_name='涨停板')
|
||
logger.info("已创建基础Excel文件")
|
||
|
||
# 2. 获取宏观新闻数据
|
||
logger.info("正在获取宏观新闻...")
|
||
logger.info(f"获取从 {news_start_date} 到今天的宏观新闻...")
|
||
news = DataReader.get_news(start_date=news_start_date)
|
||
macro_news = news[news['channels'] == '宏观'].copy()
|
||
|
||
# 如果没有新数据，退出函数
|
||
if macro_news.empty:
|
||
logger.info("没有新的宏观新闻需要分析，退出更新")
|
||
return {}
|
||
|
||
# 3. 将日期转换为统一的YYYYMMDD格式
|
||
macro_news['date'] = pd.to_datetime(macro_news['datetime']).dt.date
|
||
macro_news['date_str'] = macro_news['date'].apply(lambda x: x.strftime('%Y%m%d'))
|
||
|
||
# 4. 按日期分组整理新闻内容
|
||
result = macro_news.groupby('date_str')['content'].agg(lambda x: list(x)).reset_index()
|
||
result['date'] = pd.to_datetime(result['date_str'], format='%Y%m%d').dt.date
|
||
|
||
# 5. 初始化LLM管理器
|
||
llm = get_llm_manager()
|
||
|
||
# 6. 创建系统提示词，强调格式必须严格遵守
|
||
sector_list_text = ", ".join(sector_names)
|
||
system_prompt = """你是一位资深宏观新闻分析师和股票板块研究专家。
|
||
请分析以下一天内的新闻集合，识别出对股票板块的潜在影响。
|
||
你只能从以下列出的板块名称中选择，必须使用完全一致的板块名称，一个字都不能改变:
|
||
===输出格式要求（极其重要）===
|
||
你必须严格按照以下格式输出结果，不要添加任何额外的标点或修改格式:
|
||
## 利好板块
|
||
[板块名称]:[影响评分]-[简要解释]
|
||
[板块名称]:[影响评分]-[简要解释]
|
||
...
|
||
## 利空板块
|
||
[板块名称]:[影响评分]-[简要解释]
|
||
[板块名称]:[影响评分]-[简要解释]
|
||
===格式示例（精确参考）===
|
||
如下所示，不要修改格式，只需替换内容:
|
||
## 利好板块
|
||
[板块A]:[8]-[这是对板块A影响的解释]
|
||
[板块B]:[6]-[这是对板块B影响的解释]
|
||
## 利空板块
|
||
[板块C]:[-7]-[这是对板块C影响的解释]
|
||
[板块D]:[-9]-[这是对板块D影响的解释]
|
||
===重要规则===
|
||
1. 严格遵循上述格式，不要添加额外的冒号或任何其他标点符号
|
||
2. 方括号[]在最终输出中必须保留
|
||
3. 评分必须是-10到10之间的整数，且不为0
|
||
4. 利好板块评分为正数(1-10)，利空板块评分为负数(-1至-10)
|
||
5. 评分标准如下:
|
||
|
||
===利好/利空评分标准===
|
||
|
||
A. 政策与监管因素 (基于发布主体和政策确定性):
|
||
* 10/(-10): 国家级明确政策 - 国务院、中央部委或央行发布的正式政策、法规或措施
|
||
* 9/(-9): 省级明确政策 - 省级政府部门发布的正式政策、法规或措施
|
||
* 8/(-8): 市级明确政策 - 市级政府部门发布的正式政策、法规或措施
|
||
* 7/(-7): 行业协会明确指导 - 全国性行业协会发布的规范性文件或指导意见
|
||
* 6/(-6): 国家级政策导向 - 国家级部门的讲话、会议精神或指导性意见
|
||
* 5/(-5): 省级政策导向 - 省级部门的讲话、会议精神或指导性意见
|
||
* 4/(-4): 市级政策导向 - 市级部门的讲话、会议精神或指导性意见
|
||
* 3/(-3): 行业协会政策导向 - 行业协会的讲话、会议精神或建议
|
||
|
||
B. 市场与行业事件 (基于事件影响范围与程度):
|
||
* 10/(-10): 重大市场事件 - 对整个行业产生全局性、长期影响的重大事件
|
||
* 8/(-8): 显著市场事件 - 对行业产生明显影响的重要事件
|
||
* 6/(-6): 中等市场事件 - 对行业产生一定影响的事件
|
||
* 4/(-4): 一般市场事件 - 对行业产生有限影响的普通事件
|
||
* 2/(-2): 轻微市场事件 - 对行业产生轻微影响的小型事件
|
||
|
||
C. 信息来源可信度与影响程度 (其他情况):
|
||
* 5/(-5): 权威专家观点 - 行业公认的顶级专家或研究机构的深度分析
|
||
* 3/(-3): 一般专家观点 - 行业内专家学者的分析观点
|
||
* 2/(-2): 市场分析 - 主流媒体或分析师的市场分析
|
||
* 1/(-1): 一般信息 - 普通媒体报道或一般性市场信息
|
||
|
||
6. 根据新闻性质，从以上三类标准中选择最适合的一类进行评分
|
||
7. 只能使用提供的板块名称，不得修改一个字，确保你输出的板块名称能够在""" + sector_list_text + """中找到，并且选择的板块名称和你判断的以及有着高度相关性，不要延伸概念，实事求是"""
|
||
|
||
# 7. 存储分析结果
|
||
sector_impact_results = {}
|
||
|
||
# 8. 对每天的新闻进行分析
|
||
logger.info("开始分析每日宏观新闻对板块的影响...")
|
||
for index, row in result.iterrows():
|
||
date_str = row['date_str']
|
||
content_list = row['content']
|
||
# 合并当天的所有新闻内容
|
||
combined_content = "\n".join([f"- {item}" for item in content_list])
|
||
# 构建提问内容
|
||
query = f"以下是{row['date'].strftime('%Y年%m月%d日')}的宏观新闻汇总:\n\n{combined_content}\n\n请分析这些新闻对股票板块的具体影响。"
|
||
logger.info(f"正在分析 {date_str} 的新闻...")
|
||
# 调用LLM进行分析
|
||
analysis = llm.chat(query, prompt=system_prompt)
|
||
# 解析结果以提取板块评分
|
||
sector_scores = parse_sector_impact(analysis)
|
||
# 存储结果
|
||
sector_impact_results[date_str] = {
|
||
'date': row['date'],
|
||
'news_count': len(content_list),
|
||
'analysis': analysis,
|
||
'sector_scores': sector_scores
|
||
}
|
||
# 输出分析结果
|
||
logger.info(f"=============== {date_str} 板块影响分析 ({len(content_list)}条新闻) ===============")
|
||
logger.info("利好板块:")
|
||
for sector, score in sorted([(k, v) for k, v in sector_scores.items() if v > 0], key=lambda x: -x[1]):
|
||
logger.info(f" {sector}: +{score}")
|
||
logger.info("利空板块:")
|
||
for sector, score in sorted([(k, v) for k, v in sector_scores.items() if v < 0], key=lambda x: x[1]):
|
||
logger.info(f" {sector}: {score}")
|
||
logger.info("=" * 60)
|
||
|
||
logger.info(f"完成分析! 共分析了 {len(sector_impact_results)} 天的宏观新闻")
|
||
|
||
# 9. 将结果更新到Excel中
|
||
update_excel_with_colors(sector_impact_results)
|
||
|
||
return sector_impact_results
|
||
|
||
|
||
def parse_sector_impact(analysis_text):
|
||
"""从LLM分析文本中提取板块评分（-10到10评分范围）"""
|
||
sector_scores = {}
|
||
|
||
# 打印完整的分析文本片段以便调试
|
||
logger.debug("\n== 分析文本片段 ==")
|
||
logger.debug(analysis_text)
|
||
|
||
# 定义正则表达式模式来匹配方括号格式的评分
|
||
pattern = r'\[([^\]]+)\]:\[([+-]?\d+)\]-'
|
||
|
||
# 提取所有匹配
|
||
matches = re.findall(pattern, analysis_text)
|
||
|
||
# 处理匹配结果
|
||
logger.info("== 提取的评分 ==")
|
||
for sector_name, score in matches:
|
||
sector_name = sector_name.strip()
|
||
try:
|
||
score_value = int(score)
|
||
# 检查评分是否在合理范围内 (-10到10，非零)
|
||
if -10 <= score_value <= 10 and score_value != 0:
|
||
sector_scores[sector_name] = score_value
|
||
logger.debug(f" {sector_name}: {score_value}")
|
||
except ValueError:
|
||
continue
|
||
|
||
return sector_scores
|
||
|
||
|
||
def update_excel_with_colors(sector_impact_results):
|
||
"""更新Excel文件，根据板块评分添加颜色（-10到10分的评分范围）"""
|
||
excel_path = 'concept_analysis.xlsx'
|
||
if not os.path.exists(excel_path):
|
||
logger.info(f"错误: {excel_path} 文件不存在，无法更新颜色")
|
||
return
|
||
|
||
logger.info(f"正在更新Excel文件 {excel_path} 的颜色标注...")
|
||
|
||
# 加载Excel文件
|
||
workbook = load_workbook(excel_path)
|
||
|
||
# 记录颜色更新的统计信息
|
||
total_updates = 0
|
||
matched_dates = set()
|
||
|
||
# 只处理涨停板工作表
|
||
sheet_name = '涨停板'
|
||
if sheet_name in workbook.sheetnames:
|
||
sheet = workbook[sheet_name]
|
||
|
||
# 获取列名（板块名）和它们的列号
|
||
columns = {}
|
||
for col in range(2, sheet.max_column + 1): # 从第2列开始，第1列是日期索引
|
||
sector_name = sheet.cell(row=1, column=col).value
|
||
if sector_name:
|
||
columns[sector_name] = col
|
||
|
||
logger.info(f"工作表 {sheet_name} 共有 {len(columns)} 个板块列")
|
||
|
||
# 获取日期行，调试打印前几个日期
|
||
date_cells = []
|
||
for row in range(2, min(7, sheet.max_row + 1)): # 只打印前5个日期用于调试
|
||
cell_value = sheet.cell(row=row, column=1).value
|
||
date_cells.append(str(cell_value))
|
||
|
||
logger.info(f"工作表前几个日期单元格值示例: {date_cells}")
|
||
|
||
# 根据日期找到对应的行，并在板块对应的列添加颜色
|
||
for date_str, impact_data in sector_impact_results.items():
|
||
sector_scores = impact_data['sector_scores']
|
||
date_found = False
|
||
|
||
# 在Excel中找到对应日期的行
|
||
for row in range(2, sheet.max_row + 1):
|
||
cell_value = sheet.cell(row=row, column=1).value
|
||
excel_date_str = None
|
||
|
||
# 处理不同类型的日期值
|
||
if isinstance(cell_value, datetime):
|
||
excel_date_str = cell_value.strftime('%Y%m%d')
|
||
elif isinstance(cell_value, pd.Timestamp):
|
||
excel_date_str = cell_value.strftime('%Y%m%d')
|
||
elif isinstance(cell_value, str):
|
||
# 尝试解析字符串日期
|
||
try:
|
||
parsed_date = pd.to_datetime(cell_value)
|
||
excel_date_str = parsed_date.strftime('%Y%m%d')
|
||
except:
|
||
excel_date_str = cell_value.strip()
|
||
elif cell_value is not None:
|
||
excel_date_str = str(cell_value).strip()
|
||
|
||
if excel_date_str == date_str:
|
||
date_found = True
|
||
date_row = row
|
||
matched_dates.add(date_str)
|
||
|
||
# 为每个评分的板块添加颜色
|
||
updates_for_date = 0
|
||
for sector, score in sector_scores.items():
|
||
if sector in columns:
|
||
col = columns[sector]
|
||
cell = sheet.cell(row=date_row, column=col)
|
||
|
||
# 根据评分设置颜色深浅 (-10到10分范围)
|
||
if score > 0: # 利好，使用红色
|
||
# 分数越高，红色越深 (10分为最深红色)
|
||
intensity = max(0, 255 - int(score * 25.5)) # 10分时为0，1分时为230
|
||
fill = PatternFill(start_color=f"FF{intensity:02X}{intensity:02X}",
|
||
end_color=f"FF{intensity:02X}{intensity:02X}",
|
||
fill_type="solid")
|
||
cell.fill = fill
|
||
updates_for_date += 1
|
||
elif score < 0: # 利空，使用绿色
|
||
# 分数越低，绿色越深 (-10分为最深绿色)
|
||
intensity = max(0, 255 + int(score * 25.5)) # -10分时为0，-1分时为230
|
||
fill = PatternFill(start_color=f"{intensity:02X}FF{intensity:02X}",
|
||
end_color=f"{intensity:02X}FF{intensity:02X}",
|
||
fill_type="solid")
|
||
cell.fill = fill
|
||
updates_for_date += 1
|
||
|
||
logger.info(f"日期 {date_str}: 更新了 {updates_for_date} 个板块的颜色")
|
||
total_updates += updates_for_date
|
||
break
|
||
|
||
if not date_found:
|
||
logger.info(f"警告: 在Excel中未找到日期 {date_str}")
|
||
else:
|
||
logger.info(f"错误: 工作表 '{sheet_name}' 不存在于Excel文件中")
|
||
|
||
# 提供最终的统计信息
|
||
logger.info(f"总共匹配了 {len(matched_dates)}/{len(sector_impact_results)} 个日期")
|
||
logger.info(f"总共更新了 {total_updates} 个单元格的颜色")
|
||
|
||
# 保存更新后的Excel文件
|
||
try:
|
||
workbook.save(excel_path)
|
||
logger.info(f"Excel文件已成功保存，板块评分已用颜色标注")
|
||
except Exception as e:
|
||
logger.info(f"保存Excel文件时出错: {str(e)}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 确保Excel文件存在
|
||
create_concept_analysis_excel()
|
||
|
||
# 执行宏观新闻分析并更新Excel
|
||
sector_impact_results = analyze_sectors_from_news(force_update=False) |