Spaces:
Sleeping
Sleeping
| # 由 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 = """ | |
| <div style="overflow-x: auto; margin: 20px 0;"> | |
| <table style="width: 100%; border-collapse: collapse; font-family: Arial, sans-serif;"> | |
| <thead> | |
| <tr style="background-color: #f0f0f0;"> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: left;">股票代號</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: left;">股票名稱</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: right;">當前價格</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: right;">上漲機率(%)</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: right;">下跌機率(%)</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: right;">盤整機率(%)</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: right;">信心度(%)</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: center;">預測方向</th> | |
| <th style="border: 1px solid #ddd; padding: 12px; text-align: left;">狀態</th> | |
| </tr> | |
| </thead> | |
| <tbody> | |
| """ | |
| for result in results: | |
| # 判斷預測方向和顏色 | |
| if result['error_message']: | |
| direction = "❌ 錯誤" | |
| row_color = "#fff2f2" | |
| else: | |
| up_prob = float(result['up_probability']) | |
| down_prob = float(result['down_probability']) | |
| sideways_prob = float(result['sideways_probability']) | |
| if up_prob > down_prob and up_prob > sideways_prob: | |
| direction = "📈 看多" | |
| row_color = "#f0fff0" # 淡綠色 | |
| elif down_prob > up_prob and down_prob > sideways_prob: | |
| direction = "📉 看空" | |
| row_color = "#fff0f0" # 淡紅色 | |
| else: | |
| direction = "➡️ 盤整" | |
| row_color = "#f8f8f8" # 淡灰色 | |
| status = "✅ 成功" if not result['error_message'] else f"❌ {result['error_message'][:30]}..." | |
| table_html += f""" | |
| <tr style="background-color: {row_color};"> | |
| <td style="border: 1px solid #ddd; padding: 8px; font-weight: bold;">{result['symbol']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px;">{result['name']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: right;">{result['current_price']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: right;">{result['up_probability']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: right;">{result['down_probability']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: right;">{result['sideways_probability']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: right;">{result['confidence']}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px; text-align: center;">{direction}</td> | |
| <td style="border: 1px solid #ddd; padding: 8px;">{status}</td> | |
| </tr> | |
| """ | |
| table_html += """ | |
| </tbody> | |
| </table> | |
| </div> | |
| """ | |
| return table_html | |
| def create_batch_analysis_charts(results): | |
| """創建批次分析結果圖表""" | |
| if not results: | |
| return None, None, None, None | |
| # 過濾出成功分析的結果 | |
| success_results = [r for r in results if r['error_message'] == ''] | |
| if not success_results: | |
| return None, None, None, None | |
| # 準備數據 | |
| symbols = [r['symbol'] for r in success_results] | |
| up_probs = [float(r['up_probability']) for r in success_results] | |
| down_probs = [float(r['down_probability']) for r in success_results] | |
| sideways_probs = [float(r['sideways_probability']) for r in success_results] | |
| confidence = [float(r['confidence']) for r in success_results] | |
| # 1. 機率比較柱狀圖 | |
| fig_bar = go.Figure() | |
| fig_bar.add_trace(go.Bar(name='上漲機率', x=symbols, y=up_probs, marker_color='green', opacity=0.8)) | |
| fig_bar.add_trace(go.Bar(name='下跌機率', x=symbols, y=down_probs, marker_color='red', opacity=0.8)) | |
| fig_bar.add_trace(go.Bar(name='盤整機率', x=symbols, y=sideways_probs, marker_color='gray', opacity=0.8)) | |
| fig_bar.update_layout( | |
| title='📊 股票預測機率比較', | |
| xaxis_title='股票代號', | |
| yaxis_title='機率 (%)', | |
| barmode='group', | |
| height=500, | |
| showlegend=True, | |
| xaxis_tickangle=-45 | |
| ) | |
| # 2. 信心度散佈圖 | |
| fig_scatter = go.Figure() | |
| # 根據最高機率決定顏色 | |
| colors = [] | |
| for i in range(len(success_results)): | |
| if up_probs[i] > down_probs[i] and up_probs[i] > sideways_probs[i]: | |
| colors.append('green') # 看多 | |
| elif down_probs[i] > up_probs[i] and down_probs[i] > sideways_probs[i]: | |
| colors.append('red') # 看空 | |
| else: | |
| colors.append('gray') # 盤整 | |
| fig_scatter.add_trace(go.Scatter( | |
| x=symbols, | |
| y=confidence, | |
| mode='markers+text', | |
| marker=dict( | |
| size=[max(prob) for prob in zip(up_probs, down_probs, sideways_probs)], | |
| sizemode='diameter', | |
| sizeref=2, | |
| color=colors, | |
| opacity=0.7, | |
| line=dict(width=2, color='white') | |
| ), | |
| text=[f"{conf:.1f}%" for conf in confidence], | |
| textposition="middle center", | |
| name='信心度' | |
| )) | |
| fig_scatter.update_layout( | |
| title='🎯 預測信心度分佈 (圓圈大小=最高機率)', | |
| xaxis_title='股票代號', | |
| yaxis_title='信心度 (%)', | |
| height=500, | |
| xaxis_tickangle=-45 | |
| ) | |
| # 3. 綜合評分雷達圖 (取前6支股票) | |
| radar_data = success_results[:6] # 限制顯示數量避免過於擁擠 | |
| fig_radar = go.Figure() | |
| categories = ['上漲機率', '信心度', '綜合評分'] | |
| for i, result in enumerate(radar_data): | |
| # 計算綜合評分 (上漲機率 * 信心度 / 100) | |
| composite_score = float(result['up_probability']) * float(result['confidence']) / 100 | |
| values = [ | |
| float(result['up_probability']), | |
| float(result['confidence']), | |
| composite_score | |
| ] | |
| fig_radar.add_trace(go.Scatterpolar( | |
| r=values + [values[0]], # 閉合雷達圖 | |
| theta=categories + [categories[0]], | |
| fill='toself', | |
| name=result['symbol'], | |
| opacity=0.6 | |
| )) | |
| fig_radar.update_layout( | |
| polar=dict( | |
| radialaxis=dict( | |
| visible=True, | |
| range=[0, 100] | |
| ) | |
| ), | |
| title='📈 股票綜合評分雷達圖 (前6支)', | |
| height=500, | |
| showlegend=True | |
| ) | |
| # 4. 機率分佈餅圖統計 | |
| # 統計各種預測傾向的數量 | |
| bullish_count = sum(1 for r in success_results if float(r['up_probability']) > max(float(r['down_probability']), float(r['sideways_probability']))) | |
| bearish_count = sum(1 for r in success_results if float(r['down_probability']) > max(float(r['up_probability']), float(r['sideways_probability']))) | |
| neutral_count = len(success_results) - bullish_count - bearish_count | |
| fig_pie = go.Figure(data=[go.Pie( | |
| labels=['看多股票', '看空股票', '盤整股票'], | |
| values=[bullish_count, bearish_count, neutral_count], | |
| marker_colors=['green', 'red', 'gray'], | |
| textinfo='label+percent+value', | |
| hovertemplate='<b>%{label}</b><br>數量: %{value}<br>比例: %{percent}<extra></extra>' | |
| )]) | |
| fig_pie.update_layout( | |
| title='🥧 整體市場情緒分佈', | |
| height=400 | |
| ) | |
| return fig_bar, fig_scatter, fig_radar, fig_pie | |
| def batch_analyze_stocks(): | |
| """批次分析股票清單""" | |
| stock_list_file = "StockList.txt" | |
| # 檢查股票清單檔案是否存在 | |
| if not os.path.exists(stock_list_file): | |
| return f"❌ 找不到 {stock_list_file} 檔案!請確認檔案存在。", "", None, None, None, None, "" | |
| try: | |
| # 讀取股票清單 | |
| with open(stock_list_file, 'r', encoding='utf-8') as f: | |
| stock_symbols = [line.strip() for line in f if line.strip()] | |
| if not stock_symbols: | |
| return "❌ 股票清單檔案為空!", "", None, None, None, None, "" | |
| # 準備結果列表 | |
| results = [] | |
| progress_messages = [] | |
| progress_messages.append(f"📊 開始批次分析 {len(stock_symbols)} 支股票...") | |
| # 分析每支股票 | |
| for i, symbol in enumerate(stock_symbols, 1): | |
| progress_messages.append(f"\n🔍 正在分析 ({i}/{len(stock_symbols)}): {symbol}") | |
| try: | |
| # 獲取股票數據 | |
| result = analyzer.fetch_stock_data(symbol.upper()) | |
| if len(result) == 3: | |
| success, message, stock_name = result | |
| else: | |
| success, message = result | |
| stock_name = symbol | |
| if not success: | |
| # 記錄錯誤 | |
| results.append({ | |
| 'symbol': symbol, | |
| 'name': stock_name or symbol, | |
| 'current_price': 'N/A', | |
| 'up_probability': 'ERROR', | |
| 'down_probability': 'ERROR', | |
| 'sideways_probability': 'ERROR', | |
| 'confidence': 'ERROR', | |
| 'analysis_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), | |
| 'error_message': message | |
| }) | |
| progress_messages.append(f"❌ {symbol}: {message}") | |
| continue | |
| # 計算技術指標 | |
| df = analyzer.calculate_technical_indicators() | |
| if df is None or len(df) < 30: | |
| results.append({ | |
| 'symbol': symbol, | |
| 'name': stock_name or symbol, | |
| 'current_price': 'N/A', | |
| 'up_probability': 'ERROR', | |
| 'down_probability': 'ERROR', | |
| 'sideways_probability': 'ERROR', | |
| 'confidence': 'ERROR', | |
| 'analysis_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), | |
| 'error_message': '數據不足,無法分析' | |
| }) | |
| progress_messages.append(f"❌ {symbol}: 數據不足") | |
| continue | |
| # 獲取新聞情感 | |
| news_sentiment = analyzer.get_news_sentiment(symbol) | |
| sentiment_summary = analyzer.analyze_sentiment_summary(news_sentiment) | |
| # 計算預測機率 | |
| recent_data = df.tail(20) | |
| technical_signals = [] | |
| # 簡化的技術信號計算 | |
| latest = df.iloc[-1] | |
| if latest['Close'] > latest['MA20']: | |
| technical_signals.append("價格在20日均線之上") | |
| else: | |
| technical_signals.append("價格在20日均線之下") | |
| probabilities = analyzer.calculate_prediction_probabilities( | |
| technical_signals, sentiment_summary, recent_data | |
| ) | |
| # 獲取股票資訊 | |
| stock_info = analyzer.get_stock_info(symbol) | |
| # 記錄成功結果 | |
| results.append({ | |
| 'symbol': symbol, | |
| 'name': stock_info['name'], | |
| 'current_price': f"{latest['Close']:.2f}" if latest['Close'] else 'N/A', | |
| 'up_probability': f"{probabilities['up']:.1f}", | |
| 'down_probability': f"{probabilities['down']:.1f}", | |
| 'sideways_probability': f"{probabilities['sideways']:.1f}", | |
| 'confidence': f"{probabilities['confidence']*100:.1f}", | |
| 'analysis_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), | |
| 'error_message': '' | |
| }) | |
| progress_messages.append(f"✅ {symbol}: 分析完成") | |
| except Exception as e: | |
| # 處理未預期的錯誤 | |
| results.append({ | |
| 'symbol': symbol, | |
| 'name': symbol, | |
| 'current_price': 'N/A', | |
| 'up_probability': 'ERROR', | |
| 'down_probability': 'ERROR', | |
| 'sideways_probability': 'ERROR', | |
| 'confidence': 'ERROR', | |
| 'analysis_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), | |
| 'error_message': f'未預期錯誤: {str(e)}' | |
| }) | |
| progress_messages.append(f"❌ {symbol}: 未預期錯誤") | |
| # 統計結果 | |
| success_count = len([r for r in results if r['error_message'] == '']) | |
| error_count = len(results) - success_count | |
| summary_message = f""" | |
| 📈 批次分析完成! | |
| 📊 **分析統計:** | |
| - 總計股票數:{len(stock_symbols)} | |
| - 成功分析:{success_count} | |
| - 分析失敗:{error_count} | |
| � **圖表已生成:** | |
| - 📊 機率比較柱狀圖 | |
| - 🎯 信心度散佈圖 | |
| - 📈 綜合評分雷達圖 | |
| - 🥧 市場情緒餅圖 | |
| 🎯 **請查看下方圖表進行投資決策分析!** | |
| """ | |
| progress_log = "\n".join(progress_messages) | |
| # 創建圖表和結果表格 | |
| chart_bar, chart_scatter, chart_radar, chart_pie = create_batch_analysis_charts(results) | |
| results_table = create_results_table(results) | |
| return summary_message, progress_log, chart_bar, chart_scatter, chart_radar, chart_pie, results_table | |
| except Exception as e: | |
| return f"❌ 批次分析過程中發生錯誤:{str(e)}", "", None, None, None, None, "" | |
| # 創建 Gradio 界面 | |
| with gr.Blocks(title="AI 股票分析師", theme=gr.themes.Soft()) as app: | |
| gr.Markdown( | |
| """ | |
| # 📈 AI 股票分析師 | |
| ### 🤖 使用 Hugging Face 模型進行智能股票分析 | |
| **✨ 核心功能:** | |
| - 📊 **完整技術指標**:MA、RSI、MACD、布林通道分析 | |
| - 🧠 **AI 情感分析**:使用 FinBERT 模型分析市場情緒 | |
| - 🎯 **機率預測**:提供上漲/下跌/盤整機率百分比 | |
| - 📈 **智能建議**:根據機率給出個性化投資策略 | |
| - 🖼️ **互動圖表**:動態視覺化技術指標走勢 | |
| - 📁 **批次分析**:一次分析多支股票並匯出CSV報告 | |
| **🚀 使用方法:** 單支分析輸入股票代碼,批次分析請確保 `StockList.txt` 檔案存在! | |
| """ | |
| ) | |
| # 建立分頁 | |
| with gr.Tabs(): | |
| with gr.TabItem("🎯 單支股票分析"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| stock_input = gr.Textbox( | |
| label="股票代碼", | |
| placeholder="例如:AAPL, TSLA, 2330.TW", | |
| value="2330.TW" | |
| ) | |
| analyze_btn = gr.Button("開始分析", variant="primary", size="lg") | |
| status_output = gr.Textbox( | |
| label="分析狀態", | |
| lines=2, | |
| interactive=False | |
| ) | |
| with gr.Column(scale=2): | |
| chart_output = gr.Plot(label="股價走勢圖") | |
| prediction_output = gr.Markdown(label="AI 分析報告") | |
| # 事件綁定 | |
| analyze_btn.click( | |
| fn=analyze_stock, | |
| inputs=[stock_input], | |
| outputs=[chart_output, status_output, prediction_output] | |
| ) | |
| # 範例按鈕 | |
| gr.Examples( | |
| examples=[ | |
| ["AAPL"], | |
| ["TSLA"], | |
| ["2330.TW"], | |
| ["MSFT"], | |
| ["GOOGL"] | |
| ], | |
| inputs=[stock_input] | |
| ) | |
| with gr.TabItem("📊 批次股票分析"): | |
| gr.Markdown( | |
| """ | |
| ### 📁 批次分析功能 | |
| **📋 使用步驟:** | |
| 1. 確保 `StockList.txt` 檔案存在於專案目錄 | |
| 2. 檔案中每行一個股票代號(如:2330.TW) | |
| 3. 點擊「開始批次分析」按鈕 | |
| 4. 查看即時互動圖表分析結果 | |
| **📈 輸出內容:** | |
| - 📊 機率比較柱狀圖:直觀對比各股票預測機率 | |
| - 🎯 信心度散佈圖:顯示預測可靠性分佈 | |
| - 📈 綜合評分雷達圖:多維度股票評分比較 | |
| - 🥧 市場情緒餅圖:整體多空情緒統計 | |
| - 即時進度顯示和完整錯誤處理 | |
| """ | |
| ) | |
| with gr.Row(): | |
| batch_analyze_btn = gr.Button( | |
| "🚀 開始批次分析", | |
| variant="primary", | |
| size="lg", | |
| scale=1 | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| batch_summary = gr.Markdown(label="📊 分析摘要") | |
| with gr.Column(scale=1): | |
| batch_progress = gr.Textbox( | |
| label="📋 分析進度", | |
| lines=10, | |
| interactive=False, | |
| max_lines=15 | |
| ) | |
| # 圖表顯示區域 | |
| gr.Markdown("## 📈 視覺化分析結果") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| chart_probability = gr.Plot(label="📊 股票預測機率比較") | |
| with gr.Column(scale=1): | |
| chart_confidence = gr.Plot(label="🎯 預測信心度分佈") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| chart_radar = gr.Plot(label="📈 綜合評分雷達圖") | |
| with gr.Column(scale=1): | |
| chart_sentiment = gr.Plot(label="🥧 整體市場情緒分佈") | |
| # 詳細結果表格 | |
| gr.Markdown("## 📋 詳細分析結果") | |
| results_table = gr.HTML(label="分析結果表格") | |
| # 批次分析事件綁定 | |
| batch_analyze_btn.click( | |
| fn=batch_analyze_stocks, | |
| inputs=[], | |
| outputs=[ | |
| batch_summary, | |
| batch_progress, | |
| chart_probability, | |
| chart_confidence, | |
| chart_radar, | |
| chart_sentiment, | |
| results_table | |
| ] | |
| ) | |
| # 啟動應用 | |
| if __name__ == "__main__": | |
| print("正在啟動 AI 股票分析師...") | |
| # 簡化的啟動邏輯 | |
| try: | |
| if IS_HUGGINGFACE_SPACE: | |
| # Hugging Face Spaces 環境 - 使用預設配置 | |
| print("在 Hugging Face Spaces 中啟動...") | |
| app.launch() | |
| else: | |
| # 本地環境 - 嘗試多個端口 | |
| print("在本地環境中啟動...") | |
| ports_to_try = [7860, 7861, 7862, 7863, 7864, 7865] | |
| launched = False | |
| for port in ports_to_try: | |
| try: | |
| print(f"嘗試端口 {port}...") | |
| app.launch( | |
| share=True, | |
| server_name="0.0.0.0", | |
| server_port=port, | |
| show_error=True, | |
| quiet=False | |
| ) | |
| launched = True | |
| break | |
| except OSError as e: | |
| if "port" in str(e).lower(): | |
| print(f"端口 {port} 不可用,嘗試下一個...") | |
| continue | |
| else: | |
| raise e | |
| if not launched: | |
| print("所有預設端口都被佔用,使用隨機端口...") | |
| app.launch( | |
| share=True, | |
| server_name="0.0.0.0", | |
| server_port=0, # 0 表示自動分配端口 | |
| show_error=True | |
| ) | |
| except Exception as e: | |
| print(f"啟動失敗: {e}") | |
| print("請檢查端口使用情況或嘗試重新啟動") | |
| raise e |