# 由 Copilot 生成 - AI 股票分析師 (含批次分析功能) import subprocess import sys import os from datetime import datetime # 環境檢測 IS_HUGGINGFACE_SPACE = "SPACE_ID" in os.environ print(f"運行環境: {'Hugging Face Spaces' if IS_HUGGINGFACE_SPACE else '本地環境'}") # 檢查並安裝所需套件的函數 def install_package(package_name): try: __import__(package_name) except ImportError: print(f"正在安裝 {package_name}...") subprocess.check_call([sys.executable, "-m", "pip", "install", package_name]) # 安裝必要套件 required_packages = [ "torch>=2.0.0", "torchvision>=0.15.0", "torchaudio>=2.0.0", "yfinance>=0.2.18", "gradio>=4.0.0", "pandas>=1.5.0", "numpy>=1.21.0", "matplotlib>=3.5.0", "plotly>=5.0.0", "beautifulsoup4>=4.11.0", "requests>=2.28.0", "transformers>=4.21.0", "accelerate>=0.20.0", "tokenizers>=0.13.0" ] for package in required_packages: package_name = package.split(">=")[0].split("==")[0] if package_name == "beautifulsoup4": package_name = "bs4" try: __import__(package_name) except ImportError: print(f"正在安裝 {package}...") subprocess.check_call([sys.executable, "-m", "pip", "install", package]) # 現在導入所有套件 import gradio as gr import yfinance as yf import pandas as pd import numpy as np import matplotlib.pyplot as plt import plotly.graph_objects as go import plotly.express as px from datetime import datetime, timedelta import requests from bs4 import BeautifulSoup from transformers import pipeline import warnings warnings.filterwarnings('ignore') # 初始化 Hugging Face 模型 print("正在載入 AI 模型...") # 嘗試載入模型,如果失敗則使用較輕量的替代方案 try: sentiment_analyzer = pipeline("sentiment-analysis", model="ProsusAI/finbert") print("FinBERT 情感分析模型載入成功") except Exception as e: print(f"FinBERT 載入失敗,嘗試替代模型: {e}") try: sentiment_analyzer = pipeline("sentiment-analysis", model="nlptown/bert-base-multilingual-uncased-sentiment") print("多語言情感分析模型載入成功") except Exception as e2: print(f"替代模型載入失敗: {e2}") sentiment_analyzer = None try: summarizer = pipeline("summarization", model="facebook/bart-large-cnn") print("BART 摘要模型載入成功") except Exception as e: print(f"BART 載入失敗,嘗試替代模型: {e}") try: summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6") print("DistilBART 摘要模型載入成功") except Exception as e2: print(f"摘要模型載入失敗: {e2}") summarizer = None class StockAnalyzer: def __init__(self): self.data = None self.symbol = None def fetch_stock_data(self, symbol, period="1y"): """獲取股票歷史數據""" try: ticker = yf.Ticker(symbol) self.data = ticker.history(period=period) self.symbol = symbol # 獲取股票資訊 info = ticker.info stock_name = info.get('longName', info.get('shortName', symbol)) return True, f"成功獲取 {symbol} 的歷史數據", stock_name except Exception as e: return False, f"數據獲取失敗: {str(e)}", None def get_stock_info(self, symbol): """獲取股票基本資訊""" try: ticker = yf.Ticker(symbol) info = ticker.info current_price = self.data['Close'].iloc[-1] if self.data is not None else None stock_name = info.get('longName', info.get('shortName', symbol)) return { 'name': stock_name, 'current_price': current_price, 'symbol': symbol } except Exception as e: return { 'name': symbol, 'current_price': None, 'symbol': symbol } def calculate_technical_indicators(self): """計算技術指標""" if self.data is None: return None df = self.data.copy() # 移動平均線 df['MA5'] = df['Close'].rolling(window=5).mean() df['MA20'] = df['Close'].rolling(window=20).mean() df['MA60'] = df['Close'].rolling(window=60).mean() # RSI 相對強弱指標 delta = df['Close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['RSI'] = 100 - (100 / (1 + rs)) # MACD exp1 = df['Close'].ewm(span=12).mean() exp2 = df['Close'].ewm(span=26).mean() df['MACD'] = exp1 - exp2 df['MACD_signal'] = df['MACD'].ewm(span=9).mean() # 布林通道 df['BB_middle'] = df['Close'].rolling(window=20).mean() bb_std = df['Close'].rolling(window=20).std() df['BB_upper'] = df['BB_middle'] + (bb_std * 2) df['BB_lower'] = df['BB_middle'] - (bb_std * 2) return df def get_news_sentiment(self, symbol): """獲取並分析新聞情感""" try: # 模擬新聞標題(實際應用中需要接入新聞 API) sample_news = [ f"{symbol} 股價創新高,投資人信心大增", f"市場關注 {symbol} 最新財報表現", f"{symbol} 面臨供應鏈挑戰,股價承壓", f"分析師上調 {symbol} 目標價,看好後市", f"{symbol} 技術創新獲得市場認可" ] sentiments = [] # 檢查情感分析模型是否可用 if sentiment_analyzer is None: # 如果模型不可用,返回模擬的情感分析結果 for news in sample_news: # 簡單的關鍵詞情感分析替代方案 positive_words = ['創新高', '信心大增', '上調', '看好', '創新', '獲得認可'] negative_words = ['挑戰', '承壓', '面臨', '下滑'] score = 0.5 # 中性 sentiment = 'NEUTRAL' for word in positive_words: if word in news: score = 0.8 sentiment = 'POSITIVE' break for word in negative_words: if word in news: score = 0.8 sentiment = 'NEGATIVE' break sentiments.append({ 'text': news, 'sentiment': sentiment, 'score': score }) else: # 使用 AI 模型進行情感分析 for news in sample_news: result = sentiment_analyzer(news)[0] sentiments.append({ 'text': news, 'sentiment': result['label'], 'score': result['score'] }) return sentiments except Exception as e: return [{'text': f'新聞分析暫時無法使用: {str(e)}', 'sentiment': 'NEUTRAL', 'score': 0.5}] def analyze_sentiment_summary(self, sentiments): """分析情感摘要""" if not sentiments: return "中性" positive_count = sum(1 for s in sentiments if s['sentiment'] == 'POSITIVE') negative_count = sum(1 for s in sentiments if s['sentiment'] == 'NEGATIVE') if positive_count > negative_count: return "偏樂觀" elif negative_count > positive_count: return "偏悲觀" else: return "中性" def calculate_prediction_probabilities(self, technical_signals, sentiment, recent_data): """計算上漲和下跌機率""" # 計算技術面得分 bullish_signals = sum(1 for signal in technical_signals if "多頭" in signal or "機會" in signal) bearish_signals = sum(1 for signal in technical_signals if "空頭" in signal or "警訊" in signal) neutral_signals = len(technical_signals) - bullish_signals - bearish_signals # 技術面得分 (-1 到 1) total_signals = len(technical_signals) if total_signals > 0: tech_score = (bullish_signals - bearish_signals) / total_signals else: tech_score = 0 # 情感得分 (-1 到 1) sentiment_score = 0 if sentiment == "偏樂觀": sentiment_score = 0.6 elif sentiment == "偏悲觀": sentiment_score = -0.6 else: sentiment_score = 0 # 價格動量得分 price_change = ((recent_data['Close'].iloc[-1] - recent_data['Close'].iloc[-5]) / recent_data['Close'].iloc[-5]) * 100 momentum_score = np.tanh(price_change / 10) # 標準化到 -1 到 1 # RSI 得分 latest = recent_data.iloc[-1] rsi = latest.get('RSI', 50) if rsi > 70: rsi_score = -0.5 # 超買,偏空 elif rsi < 30: rsi_score = 0.5 # 超賣,偏多 else: rsi_score = (50 - rsi) / 100 # 標準化 # MACD 得分 macd_score = 0 if 'MACD' in latest and 'MACD_signal' in latest: if latest['MACD'] > latest['MACD_signal']: macd_score = 0.3 else: macd_score = -0.3 # 綜合得分計算(加權平均) weights = { 'tech': 0.25, 'sentiment': 0.20, 'momentum': 0.25, 'rsi': 0.15, 'macd': 0.15 } total_score = ( tech_score * weights['tech'] + sentiment_score * weights['sentiment'] + momentum_score * weights['momentum'] + rsi_score * weights['rsi'] + macd_score * weights['macd'] ) # 將得分轉換為機率 (使用 sigmoid 函數) def sigmoid(x): return 1 / (1 + np.exp(-x * 3)) # 放大 3 倍讓機率更明顯 up_probability = sigmoid(total_score) * 100 down_probability = sigmoid(-total_score) * 100 sideways_probability = 100 - up_probability - down_probability # 確保機率總和為 100% total_prob = up_probability + down_probability + sideways_probability up_probability = (up_probability / total_prob) * 100 down_probability = (down_probability / total_prob) * 100 sideways_probability = (sideways_probability / total_prob) * 100 return { 'up': max(15, min(75, up_probability)), # 限制在 15%-75% 範圍內 'down': max(15, min(75, down_probability)), # 限制在 15%-75% 範圍內 'sideways': max(10, sideways_probability), # 至少 10% 'confidence': abs(total_score) # 信心度 } def generate_comprehensive_prediction(self, technical_signals, sentiment, recent_data): """生成綜合預測報告""" # 計算價格變化 price_change = ((recent_data['Close'].iloc[-1] - recent_data['Close'].iloc[-5]) / recent_data['Close'].iloc[-5]) * 100 # 計算預測機率 probabilities = self.calculate_prediction_probabilities(technical_signals, sentiment, recent_data) # 確定主要預測方向 max_prob = max(probabilities['up'], probabilities['down'], probabilities['sideways']) if probabilities['up'] == max_prob: main_direction = "看多" direction_emoji = "📈" elif probabilities['down'] == max_prob: main_direction = "看空" direction_emoji = "📉" else: main_direction = "盤整" direction_emoji = "➡️" # 信心度描述 confidence = probabilities['confidence'] if confidence > 0.4: confidence_desc = "高信心" elif confidence > 0.2: confidence_desc = "中等信心" else: confidence_desc = "低信心" report = f""" ## 📊 {self.symbol} AI 分析報告 ### 📈 技術面分析: {chr(10).join(f"• {signal}" for signal in technical_signals)} ### 💭 市場情感:{sentiment} ### 📊 近期表現: - 5日漲跌幅:{price_change:+.2f}% - 當前價位:${recent_data['Close'].iloc[-1]:.2f} ### 🤖 AI 預測機率(短期 1-7天): | 方向 | 機率 | 說明 | |------|------|------| | 📈 **上漲** | **{probabilities['up']:.1f}%** | 股價向上突破的可能性 | | 📉 **下跌** | **{probabilities['down']:.1f}%** | 股價向下修正的可能性 | | ➡️ **盤整** | **{probabilities['sideways']:.1f}%** | 股價維持震盪的可能性 | ### 🎯 主要預測方向: {direction_emoji} **{main_direction}** ({confidence_desc} - {confidence*100:.0f}%) ### 📋 投資建議: """ # 根據最高機率給出建議 if probabilities['up'] > 50: report += """ - 💡 **多頭策略**:考慮逢低加碼或持有現有部位 - 🎯 **目標設定**:關注上方阻力位,設定合理獲利目標 - 🛡️ **風險管理**:設置止損點保護資本""" elif probabilities['down'] > 50: report += """ - 💡 **防守策略**:考慮減碼或等待更佳進場點 - 🎯 **支撐觀察**:留意下方支撐位是否守住 - 🛡️ **風險管理**:避免追高,控制倉位大小""" else: report += """ - 💡 **中性策略**:保持觀望,等待明確方向訊號 - 🎯 **區間操作**:可考慮在支撐阻力區間內操作 - 🛡️ **風險管理**:小部位測試,嚴格執行停損""" report += f""" ### 📅 中期展望(1個月): 基於當前技術面和市場情緒分析,建議持續關注: - 關鍵技術位:支撐與阻力區間 - 市場情緒變化:新聞面和資金流向 - 整體大盤走勢:系統性風險評估 ⚠️ **風險提醒**:此分析基於歷史數據和 AI 模型預測,僅供參考。投資有風險,請謹慎評估並做好風險管理! --- *預測信心度:{confidence*100:.0f}% | 分析時間:{datetime.now().strftime('%Y-%m-%d %H:%M')}* """ return report def generate_prediction(self, df, news_sentiment): """生成預測分析""" if df is None or len(df) < 30: return "數據不足,無法進行預測分析" # 獲取最新數據 latest = df.iloc[-1] recent_data = df.tail(20) # 技術分析信號 technical_signals = [] # 價格趋势 if latest['Close'] > latest['MA20']: technical_signals.append("價格在20日均線之上(多頭信號)") else: technical_signals.append("價格在20日均線之下(空頭信號)") # RSI 分析 rsi = latest['RSI'] if rsi > 70: technical_signals.append(f"RSI({rsi:.1f}) 超買警訊") elif rsi < 30: technical_signals.append(f"RSI({rsi:.1f}) 超賣機會") else: technical_signals.append(f"RSI({rsi:.1f}) 正常範圍") # MACD 分析 if latest['MACD'] > latest['MACD_signal']: technical_signals.append("MACD 呈現多頭排列") else: technical_signals.append("MACD 呈現空頭排列") # 新聞情感分析 sentiment_summary = self.analyze_sentiment_summary(news_sentiment) # 綜合預測 prediction = self.generate_comprehensive_prediction(technical_signals, sentiment_summary, recent_data) return prediction # 創建分析器實例 analyzer = StockAnalyzer() def analyze_stock(symbol): """主要分析函數""" if not symbol.strip(): return None, "請輸入股票代碼", "" # 獲取數據 result = analyzer.fetch_stock_data(symbol.upper()) if len(result) == 3: success, message, stock_name = result else: success, message = result stock_name = None if not success: return None, message, "" # 計算技術指標 df = analyzer.calculate_technical_indicators() # 創建價格圖表 fig = go.Figure() # 添加K線圖 fig.add_trace(go.Candlestick( x=df.index, open=df['Open'], high=df['High'], low=df['Low'], close=df['Close'], name='價格' )) # 添加移動平均線 fig.add_trace(go.Scatter(x=df.index, y=df['MA5'], name='MA5', line=dict(color='orange'))) fig.add_trace(go.Scatter(x=df.index, y=df['MA20'], name='MA20', line=dict(color='blue'))) fig.update_layout( title=f'{symbol} 股價走勢與技術指標', xaxis_title='日期', yaxis_title='價格', height=600 ) # 獲取新聞情感 news_sentiment = analyzer.get_news_sentiment(symbol) # 生成預測 prediction = analyzer.generate_prediction(df, news_sentiment) return fig, "分析完成!", prediction def create_results_table(results): """創建結果表格""" if not results: return "" # 創建表格 HTML table_html = """
| 股票代號 | 股票名稱 | 當前價格 | 上漲機率(%) | 下跌機率(%) | 盤整機率(%) | 信心度(%) | 預測方向 | 狀態 |
|---|---|---|---|---|---|---|---|---|
| {result['symbol']} | {result['name']} | {result['current_price']} | {result['up_probability']} | {result['down_probability']} | {result['sideways_probability']} | {result['confidence']} | {direction} | {status} |