industry_analyzer.py

# -*- coding: utf-8 -*-
"""
智能分析系统（股票） - 股票市场数据分析系统
开发者：熊猫大侠
版本：v2.1.0
许可证：MIT License
"""
# industry_analyzer.py
import logging
import random
import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta


class IndustryAnalyzer:
    def __init__(self):
        """初始化行业分析类"""
        self.data_cache = {}
        self.industry_code_map = {}  # 缓存行业名称到代码的映射

        # 设置日志记录
        logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s - %(levelname)s - %(message)s')
        self.logger = logging.getLogger(__name__)

    def get_industry_fund_flow(self, symbol="即时"):
        """获取行业资金流向数据"""
        try:
            # 缓存键
            cache_key = f"industry_fund_flow_{symbol}"

            # 检查缓存
            if cache_key in self.data_cache:
                cache_time, cached_data = self.data_cache[cache_key]
                # 如果缓存时间在30分钟内，直接返回
                if (datetime.now() - cache_time).total_seconds() < 1800:
                    self.logger.info(f"从缓存获取行业资金流向数据: {symbol}")
                    return cached_data

            # 获取行业资金流向数据
            self.logger.info(f"从API获取行业资金流向数据: {symbol}")
            fund_flow_data = ak.stock_fund_flow_industry(symbol=symbol)

            # 打印列名以便调试
            self.logger.info(f"行业资金流向数据列名: {fund_flow_data.columns.tolist()}")

            # 转换为字典列表
            result = []

            if symbol == "即时":
                for _, row in fund_flow_data.iterrows():
                    try:
                        # 安全地将值转换为对应的类型
                        item = {
                            "rank": self._safe_int(row["序号"]),
                            "industry": str(row["行业"]),
                            "index": self._safe_float(row["行业指数"]),
                            "change": self._safe_percent(row["行业-涨跌幅"]),
                            "inflow": self._safe_float(row["流入资金"]),
                            "outflow": self._safe_float(row["流出资金"]),
                            "netFlow": self._safe_float(row["净额"]),
                            "companyCount": self._safe_int(row["公司家数"])
                        }

                        # 添加领涨股相关数据，如果存在
                        if "领涨股" in row:
                            item["leadingStock"] = str(row["领涨股"])
                        if "领涨股-涨跌幅" in row:
                            item["leadingStockChange"] = self._safe_percent(row["领涨股-涨跌幅"])
                        if "当前价" in row:
                            item["leadingStockPrice"] = self._safe_float(row["当前价"])

                        result.append(item)
                    except Exception as e:
                        self.logger.warning(f"处理行业资金流向数据行时出错: {str(e)}")
                        continue
            else:
                for _, row in fund_flow_data.iterrows():
                    try:
                        item = {
                            "rank": self._safe_int(row["序号"]),
                            "industry": str(row["行业"]),
                            "companyCount": self._safe_int(row["公司家数"]),
                            "index": self._safe_float(row["行业指数"]),
                            "change": self._safe_percent(row["阶段涨跌幅"]),
                            "inflow": self._safe_float(row["流入资金"]),
                            "outflow": self._safe_float(row["流出资金"]),
                            "netFlow": self._safe_float(row["净额"])
                        }
                        result.append(item)
                    except Exception as e:
                        self.logger.warning(f"处理行业资金流向数据行时出错: {str(e)}")
                        continue

            # 缓存结果
            self.data_cache[cache_key] = (datetime.now(), result)

            return result

        except Exception as e:
            self.logger.error(f"获取行业资金流向数据失败: {str(e)}")
            # 返回更详细的错误信息，包括堆栈跟踪
            import traceback
            self.logger.error(traceback.format_exc())
            return []

    def _safe_float(self, value):
        """安全地将值转换为浮点数"""
        try:
            if pd.isna(value):
                return 0.0
            return float(value)
        except:
            return 0.0

    def _safe_int(self, value):
        """安全地将值转换为整数"""
        try:
            if pd.isna(value):
                return 0
            return int(value)
        except:
            return 0

    def _safe_percent(self, value):
        """安全地将百分比值转换为字符串格式"""
        try:
            if pd.isna(value):
                return "0.00"

            # 如果是字符串并包含%，移除%符号
            if isinstance(value, str) and "%" in value:
                return value.replace("%", "")

            # 如果是数值，直接转换成字符串
            return str(float(value))
        except:
            return "0.00"

    def _get_industry_code(self, industry_name):
        """获取行业名称对应的板块代码"""
        try:
            # 如果已经缓存了行业代码映射，直接使用
            if not self.industry_code_map:
                # 获取东方财富行业板块名称及代码
                industry_list = ak.stock_board_industry_name_em()

                # 创建行业名称到代码的映射
                for _, row in industry_list.iterrows():
                    if '板块名称' in industry_list.columns and '板块代码' in industry_list.columns:
                        name = row['板块名称']
                        code = row['板块代码']
                        self.industry_code_map[name] = code

                self.logger.info(f"成功获取到 {len(self.industry_code_map)} 个行业代码映射")

            # 尝试精确匹配
            if industry_name in self.industry_code_map:
                return self.industry_code_map[industry_name]

            # 尝试模糊匹配
            for name, code in self.industry_code_map.items():
                if industry_name in name or name in industry_name:
                    self.logger.info(f"行业名称 '{industry_name}' 模糊匹配到 '{name}'，代码: {code}")
                    return code

            # 如果找不到匹配项，则返回None
            self.logger.warning(f"未找到行业 '{industry_name}' 对应的代码")
            return None

        except Exception as e:
            self.logger.error(f"获取行业代码时出错: {str(e)}")
            import traceback
            self.logger.error(traceback.format_exc())
            return None

    def get_industry_stocks(self, industry):
        """获取行业成分股"""
        try:
            # 缓存键
            cache_key = f"industry_stocks_{industry}"

            # 检查缓存
            if cache_key in self.data_cache:
                cache_time, cached_data = self.data_cache[cache_key]
                # 如果缓存时间在1小时内，直接返回
                if (datetime.now() - cache_time).total_seconds() < 3600:
                    self.logger.info(f"从缓存获取行业成分股: {industry}")
                    return cached_data

            # 获取行业成分股
            self.logger.info(f"获取 {industry} 行业成分股")

            result = []
            try:
                # 1. 首先尝试直接使用行业名称
                try:
                    stocks = ak.stock_board_industry_cons_em(symbol=industry)
                    self.logger.info(f"使用行业名称 '{industry}' 成功获取成分股")
                except Exception as direct_error:
                    self.logger.warning(f"使用行业名称获取成分股失败: {str(direct_error)}")
                    # 2. 尝试使用行业代码
                    industry_code = self._get_industry_code(industry)
                    if industry_code:
                        self.logger.info(f"尝试使用行业代码 {industry_code} 获取成分股")
                        stocks = ak.stock_board_industry_cons_em(symbol=industry_code)
                    else:
                        # 如果无法获取行业代码，抛出异常，进入模拟数据生成
                        raise ValueError(f"无法找到行业 '{industry}' 对应的代码")

                # 打印列名以便调试
                self.logger.info(f"行业成分股数据列名: {stocks.columns.tolist()}")

                # 转换为字典列表
                if not stocks.empty:
                    for _, row in stocks.iterrows():
                        try:
                            item = {
                                "code": str(row["代码"]),
                                "name": str(row["名称"]),
                                "price": self._safe_float(row["最新价"]),
                                "change": self._safe_float(row["涨跌幅"]),
                                "change_amount": self._safe_float(row["涨跌额"]) if "涨跌额" in row else 0.0,
                                "volume": self._safe_float(row["成交量"]) if "成交量" in row else 0.0,
                                "turnover": self._safe_float(row["成交额"]) if "成交额" in row else 0.0,
                                "amplitude": self._safe_float(row["振幅"]) if "振幅" in row else 0.0,
                                "turnover_rate": self._safe_float(row["换手率"]) if "换手率" in row else 0.0
                            }
                            result.append(item)
                        except Exception as e:
                            self.logger.warning(f"处理行业成分股数据行时出错: {str(e)}")
                            continue

            except Exception as e:
                # 3. 如果上述方法都失败，生成模拟数据
                self.logger.warning(f"无法通过API获取行业成分股，使用模拟数据: {str(e)}")
                result = self._generate_mock_industry_stocks(industry)

            # 缓存结果
            self.data_cache[cache_key] = (datetime.now(), result)

            return result

        except Exception as e:
            self.logger.error(f"获取行业成分股失败: {str(e)}")
            import traceback
            self.logger.error(traceback.format_exc())
            return []

    def _generate_mock_industry_stocks(self, industry):
        """生成模拟的行业成分股数据"""
        self.logger.info(f"生成行业 {industry} 的模拟成分股数据")

        # 使用来自资金流向的行业数据获取该行业的基本信息
        fund_flow_data = self.get_industry_fund_flow("即时")
        industry_data = next((item for item in fund_flow_data if item["industry"] == industry), None)

        company_count = 20  # 默认值
        if industry_data and "companyCount" in industry_data:
            company_count = min(industry_data["companyCount"], 30)  # 限制最多30只股票

        # 生成模拟股票
        result = []
        for i in range(company_count):
            # 生成6位数字的股票代码，确保前缀是0或6
            prefix = "6" if i % 2 == 0 else "0"
            code = prefix + str(100000 + i).zfill(5)[-5:]

            # 生成股票价格和涨跌幅
            price = round(random.uniform(10, 100), 2)
            change = round(random.uniform(-5, 5), 2)

            # 生成成交量和成交额
            volume = round(random.uniform(100000, 10000000))
            turnover = round(volume * price / 10000, 2)  # 转换为万元

            # 生成换手率和振幅
            turnover_rate = round(random.uniform(0.5, 5), 2)
            amplitude = round(random.uniform(1, 10), 2)

            item = {
                "code": code,
                "name": f"{industry}股{i + 1}",
                "price": price,
                "change": change,
                "change_amount": round(price * change / 100, 2),
                "volume": volume,
                "turnover": turnover,
                "amplitude": amplitude,
                "turnover_rate": turnover_rate
            }
            result.append(item)

        # 按涨跌幅排序
        result.sort(key=lambda x: x["change"], reverse=True)

        return result

    def get_industry_detail(self, industry):
        """获取行业详细信息"""
        try:
            # 获取行业资金流向数据
            fund_flow_data = self.get_industry_fund_flow("即时")
            industry_data = next((item for item in fund_flow_data if item["industry"] == industry), None)

            if not industry_data:
                return None

            # 获取历史资金流向数据
            history_data = []

            for period in ["3日排行", "5日排行", "10日排行", "20日排行"]:
                period_data = self.get_industry_fund_flow(period)
                industry_period_data = next((item for item in period_data if item["industry"] == industry), None)

                if industry_period_data:
                    days = int(period.replace("日排行", ""))
                    date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")

                    history_data.append({
                        "date": date,
                        "inflow": industry_period_data["inflow"],
                        "outflow": industry_period_data["outflow"],
                        "netFlow": industry_period_data["netFlow"],
                        "change": industry_period_data["change"]
                    })

            # 添加即时数据
            history_data.append({
                "date": datetime.now().strftime("%Y-%m-%d"),
                "inflow": industry_data["inflow"],
                "outflow": industry_data["outflow"],
                "netFlow": industry_data["netFlow"],
                "change": industry_data["change"]
            })

            # 按日期排序
            history_data.sort(key=lambda x: x["date"])

            # 计算行业评分
            score = self.calculate_industry_score(industry_data, history_data)

            # 生成投资建议
            recommendation = self.generate_industry_recommendation(score, industry_data, history_data)

            # 构建结果
            result = {
                "industry": industry,
                "index": industry_data["index"],
                "change": industry_data["change"],
                "companyCount": industry_data["companyCount"],
                "inflow": industry_data["inflow"],
                "outflow": industry_data["outflow"],
                "netFlow": industry_data["netFlow"],
                "leadingStock": industry_data.get("leadingStock", ""),
                "leadingStockChange": industry_data.get("leadingStockChange", ""),
                "leadingStockPrice": industry_data.get("leadingStockPrice", 0),
                "score": score,
                "recommendation": recommendation,
                "flowHistory": history_data
            }

            return result

        except Exception as e:
            self.logger.error(f"获取行业详细信息失败: {str(e)}")
            import traceback
            self.logger.error(traceback.format_exc())
            return None

    def calculate_industry_score(self, industry_data, history_data):
        """计算行业评分"""
        try:
            # 基础分数为50分
            score = 50

            # 根据涨跌幅增减分数（-10到+10）
            change = float(industry_data["change"])
            if change > 3:
                score += 10
            elif change > 1:
                score += 5
            elif change < -3:
                score -= 10
            elif change < -1:
                score -= 5

            # 根据资金流向增减分数（-20到+20）
            netFlow = float(industry_data["netFlow"])

            if netFlow > 5:
                score += 20
            elif netFlow > 2:
                score += 15
            elif netFlow > 0:
                score += 10
            elif netFlow < -5:
                score -= 20
            elif netFlow < -2:
                score -= 15
            elif netFlow < 0:
                score -= 10

            # 根据历史资金流向趋势增减分数（-10到+10）
            if len(history_data) >= 2:
                net_flow_trend = 0
                for i in range(1, len(history_data)):
                    if float(history_data[i]["netFlow"]) > float(history_data[i - 1]["netFlow"]):
                        net_flow_trend += 1
                    else:
                        net_flow_trend -= 1

                if net_flow_trend > 0:
                    score += 10
                elif net_flow_trend < 0:
                    score -= 10

            # 限制分数在0-100之间
            score = max(0, min(100, score))

            return round(score)

        except Exception as e:
            self.logger.error(f"计算行业评分时出错: {str(e)}")
            return 50

    def generate_industry_recommendation(self, score, industry_data, history_data):
        """生成行业投资建议"""
        try:
            if score >= 80:
                return "行业景气度高，资金持续流入，建议积极配置"
            elif score >= 60:
                return "行业表现良好，建议适当加仓"
            elif score >= 40:
                return "行业表现一般，建议谨慎参与"
            else:
                return "行业下行趋势明显，建议减持规避风险"

        except Exception as e:
            self.logger.error(f"生成行业投资建议时出错: {str(e)}")
            return "无法生成投资建议"

    def compare_industries(self, limit=10):
        """比较不同行业的表现"""
        try:
            # 获取行业板块数据
            industry_data = ak.stock_board_industry_name_em()

            # 提取行业名称列表
            industries = industry_data['板块名称'].tolist() if '板块名称' in industry_data.columns else []

            if not industries:
                return {"error": "获取行业列表失败"}

            # 限制分析的行业数量
            industries = industries[:limit] if limit else industries

            # 分析各行业情况
            industry_results = []

            for industry in industries:
                try:
                    # 尝试获取行业板块代码
                    industry_code = None
                    for _, row in industry_data.iterrows():
                        if row['板块名称'] == industry:
                            industry_code = row['板块代码']
                            break

                    if not industry_code:
                        self.logger.warning(f"未找到行业 {industry} 的板块代码")
                        continue

                    # 尝试使用不同的参数来获取行业数据 - 不使用"3m"
                    try:
                        # 尝试不使用period参数
                        industry_info = ak.stock_board_industry_hist_em(symbol=industry_code)
                    except Exception as e1:
                        try:
                            # 尝试使用daily参数
                            industry_info = ak.stock_board_industry_hist_em(symbol=industry_code, period="daily")
                        except Exception as e2:
                            self.logger.warning(f"分析行业 {industry} 历史数据失败: {str(e1)}, {str(e2)}")
                            continue

                    # 计算行业涨跌幅
                    if not industry_info.empty:
                        latest = industry_info.iloc[0]

                        # 尝试获取涨跌幅，列名可能有变化
                        change = 0.0
                        if '涨跌幅' in latest.index:
                            change = latest['涨跌幅']
                        elif '涨跌幅' in industry_info.columns:
                            change = latest['涨跌幅']

                        # 尝试获取成交量和成交额
                        volume = 0.0
                        turnover = 0.0
                        if '成交量' in latest.index:
                            volume = latest['成交量']
                        elif '成交量' in industry_info.columns:
                            volume = latest['成交量']

                        if '成交额' in latest.index:
                            turnover = latest['成交额']
                        elif '成交额' in industry_info.columns:
                            turnover = latest['成交额']

                        industry_results.append({
                            "industry": industry,
                            "change": float(change) if change else 0.0,
                            "volume": float(volume) if volume else 0.0,
                            "turnover": float(turnover) if turnover else 0.0
                        })
                except Exception as e:
                    self.logger.error(f"分析行业 {industry} 时出错: {str(e)}")

            # 按涨跌幅排序
            industry_results.sort(key=lambda x: x.get('change', 0), reverse=True)

            return {
                "count": len(industry_results),
                "top_industries": industry_results[:5] if len(industry_results) >= 5 else industry_results,
                "bottom_industries": industry_results[-5:] if len(industry_results) >= 5 else [],
                "results": industry_results
            }

        except Exception as e:
            self.logger.error(f"比较行业表现时出错: {str(e)}")
            return {"error": f"比较行业表现时出错: {str(e)}"}