""" 行业估值分析模块 提供行业历史PE、PB、PS分位数分析功能以及行业拥挤度指标，包括： 1. 行业历史PE、PB、PS数据获取 2. 分位数计算 3. 行业交易拥挤度计算 """ import pandas as pd import numpy as np from sqlalchemy import create_engine, text import datetime import logging import json import redis import time from typing import Tuple, Dict, List, Optional, Union from .config import DB_URL, OUTPUT_DIR, LOG_FILE # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() ] ) logger = logging.getLogger("industry_analysis") # 添加Redis客户端 redis_client = redis.Redis( host='192.168.18.208', # Redis服务器地址，根据实际情况调整 port=6379, password='wlkj2018', db=13, socket_timeout=5, decode_responses=True ) class IndustryAnalyzer: """行业估值分析器类""" def __init__(self, db_url: str = DB_URL): """ 初始化行业估值分析器 Args: db_url: 数据库连接URL """ self.engine = create_engine( db_url, pool_size=5, max_overflow=10, pool_recycle=3600 ) logger.info("行业估值分析器初始化完成") def get_industry_list(self) -> List[Dict]: """ 获取所有行业列表 Returns: 行业列表，每个行业为一个字典，包含code和name """ try: query = text(""" SELECT DISTINCT bk_code, bk_name FROM gp_hybk ORDER BY bk_name """) with self.engine.connect() as conn: result = conn.execute(query).fetchall() if result: return [{"code": str(row[0]), "name": row[1]} for row in result] else: logger.warning("未找到行业数据") return [] except Exception as e: logger.error(f"获取行业列表失败: {e}") return [] def get_concept_list(self) -> List[Dict]: """ 获取所有概念板块列表 Returns: 概念板块列表，每个概念板块为一个字典，包含code和name """ try: query = text(""" SELECT DISTINCT bk_code, bk_name FROM gp_gnbk ORDER BY bk_name """) with self.engine.connect() as conn: result = conn.execute(query).fetchall() if result: return [{"code": str(row[0]), "name": row[1]} for row in result] else: logger.warning("未找到概念板块数据") return [] except Exception as e: logger.error(f"获取概念板块列表失败: {e}") return [] def get_industry_stocks(self, industry_name: str) -> List[str]: """ 获取指定行业的所有股票代码 Args: industry_name: 行业名称 Returns: 股票代码列表 """ try: query = text(""" SELECT DISTINCT gp_code FROM gp_hybk WHERE bk_name = :industry_name """) with self.engine.connect() as conn: result = conn.execute(query, {"industry_name": industry_name}).fetchall() if result: return [row[0] for row in result] else: logger.warning(f"未找到行业 {industry_name} 的股票") return [] except Exception as e: logger.error(f"获取行业股票失败: {e}") return [] def get_industry_valuation_data(self, industry_name: str, start_date: str, metric: str = 'pe') -> pd.DataFrame: """ 获取行业估值数据，返回每日行业平均PE/PB/PS 说明： - 行业估值数据是指行业内所有股票的平均PE/PB/PS的历史数据 - 在计算过程中会剔除负值和极端值(如PE>1000) Args: industry_name: 行业名称 start_date: 开始日期 metric: 估值指标（pe、pb或ps） Returns: 包含行业估值数据的DataFrame，主要包含以下列： - timestamp: 日期 - avg_{metric}: 行业平均值 - stock_count: 参与计算的股票数量 """ try: # 验证metric参数 if metric not in ['pe', 'pb', 'ps']: logger.error(f"不支持的估值指标: {metric}") return pd.DataFrame() # 获取行业所有股票 stock_codes = self.get_industry_stocks(industry_name) if not stock_codes: return pd.DataFrame() # 构建查询 - 只计算每天的行业平均值和参与计算的股票数量 query = text(f""" WITH valid_data AS ( SELECT `timestamp`, symbol, {metric} FROM gp_day_data WHERE symbol IN :stock_codes AND `timestamp` >= :start_date AND {metric} > 0 AND {metric} < 1000 -- 过滤掉极端异常值 ) SELECT `timestamp`, AVG({metric}) as avg_{metric}, COUNT(*) as stock_count FROM valid_data GROUP BY `timestamp` ORDER BY `timestamp` """) with self.engine.connect() as conn: # 获取汇总数据 df = pd.read_sql( query, conn, params={"stock_codes": tuple(stock_codes), "start_date": start_date} ) if df.empty: logger.warning(f"未找到行业 {industry_name} 的估值数据") return pd.DataFrame() logger.info(f"成功获取行业 {industry_name} 的{metric.upper()}数据，共 {len(df)} 条记录") return df except Exception as e: logger.error(f"获取行业估值数据失败: {e}") return pd.DataFrame() def calculate_industry_percentiles(self, data: pd.DataFrame, metric: str = 'pe') -> Dict: """ 计算行业估值指标的当前分位数及历史统计值 计算的是行业平均PE/PB/PS在其历史分布中的百分位，以及历史最大值、最小值、四分位数等 Args: data: 历史数据DataFrame metric: 估值指标，pe、pb或ps Returns: 包含分位数信息的字典 """ if data.empty: logger.warning(f"数据为空，无法计算行业{metric}分位数") return {} # 验证metric参数 if metric not in ['pe', 'pb', 'ps']: logger.error(f"不支持的估值指标: {metric}") return {} # 列名 avg_col = f'avg_{metric}' # 获取最新值 latest_data = data.iloc[-1] # 计算当前均值在历史分布中的百分位 # 使用 <= 是为了计算当前值在历史数据中的累积分布函数值 # 这样可以得到，有多少比例的历史数据小于等于当前值，即当前值的百分位 percentile = (data[avg_col] <= latest_data[avg_col]).mean() * 100 # 计算行业平均PE的历史最小值、最大值、四分位数等 min_value = float(data[avg_col].min()) max_value = float(data[avg_col].max()) mean_value = float(data[avg_col].mean()) median_value = float(data[avg_col].median()) q1_value = float(data[avg_col].quantile(0.25)) q3_value = float(data[avg_col].quantile(0.75)) # 计算各种分位数 result = { 'date': latest_data['timestamp'].strftime('%Y-%m-%d'), 'current': float(latest_data[avg_col]), 'min': min_value, 'max': max_value, 'mean': mean_value, 'median': median_value, 'q1': q1_value, 'q3': q3_value, 'percentile': float(percentile), 'stock_count': int(latest_data['stock_count']) } logger.info(f"计算行业 {metric} 分位数完成: 当前{metric}={result['current']:.2f}, 百分位={result['percentile']:.2f}%") return result def get_industry_crowding_index(self, industry_name: str, start_date: str = None, end_date: str = None, use_cache: bool = True) -> pd.DataFrame: """ 计算行业交易拥挤度指标，并使用Redis缓存结果 对于拥挤度指标，固定使用3年数据，不受start_date影响 缓存时间为1天 Args: industry_name: 行业名称 start_date: 不再使用此参数，保留是为了兼容性 end_date: 结束日期（默认为当前日期） use_cache: 是否使用缓存，默认为True Returns: 包含行业拥挤度指标的DataFrame """ try: # 始终使用3年前作为开始日期 three_years_ago = (datetime.datetime.now() - datetime.timedelta(days=3*365)).strftime('%Y-%m-%d') if end_date is None: end_date = datetime.datetime.now().strftime('%Y-%m-%d') # 检查缓存 if use_cache: cache_key = f"industry_crowding:{industry_name}" cached_data = redis_client.get(cache_key) if cached_data: try: # 尝试解析缓存的JSON数据 cached_df_dict = json.loads(cached_data) logger.info(f"从缓存获取行业 {industry_name} 的拥挤度数据") # 将缓存的字典转换回DataFrame df = pd.DataFrame(cached_df_dict) # 确保trade_date列是日期类型 df['trade_date'] = pd.to_datetime(df['trade_date']) return df except Exception as cache_error: logger.warning(f"解析缓存的拥挤度数据失败，将重新查询: {cache_error}") # 获取行业所有股票 stock_codes = self.get_industry_stocks(industry_name) if not stock_codes: return pd.DataFrame() # 优化方案：分别查询市场总成交额和行业成交额，然后在Python中计算比率 # 查询1：获取每日总成交额 query_total = text(""" SELECT `timestamp` AS trade_date, SUM(amount) AS total_market_amount FROM gp_day_data WHERE `timestamp` BETWEEN :start_date AND :end_date GROUP BY `timestamp` ORDER BY `timestamp` """) # 查询2：获取行业每日成交额 query_industry = text(""" SELECT `timestamp` AS trade_date, SUM(amount) AS industry_amount FROM gp_day_data WHERE symbol IN :stock_codes AND `timestamp` BETWEEN :start_date AND :end_date GROUP BY `timestamp` ORDER BY `timestamp` """) with self.engine.connect() as conn: # 执行两个独立的查询 df_total = pd.read_sql( query_total, conn, params={"start_date": three_years_ago, "end_date": end_date} ) df_industry = pd.read_sql( query_industry, conn, params={ "stock_codes": tuple(stock_codes), "start_date": three_years_ago, "end_date": end_date } ) # 检查查询结果 if df_total.empty or df_industry.empty: logger.warning(f"未找到行业 {industry_name} 的交易数据") return pd.DataFrame() # 在Python中合并数据并计算比率 df = pd.merge(df_total, df_industry, on='trade_date', how='inner') # 计算行业成交额占比 df['industry_amount_ratio'] = (df['industry_amount'] / df['total_market_amount']) * 100 # 在Python中计算百分位 df['percentile'] = df['industry_amount_ratio'].rank(pct=True) * 100 # 添加拥挤度评级 df['crowding_level'] = pd.cut( df['percentile'], bins=[0, 20, 40, 60, 80, 100], labels=['不拥挤', '较不拥挤', '中性', '较为拥挤', '极度拥挤'] ) # 将DataFrame转换为字典，以便缓存 df_dict = df.to_dict(orient='records') # 缓存结果，有效期1天（86400秒） if use_cache: try: redis_client.set( cache_key, json.dumps(df_dict, default=str), # 使用default=str处理日期等特殊类型 ex=86400 # 1天的秒数 ) logger.info(f"已缓存行业 {industry_name} 的拥挤度数据，有效期为1天") except Exception as cache_error: logger.warning(f"缓存行业拥挤度数据失败: {cache_error}") logger.info(f"成功计算行业 {industry_name} 的拥挤度指标，共 {len(df)} 条记录") return df except Exception as e: logger.error(f"计算行业拥挤度指标失败: {e}") return pd.DataFrame() def get_industry_analysis(self, industry_name: str, metric: str = 'pe', start_date: str = None) -> Dict: """ 获取行业综合分析结果 Args: industry_name: 行业名称 metric: 估值指标（pe、pb或ps） start_date: 开始日期（默认为3年前） Returns: 行业分析结果字典，包含以下内容： - success: 是否成功 - industry_name: 行业名称 - metric: 估值指标 - analysis_date: 分析日期 - valuation: 估值数据，包含： - dates: 日期列表 - avg_values: 行业平均值列表 - stock_counts: 参与计算的股票数量列表 - percentiles: 分位数信息，包含行业平均值的历史最大值、最小值、四分位数等 - crowding（如有）: 拥挤度数据，包含： - dates: 日期列表 - ratios: 拥挤度比例列表 - percentiles: 拥挤度百分位列表 - current: 当前拥挤度信息 """ try: # 默认查询近3年数据 if start_date is None: start_date = (datetime.datetime.now() - datetime.timedelta(days=3*365)).strftime('%Y-%m-%d') # 获取估值数据 valuation_data = self.get_industry_valuation_data(industry_name, start_date, metric) if valuation_data.empty: return {"success": False, "message": f"无法获取行业 {industry_name} 的估值数据"} # 计算估值分位数 percentiles = self.calculate_industry_percentiles(valuation_data, metric) if not percentiles: return {"success": False, "message": f"无法计算行业 {industry_name} 的估值分位数"} # 获取拥挤度指标（始终使用3年数据，不受start_date影响） crowding_data = self.get_industry_crowding_index(industry_name) # 为了兼容前端，准备一些行业平均值的历史统计数据 avg_values = valuation_data[f'avg_{metric}'].tolist() # 准备返回结果 result = { "success": True, "industry_name": industry_name, "metric": metric.upper(), "analysis_date": datetime.datetime.now().strftime('%Y-%m-%d'), "valuation": { "dates": valuation_data['timestamp'].dt.strftime('%Y-%m-%d').tolist(), "avg_values": avg_values, # 填充行业平均值的历史统计线 "min_values": [percentiles['min']] * len(avg_values), # 行业平均PE历史最小值 "max_values": [percentiles['max']] * len(avg_values), # 行业平均PE历史最大值 "q1_values": [percentiles['q1']] * len(avg_values), # 行业平均PE历史第一四分位数 "q3_values": [percentiles['q3']] * len(avg_values), # 行业平均PE历史第三四分位数 "median_values": [percentiles['median']] * len(avg_values), # 行业平均PE历史中位数 "stock_counts": valuation_data['stock_count'].tolist(), "percentiles": percentiles } } # 添加拥挤度数据（如果有） if not crowding_data.empty: current_crowding = crowding_data.iloc[-1] result["crowding"] = { "dates": crowding_data['trade_date'].dt.strftime('%Y-%m-%d').tolist(), "ratios": crowding_data['industry_amount_ratio'].tolist(), "percentiles": crowding_data['percentile'].tolist(), "current": { "date": current_crowding['trade_date'].strftime('%Y-%m-%d'), "ratio": float(current_crowding['industry_amount_ratio']), "percentile": float(current_crowding['percentile']), "level": current_crowding['crowding_level'], # 添加行业成交额比例的历史分位信息 "ratio_stats": { "min": float(crowding_data['industry_amount_ratio'].min()), "max": float(crowding_data['industry_amount_ratio'].max()), "mean": float(crowding_data['industry_amount_ratio'].mean()), "median": float(crowding_data['industry_amount_ratio'].median()), "q1": float(crowding_data['industry_amount_ratio'].quantile(0.25)), "q3": float(crowding_data['industry_amount_ratio'].quantile(0.75)), } } } return result except Exception as e: logger.error(f"获取行业综合分析失败: {e}") return {"success": False, "message": f"获取行业综合分析失败: {e}"} def get_stock_concepts(self, stock_code: str) -> List[str]: """ 获取指定股票所属的概念板块列表 Args: stock_code: 股票代码 Returns: 概念板块名称列表 """ try: # 转换股票代码格式 formatted_code = self._convert_stock_code_format(stock_code) query = text(""" SELECT DISTINCT bk_name FROM gp_gnbk WHERE gp_code = :stock_code """) with self.engine.connect() as conn: result = conn.execute(query, {"stock_code": formatted_code}).fetchall() if result: return [row[0] for row in result] else: logger.warning(f"未找到股票 {stock_code} 的概念板块数据") return [] except Exception as e: logger.error(f"获取股票概念板块失败: {e}") return [] def _convert_stock_code_format(self, stock_code: str) -> str: """ 转换股票代码格式 Args: stock_code: 原始股票代码，格式如 "600519.SH" Returns: 转换后的股票代码，格式如 "SH600519" """ try: code, market = stock_code.split('.') return f"{market}{code}" except Exception as e: logger.error(f"转换股票代码格式失败: {str(e)}") return stock_code