StockRecommander / app_batch.py
54justin's picture
Update app_batch.py
62f4258 verified
# 由 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