import os import traceback import matplotlib.pyplot as plt import numpy as np import pandas as pd from config_manager import get_config_manager from database_manager import DatabaseManager from logger_manager import get_logger class MoneyflowAnalyzer: """ 市场资金流分析器，分析各类资金流向指标对行业在随后1-N天表现的影响 """ def __init__(self): self.db_manager = DatabaseManager() self.config = get_config_manager() self.logger = get_logger() def main_flow_analyze(self, days_forward=10, use_consistent_samples=False, max_days=10): """ 分析各类资金流向指标对行业在随后1-N天表现的影响 参数: days_forward (int): 要分析的未来天数，默认为10天 """ self.logger.info(f"开始分析资金流数据，未来{days_forward}天的表现...") self.logger.info(f"使用一致样本集: {use_consistent_samples}") # 读取资金流数据 try: df = self.db_manager.load_df_from_db('moneyflow_ind_dc') self.logger.info(f"成功从数据库加载资金流数据，共计{len(df)}条记录") except Exception as e: self.logger.error(f"从数据库读取数据失败: {e}") self.logger.debug(f"完整的错误追踪信息:\n{traceback.format_exc()}") return False # 检查是否成功获取数据 if df.empty: self.logger.info("获取的资金流数据为空，无法进行分析") return False # 数据预处理 - 计算主力资金 df['main_force_amount'] = df['buy_elg_amount'] + df['buy_lg_amount'] # 将日期格式转换为datetime df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d', errors='coerce') df = df[~df['trade_date'].isna()] # 按日期排序 df = df.sort_values('trade_date') # 获取所有交易日期 all_dates = df['trade_date'].unique() # 确定样本范围 if use_consistent_samples: # 如果max_days大于days_forward，则预测天数使用max_days if max_days > days_forward: days_forward = max_days self.logger.info(f"max_days大于days_forward，使用max_days进行分析") # 使用固定的最大预测天数来确保所有分析使用相同的样本集 sample_limit = len(all_dates) - max_days self.logger.info(f"使用统一样本集进行分析，样本量: {sample_limit}") else: # 使用当前分析所需的预测天数 sample_limit = len(all_dates) - days_forward self.logger.info(f"使用动态样本集进行分析，样本量: {sample_limit}") # 定义要分析的资金流指标 # 格式: (指标名, 排序方向, 关联性) # 关联性: 正相关=1, 负相关=-1 (用于确定是取最高还是最低) flow_indicators = [ ('main_force_amount', 1, '主力净额') ] # 确保结果目录存在 os.makedirs('result', exist_ok=True) # 为每个指标进行分析 for indicator, correlation, indicator_name in flow_indicators: self.logger.info(f"分析 {indicator_name} 与未来指数关系...") # 创建结果数据结构 results = [] # 遍历每个交易日期（固定样本集大小） for i in range(sample_limit): current_date = all_dates[i] # 获取当前日期的数据 current_day_data = df[df['trade_date'] == current_date] # 找出该指标排名靠前的行业 if correlation > 0: # 正相关，找最高值 top_sectors = current_day_data.sort_values(indicator, ascending=False).head(1)['name'].tolist() else: # 负相关，找最低值 top_sectors = current_day_data.sort_values(indicator, ascending=True).head(1)['name'].tolist() # 分析每个行业在随后1-N天的表现 for sector in top_sectors: # 获取该行业当天的指数变化和指标值 sector_current = current_day_data[current_day_data['name'] == sector] if sector_current.empty: continue current_pct_change = sector_current['pct_change'].values[0] current_indicator_value = sector_current[indicator].values[0] # 分析随后1-N天的表现 future_changes = [] for day_offset in range(1, days_forward + 1): if i + day_offset < len(all_dates): future_date = all_dates[i + day_offset] future_data = df[(df['trade_date'] == future_date) & (df['name'] == sector)] if not future_data.empty: future_changes.append(future_data['pct_change'].values[0]) else: future_changes.append(None) else: future_changes.append(None) # 如果至少有一个未来日期有数据 if any(x is not None for x in future_changes): result_entry = { 'date': current_date.strftime('%Y%m%d'), # 将日期格式化为YYYYMMDD字符串 'sector': sector, f'{indicator}': current_indicator_value, 'current_pct_change': current_pct_change, } # 添加1-N天的变化 for day in range(1, days_forward + 1): result_entry[f'day{day}_change'] = future_changes[day - 1] # 计算平均变化 result_entry[f'avg_{days_forward}day_change'] = np.nanmean( [x for x in future_changes if x is not None]) results.append(result_entry) # 转换为DataFrame results_df = pd.DataFrame(results) if results_df.empty: self.logger.info(f"没有足够的数据来分析{indicator_name}与后续表现的关系") continue # 保存结果 os.makedirs('result', exist_ok=True) output_file = f'result/{indicator}_performance_{days_forward}days.xlsx' results_df.to_excel(output_file, index=False) self.logger.info(f"{indicator_name}表现分析已保存至{output_file}") # 分析整体表现 avg_performance = {} success_rates = {} for day in range(1, days_forward + 1): # 计算每一天的变化均值 avg_performance[f'day{day}'] = results_df[f'day{day}_change'].mean() # 计算每一天的胜率（正收益的概率） success_rates[f'day{day}'] = (results_df[f'day{day}_change'] > 0).mean() * 100 avg_performance[f'avg_{days_forward}day'] = results_df[f'avg_{days_forward}day_change'].mean() # 输出每天的表现数据 self.logger.info(f"{indicator_name}极值行业在未来{days_forward}天的表现分析：") self.logger.info("日期\t\t平均收益率\t\t胜率") for day in range(1, days_forward + 1): self.logger.info(f"T+{day}\t\t{avg_performance[f'day{day}']:.5f}%\t\t{success_rates[f'day{day}']:.2f}%") self.logger.info(f"{days_forward}天平均: {avg_performance[f'avg_{days_forward}day']:.5f}%") # 绘制折线图显示未来N天的平均表现和胜率 plt.figure(figsize=(14, 8)) # 设置中文字体 try: plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'WenQuanYi Micro Hei'] plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题 days_labels = ['当天'] + [f'第{i}天' for i in range(1, days_forward + 1)] direction = "最高" if correlation > 0 else "最低" title = f'{indicator_name}{direction}后的平均表现 ({days_forward}天)' xlabel = '时间' ylabel = '指数变化率 (%)' except: # 如果没有中文字体，使用英文 days_labels = ['Current'] + [f'Day+{i}' for i in range(1, days_forward + 1)] direction = "Highest" if correlation > 0 else "Lowest" title = f'Average Performance After {indicator_name} {direction} ({days_forward} Days)' xlabel = 'Time' ylabel = 'Index Change Rate (%)' # 创建双y轴图表 fig, ax1 = plt.subplots(figsize=(14, 8)) # 设置收益率曲线 (左y轴) values = [results_df['current_pct_change'].mean()] + [avg_performance[f'day{i}'] for i in range(1, days_forward + 1)] ax1.plot(days_labels, values, marker='o', linewidth=2, color='blue', label='平均收益率') ax1.axhline(y=0, color='r', linestyle='--') ax1.set_xlabel(xlabel) ax1.set_ylabel('收益率 (%)', color='blue') ax1.tick_params(axis='y', labelcolor='blue') # 设置胜率曲线 (右y轴) ax2 = ax1.twinx() win_rates = [50] # 当天默认为50% for i in range(1, days_forward + 1): win_rates.append(success_rates[f'day{i}']) ax2.plot(days_labels, win_rates, marker='s', linewidth=2, color='green', label='胜率') ax2.set_ylabel('胜率 (%)', color='green') ax2.tick_params(axis='y', labelcolor='green') ax2.axhline(y=50, color='green', linestyle='--', alpha=0.5) # 添加图例 lines1, labels1 = ax1.get_legend_handles_labels() lines2, labels2 = ax2.get_legend_handles_labels() ax1.legend(lines1 + lines2, labels1 + labels2, loc='best') plt.title(title, fontsize=14) plt.grid(True) plt.xticks(rotation=45) # 旋转x轴标签以避免重叠 # 保存图表 output_image = f'result/{indicator}_performance_{days_forward}days.png' plt.savefig(output_image, dpi=300, bbox_inches='tight') self.logger.info(f"{indicator_name}表现图表已保存至{output_image}") # 额外保存胜率图表 plt.figure(figsize=(14, 8)) plt.bar(range(1, days_forward + 1), [success_rates[f'day{i}'] for i in range(1, days_forward + 1)], color='green', alpha=0.7) plt.axhline(y=50, color='r', linestyle='--') plt.title(f'{indicator_name}{direction}后的未来胜率分布 ({days_forward}天)', fontsize=14) plt.xlabel('未来天数') plt.ylabel('胜率 (%)') plt.xticks(range(1, days_forward + 1), [f'T+{i}' for i in range(1, days_forward + 1)]) plt.grid(True, axis='y') # 在柱状图上标注具体数值 for i in range(1, days_forward + 1): plt.text(i, success_rates[f'day{i}'] + 1, f"{success_rates[f'day{i}']:.1f}%", ha='center', va='bottom', fontsize=9) win_rate_image = f'result/{indicator}_win_rate_{days_forward}days.png' plt.savefig(win_rate_image, dpi=300, bbox_inches='tight') self.logger.info(f"{indicator_name}胜率分布图已保存至{win_rate_image}") return True