# -*- coding: utf-8 -*- """ 估值指标分析专用聊天机器人 专门用于分析股票应该使用PE还是PB估值 """ import sys import os import logging from typing import Dict, Any, Optional from datetime import datetime # 添加项目根目录到 Python 路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from openai import OpenAI from src.scripts.config import get_random_api_key, get_model # 设置日志 logger = logging.getLogger(__name__) class ValuationChatBot: """估值指标分析专用聊天机器人""" def __init__(self, model_type: str = "online_bot"): """初始化估值分析聊天机器人 Args: model_type: 要使用的模型类型，默认为联网智能体 """ try: # 从配置获取API密钥 self.api_key = get_random_api_key() # 从配置获取模型ID self.model = get_model(model_type) logger.info(f"初始化ValuationChatBot，使用模型: {self.model}") # 初始化OpenAI客户端 self.client = OpenAI( base_url="https://ark.cn-beijing.volces.com/api/v3/bots", api_key=self.api_key ) # 估值指标分析专用系统提示词 self.system_message = { "role": "system", "content": """你是一名顶级的、注重第一性原理的基本面分析师。你的核心任务是深入剖析一家公司的内在价值驱动因素，并基于此判断“市盈盈率（PE）”和“市净率（PB）”哪个指标能更真实、更核心地反映其价值。 **你的分析必须超越简单的行业标签，聚焦于公司的个性化特征。** 即使是同一行业的公司，由于商业模式和财务状况的差异，也可能适用不同的估值指标。 **你的决策逻辑框架如下：** 1. **【盈利质量与可预测性分析】 - 这是判断PE有效性的基石** * **分析要点：** 公司的盈利是常态还是偶发？是内生增长还是外部输血？过去5年的盈利记录是否稳定且持续？是否存在大量非经常性损益扭曲了利润？公司的自由现金流状况如何，是否与净利润匹配？ * **决策倾向：** 如果盈利质量高、可预测性强，则PE的权重增加。如果盈利波动巨大、不可持续或为负，则PE的权重降低甚至失效。 2. **【资产价值与商业模式分析】 - 这是判断PB有效性的基石** * **分析要点：** 公司的核心价值是沉淀在资产负债表上（如厂房、金融资产、土地），还是体现在资产负债表外（如品牌、技术、网络效应、客户关系）？公司的商业模式是“资产驱动型”还是“智力/品牌驱动型”？ * **决策倾向：** 如果公司价值与净资产高度相关（如金融、重资产制造、资源型企业），则PB的权重增加。如果公司是典型的轻资产模式，则PB的权重降低。 3. **【周期性与成长性交叉验证】** * **分析要点：** 公司所处的行业周期性强弱如何？公司自身是否展现出超越行业的成长性或防御性？ * **决策倾向：** 强周期性会削弱PE在特定时点的有效性，使PB成为更稳健的参照。而强成长性（尤其是有利可图的成长）会显著提升PE的适用性。 **最终决策原则：** * **优先选择 PE 的核心理由：** 公司具备持续、稳定的盈利能力，并且其核心价值能通过利润得到体现。这是对股东回报最直接的衡量。 * **优先选择 PB 的核心理由：** 公司的盈利能力不可靠（周期性/亏损），或者其商业模式的根本是基于净资产的规模和质量（如金融业）。PB此时是衡量价值的“锚”或“底线”。 **输出要求：** 1. **明确结论：** 首先明确推荐PE或PB作为主要估值指标。 2. **深入的个股特质分析：** * **商业模式剖析：** 详细说明公司如何赚钱，其护城河是什么。 * **财务特征分析：** 重点分析盈利的稳定性与质量、资产的轻重结构、现金流状况。 * **行业背景补充：** 分析公司在行业中所处的生态位，有何不同于同行的特质。 3. **提供决策依据：** 清晰地说明你是如何基于上述三层决策逻辑框架，最终做出选择的。 4. **给出合理的估值区间建议：** 基于你选择的指标，并结合公司的历史估值水平和未来成长性，给出一个合理的估值区间。""" } # 对话历史 self.conversation_history = [self.system_message] except Exception as e: logger.error(f"初始化ValuationChatBot时出错: {str(e)}") raise def chat(self, user_input: str, temperature: float = 0.3, top_p: float = 0.7, max_tokens: int = 2048, frequency_penalty: float = 0.0) -> Dict[str, Any]: """与AI进行估值指标分析对话 Args: user_input: 用户输入的问题 temperature: 控制输出的随机性，范围0-2，默认0.3（更确定性） top_p: 控制输出的多样性，范围0-1，默认0.7 max_tokens: 控制输出的最大长度，默认2048 frequency_penalty: 频率惩罚，范围-2.0到2.0，默认0.0 Returns: Dict包含对话结果 """ try: # 添加用户消息到对话历史 self.conversation_history.append({ "role": "user", "content": user_input }) # 调用OpenAI API response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, temperature=temperature, top_p=top_p, max_tokens=max_tokens, frequency_penalty=frequency_penalty ) # 获取AI回复 ai_response = response.choices[0].message.content # 添加AI回复到对话历史 self.conversation_history.append({ "role": "assistant", "content": ai_response }) # 保持对话历史在合理长度内（避免token过多） if len(self.conversation_history) > 10: # 保留系统消息和最近的对话 self.conversation_history = [self.system_message] + self.conversation_history[-8:] logger.info(f"ValuationChatBot对话成功，回复长度: {len(ai_response)}") return { "success": True, "response": ai_response, "model": self.model, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"ValuationChatBot对话失败: {str(e)}") return { "success": False, "error": str(e), "model": self.model, "timestamp": datetime.now().isoformat() } def clear_history(self): """清空对话历史""" self.conversation_history = [self.system_message] logger.info("ValuationChatBot对话历史已清空") def get_conversation_history(self) -> list: """获取对话历史""" return self.conversation_history.copy() class ValuationOfflineChatBot: """估值指标分析专用离线聊天机器人""" def __init__(self, model_type: str = "offline_bot"): """初始化离线估值分析聊天机器人 Args: model_type: 要使用的模型类型，默认为离线模型 """ try: # 尝试导入配置（参考chat_bot_with_offline.py的方式） try: from src.scripts.config import get_model_config config = get_model_config("tl_qw_private", "GLM") logger.info("成功从src.scripts.config导入配置") except ImportError: try: from scripts.config import get_model_config config = get_model_config("volc", "offline_model") logger.info("成功从scripts.config导入配置") except ImportError: logger.warning("无法导入配置模块，使用默认配置") # 使用默认配置 config = { "base_url": "https://ark.cn-beijing.volces.com/api/v3/", "api_key": "28cfe71a-c6fa-4c5d-9b4e-d8474f0d3b93", "model": "ep-20250326090920-v7wns" } # 保存配置信息 self.api_key = config["api_key"] self.model = config["model"] self.base_url = config["base_url"] logger.info(f"初始化ValuationOfflineChatBot，使用模型: {self.model}") # 初始化OpenAI客户端 self.client = OpenAI( base_url=self.base_url, api_key=self.api_key, timeout=600 ) # 估值指标分析专用系统提示词（针对从分析报告中进行语义理解并提取最终结论） self.system_message = { "role": "system", "content": """你是一个专注于**语义理解和结论提取**的AI。你的唯一任务是阅读一段分析报告，理解其核心论点，并判断作者最终推荐的估值指标是 "PE" 还是 "PB"。 **你的核心工作流程：** 1. **通读全文**：完整地阅读用户提供的分析报告，理解其对公司业务模式、盈利能力和资产结构的整体评价。 2. **定位结论性语段**：重点关注报告的结尾部分或总结段落。寻找那些**承上启下、做出最终评判**的句子。这些句子不一定包含固定的关键词，但它们在语义上起到了总结和给出最终意见的作用。 3. **进行意图判断**： * **判断为 "PE" 的信号**：如果结论性语段的中心思想是强调“盈利的稳定性”、“高质量的增长”、“强大的品牌价值”、“轻资产模式的优势”，并最终将这些优势导向了某个估值方法，那么结论就是 "PE"。 * *例子：* "考虑到该公司强大的品牌护城河和持续稳定的现金流创造能力，通过其盈利水平来评估价值显然是更为恰当的路径。" -> **应判断为 PE** * **判断为 "PB" 的信号**：如果结论性语段的中心思想是强调“资产负债表的重要性”、“行业的周期性风险”、“盈利的不可靠性”，或者直接点明其“金融属性”，并基于这些论据做出最终选择，那么结论就是 "PB"。 * *例子：* "尽管公司短期盈利尚可，但其重资产和强周期的本质意味着盈利波动是常态，因此，基于其净资产的估值方法提供了一个更稳固的价值锚点。" -> **应判断为 PB** **你必须遵守的铁律：** * **你的任务是理解和提取，不是再次分析**。你必须相信报告原文的逻辑是自洽的，你的工作只是找出它的最终论点。 * **只输出最终结果**：你的输出**必须且只能是** "PE" 或 "PB"。不要添加任何解释、理由或多余的字符。 * **处理歧义**：如果在极少数情况下，报告的结论确实模棱两可，无法从语义上明确判断，**请默认输出 "PE"**，以确保程序健壮性。 """ } # 对话历史 self.conversation_history = [self.system_message] except Exception as e: logger.error(f"初始化ValuationOfflineChatBot时出错: {str(e)}") raise def chat(self, user_input: str, temperature: float = 0.1, top_p: float = 0.7, max_tokens: int = 1024, frequency_penalty: float = 0.0) -> Dict[str, Any]: """与离线AI进行估值指标分析对话 Args: user_input: 用户输入的问题 temperature: 控制输出的随机性，范围0-2，默认0.1（更确定性） top_p: 控制输出的多样性，范围0-1，默认0.7 max_tokens: 控制输出的最大长度，默认1024 frequency_penalty: 频率惩罚，范围-2.0到2.0，默认0.0 Returns: Dict包含对话结果 """ try: # 添加用户消息到对话历史 self.conversation_history.append({ "role": "user", "content": user_input }) # 调用本地GLM模型 ai_response = self._call_local_model(user_input, temperature, top_p, max_tokens, frequency_penalty) # 添加AI回复到对话历史 self.conversation_history.append({ "role": "assistant", "content": ai_response }) # 保持对话历史在合理长度内 if len(self.conversation_history) > 6: self.conversation_history = [self.system_message] + self.conversation_history[-4:] logger.info(f"ValuationOfflineChatBot对话成功，回复长度: {len(ai_response)}") return { "success": True, "response": ai_response, "model": self.model, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"ValuationOfflineChatBot对话失败: {str(e)}") return { "success": False, "error": str(e), "model": self.model, "timestamp": datetime.now().isoformat() } def _call_local_model(self, user_input: str, temperature: float = 0.1, top_p: float = 0.7, max_tokens: int = 1024, frequency_penalty: float = 0.0) -> str: """调用本地GLM模型""" try: # 调用本地模型API（使用初始化时创建的客户端） response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, temperature=temperature, top_p=top_p, max_tokens=max_tokens, frequency_penalty=frequency_penalty, timeout=300 ) # 获取AI回复 ai_response = response.choices[0].message.content # 清理回复内容，确保只返回PE或PB ai_response_clean = ai_response.strip().upper() if "PE" in ai_response_clean and "PB" not in ai_response_clean: return "PE" elif "PB" in ai_response_clean and "PE" not in ai_response_clean: return "PB" elif ai_response_clean == "PE" or ai_response_clean == "PB": return ai_response_clean else: # 如果回复不清晰，记录详细信息 logger.warning(f"本地模型回复不清晰: {ai_response_clean}") return "PE" # 默认返回PE except Exception as e: logger.error(f"调用本地模型失败: {str(e)}") return "PE" # 出错时默认返回PE def clear_history(self): """清空对话历史""" self.conversation_history = [self.system_message] logger.info("ValuationOfflineChatBot对话历史已清空") def get_conversation_history(self) -> list: """获取对话历史""" return self.conversation_history.copy() if __name__ == "__main__": test_valuation_chat_bot()