# -*- coding: utf-8 -*- """ 智能分析系统（股票） - 股票市场数据分析系统 开发者：熊猫大侠 版本：v2.1.0 许可证：MIT License """ # scenario_predictor.py import os import numpy as np import pandas as pd from datetime import datetime, timedelta import openai """ """ class ScenarioPredictor: def __init__(self, analyzer, openai_api_key=None, openai_model=None): self.analyzer = analyzer self.openai_api_key = os.getenv('OPENAI_API_KEY', os.getenv('OPENAI_API_KEY')) self.openai_api_url = os.getenv('OPENAI_API_URL') self.openai_model = os.getenv('OPENAI_API_MODEL', 'gemini-2.0-pro-exp-02-05') def generate_scenarios(self, stock_code, market_type='A', days=60): """生成乐观、中性、悲观三种市场情景预测""" try: # 获取股票数据和技术指标 df = self.analyzer.get_stock_data(stock_code, market_type) df = self.analyzer.calculate_indicators(df) # 获取股票信息 stock_info = self.analyzer.get_stock_info(stock_code) # 计算基础数据 current_price = df.iloc[-1]['close'] avg_volatility = df['Volatility'].mean() # 根据历史波动率计算情景 scenarios = self._calculate_scenarios(df, days) # 使用AI生成各情景的分析 if self.openai_api_key: ai_analysis = self._generate_ai_analysis(stock_code, stock_info, df, scenarios) scenarios.update(ai_analysis) return scenarios except Exception as e: print(f"生成情景预测出错: {str(e)}") return {} def _calculate_scenarios(self, df, days): """基于历史数据计算三种情景的价格预测""" current_price = df.iloc[-1]['close'] # 计算历史波动率和移动均线 volatility = df['Volatility'].mean() / 100 # 转换为小数 daily_volatility = volatility / np.sqrt(252) # 转换为日波动率 ma20 = df.iloc[-1]['MA20'] ma60 = df.iloc[-1]['MA60'] # 计算乐观情景（上涨至压力位或突破） optimistic_return = 0.15 # 15%上涨 if df.iloc[-1]['BB_upper'] > current_price: optimistic_target = df.iloc[-1]['BB_upper'] * 1.05 # 突破上轨5% else: optimistic_target = current_price * (1 + optimistic_return) # 计算中性情景（震荡，围绕当前价格或20日均线波动） neutral_target = (current_price + ma20) / 2 # 计算悲观情景（下跌至支撑位或跌破） pessimistic_return = -0.12 # 12%下跌 if df.iloc[-1]['BB_lower'] < current_price: pessimistic_target = df.iloc[-1]['BB_lower'] * 0.95 # 跌破下轨5% else: pessimistic_target = current_price * (1 + pessimistic_return) # 计算预期时间 time_periods = np.arange(1, days + 1) # 生成乐观路径 opt_path = [current_price] for _ in range(days): daily_return = (optimistic_target / current_price) ** (1 / days) - 1 random_component = np.random.normal(0, daily_volatility) new_price = opt_path[-1] * (1 + daily_return + random_component / 2) opt_path.append(new_price) # 生成中性路径 neu_path = [current_price] for _ in range(days): daily_return = (neutral_target / current_price) ** (1 / days) - 1 random_component = np.random.normal(0, daily_volatility) new_price = neu_path[-1] * (1 + daily_return + random_component) neu_path.append(new_price) # 生成悲观路径 pes_path = [current_price] for _ in range(days): daily_return = (pessimistic_target / current_price) ** (1 / days) - 1 random_component = np.random.normal(0, daily_volatility) new_price = pes_path[-1] * (1 + daily_return + random_component / 2) pes_path.append(new_price) # 生成日期序列 start_date = datetime.now() dates = [(start_date + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(days + 1)] # 组织结果 return { 'current_price': current_price, 'optimistic': { 'target_price': optimistic_target, 'change_percent': (optimistic_target / current_price - 1) * 100, 'path': dict(zip(dates, opt_path)) }, 'neutral': { 'target_price': neutral_target, 'change_percent': (neutral_target / current_price - 1) * 100, 'path': dict(zip(dates, neu_path)) }, 'pessimistic': { 'target_price': pessimistic_target, 'change_percent': (pessimistic_target / current_price - 1) * 100, 'path': dict(zip(dates, pes_path)) } } def _generate_ai_analysis(self, stock_code, stock_info, df, scenarios): """使用AI生成各情景的分析说明""" try: openai.api_key = self.openai_api_key openai.api_base = self.openai_api_url # 提取关键数据 current_price = df.iloc[-1]['close'] ma5 = df.iloc[-1]['MA5'] ma20 = df.iloc[-1]['MA20'] ma60 = df.iloc[-1]['MA60'] rsi = df.iloc[-1]['RSI'] macd = df.iloc[-1]['MACD'] signal = df.iloc[-1]['Signal'] # 构建提示词 prompt = f"""分析股票{stock_code}（{stock_info.get('股票名称', '未知')}）的三种市场情景: 1. 当前数据: - 当前价格: {current_price} - 均线: MA5={ma5}, MA20={ma20}, MA60={ma60} - RSI: {rsi} - MACD: {macd}, Signal: {signal} 2. 预测目标价: - 乐观情景: {scenarios['optimistic']['target_price']:.2f} ({scenarios['optimistic']['change_percent']:.2f}%) - 中性情景: {scenarios['neutral']['target_price']:.2f} ({scenarios['neutral']['change_percent']:.2f}%) - 悲观情景: {scenarios['pessimistic']['target_price']:.2f} ({scenarios['pessimistic']['change_percent']:.2f}%) 请为每种情景提供简短分析(每种情景100字以内)，包括可能的触发条件和风险因素。格式为JSON: {{ "optimistic_analysis": "乐观情景分析...", "neutral_analysis": "中性情景分析...", "pessimistic_analysis": "悲观情景分析..." }} """ # 调用AI API response = openai.ChatCompletion.create( model=self.openai_model, messages=[ {"role": "system", "content": "你是专业的股票分析师，擅长技术分析和情景预测。"}, {"role": "user", "content": prompt} ], temperature=0.7 ) # 解析AI回复 import json try: analysis = json.loads(response.choices[0].message.content) return analysis except: # 如果解析失败，尝试从文本中提取JSON import re json_match = re.search(r'```json\s*([\s\S]*?)\s*```', response.choices[0].message.content) if json_match: json_str = json_match.group(1) return json.loads(json_str) else: return { "optimistic_analysis": "乐观情景分析暂无", "neutral_analysis": "中性情景分析暂无", "pessimistic_analysis": "悲观情景分析暂无" } except Exception as e: print(f"生成AI分析出错: {str(e)}") return { "optimistic_analysis": "乐观情景分析暂无", "neutral_analysis": "中性情景分析暂无", "pessimistic_analysis": "悲观情景分析暂无" }