✨ feat(news_analyser): 添加股票板块新闻分析模块，实现宏观新闻对板块影响评估和可视化
This commit is contained in:
parent
8f71e65f8c
commit
a2e85ff1a6
543
news_analyser.py
Normal file
543
news_analyser.py
Normal file
@ -0,0 +1,543 @@
|
||||
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')]
|
||||
else:
|
||||
# 如果没有新数据，仍使用现有数据
|
||||
z_t_df = existing_z_t_df.copy()
|
||||
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
|
||||
|
||||
# 确保透视表包含所有日期
|
||||
# 首先获取唯一的交易日
|
||||
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()
|
||||
|
||||
# 创建Excel文件并写入数据
|
||||
with pd.ExcelWriter(excel_path) as writer:
|
||||
z_t_df.to_excel(writer, 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. 只能使用提供的板块名称，不得修改一个字""".format(sector_list_text)
|
||||
|
||||
# 7. 存储分析结果
|
||||
sector_impact_results = {}
|
||||
|
||||
# 8. 对每天的新闻进行分析
|
||||
logger.info("\n开始分析每日宏观新闻对板块的影响...")
|
||||
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"\n=============== {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("\n利空板块:")
|
||||
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"\n完成分析! 共分析了 {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("\n== 提取的评分 ==")
|
||||
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"\n正在更新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()
|
||||
Loading…
Reference in New Issue
Block a user