# HUGING_FACE_V3.2.0.py (整合 Bert_predict 和 XGBoost 版本 - 新特徵版本)
# 系統套件
import os
from datetime import datetime, timedelta
import google.generativeai as genai
import pandas as pd
import numpy as np
import yfinance as yf
from dash import Dash, dcc, html, callback, Input, Output, State
import dash
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import re
from bs4 import BeautifulSoup
import requests
import time # 引用 time 模組以處理時間戳
import warnings
warnings.filterwarnings('ignore')
# ========================= 引用外部模組 START =========================
# 引用您組員的預測器程式
from Bert_predict import BertPredictor
# 引用新的模型預測器
from model_predictor import XGBoostModel
# ========================== 引用外部模組 END ==========================
# ========================= 新增：交易回測模組 START =========================
class TradingBacktester:
def __init__(self, initial_capital=1000000, max_position_ratio=0.8, batch_ratio=0.2):
"""
初始化交易回測器
Args:
initial_capital: 初始資金 (預設100萬)
max_position_ratio: 最大持倉比例 (預設80%)
batch_ratio: 每次分批交易比例 (預設20%)
"""
self.initial_capital = initial_capital
self.max_position_ratio = max_position_ratio
self.batch_ratio = batch_ratio
# 交易記錄
self.trades = []
self.portfolio_value = []
self.positions = []
self.cash_history = []
# 策略參數
self.trend_threshold = 0.5 # 趨勢判斷閾值 (0.5%)
self.min_trend_days = 3 # 最少連續趨勢天數
def get_trend_signal(self, predictions):
"""
根據預測結果判斷趨勢信號
Args:
predictions: dict包含1,5,10,20日預測結果
Returns:
signal: 1(買進), -1(賣出), 0(持有)
strength: 信號強度 (0-1)
"""
# 提取預測漲跌幅
pred_1d = predictions.get('1d', 0)
pred_5d = predictions.get('5d', 0)
pred_10d = predictions.get('10d', 0)
pred_20d = predictions.get('20d', 0)
# 計算趨勢分數
trend_score = 0
total_weight = 0
# 權重設計：近期權重較高
weights = {'1d': 0.4, '5d': 0.3, '10d': 0.2, '20d': 0.1}
for period, pred in [('1d', pred_1d), ('5d', pred_5d), ('10d', pred_10d), ('20d', pred_20d)]:
if abs(pred) > self.trend_threshold:
trend_score += np.sign(pred) * weights[period]
total_weight += weights[period]
# 正規化趨勢分數
if total_weight > 0:
trend_score = trend_score / total_weight
# 判斷信號強度
strength = abs(trend_score)
# 決定交易信號
if trend_score > 0.3: # 明顯上漲趨勢
return 1, strength
elif trend_score < -0.3: # 明顯下跌趨勢
return -1, strength
else: # 盤整或趨勢不明
return 0, strength
def calculate_position_size(self, signal, strength, current_cash, current_price, current_position):
"""
計算交易部位大小
Args:
signal: 交易信號
strength: 信號強度
current_cash: 當前現金
current_price: 當前價格
current_position: 當前持股數量
Returns:
shares_to_trade: 交易股數 (正數買入，負數賣出)
"""
max_position_value = self.initial_capital * self.max_position_ratio
max_shares = int(max_position_value / current_price)
if signal == 1: # 買進信號
# 計算可買進的最大股數
available_cash = current_cash * self.batch_ratio * strength
max_buy_shares = int(available_cash / current_price)
# 確保不超過最大持倉限制
remaining_capacity = max_shares - current_position
shares_to_buy = min(max_buy_shares, remaining_capacity)
return max(0, shares_to_buy)
elif signal == -1: # 賣出信號
# 計算要賣出的股數
sell_ratio = self.batch_ratio * strength
shares_to_sell = int(current_position * sell_ratio)
return -min(shares_to_sell, current_position)
return 0
def simulate_predictions(self, data, predictor_func):
"""
模擬歷史預測結果
Args:
data: 股價歷史資料
predictor_func: 預測函數
Returns:
predictions_history: 歷史預測結果字典
"""
predictions_history = {}
# 為每個交易日生成預測
for i in range(60, len(data)): # 從第60天開始，確保有足夠歷史資料，
current_date = data.index[i]
historical_data = data.iloc[:i] # 到當前日期的歷史資料
try:
# 呼叫預測函數
predictions = {}
for days in [1, 5, 10, 20]:
pred_result = predictor_func(historical_data, days)
if pred_result:
predictions[f'{days}d'] = pred_result.get('change_pct', 0)
else:
predictions[f'{days}d'] = 0
predictions_history[current_date] = predictions
except Exception as e:
# print(f"預測失敗 {current_date}: {e}")
predictions_history[current_date] = {
'1d': 0, '5d': 0, '10d': 0, '20d': 0
}
return predictions_history
def run_backtest(self, stock_data, predictor_func, start_date=None, end_date=None):
"""
執行回測
Args:
stock_data: 股價資料
predictor_func: 預測函數
start_date: 回測開始日期
end_date: 回測結束日期
Returns:
results: 回測結果字典
"""
# 重置交易記錄
self.trades = []
self.portfolio_value = []
self.positions = []
self.cash_history = []
# 設定回測期間
if start_date:
stock_data = stock_data[stock_data.index >= start_date]
if end_date:
stock_data = stock_data[stock_data.index <= end_date]
if len(stock_data) < 100:
raise ValueError("資料不足，無法進行回測")
print("開始生成歷史預測...")
predictions_history = self.simulate_predictions(stock_data, predictor_func)
# 初始化
current_cash = self.initial_capital
current_position = 0
print("開始執行回測...")
# 逐日回測
for date in stock_data.index:
if date not in predictions_history:
continue
current_price = stock_data.loc[date, 'Close']
predictions = predictions_history[date]
# 獲取交易信號
signal, strength = self.get_trend_signal(predictions)
# 計算交易量
shares_to_trade = self.calculate_position_size(
signal, strength, current_cash, current_price, current_position
)
# 執行交易
if shares_to_trade != 0:
trade_value = shares_to_trade * current_price
# 更新現金和持倉
current_cash -= trade_value
current_position += shares_to_trade
# 記錄交易
self.trades.append({
'date': date,
'signal': signal,
'shares': shares_to_trade,
'price': current_price,
'value': trade_value,
'strength': strength,
'predictions': predictions.copy()
})
# 計算投資組合價值
portfolio_val = current_cash + current_position * current_price
# 記錄每日狀態
self.portfolio_value.append({
'date': date,
'portfolio_value': portfolio_val,
'cash': current_cash,
'position_value': current_position * current_price,
'position_shares': current_position,
'price': current_price
})
# 計算績效指標
results = self._calculate_performance_metrics(stock_data)
print(f"回測完成！總交易次數: {len(self.trades)}")
return results
def _calculate_performance_metrics(self, stock_data):
"""計算績效指標"""
if not self.portfolio_value:
return {}
portfolio_df = pd.DataFrame(self.portfolio_value)
portfolio_df.set_index('date', inplace=True)
# 基本績效
final_value = portfolio_df['portfolio_value'].iloc[-1]
total_return = (final_value / self.initial_capital - 1) * 100
# 基準比較（買入持有策略）
initial_price = stock_data['Close'].iloc[0]
final_price = stock_data['Close'].iloc[-1]
benchmark_return = (final_price / initial_price - 1) * 100
# 計算波動率
portfolio_returns = portfolio_df['portfolio_value'].pct_change().dropna()
annual_volatility = portfolio_returns.std() * np.sqrt(252) * 100
# 最大回撤
rolling_max = portfolio_df['portfolio_value'].expanding().max()
drawdown = (portfolio_df['portfolio_value'] - rolling_max) / rolling_max
max_drawdown = drawdown.min() * 100
# 夏普比率 (假設無風險利率為2%)
risk_free_rate = 0.02
excess_return = total_return/100 - risk_free_rate
sharpe_ratio = excess_return / (annual_volatility/100) if annual_volatility > 0 else 0
# 交易統計
trades_df = pd.DataFrame(self.trades) if self.trades else pd.DataFrame()
buy_trades = len(trades_df[trades_df['signal'] == 1]) if not trades_df.empty else 0
sell_trades = len(trades_df[trades_df['signal'] == -1]) if not trades_df.empty else 0
results = {
'final_value': final_value,
'total_return': total_return,
'benchmark_return': benchmark_return,
'excess_return': total_return - benchmark_return,
'annual_volatility': annual_volatility,
'max_drawdown': max_drawdown,
'sharpe_ratio': sharpe_ratio,
'total_trades': len(self.trades),
'buy_trades': buy_trades,
'sell_trades': sell_trades,
'win_rate': self._calculate_win_rate(),
'portfolio_df': portfolio_df,
'trades_df': trades_df,
'initial_capital': self.initial_capital
}
return results
def _calculate_win_rate(self):
"""計算勝率"""
if len(self.trades) < 2:
return 0
# 簡化的勝率計算：檢查每筆交易後的價格變化
winning_trades = 0
total_closed_trades = 0
for i in range(len(self.trades) - 1):
current_trade = self.trades[i]
next_trade = self.trades[i + 1]
if current_trade['signal'] == 1: # 買入交易
price_change = (next_trade['price'] - current_trade['price']) / current_trade['price']
if price_change > 0:
winning_trades += 1
total_closed_trades += 1
return (winning_trades / total_closed_trades * 100) if total_closed_trades > 0 else 0
def create_backtest_chart(self, results, stock_data):
"""創建回測結果圖表"""
if 'portfolio_df' not in results:
return go.Figure()
portfolio_df = results['portfolio_df']
trades_df = results['trades_df']
# 創建子圖
fig = make_subplots(
rows=3, cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
row_heights=[0.5, 0.25, 0.25],
subplot_titles=('投資組合價值 vs 基準', '持倉變化', '交易信號')
)
# 第一個子圖：投資組合價值對比
fig.add_trace(
go.Scatter(
x=portfolio_df.index,
y=portfolio_df['portfolio_value'],
mode='lines',
name='投資組合價值',
line=dict(color='blue', width=2)
),
row=1, col=1
)
# 基準線（買入持有）
benchmark_values = []
initial_shares = self.initial_capital / stock_data['Close'].iloc[0]
for date in portfolio_df.index:
if date in stock_data.index:
benchmark_val = initial_shares * stock_data.loc[date, 'Close']
benchmark_values.append(benchmark_val)
else:
benchmark_values.append(np.nan)
fig.add_trace(
go.Scatter(
x=portfolio_df.index,
y=benchmark_values,
mode='lines',
name='買入持有基準',
line=dict(color='gray', width=2, dash='dash')
),
row=1, col=1
)
# 第二個子圖：持倉變化
fig.add_trace(
go.Scatter(
x=portfolio_df.index,
y=portfolio_df['position_shares'],
mode='lines',
name='持股數量',
line=dict(color='green', width=2)
),
row=2, col=1
)
# 第三個子圖：價格和交易信號
# 先畫價格線
price_data = stock_data.reindex(portfolio_df.index, method='ffill')
fig.add_trace(
go.Scatter(
x=portfolio_df.index,
y=price_data['Close'],
mode='lines',
name='股價',
line=dict(color='black', width=1)
),
row=3, col=1
)
# 添加交易點
if not trades_df.empty:
buy_trades = trades_df[trades_df['signal'] == 1]
sell_trades = trades_df[trades_df['signal'] == -1]
if not buy_trades.empty:
fig.add_trace(
go.Scatter(
x=buy_trades['date'],
y=buy_trades['price'],
mode='markers',
name='買入',
marker=dict(color='red', size=8, symbol='triangle-up')
),
row=3, col=1
)
if not sell_trades.empty:
fig.add_trace(
go.Scatter(
x=sell_trades['date'],
y=sell_trades['price'],
mode='markers',
name='賣出',
marker=dict(color='green', size=8, symbol='triangle-down')
),
row=3, col=1
)
# 更新布局
fig.update_layout(
title=f"交易策略回測結果",
height=800,
showlegend=True,
xaxis3_title="日期"
)
fig.update_yaxes(title_text="價值 (TWD)", row=1, col=1)
fig.update_yaxes(title_text="股數", row=2, col=1)
fig.update_yaxes(title_text="股價 (TWD)", row=3, col=1)
return fig
def create_backtest_summary_card(results):
"""創建回測摘要卡片"""
if not results:
return html.Div("回測結果為空", style={'color': 'red'})
# 決定績效顏色
return_color = 'red' if results['total_return'] > 0 else 'green'
excess_color = 'red' if results['excess_return'] > 0 else 'green'
return html.Div([
html.Div([
html.H4("交易策略回測摘要", style={'color': '#2C3E50', 'margin-bottom': '20px'}),
# 核心績效指標
html.Div([
html.Div([
html.H5("總報酬率", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['total_return']:+.2f}%",
style={'margin': '5px 0', 'color': return_color, 'font-weight': 'bold'})
], className="metric-item"),
html.Div([
html.H5("vs 買入持有", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['excess_return']:+.2f}%",
style={'margin': '5px 0', 'color': excess_color, 'font-weight': 'bold'})
], className="metric-item"),
html.Div([
html.H5("夏普比率", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['sharpe_ratio']:.2f}",
style={'margin': '5px 0', 'color': '#3498DB', 'font-weight': 'bold'})
], className="metric-item"),
html.Div([
html.H5("最大回撤", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['max_drawdown']:.2f}%",
style={'margin': '5px 0', 'color': '#E74C3C', 'font-weight': 'bold'})
], className="metric-item")
], style={'display': 'flex', 'justify-content': 'space-around', 'margin-bottom': '20px'}),
# 詳細統計
html.Hr(),
html.Table([
html.Tr([html.Td("最終投資組合價值"), html.Td(f"${results['final_value']:,.0f}")]),
html.Tr([html.Td("初始資金"), html.Td(f"${results.get('initial_capital', 1000000):,.0f}")]),
html.Tr([html.Td("年化波動率"), html.Td(f"{results['annual_volatility']:.2f}%")]),
html.Tr([html.Td("總交易次數"), html.Td(f"{results['total_trades']}")]),
html.Tr([html.Td("買入次數"), html.Td(f"{results['buy_trades']}")]),
html.Tr([html.Td("賣出次數"), html.Td(f"{results['sell_trades']}")]),
html.Tr([html.Td("交易勝率"), html.Td(f"{results['win_rate']:.1f}%")])
], style={'width': '100%', 'margin-top': '10px'})
])
], style={
'padding': '25px',
'background': 'white',
'border-radius': '12px',
'box-shadow': '0 4px 20px rgba(0,0,0,0.08)',
'border': '1px solid #e9ecef',
'margin-bottom': '20px'
})
# ========================= 新增：交易回測模組 END =========================
# ========================== 引用外部模組 END ==========================
# ========================= 全域設定 START =========================
# 【【【模型切換開關】】】
# False: 使用簡易統計模型 (預設)
# True: 使用 model_predictor.py 中的進階 XGBoost 模型
USE_ADVANCED_MODEL = True
# ========================= CACHE 設定 START =========================
# 分析結果的快取字典
ANALYSIS_CACHE = {}
ANALYSIS_CACHE_DURATION = 60 # 分析結果緩存60秒
STOCK_DATA_CACHE = {}
CACHE_EXPIRE_SECONDS = 60 # 改為1分鐘，確保數據更及時
# 快取有效時間（秒），例如：8 小時 = 8 * 60 * 60 = 28800 秒
CACHE_DURATION_SECONDS = 60 # 60秒緩存
# ========================== CACHE 設定 END ==========================
# ========================== 全域設定 END ==========================
# 台股代號對應表 (移除台指期，因為它現在是獨立個塊)
TAIWAN_STOCKS = {
'元大台灣50': '0050.TW',
'台積電': '2330.TW',
'聯發科': '2454.TW',
'鴻海': '2317.TW',
'台達電': '2308.TW',
'廣達': '2382.TW',
'富邦金': '2881.TW',
'中信金': '2891.TW',
'國泰金': '2882.TW',
'聯電': '2303.TW',
'中華電': '2412.TW',
'玉山金': '2884.TW',
'兆豐金': '2886.TW',
'日月光投控': '3711.TW',
'華碩': '2357.TW',
'統一': '1216.TW',
'元大金': '2885.TW',
'智邦': '2345.TW',
'緯創': '3231.TW',
'華邦': '3034.TW',
'第一金': '2892.TW',
'瑞昱': '2379.TW',
'緯穎': '6669.TW',
'永豐金': '2890.TW',
'合庫金': '5880.TW',
'臺南金': '2880.TW',
'台光電': '2383.TW',
'世芯-KY': '3661.TW',
'奇鋐': '3017.TW',
'凱基金': '2883.TW',
'大立光': '3008.TW',
'長榮': '2603.TW',
'光寶科': '2301.TW',
'中鋼': '2002.TW',
'中租-KY': '5871.TW',
'國巨': '2327.TW',
'台新金': '2887.TW',
'上海商銀': '5876.TW',
'台泥': '1101.TW',
'台灣大': '3045.TW',
'和碩': '4938.TW',
'遠傳': '4904.TW',
'和泰車': '2207.TW',
'研華': '2395.TW',
'台塑': '1301.TW',
'統一超': '2912.TW',
'藥華藥': '6446.TW',
'南亞': '1303.TW',
'陽明': '2609.TW',
'謝海': '2615.TW',
'台塑化': '6505.TW',
'慧洋-KY': '2637.TW',
'上銀': '2049.TW',
'南亞科': '2408.TW',
'旺宏': '2337.TW',
'譜瑞-KY': '4966.TWO',
'貿聯-KY': '3665.TW',
'驊訊': '6870.TWO',
'穩懋': '3105.TWO',
'金居': '8358.TWO',
'緯軟': '4953.TWO',
'宏捷科': '8086.TWO',
'漢磊': '3707.TWO',
'茂矽': '2342.TW',
'騰雲': '6870.TWO',
'順德': '2351.TW',
'明泰': '3380.TW',
}
# 產業分類
INDUSTRY_MAPPING = {
'0050.TW': 'ETF',
'2330.TW': '半導體',
'2454.TW': '半導體',
'2317.TW': '電子組件',
'2308.TW': '電子',
'2382.TW': '電子',
'2881.TW': '金融',
'2891.TW': '金融',
'2882.TW': '金融',
'2303.TW': '半導體',
'2412.TW': '電信',
'2884.TW': '金融',
'2886.TW': '金融',
'3711.TW': '半導體',
'2357.TW': '電子',
'1216.TW': '食品',
'2885.TW': '金融',
'2345.TW': '網通設備',
'3231.TW': '電子',
'3034.TW': '半導體',
'2892.TW': '金融',
'2379.TW': '半導體',
'6669.TW': '電子',
'2890.TW': '金融',
'5880.TW': '金融',
'2880.TW': '金融',
'2383.TW': '電子',
'3661.TW': '半導體',
'3017.TW': '電子',
'2883.TW': '金融',
'3008.TW': '光學',
'2603.TW': '航運',
'2301.TW': '電子',
'2002.TW': '鋼鐵',
'5871.TW': '金融',
'2327.TW': '電子被動元件',
'2887.TW': '金融',
'5876.TW': '金融',
'1101.TW': '營建',
'3045.TW': '電信',
'4938.TW': '電子',
'4904.TW': '電信',
'2207.TW': '汽車',
'2395.TW': '電腦周邊',
'1301.TW': '塑膠',
'2912.TW': '百貨',
'6446.TW': '生技',
'1303.TW': '塑膠',
'2609.TW': '航運',
'2615.TW': '航運',
'6505.TW': '塑膠',
'2637.TW': '散裝航運',
'2049.TW': '工具機',
'2408.TW': 'DRAM',
'2337.TW': 'NFLSH',
'4966.TWO': '高速傳輸',
'3665.TW': '連接器',
'6870.TWO': '軟體整合',
'3105.TWO': 'PA功率',
'8358.TWO': '銅箔',
'4953.TWO': '軟體',
'8086.TWO': 'PA功率',
'3707.TWO': '矽晶圓',
'2342.TW': '矽晶圓',
'2351.TW': '導線架',
'6870.TWO': '軟體整合',
'3380.TW': '網通'
}
# ========================= 風險管理模組 START =========================
class RiskAnalyzer:
def __init__(self):
self.risk_free_rate = 0.01 # 假設無風險利率為1%
def calculate_var(self, returns, confidence_level=0.05):
"""計算風險價值 (Value at Risk)"""
if returns is None or len(returns) < 30:
return None
return np.percentile(returns.dropna(), confidence_level * 100)
def calculate_sharpe_ratio(self, returns):
"""計算夏普比率"""
if returns is None or returns.std() == 0:
return 0
excess_returns = returns - self.risk_free_rate / 252
return excess_returns.mean() / excess_returns.std() * np.sqrt(252)
def calculate_max_drawdown(self, prices):
"""計算最大回撤"""
if prices is None or prices.empty:
return None
# 使用 pct_change() 計算每日報酬率，並加 1
daily_returns = prices.pct_change() + 1
# 計算累積乘積，填充 NA 值為 1
cumulative = daily_returns.cumprod().fillna(1)
rolling_max = cumulative.expanding().max()
drawdown = (cumulative - rolling_max) / rolling_max
return drawdown.min()
def calculate_beta(self, stock_returns, market_returns):
"""計算貝塔值"""
if stock_returns is None or market_returns is None or len(stock_returns) != len(market_returns) or len(
stock_returns) < 30:
return None
# 合併並去除NA值以對齊數據
combined = pd.DataFrame({'stock': stock_returns, 'market': market_returns}).dropna()
if len(combined) < 30:
return None
covariance = np.cov(combined['stock'], combined['market'])[0][1]
market_variance = np.var(combined['market'])
return covariance / market_variance if market_variance != 0 else 0
# ========================== 風險管理模組 END ==========================
def clear_all_cache():
"""清理所有緩存"""
global ANALYSIS_CACHE, BACKTEST_CACHE
ANALYSIS_CACHE.clear()
if 'BACKTEST_CACHE' in globals():
BACKTEST_CACHE.clear()
print("所有緩存已清空")
def get_stock_data(symbol, period='1y'):
"""獲取股票資料 - 添加調試信息"""
try:
current_time = datetime.now()
print(f"[{current_time.strftime('%H:%M:%S')}] 正在獲取 {symbol} 數據...")
stock = yf.Ticker(symbol)
data = stock.history(period=period)
if data.empty and symbol == 'TXF=F':
stock = yf.Ticker('0050.TW')
data = stock.history(period=period)
if data.empty:
stock = yf.Ticker('^TWII')
data = stock.history(period=period)
if not data.empty:
latest_close = data['Close'].iloc[-1]
latest_date = data.index[-1].strftime('%Y-%m-%d')
print(f"[{current_time.strftime('%H:%M:%S')}] {symbol} 數據獲取成功: {latest_date}, 收盤價: {latest_close:.2f}")
else:
print(f"[{current_time.strftime('%H:%M:%S')}] 警告: {symbol} 數據為空")
return data
except Exception as e:
print(f"獲取 {symbol} 數據時發生錯誤: {e}")
return pd.DataFrame()
def get_us_market_data():
"""獲取美股指數數據"""
try:
indices = {
'DJI': '^DJI', # 道瓊指數
'NAS': '^IXIC', # 那斯達克
'SOX': '^SOX', # 費城半導體
'S&P_500': '^GSPC', # S&P 500
'TSM_ADR': 'TSM' # 台積電ADR
}
market_data = {}
for name, symbol in indices.items():
try:
data = yf.Ticker(symbol).history(period='5d')
if not data.empty:
market_data[name] = data['Close'].iloc[-1]
else:
market_data[name] = 0
except:
market_data[name] = 0
return market_data
except Exception as e:
print(f"獲取美股數據時發生錯誤: {e}")
return {'DJI': 0, 'NAS': 0, 'SOX': 0, 'S&P_500': 0, 'TSM_ADR': 0}
def get_exchange_rate():
"""獲取台幣匯率 (USD/TWD)"""
try:
data = yf.Ticker('USDTWD=X').history(period='5d')
if not data.empty:
return data['Close'].iloc[-1]
else:
return 31.5 # 預設值
except:
return 31.5
def simple_statistical_predict(data, predict_days=5):
"""簡化的統計預測模型 - 基於真實市場時間變化"""
if len(data) < 60:
return None
prices = data['Close'].values
current_price = prices[-1]
ma_short = np.mean(prices[-5:])
ma_medium = np.mean(prices[-20:])
ma_long = np.mean(prices[-60:])
recent_trend = np.polyfit(range(20), prices[-20:], 1)[0]
volatility = np.std(prices[-20:]) / np.mean(prices[-20:])
# 使用系統時間作為市場變化因子
now = datetime.now()
# 時間相關因子（基於當前時間）
hour_factor = (now.hour % 24) * 0.0001 # 小時因子
minute_factor = now.minute * 0.000001 # 分鐘因子
second_factor = now.second * 0.0000001 # 秒因子
time_factor = hour_factor + minute_factor + second_factor
# 市場開盤時間調整
if 9 <= now.hour <= 13: # 台股交易時間
market_active_factor = 0.001
else:
market_active_factor = -0.0005
base_change = recent_trend * predict_days
trend_factor = 1.0 + (0.02 if ma_short > ma_medium > ma_long else -0.02 if ma_short < ma_medium < ma_long else 0)
# 結合時間因子
total_time_factor = time_factor + market_active_factor
predicted_price = current_price * trend_factor + base_change + (current_price * total_time_factor)
# 計算漲幅百分比
change_pct = ((predicted_price - current_price) / current_price) * 100
change_pct = np.clip(change_pct, -15.0, 15.0)
confidence = max(0.6, 1 - volatility * 2)
print(f"[{now.strftime('%H:%M:%S')}] 統計預測 - 時間因子: {total_time_factor:.8f}, 預測漲幅: {change_pct:+.2f}%")
return {
'predicted_price': predicted_price,
'change_pct': change_pct,
'confidence': confidence
}
def calculate_new_features(df):
"""
計算新的技術指標特徵 - 針對新特徵需求 (已修正)
"""
if df.empty:
return df
# 1. return_t-1 – 前一日報酬率
df['return_t-1'] = df['Close'].pct_change()
# 2. return_t-5 – 過去 5 日累積報酬率
df['return_t-5'] = (df['Close'] / df['Close'].shift(5) - 1)
# 3. MA5_close – 5 日移動平均價
df['MA5_close'] = df['Close'].rolling(window=5).mean()
# 4. volatility_5d – 5 日報酬標準差（短期波動）
df['volatility_5d'] = df['return_t-1'].rolling(window=5).std()
# 5. volume_ratio_5d – 今日成交量 ÷ 5 日均量
# Note: Use 'Volume' with a capital V
df['volume_5d_avg'] = df['Volume'].rolling(window=5).mean()
df['volume_ratio_5d'] = df['Volume'] / df['volume_5d_avg']
# 6. MACD_diff – MACD - signal（趨勢強弱）
# This is already calculated as 'MACD_Histogram' in calculate_technical_indicators, but we'll recalculate to be safe.
exp1 = df['Close'].ewm(span=12, adjust=False).mean()
exp2 = df['Close'].ewm(span=26, adjust=False).mean()
macd_line = exp1 - exp2
signal_line = macd_line.ewm(span=9, adjust=False).mean()
df['MACD_diff'] = macd_line - signal_line
# 7. volume_weighted_return - 成交量加權報酬率
# 【【ERROR FIX】】 Use 'Volume' with a capital V instead of 'volume'
df['volume_weighted_return'] = abs(df['return_t-1']) * df['Volume']
# 移除輔助欄位
if 'volume_5d_avg' in df.columns:
df = df.drop('volume_5d_avg', axis=1)
return df
def advanced_xgboost_predict(predict_days=5):
"""使用 XGBoost 模型進行預測 - 強制刷新數據版本"""
try:
print(f"開始XGBoost預測 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
xgb_model = XGBoostModel()
# 強制重新獲取台指數據 - 不使用緩存
print("正在獲取最新台指數據...")
taiex_data = get_stock_data('^TWII', '2y')
if taiex_data.empty or len(taiex_data) < 60:
print("台指期數據不足，無法進行XGBoost預測")
return None
taiex_data = calculate_technical_indicators(taiex_data)
taiex_data = calculate_new_features(taiex_data)
# 新增 volume_weighted_return 計算
# if 'return_t-1' in taiex_data.columns and 'Volume' in taiex_data.columns:
# taiex_data['volume_weighted_return'] = abs(taiex_data['return_t-1']) * taiex_data['Volume']
# else:
# taiex_data['volume_weighted_return'] = 0
print("正在獲取美股數據...")
us_market_data = get_us_market_data()
try:
if predictor is not None:
sentiment_score_raw = predictor.get_news_index()
if sentiment_score_raw is None: sentiment_score_raw = 0
print(f"新聞情緒分數: {sentiment_score_raw}")
else:
sentiment_score_raw = 0
except:
sentiment_score_raw = 0
latest_data = taiex_data.iloc[-1]
# 【【修改點】】定義完整的特徵列表
feature_columns_map = {
'close': latest_data.get('Close', 0),
'return_t-1': latest_data.get('return_t-1', 0),
'return_t-5': latest_data.get('return_t-5', 0),
'MA5_close': latest_data.get('MA5_close', 0),
'volatility_5d': latest_data.get('volatility_5d', 0),
'volume_ratio_5d': latest_data.get('volume_ratio_5d', 0),
'MACD_diff': latest_data.get('MACD_diff', 0),
'NEWS': sentiment_score_raw,
'MACDvol': latest_data.get('MACDvol', 0),
'RSI_14': latest_data.get('RSI', 0), # 注意: app.py中計算的欄位名為 'RSI'
'ADX': latest_data.get('ADX', 0),
'volume_weighted_return': latest_data.get('volume_weighted_return', 0)
}
dji_return = 0
sox_return = 0
try:
dji_data = get_stock_data('^DJI', '5d')
if not dji_data.empty and len(dji_data) >= 2:
dji_return = (dji_data['Close'].iloc[-1] / dji_data['Close'].iloc[-2] - 1)
except: pass
try:
sox_data = get_stock_data('^SOX', '5d')
if not sox_data.empty and len(sox_data) >= 2:
sox_return = (sox_data['Close'].iloc[-1] / sox_data['Close'].iloc[-2] - 1)
except: pass
feature_columns_map['dji_return_t-1'] = dji_return
feature_columns_map['sox_return_t-1'] = sox_return
# 處理可能的 NaN 值
for key, value in feature_columns_map.items():
if pd.isna(value):
feature_columns_map[key] = 0
input_df = pd.DataFrame([feature_columns_map])
print("\n=== 📊 本次預測輸入特徵 DataFrame ===")
print(input_df)
print("======================================\n")
predictions = xgb_model.predict('xgboost_model', input_df)
if predictions is None:
return None
pred_mapping = {1: 'Change_pct_t1_pred', 5: 'Change_pct_t5_pred', 10: 'Change_pct_t10_pred', 20: 'Change_pct_t20_pred'}
closest_day = min(pred_mapping.keys(), key=lambda x: abs(x - predict_days))
pred_key = pred_mapping[closest_day]
predicted_change_pct = predictions[pred_key]
current_price = latest_data['Close']
predicted_price = current_price * (1 + predicted_change_pct / 100)
print(f"XGBoost預測結果 - 漲幅: {predicted_change_pct:+.2f}%, 時間: {datetime.now().strftime('%H:%M:%S')}")
return {'predicted_price': predicted_price, 'change_pct': predicted_change_pct, 'confidence': 0.75}
except Exception as e:
print(f"XGBoost 預測錯誤: {e}")
import traceback
traceback.print_exc()
return None
# 呼叫模型預測
predictions = model.predict('xgboost_model', input_df)
def clear_old_cache():
"""清理過期的緩存數據"""
current_time = time.time()
expired_keys = []
for key, (data, cache_time) in STOCK_DATA_CACHE.items():
if current_time - cache_time > CACHE_EXPIRE_SECONDS:
expired_keys.append(key)
for key in expired_keys:
del STOCK_DATA_CACHE[key]
print(f"清理過期緩存: {key}")
def get_prediction(data, predict_days=5):
"""
【【模型預測控制器】】
根據 USE_ADVANCED_MODEL 的設定，呼叫對應的預測模型。
"""
if USE_ADVANCED_MODEL:
print(f"模式: 進階XGBoost模型 | 預測天期: {predict_days}天")
prediction = advanced_xgboost_predict(predict_days)
# 如果進階模型預測失敗，則自動降級使用簡易模型
if prediction is not None:
return prediction
else:
print("進階模型預測失敗，自動降級為簡易統計模型。")
# 預設或降級時執行簡易模型
print(f"模式: 簡易統計模型 | 預測天期: {predict_days}天")
return simple_statistical_predict(data, predict_days)
def calculate_technical_indicators(df):
"""計算技術指標"""
if df.empty: return df
# 移動平均線
df['MA5'] = df['Close'].rolling(window=5).mean()
df['MA20'] = df['Close'].rolling(window=20).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['MACD_Histogram'] = df['MACD'] - df['MACD_Signal']
# 布林通道
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)
# KD 指標
low_min = df['Low'].rolling(window=9).min()
high_max = df['High'].rolling(window=9).max()
rsv = (df['Close'] - low_min) / (high_max - low_min) * 100
df['K'] = rsv.ewm(com=2).mean()
df['D'] = df['K'].ewm(com=2).mean()
# 威廉指標
low_min_14 = df['Low'].rolling(window=14).min()
high_max_14 = df['High'].rolling(window=14).max()
df['Williams_R'] = -100 * (high_max_14 - df['Close']) / (high_max_14 - low_min_14)
# DMI 指標
df['up_move'] = df['High'] - df['High'].shift(1)
df['down_move'] = df['Low'].shift(1) - df['Low']
df['+DM'] = np.where((df['up_move'] > df['down_move']) & (df['up_move'] > 0), df['up_move'], 0)
df['-DM'] = np.where((df['down_move'] > df['up_move']) & (df['down_move'] > 0), df['down_move'], 0)
df['TR'] = np.max([df['High'] - df['Low'], abs(df['High'] - df['Close'].shift(1)), abs(df['Low'] - df['Close'].shift(1))], axis=0)
df['+DI'] = (df['+DM'].ewm(com=13, adjust=False).mean() / df['TR'].ewm(com=13, adjust=False).mean()) * 100
df['-DI'] = (df['-DM'].ewm(com=13, adjust=False).mean() / df['TR'].ewm(com=13, adjust=False).mean()) * 100
df['DX'] = abs(df['+DI'] - df['-DI']) / (df['+DI'] + df['-DI']) * 100
df['ADX'] = df['DX'].ewm(com=13, adjust=False).mean()
# 【【新增】】計算成交量MACD (MACDvol)
if 'Volume' in df.columns and not df['Volume'].isnull().all():
exp1_vol = df['Volume'].ewm(span=12, adjust=False).mean()
exp2_vol = df['Volume'].ewm(span=26, adjust=False).mean()
df['MACDvol'] = exp1_vol - exp2_vol
else:
df['MACDvol'] = 0 # 如果沒有成交量數據，則設為0
return df
def calculate_volume_profile(df, num_bins=50):
if df.empty or 'High' not in df.columns or 'Low' not in df.columns or 'Volume' not in df.columns:
return None, None, None
all_prices = np.concatenate([df['High'].values, df['Low'].values])
min_price, max_price = all_prices.min(), all_prices.max()
price_for_volume = (df['High'] + df['Low'] + df['Close']) / 3
df_vol_profile = df.copy()
df_vol_profile['Price_Indicator'] = price_for_volume
hist, bin_edges = np.histogram(df_vol_profile['Price_Indicator'], bins=num_bins, range=(min_price, max_price), weights=df_vol_profile['Volume'])
price_centers = (bin_edges[:-1] + bin_edges[1:]) / 2
return bin_edges, hist, price_centers
def get_business_climate_data():
try:
if not os.path.exists('business_climate.csv'): return pd.DataFrame()
df = pd.read_csv('business_climate.csv')
if 'Date' not in df.columns: df.columns = ['Date', 'Index'] if len(df.columns) == 2 else df.columns
if 'Date' in df.columns:
try: df['Date'] = pd.to_datetime(df['Date'] + '-01', format='%Y-%m-%d', errors='coerce')
except: df['Date'] = pd.to_datetime(df['Date'], errors='coerce')
df = df.dropna(subset=['Date'])
return df
except Exception as e:
print(f"無法獲取景氣燈號資料: {str(e)}")
return pd.DataFrame()
def get_pmi_data():
try:
if not os.path.exists('taiwan_pmi.csv'): return pd.DataFrame()
df = pd.read_csv('taiwan_pmi.csv')
if 'DATE' in df.columns: df = df.rename(columns={'DATE': 'Date', 'INDEX': 'Index'})
elif len(df.columns) == 2: df.columns = ['Date', 'Index']
if 'Date' in df.columns:
try: df['Date'] = pd.to_datetime(df['Date'] + '-01', format='%Y-%m-%d', errors='coerce')
except: df['Date'] = pd.to_datetime(df['Date'], errors='coerce')
df = df.dropna(subset=['Date'])
return df
except Exception as e:
print(f"無法獲取 PMI 資料: {str(e)}")
return pd.DataFrame()
def generate_gemini_analysis(stock_name, stock_symbol, period, data):
"""
使用 Gemini API 生成基本面和市場展望分析。
"""
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
return "無法讀取 GEMINI API 金鑰", "請在系統環境變數中設定您的金鑰"
try:
genai.configure(api_key=api_key)
model = genai.GenerativeModel('gemini-2.0-flash')
price_change = ((data['Close'].iloc[-1] - data['Close'].iloc[0]) / data['Close'].iloc[0]) * 100
rsi_current = data['RSI'].iloc[-1]
macd_current = data['MACD'].iloc[-1]
macd_signal_current = data['MACD_Signal'].iloc[-1]
industry = INDUSTRY_MAPPING.get(stock_symbol, '綜合')
prompt = f"""
請扮演一位專業、資深的台灣股市金融分析師。
我將提供一檔台股的即時技術指標數據，請你基於這些數據，結合你對這家公司、其所在產業以及當前市場趨勢的理解，為我生成一段專業的「基本面分析」和一段「市場展望與投資建議」。
**股票資訊:**
- **公司名稱:** {stock_name} ({stock_symbol})
- **分析期間:** 最近 {period}
- **所屬產業:** {industry}
- **期間價格變動:** {price_change:+.2f}%
- **目前 RSI 指標:** {rsi_current:.2f}
- **目前 MACD 指標:** MACD線為 {macd_current:.3f}, 信號線為 {macd_signal_current:.3f}
**你的任務:**
1. **基本面分析 (約 150 字):**
- 回覆前都先搜尋資料。
- 評論這家公司的產業地位、近期營運亮點或挑戰。
- 提及任何可能影響其基本面的關鍵因素 (例如：財報、法說會、政策、供應鏈變化等)。
- 請用專業、客觀的語氣撰寫。
2. **市場展望與投資建議 (約 150 字):**
- 基於上述所有資訊，提供對該股票的短期和中期市場展望(例如:是否有沉重賣壓，或是換手發生)。
- 提出具體的投資建議，例如：適合何種類型的投資人、潛在的風險點。
- 請直接提供分析內容，不要包含任何問候語。
**輸出格式:**
請嚴格按照以下格式回傳，使用"$$"作為兩個段落之間的分隔符：
[基本面分析內容]$$[市場展望與投資建議內容]
"""
response = model.generate_content(prompt)
parts = response.text.split('$$')
if len(parts) == 2:
fundamental_analysis = parts[0].strip()
market_outlook = parts[1].strip()
return dcc.Markdown(fundamental_analysis), dcc.Markdown(market_outlook)
else:
# Fallback for unexpected response format
return dcc.Markdown("無法解析 Gemini 回應，請稍後再試。"), dcc.Markdown(response.text)
except Exception as e:
error_message = f"呼叫 Gemini API 時發生錯誤: {str(e)}"
print(error_message)
return dcc.Markdown(error_message), dcc.Markdown("請檢查後台日誌或 API 金鑰設定")
# 建立 Dash 應用程式
app = dash.Dash(__name__, suppress_callback_exceptions=True)
# 初始化模型
try:
print("正在初始化新聞情緒分析模型...")
predictor = BertPredictor(max_news_per_keyword=5)
print("新聞情緒分析模型初始化成功。")
except Exception as e:
print(f"錯誤：新聞情緒分析模型初始化失敗 - {e}")
predictor = None
# 應用程式佈局
# 完整的 app.layout
app.layout = html.Div([
html.H1("台股分析儀表板", style={'text-align': 'center', 'margin-bottom': '30px'}),
# AI助手與XGboost機器學習預測區塊
html.Div([
html.H2("AI助手與XGboost機器學習預測 - 加權指數", style={'text-align': 'center','color': '#FFCC22','margin-bottom': '25px'}),
html.Div([
html.Div([
html.Label("預測期間:", style={'font-weight': 'bold', 'color': '#FFCC22'}),
dcc.Dropdown(id='taiex-prediction-period',
options=[
{'label': '1日後預測', 'value': 1},{'label': '5日後預測', 'value': 5},
{'label': '10日後預測', 'value': 10},{'label': '20日後預測', 'value': 20},
{'label': '60日後預測', 'value': 60}], value=5,
style={'margin-bottom': '10px', 'color': '#272727'})
], style={'width': '30%', 'display': 'inline-block'}),
html.Div(id='taiex-prediction-results', style={'width': '65%', 'display': 'inline-block', 'margin-left': '5%'})
]),
html.Div([dcc.Graph(id='taiex-prediction-chart')], style={'margin-top': '20px'})
], style={'background': 'linear-gradient(135deg, #667eea 0%, #764ba2 100%)','padding': '25px','border-radius': '15px','box-shadow': '0 8px 25px rgba(0,0,0,0.15)','color': 'white','margin-bottom': '40px'}),
# 市場情緒與新聞分析區塊
html.Div([
html.Div([
html.H4("市場情緒指標", style={'color': '#8E44AD'}),
html.Div(id='sentiment-gauge')
], style={'width': '48%', 'display': 'inline-block', 'vertical-align': 'top'}),
html.Div([
html.H4("關鍵新聞摘要", style={'color': '#27AE60'}),
html.Div(id='news-summary', style={'background': '#f8f9fa','padding': '15px','border-radius': '8px','max-height': '200px','overflow-y': 'auto'})
], style={'width': '48%', 'display': 'inline-block', 'margin-left': '4%', 'vertical-align': 'top'})
], style={'margin-top': '30px','padding': '20px','background': 'white','border-radius': '10px','box-shadow': '0 2px 10px rgba(0,0,0,0.1)'}),
# 分析師觀點與市場解讀區塊
html.Div([
html.H3("分析師觀點與市場解讀", style={'color': '#2E86AB', 'margin-bottom': '20px'}),
html.Div([
html.Div([
html.H4("技術面分析", style={'color': '#A23B72', 'margin-bottom': '15px'}),
html.Div(id='technical-analysis-text', style={'background': '#f8f9fa','padding': '15px','border-radius': '8px','border-left': '4px solid #A23B72','min-height': '150px','font-size': '14px','line-height': '1.6'})
], style={'width': '48%', 'display': 'inline-block', 'vertical-align': 'top'}),
html.Div([
html.H4("基本面分析 (AI 生成)", style={'color': '#F18F01', 'margin-bottom': '15px'}),
html.Div(id='fundamental-analysis-text', style={'background': '#f8f9fa','padding': '15px','border-radius': '8px','border-left': '4px solid #F18F01','min-height': '150px','font-size': '14px','line-height': '1.6'})
], style={'width': '48%', 'display': 'inline-block', 'margin-left': '4%', 'vertical-align': 'top'})
]),
html.Div([
html.H4("市場展望與投資建議 (AI 生成)", style={'color': '#C73E1D', 'margin-bottom': '15px', 'margin-top': '25px'}),
html.Div(id='market-outlook-text', style={'background': 'linear-gradient(135deg, #667eea 0%, #764ba2 100%)','color': 'white','padding': '20px','border-radius': '10px','min-height': '100px','font-size': '15px','line-height': '1.7','box-shadow': '0 4px 15px rgba(0,0,0,0.1)'})
])
], style={'margin-top': '30px','padding': '25px','background': 'white','border-radius': '12px','box-shadow': '0 4px 20px rgba(0,0,0,0.08)','border': '1px solid #e9ecef'}),
# 股票選擇與設定區塊
html.Div([
html.Div([
html.Label("選擇股票:"),
dcc.Dropdown(id='stock-dropdown', options=[{'label': name, 'value': symbol} for name, symbol in TAIWAN_STOCKS.items()], value='2330.TW', style={'margin-bottom': '10px'})
], style={'width': '30%', 'display': 'inline-block', 'vertical-align': 'top'}),
html.Div([
html.Label("時間範圍:"),
dcc.Dropdown(id='period-dropdown',
options=[{'label': '1個月', 'value': '1mo'},{'label': '3個月', 'value': '3mo'},{'label': '6個月', 'value': '6mo'},{'label': '1年', 'value': '1y'},{'label': '2年', 'value': '2y'}],
value='1mo', style={'margin-bottom': '10px'})
], style={'width': '30%', 'display': 'inline-block', 'margin-left': '5%', 'vertical-align': 'top'}),
html.Div([
html.Label("圖表類型:"),
dcc.Dropdown(id='chart-type', options=[{'label': '線圖', 'value': 'line'},{'label': '蠟燭圖', 'value': 'candlestick'}], value='candlestick', style={'margin-bottom': '10px'})
], style={'width': '30%', 'display': 'inline-block', 'margin-left': '5%', 'vertical-align': 'top'})
], style={'margin-bottom': '30px'}),
# 股票資訊卡片
html.Div(id='stock-info-cards', style={'margin-bottom': '30px'}),
# 價格圖表
html.Div([html.Div([dcc.Graph(id='price-chart')], style={'width': '100%', 'display': 'inline-block', 'vertical-align': 'top'})]),
# 進階技術指標分析區塊
html.Div([
html.H3("進階技術指標分析", style={'margin-bottom': '20px'}),
html.Div([
html.Label("選擇技術指標:", style={'font-weight': 'bold', 'margin-right': '10px'}),
dcc.Dropdown(id='technical-indicator-selector',
options=[{'label': 'RSI 相對強弱指標', 'value': 'RSI'},{'label': 'MACD 指數平滑異同移動平均線', 'value': 'MACD'},{'label': '布林通道 Bollinger Bands', 'value': 'BB'},
{'label': 'KD 隨機指標', 'value': 'KD'},{'label': '威廉指標 %R', 'value': 'WR'},{'label': 'DMI 動向指標', 'value': 'DMI'}],
value='RSI', style={'width': '100%'})
], style={'margin-bottom': '20px'}),
html.Div([dcc.Graph(id='advanced-technical-chart')])
], style={'margin-top': '20px','padding': '20px','background': 'white','border-radius': '10px','box-shadow': '0 2px 10px rgba(0,0,0,0.1)'}),
# 成交量圖表
html.Div([dcc.Graph(id='volume-chart')], style={'margin-top': '20px'}),
# 產業表現分析
html.Div([html.H3("產業表現分析"), dcc.Graph(id='industry-analysis')], style={'margin-top': '30px'}),
# 景氣燈號與PMI分析區塊
html.Div([
html.H3("景氣燈號與 PMI 分析"),
html.Div([
html.Div([dcc.Graph(id='business-climate-chart')], style={'width': '48%', 'display': 'inline-block'}),
html.Div([dcc.Graph(id='pmi-chart')], style={'width': '48%', 'display': 'inline-block', 'margin-left': '2%'})
])
], style={'margin-top': '30px'}),
# 多檔股票比較分析區塊
html.Div([
html.H3("多檔股票比較分析", style={'margin-bottom': '20px'}),
html.Div([
html.Div([
html.Label("選擇比較股票（最多5檔）:", style={'font-weight': 'bold'}),
dcc.Dropdown(id='comparison-stocks', options=[{'label': name, 'value': symbol} for name, symbol in TAIWAN_STOCKS.items()], value=['0050.TW', '2330.TW', '2454.TW'], multi=True, style={'margin-bottom': '5px'}),
html.Small('（元大台灣50 (0050.TW) 為固定比較基準，不可移除）', style={'display': 'block', 'font-style': 'italic', 'color': 'gray'})
], style={'width': '60%', 'display': 'inline-block'}),
html.Div([
html.Label("比較期間:", style={'font-weight': 'bold'}),
dcc.Dropdown(id='comparison-period', options=[{'label': '1個月', 'value': '1mo'},{'label': '3個月', 'value': '3mo'},{'label': '6個月', 'value': '6mo'},{'label': '1年', 'value': '1y'}], value='3mo')
], style={'width': '35%', 'display': 'inline-block', 'margin-left': '5%', 'vertical-align': 'top'})
]),
html.Div([
html.Div([dcc.Graph(id='comparison-chart')], style={'width': '65%', 'display': 'inline-block'}),
html.Div([html.H4("比較結果", style={'color': '#2E86AB'}), html.Div(id='comparison-table')], style={'width': '33%', 'display': 'inline-block', 'margin-left': '2%', 'vertical-align': 'top'})
])
], style={'margin-top': '30px','padding': '20px','background': 'white','border-radius': '10px','box-shadow': '0 2px 10px rgba(0,0,0,0.1)'}),
# 風險管理與投資組合分析區塊
html.Div([
html.H3("風險管理與投資組合分析", style={'margin-bottom': '20px', 'color': '#C0392B'}),
html.Div([
html.Div([
html.Label("選擇投資組合股票（最多5檔）:", style={'font-weight': 'bold'}),
dcc.Dropdown(
id='portfolio-stocks',
options=[{'label': name, 'value': symbol} for name, symbol in TAIWAN_STOCKS.items()],
value=['0050.TW', '2330.TW', '2454.TW'],
multi=True
)
], style={'width': '60%', 'display': 'inline-block', 'vertical-align': 'top'}),
html.Div([
html.Label("分析期間:", style={'font-weight': 'bold'}),
dcc.Dropdown(
id='risk-period-dropdown',
options=[
{'label': '6個月', 'value': '6mo'},
{'label': '1年', 'value': '1y'},
{'label': '2年', 'value': '2y'}
],
value='1y'
)
], style={'width': '35%', 'display': 'inline-block', 'margin-left': '5%', 'vertical-align': 'top'})
], style={'margin-bottom': '20px'}),
html.Div(id='risk-metrics-display', style={'margin-bottom': '20px'}),
dcc.Graph(id='risk-analysis-chart')
], style={'margin-top': '30px', 'padding': '20px', 'background': 'white', 'border-radius': '10px', 'box-shadow': '0 2px 10px rgba(0,0,0,0.1)'}),
# 交易回測功能區塊
html.Div([
html.H3("AI驅動交易策略回測", style={'margin-bottom': '20px', 'color': '#8E44AD'}),
# 回測參數設定區
html.Div([
html.Div([
html.Label("選擇回測股票:", style={'font-weight': 'bold'}),
dcc.Dropdown(
id='backtest-stock-dropdown',
options=[{'label': name, 'value': symbol} for name, symbol in TAIWAN_STOCKS.items()],
value='0050.TW'
)
], style={'width': '30%', 'display': 'inline-block'}),
html.Div([
html.Label("回測期間:", style={'font-weight': 'bold'}),
dcc.Dropdown(
id='backtest-period-dropdown',
options=[
{'label': '6個月', 'value': '6mo'},
{'label': '1年', 'value': '1y'},
{'label': '2年', 'value': '2y'}
],
value='1y'
)
], style={'width': '25%', 'display': 'inline-block', 'margin-left': '3%'}),
html.Div([
html.Label("初始資金 (萬元):", style={'font-weight': 'bold'}),
dcc.Input(
id='initial-capital-input',
type='number',
value=100,
min=10,
max=1000,
step=10,
style={'width': '100%', 'padding': '5px'}
)
], style={'width': '20%', 'display': 'inline-block', 'margin-left': '3%'}),
html.Div([
html.Button(
'開始回測',
id='run-backtest-button',
n_clicks=0,
style={
'background': 'linear-gradient(45deg, #667eea 0%, #764ba2 100%)',
'color': 'white',
'border': 'none',
'padding': '10px 20px',
'border-radius': '5px',
'font-weight': 'bold',
'width': '100%',
'cursor': 'pointer'
}
)
], style={'width': '15%', 'display': 'inline-block', 'margin-left': '4%', 'vertical-align': 'bottom'})
], style={'margin-bottom': '20px'}),
# 結果顯示區
html.Div(id='backtest-progress'),
html.Div(id='backtest-summary-card'),
dcc.Graph(id='backtest-chart'),
html.Div(id='trading-details-table')
], style={
'margin-top': '30px',
'padding': '25px',
'background': 'white',
'border-radius': '12px',
'box-shadow': '0 4px 20px rgba(0,0,0,0.08)',
'border': '1px solid #e9ecef'
})
])
# 修改台指期預測的回調函數
@app.callback(
[Output('taiex-prediction-results', 'children'),
Output('taiex-prediction-chart', 'figure')],
[Input('taiex-prediction-period', 'value')]
)
def update_taiex_prediction(predict_days):
data = get_stock_data('^TWII', '2y')
if data.empty: return html.Div("無法獲取台指期資料"), {}
# === 修改點：統一呼叫 get_prediction 控制器 ===
final_prediction = get_prediction(data, predict_days)
if final_prediction is None: return html.Div("資料不足，無法進行預測"), {}
current_price, last_date = data['Close'].iloc[-1], data.index[-1]
# 【重要更新】現在 change_pct 已經是正確的漲幅百分比
predicted_price, change_pct, confidence = final_prediction['predicted_price'], final_prediction['change_pct'], final_prediction['confidence']
# 生成預測路徑（為了圖表顯示）
prediction_paths = {1: [1], 5: [1, 5], 10: [1, 5, 10], 20: [1, 10, 20], 60: [1, 10, 20, 60]}
intervals_to_predict = prediction_paths.get(predict_days, [predict_days])
prediction_dates, prediction_prices = [last_date], [current_price]
for days in intervals_to_predict:
interim_prediction = get_prediction(data, days)
if interim_prediction:
prediction_dates.append(last_date + timedelta(days=days))
prediction_prices.append(interim_prediction['predicted_price'])
# 後續繪圖邏輯不變，但現在 change_pct 是真正的漲幅百分比
color, arrow = ('red', '📈') if change_pct >= 0 else ('green', '📉')
result_card = html.Div([
html.H4(f"{predict_days}日後預測結果", style={'margin': '0 0 15px 0', 'color': 'white'}),
html.Div([
html.Span(f"{arrow} ", style={'font-size': '24px'}),
html.Span(f"{change_pct:+.2f}%", style={'font-size': '28px','font-weight': 'bold','color': color})
], style={'margin': '10px 0'}),
html.P(f"目前價格: {current_price:.2f}", style={'margin': '5px 0'}),
html.P(f"預測價格: {predicted_price:.2f}", style={'margin': '5px 0'}),
html.P(f"預測漲幅: {change_pct:+.2f}%", style={'margin': '5px 0', 'font-weight': 'bold'}), # 【新增】
html.P(f"信心度: {confidence:.1%}", style={'margin': '5px 0', 'font-size': '14px'})
], style={'background': 'rgba(255,255,255,0.1)','padding': '20px','border-radius': '10px','border': '1px solid rgba(255,255,255,0.2)'})
# 繪圖部分保持不變
fig = go.Figure()
recent_data = data.tail(30)
fig.add_trace(go.Scatter(x=recent_data.index, y=recent_data['Close'], mode='lines', name='歷史價格', line=dict(color='#FFA726', width=2)))
fig.add_trace(go.Scatter(x=prediction_dates, y=prediction_prices, mode='lines+markers', name=f'{predict_days}日預測路徑', line=dict(color=color, width=3, dash='dash'), marker=dict(size=8)))
fig.update_layout(title=f'台指期 {predict_days}日預測走勢 (預測漲幅: {change_pct:+.2f}%)', xaxis_title='日期', yaxis_title='指數點位', height=350, plot_bgcolor='rgba(0,0,0,0)', paper_bgcolor='rgba(0,0,0,0)', font=dict(color='white'))
return result_card, fig
@app.callback(
Output('stock-info-cards', 'children'),
[Input('stock-dropdown', 'value')]
)
def update_stock_info(selected_stock):
data = get_stock_data(selected_stock, '5d')
if data.empty: return html.Div("無法獲取股票資料")
current_price = data['Close'].iloc[-1]
prev_price = data['Close'].iloc[-2] if len(data) > 1 else current_price
change = current_price - prev_price
change_pct = (change / prev_price) * 100
stock_name = [name for name, symbol in TAIWAN_STOCKS.items() if symbol == selected_stock][0]
color, arrow = ('red', '▲') if change >= 0 else ('green', '▼')
return html.Div([
html.Div([
html.H3(f"{stock_name} ({selected_stock})", style={'margin': '0'}),
html.H2(f"${current_price:.2f}", style={'margin': '5px 0', 'color': color}),
html.P(f"{arrow} {change:+.2f} ({change_pct:+.2f}%)", style={'margin': '0', 'color': color, 'font-weight': 'bold'})
], style={'background': 'white','padding': '20px','border-radius': '10px','box-shadow': '0 2px 10px rgba(0,0,0,0.1)','display': 'inline-block','margin-right': '20px'}),
html.Div([
html.H4("今日統計", style={'margin': '0 0 10px 0'}),
html.P(f"最高: ${data['High'].iloc[-1]:.2f}", style={'margin': '5px 0'}),
html.P(f"最低: ${data['Low'].iloc[-1]:.2f}", style={'margin': '5px 0'}),
html.P(f"成交量: {data['Volume'].iloc[-1]:,.0f}", style={'margin': '5px 0'})
], style={'background': 'white','padding': '20px','border-radius': '10px','box-shadow': '0 2px 10px rgba(0,0,0,0.1)','display': 'inline-block'})
])
@app.callback(
Output('price-chart', 'figure'),
[Input('stock-dropdown', 'value'),
Input('period-dropdown', 'value'),
Input('chart-type', 'value')]
)
def update_price_chart(selected_stock, period, chart_type):
data = get_stock_data(selected_stock, period)
if data.empty: return {}
data = calculate_technical_indicators(data)
stock_name = [name for name, symbol in TAIWAN_STOCKS.items() if symbol == selected_stock][0]
fig = make_subplots(rows=1, cols=2, shared_yaxes=True, column_widths=[0.8, 0.2], horizontal_spacing=0.01)
if chart_type == 'candlestick':
fig.add_trace(go.Candlestick(x=data.index, open=data['Open'], high=data['High'], low=data['Low'], close=data['Close'], name=stock_name, increasing_line_color='red', decreasing_line_color='green'), row=1, col=1)
else:
fig.add_trace(px.line(data, y='Close').data[0], row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['MA5'], mode='lines', name='MA5', line=dict(color='orange')), row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['MA20'], mode='lines', name='MA20', line=dict(color='blue')), row=1, col=1)
bin_edges, volume_per_bin, price_centers = calculate_volume_profile(data, num_bins=50)
if volume_per_bin is not None:
fig.add_trace(go.Bar(orientation='h', y=price_centers, x=volume_per_bin, name='Volume Profile', text=[f'{vol/1000:.0f}k' for vol in volume_per_bin], textposition='auto', marker=dict(color='rgba(173, 216, 230, 0.6)', line=dict(color='rgba(30, 144, 255, 0.8)', width=1))), row=1, col=2)
fig.update_layout(title_text=f'{stock_name} 股價走勢與成交量分佈', height=500, showlegend=True, xaxis1=dict(title='日期', type='date', rangeslider_visible=False), yaxis1=dict(title='價格 (TWD)'), xaxis2=dict(title='成交量', showticklabels=True), yaxis2=dict(showticklabels=False), bargap=0.05)
return fig
@app.callback(
Output('advanced-technical-chart', 'figure'),
[Input('technical-indicator-selector', 'value'),
Input('stock-dropdown', 'value'),
Input('period-dropdown', 'value')]
)
def update_advanced_technical_chart(indicator, selected_stock, period):
data = get_stock_data(selected_stock, period)
if data.empty: return {}
data = calculate_technical_indicators(data)
stock_name = [name for name, symbol in TAIWAN_STOCKS.items() if symbol == selected_stock][0]
fig = go.Figure()
if indicator == 'RSI':
fig = go.Figure()
fig.add_trace(go.Scatter(x=data.index, y=data['RSI'], mode='lines', name='RSI', line=dict(color='purple', width=2)))
fig.add_hline(y=70, line_dash="dash", line_color="green", annotation_text="超買線(70)")
fig.add_hline(y=30, line_dash="dash", line_color="red", annotation_text="超賣線(30)")
fig.add_hline(y=50, line_dash="dot", line_color="gray", annotation_text="中線(50)")
fig.update_layout(title=f'{stock_name} - RSI 相對強弱指標', xaxis_title='日期', yaxis_title='RSI', height=450, yaxis=dict(range=[0, 100]))
elif indicator == 'MACD':
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.1, row_heights=[0.7, 0.3], subplot_titles=('價格走勢', 'MACD 指標'))
fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='收盤價', line=dict(color='black', width=1.5)), row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['MACD'], mode='lines', name='MACD (快線)', line=dict(color='blue', width=2)), row=2, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['MACD_Signal'], mode='lines', name='Signal (慢線)', line=dict(color='red', width=2)), row=2, col=1)
colors = ['red' if x >= 0 else 'green' for x in data['MACD_Histogram']]
fig.add_trace(go.Bar(x=data.index, y=data['MACD_Histogram'], name='MACD柱狀圖', marker_color=colors), row=2, col=1)
fig.update_layout(title_text=f'{stock_name} - MACD 指數平滑異同移動平均線', height=550)
elif indicator == 'BB':
fig = go.Figure()
fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='收盤價', line=dict(color='black', width=2)))
fig.add_trace(go.Scatter(x=data.index, y=data['BB_Upper'], mode='lines', name='上軌', line=dict(color='red', width=1, dash='dash')))
fig.add_trace(go.Scatter(x=data.index, y=data['BB_Middle'], mode='lines', name='中軌(MA20)', line=dict(color='blue', width=1)))
fig.add_trace(go.Scatter(x=data.index, y=data['BB_Lower'], mode='lines', name='下軌', line=dict(color='green', width=1, dash='dash')))
fig.update_layout(title=f'{stock_name} - 布林通道 (20日, 2σ)', xaxis_title='日期', yaxis_title='價格 (TWD)', height=450)
elif indicator == 'KD':
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.1, row_heights=[0.6, 0.4], subplot_titles=('價格走勢', 'KD指標'))
fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='收盤價', line=dict(color='black', width=1)), row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['K'], mode='lines', name='K線', line=dict(color='blue', width=2)), row=2, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['D'], mode='lines', name='D線', line=dict(color='red', width=2)), row=2, col=1)
fig.add_hline(y=80, line_dash="dash", line_color="green", annotation_text="超買線(80)", row=2, col=1)
fig.add_hline(y=20, line_dash="dash", line_color="red", annotation_text="超賣線(20)", row=2, col=1)
fig.update_layout(title=f'{stock_name} - KD 隨機指標 (9,3,3)', height=500, yaxis2_range=[0, 100])
elif indicator == 'WR':
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.1, row_heights=[0.6, 0.4], subplot_titles=('價格走勢', '威廉指標 %R'))
fig.add_trace(go.Scatter(x=data.index, y=data['Close'], mode='lines', name='收盤價', line=dict(color='black', width=1)), row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['Williams_R'], mode='lines', name='威廉%R', line=dict(color='purple', width=2)), row=2, col=1)
fig.add_hline(y=-20, line_dash="dash", line_color="green", annotation_text="超買線(-20)", row=2, col=1)
fig.add_hline(y=-80, line_dash="dash", line_color="red", annotation_text="超賣線(-80)", row=2, col=1)
fig.update_layout(title=f'{stock_name} - 威廉指標 %R (14日)', height=500, yaxis2_range=[-100, 0])
elif indicator == 'DMI':
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.1, row_heights=[0.6, 0.4], subplot_titles=('價格走勢', 'DMI 指標'))
data_filtered = data.iloc[14:]
fig.add_trace(go.Scatter(x=data_filtered.index, y=data_filtered['Close'], mode='lines', name='收盤價', line=dict(color='black', width=1)), row=1, col=1)
fig.add_trace(go.Scatter(x=data_filtered.index, y=data_filtered['+DI'], mode='lines', name='+DI', line=dict(color='red', width=2)), row=2, col=1)
fig.add_trace(go.Scatter(x=data_filtered.index, y=data_filtered['-DI'], mode='lines', name='-DI', line=dict(color='green', width=2)), row=2, col=1)
fig.add_trace(go.Scatter(x=data_filtered.index, y=data_filtered['ADX'], mode='lines', name='ADX', line=dict(color='blue', width=2, dash='dot')), row=2, col=1)
fig.update_layout(title=f'{stock_name} - DMI 動向指標 (14日)', height=500, showlegend=True, yaxis2_range=[0, 100])
return fig
@app.callback(
Output('volume-chart', 'figure'),
[Input('stock-dropdown', 'value'),
Input('period-dropdown', 'value')]
)
def update_volume_chart(selected_stock, period):
data = get_stock_data(selected_stock, period)
if data.empty: return {}
stock_name = [name for name, symbol in TAIWAN_STOCKS.items() if symbol == selected_stock][0]
colors = ['red' if data['Close'].iloc[i] > data['Open'].iloc[i] else 'green' for i in range(len(data))]
fig = go.Figure(go.Bar(x=data.index, y=data['Volume'], marker_color=colors, name='成交量'))
fig.update_layout(title=f'{stock_name} 成交量', xaxis_title='日期', yaxis_title='成交量', height=300)
return fig
@app.callback(
Output('industry-analysis', 'figure'),
[Input('stock-dropdown', 'value')]
)
def update_industry_analysis(selected_stock):
performance_data = []
for name, symbol in TAIWAN_STOCKS.items():
data = get_stock_data(symbol, '1mo')
if not data.empty and len(data) > 1:
return_pct = ((data['Close'].iloc[-1] - data['Close'].iloc[0]) / data['Close'].iloc[0]) * 100
performance_data.append({
'股票': name,
'代碼': symbol,
'月報酬率(%)': return_pct,
'絕對波動': abs(return_pct)
})
if not performance_data:
fig = go.Figure().add_annotation(text="無法計算產業資料", showarrow=False)
fig.update_layout(title="近一月市場表現分析", height=400)
return fig
df_performance = pd.DataFrame(performance_data)
# 分離漲跌幅數據
gainers = df_performance[df_performance['月報酬率(%)'] > 0].copy()
losers = df_performance[df_performance['月報酬率(%)'] < 0].copy()
# 按報酬率排序並取前5名
top_gainers = gainers.sort_values(by='月報酬率(%)', ascending=False).head(5)
top_losers = losers.sort_values(by='月報酬率(%)', ascending=True).head(5)
# 創建子圖布局 - 1行2列
fig = make_subplots(
rows=1, cols=2,
specs=[[{"type": "pie"}, {"type": "pie"}]],
subplot_titles=('📈 近一月漲幅排行 Top 5', '📉 近一月跌幅排行 Top 5'),
horizontal_spacing=0.1
)
# 如果有上漲的股票，添加漲幅圓餅圖
if not top_gainers.empty:
fig.add_trace(go.Pie(
labels=top_gainers['股票'],
values=top_gainers['月報酬率(%)'],
name="漲幅",
textinfo='label+percent',
textposition='inside',
marker=dict(colors=['#FF6B6B', '#FF8E53', '#FF6B9D', '#C44569', '#F8B500']),
hovertemplate="%{label}
漲幅: +%{value:.1f}%",
textfont=dict(size=12)
), row=1, col=1)
else:
# 如果沒有上漲股票，顯示提示
fig.add_annotation(
text="本月無上漲股票",
x=0.25, y=0.5,
showarrow=False,
font=dict(size=16, color="gray")
)
# 如果有下跌的股票，添加跌幅圓餅圖（使用絕對值）
if not top_losers.empty:
fig.add_trace(go.Pie(
labels=top_losers['股票'],
values=abs(top_losers['月報酬率(%)']), # 使用絕對值讓圓餅圖正常顯示
name="跌幅",
textinfo='label+percent',
textposition='inside',
marker=dict(colors=['#20BF6B', '#26DE81', '#2BCBBA', '#45AAF2', '#4834D4']),
hovertemplate="%{label}
跌幅: %{customdata:.1f}%",
customdata=top_losers['月報酬率(%)'], # 顯示實際的負值
textfont=dict(size=12)
), row=1, col=2)
else:
# 如果沒有下跌股票，顯示提示
fig.add_annotation(
text="本月無下跌股票",
x=0.75, y=0.5,
showarrow=False,
font=dict(size=16, color="gray")
)
# 更新布局
fig.update_layout(
title_text="近一月市場表現分析 - 漲跌分佈",
height=500,
showlegend=False,
font=dict(size=11),
title_font_size=16,
annotations=[
dict(text=f"統計範圍：{len(performance_data)}檔股票",
x=0.5, y=-0.1,
showarrow=False,
xanchor="center",
font=dict(size=12, color="gray"))
]
)
return fig
@app.callback(
Output('business-climate-chart', 'figure'),
[Input('stock-dropdown', 'value')]
)
def update_business_climate_chart(selected_stock):
df = get_business_climate_data()
if df.empty:
fig = go.Figure().add_annotation(text="無法載入景氣燈號資料", showarrow=False)
fig.update_layout(title="台灣景氣燈號", height=300)
return fig
def get_light_color(score):
if score >= 32: return 'red'
elif score >= 24: return 'orange'
elif score >= 17: return 'yellow'
elif score >= 10: return 'lightgreen'
else: return 'blue'
colors = [get_light_color(score) for score in df['Index']]
fig = go.Figure()
fig.add_trace(go.Scatter(x=df['Date'], y=df['Index'], mode='lines+markers', name='景氣燈號', line=dict(color='darkblue', width=2), marker=dict(size=8, color=colors, line=dict(width=2, color='darkblue'))))
fig.add_hline(y=32, line_dash="dash", line_color="red", annotation_text="紅燈(32)")
fig.add_hline(y=17, line_dash="dash", line_color="yellow", annotation_text="黃燈(17)")
fig.update_layout(title="台灣景氣燈號走勢", xaxis_title='日期', yaxis_title='燈號分數', height=300, yaxis=dict(range=[0, 40]))
return fig
# ========================= MODIFIED SECTION START (CACHE INTEGRATED) =========================
@app.callback(
[Output('technical-analysis-text', 'children'),
Output('fundamental-analysis-text', 'children'),
Output('market-outlook-text', 'children')],
[Input('stock-dropdown', 'value'),
Input('period-dropdown', 'value')]
)
def update_analysis_text(selected_stock, period):
"""修改版分析文本更新 - 使用正確的變數名稱"""
# 建立快取的唯一鍵值
cache_key = f"{selected_stock}-{period}"
current_time = time.time()
# 檢查快取 - 使用正確定義的變數
if cache_key in ANALYSIS_CACHE:
cached_data = ANALYSIS_CACHE[cache_key]
# 使用全域變數 ANALYSIS_CACHE_DURATION
if current_time - cached_data['timestamp'] < ANALYSIS_CACHE_DURATION:
print(f"從快取載入分析: {cache_key}")
return cached_data['technical'], cached_data['fundamental'], cached_data['outlook']
print(f"重新生成分析: {cache_key}")
# 獲取股票數據
data = get_stock_data(selected_stock, period)
stock_name = [name for name, symbol in TAIWAN_STOCKS.items() if symbol == selected_stock][0]
if data.empty or len(data) < 20:
return "資料不足，無法分析", "資料不足，無法分析", "資料不足，無法分析"
data = calculate_technical_indicators(data)
# 技術面分析
price_change = ((data['Close'].iloc[-1] - data['Close'].iloc[0]) / data['Close'].iloc[0]) * 100
rsi_current = data['RSI'].iloc[-1] if not pd.isna(data['RSI'].iloc[-1]) else 50
macd_current = data['MACD'].iloc[-1] if not pd.isna(data['MACD'].iloc[-1]) else 0
macd_signal_current = data['MACD_Signal'].iloc[-1] if not pd.isna(data['MACD_Signal'].iloc[-1]) else 0
# 添加時間戳顯示數據更新時間
update_time = datetime.now().strftime('%H:%M:%S')
latest_date = data.index[-1].strftime('%Y-%m-%d')
technical_text = html.Div([
html.P([html.Strong("數據更新時間："), f"{latest_date} {update_time}"]),
html.P([html.Strong("價格趨勢："), f"在最近 {period} 期間內，{stock_name} 股價呈現",
html.Span(f"{'上漲' if price_change > 5 else '下跌' if price_change < -5 else '盤整'}",
style={'color': 'red' if price_change > 5 else 'green' if price_change < -5 else 'orange',
'font-weight': 'bold'}),
f"走勢，累計變動 {price_change:+.1f}%。"]),
html.P([html.Strong("RSI 指標："), f"目前的 RSI 值為 {rsi_current:.1f}，",
html.Span("處於超買區(>70)" if rsi_current > 70 else "處於超賣區(<30)" if rsi_current < 30 else "在正常範圍內",
style={'color': 'green' if rsi_current > 70 else 'red' if rsi_current < 30 else 'blue',
'font-weight': 'bold'}), "。"]),
html.P([html.Strong("MACD 指標："), f"MACD 快線 ({macd_current:.3f}) 目前",
html.Span("高於" if macd_current > macd_signal_current else "低於",
style={'color': 'red' if macd_current > macd_signal_current else 'green',
'font-weight': 'bold'}),
f" Signal 慢線 ({macd_signal_current:.3f})。"]),
])
# 基本面與展望分析 (調用 Gemini)
fundamental_text, market_outlook_text = generate_gemini_analysis(stock_name, selected_stock, period, data)
# 將新產生的結果存入快取
ANALYSIS_CACHE[cache_key] = {
'technical': technical_text,
'fundamental': fundamental_text,
'outlook': market_outlook_text,
'timestamp': current_time
}
return technical_text, fundamental_text, market_outlook_text
# ========================== MODIFIED SECTION END ==========================
@app.callback(
Output('pmi-chart', 'figure'),
[Input('stock-dropdown', 'value')]
)
def update_pmi_chart(selected_stock):
df = get_pmi_data()
if df.empty:
fig = go.Figure().add_annotation(text="無法載入PMI資料", showarrow=False)
fig.update_layout(title="台灣PMI指數", height=300)
return fig
colors = ['red' if value >= 50 else 'green' for value in df['Index']]
fig = go.Figure()
fig.add_trace(go.Scatter(x=df['Date'], y=df['Index'], mode='lines+markers', name='PMI指數', line=dict(color='darkblue', width=2), marker=dict(size=8, color=colors, line=dict(width=2, color='darkblue'))))
fig.add_hline(y=50, line_dash="dash", line_color="black", annotation_text="榮枯線(50)")
fig.update_layout(title="台灣PMI指數走勢", xaxis_title='日期', yaxis_title='PMI指數', height=300, yaxis=dict(range=[35, 60]))
return fig
def summarize_news_with_gemini(news_list: list) -> str:
"""
使用 Gemini API 將英文新聞標題列表摘要成一段繁體中文。
"""
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
return "錯誤：找不到 GEMINI_API_KEY。請在 Hugging Face Secrets 中設定。"
try:
genai.configure(api_key=api_key)
model = genai.GenerativeModel('gemini-2.0-flash')
formatted_news = "\n".join([f"- {news}" for news in news_list])
prompt = f"""
請扮演一位專業的金融市場分析師。
以下是幾則最新的英文財經新聞標題，請將它們整合成一段簡潔、流暢、約 200 字的繁體中文市場動態摘要，與利多哪些產業，利空哪些產業。
提供3段重點，
請專注於可能影響市場情緒和股價的關鍵資訊，並直接提供摘要內容，不要包含任何額外的問候語或說明。
英文新聞標題如下：
{formatted_news}
"""
response = model.generate_content(prompt)
return response.text
except Exception as e:
print(f"呼叫 Gemini API 時發生錯誤: {e}")
return f"無法生成新聞摘要，請稍後再試。錯誤訊息：{e}"
@app.callback(
[Output('comparison-chart', 'figure'),
Output('comparison-table', 'children')],
[Input('comparison-stocks', 'value'),
Input('comparison-period', 'value')]
)
def update_comparison_analysis(selected_stocks, period):
fixed_stock = '0050.TW'
if not selected_stocks: selected_stocks = [fixed_stock]
elif fixed_stock not in selected_stocks: selected_stocks.insert(0, fixed_stock)
selected_stocks = selected_stocks[:5]
fig = go.Figure()
comparison_data = []
for stock in selected_stocks:
data = get_stock_data(stock, period)
if not data.empty:
stock_name = next((name for name, symbol in TAIWAN_STOCKS.items() if symbol == stock), stock)
normalized_prices = (data['Close'] / data['Close'].iloc[0]) * 100
fig.add_trace(go.Scatter(x=data.index, y=normalized_prices, mode='lines', name=stock_name, line=dict(width=2)))
total_return = ((data['Close'].iloc[-1] / data['Close'].iloc[0]) - 1) * 100
volatility = data['Close'].pct_change().std() * np.sqrt(252) * 100
comparison_data.append({'name': stock_name, 'return': total_return, 'volatility': volatility, 'current_price': data['Close'].iloc[-1]})
fig.update_layout(title=f'股票績效比較 - {period}', xaxis_title='日期', yaxis_title='相對績效 (基期=100)', height=400, hovermode='x unified')
if comparison_data:
table_rows = []
for item in sorted(comparison_data, key=lambda x: x['return'], reverse=True):
color = 'red' if item['return'] > 0 else 'green'
table_rows.append(html.Tr([html.Td(item['name'], style={'font-weight': 'bold'}), html.Td(f"{item['return']:+.1f}%", style={'color': color, 'font-weight': 'bold'}), html.Td(f"{item['volatility']:.1f}%"), html.Td(f"${item['current_price']:.2f}")]))
table = html.Table([html.Thead(html.Tr([html.Th("股票"), html.Th("報酬率"), html.Th("波動率"), html.Th("現價")])), html.Tbody(table_rows)], style={'width': '100%'})
return fig, table
return fig, html.Div("無可比較資料")
@app.callback(
[Output('sentiment-gauge', 'children'),
Output('news-summary', 'children')],
[Input('stock-dropdown', 'value')]
)
def update_sentiment_analysis(selected_stock):
if predictor is None:
error_fig = go.Figure().add_annotation(text="情緒指標模型載入失敗", showarrow=False)
error_fig.update_layout(height=200)
return dcc.Graph(figure=error_fig), html.P("新聞分析模型載入失敗，請檢查後台日誌。")
sentiment_score_raw = predictor.get_news_index()
if sentiment_score_raw is not None:
sentiment_score_normalized = (sentiment_score_raw + 1) * 50
sentiment_score_normalized = max(0, min(100, sentiment_score_normalized))
if sentiment_score_normalized >= 65:
bar_color, level_text = "#5cb85c", "樂觀"
elif sentiment_score_normalized >= 35:
bar_color, level_text = "#f0ad4e", "中性"
else:
bar_color, level_text = "#d9534f", "悲觀"
gauge_fig = go.Figure(go.Indicator(
mode = "gauge+number", value = sentiment_score_normalized,
domain = {'x': [0, 1], 'y': [0, 1]},
title = {'text': f"昨日市場情緒: {level_text}", 'font': {'size': 18}},
gauge = {'axis': {'range': [0, 100]}, 'bar': {'color': bar_color},
'steps': [{'range': [0, 35], 'color': "rgba(217, 83, 79, 0.2)"},
{'range': [35, 65], 'color': "rgba(240, 173, 78, 0.2)"},
{'range': [65, 100], 'color': "rgba(92, 184, 92, 0.2)"}]}
))
gauge_fig.update_layout(height=200, margin=dict(l=30, r=30, t=50, b=20))
gauge_content = dcc.Graph(figure=gauge_fig)
else:
error_fig = go.Figure().add_annotation(text="今日尚無情緒分數", showarrow=False)
error_fig.update_layout(height=200)
gauge_content = dcc.Graph(figure=error_fig)
top_news_list = predictor.get_news()
news_content = None
if top_news_list and isinstance(top_news_list, list):
summary_text = summarize_news_with_gemini(top_news_list)
news_content = dcc.Markdown(summary_text, style={
'margin': '8px 0', 'padding-left': '5px',
'font-size': '15px', 'line-height': '1.7'
})
elif top_news_list == []:
news_content = html.P("昨日無重大相關新聞。", style={'text-align': 'center', 'padding-top': '50px'})
else:
news_content = html.P("讀取新聞時發生錯誤。", style={'text-align': 'center', 'padding-top': '50px'})
return gauge_content, news_content
# ========================= 新增：風險管理模組回調函數 START =========================
@app.callback(
[Output('risk-metrics-display', 'children'),
Output('risk-analysis-chart', 'figure')],
[Input('portfolio-stocks', 'value'),
Input('risk-period-dropdown', 'value')]
)
def update_risk_analysis(selected_stocks, period):
if not selected_stocks:
return html.Div("請至少選擇一檔股票進行分析。", style={'color': 'red', 'text-align': 'center'}), {}
selected_stocks = selected_stocks[:5] # 限制最多5檔
analyzer = RiskAnalyzer()
# 獲取大盤數據
market_data = get_stock_data('^TWII', period)
if market_data.empty:
return html.Div("無法獲取大盤數據 (^TWII)，無法計算 Beta 值。", style={'color': 'red'}), {}
market_returns = market_data['Close'].pct_change().dropna()
all_stock_data = {}
all_returns_data = {}
individual_metrics_cards = []
# 計算個股指標
for stock_symbol in selected_stocks:
stock_data = get_stock_data(stock_symbol, period)
stock_name = next((name for name, symbol in TAIWAN_STOCKS.items() if symbol == stock_symbol), stock_symbol)
if stock_data.empty or len(stock_data) < 30:
card = html.Div([
html.H5(f"{stock_name} ({stock_symbol})", style={'margin-bottom': '10px', 'color': '#34495E'}),
html.P("資料不足，無法計算風險指標。")
], className="metric-card", style={'border-left': '5px solid #E74C3C'})
individual_metrics_cards.append(card)
continue
all_stock_data[stock_symbol] = stock_data
stock_returns = stock_data['Close'].pct_change().dropna()
all_returns_data[stock_symbol] = stock_returns
# 計算指標
beta = analyzer.calculate_beta(stock_returns, market_returns)
sharpe = analyzer.calculate_sharpe_ratio(stock_returns)
max_dd = analyzer.calculate_max_drawdown(stock_data['Close'])
var_95 = analyzer.calculate_var(stock_returns, 0.05)
# 建立個股卡片
card = html.Div([
html.H5(f"{stock_name} ({stock_symbol})", style={'margin-bottom': '10px', 'color': '#34495E'}),
html.Table([
html.Tr([html.Td("Beta (與大盤相關性)"), html.Td(f"{beta:.2f}" if beta is not None else "N/A")]),
html.Tr([html.Td("年化夏普比率"), html.Td(f"{sharpe:.2f}" if sharpe is not None else "N/A")]),
html.Tr([html.Td("最大回撤 (MDD)"), html.Td(f"{max_dd:.2%}" if max_dd is not None else "N/A")]),
html.Tr([html.Td("每日 VaR (95%)"), html.Td(f"{var_95:.2%}" if var_95 is not None else "N/A")]),
])
], className="metric-card", style={'border-left': '5px solid #3498DB', 'width': '32%', 'display': 'inline-block', 'margin': '0.5%', 'padding': '15px', 'background': '#f9f9f9', 'border-radius': '5px', 'box-shadow': '0 1px 3px rgba(0,0,0,0.1)'})
individual_metrics_cards.append(card)
# 計算投資組合指標 (假設等權重)
if not all_returns_data:
return html.Div(individual_metrics_cards), {}
returns_df = pd.DataFrame(all_returns_data).dropna()
num_assets = len(returns_df.columns)
weights = np.array([1/num_assets] * num_assets)
portfolio_returns = returns_df.dot(weights)
portfolio_sharpe = analyzer.calculate_sharpe_ratio(portfolio_returns)
portfolio_volatility = portfolio_returns.std() * np.sqrt(252)
portfolio_var_95 = analyzer.calculate_var(portfolio_returns, 0.05)
portfolio_var_99 = analyzer.calculate_var(portfolio_returns, 0.01)
# 建立投資組合總結卡片
portfolio_card = html.Div([
html.H4("投資組合整體風險 (等權重)", style={'color': '#2C3E50', 'border-bottom': '2px solid #2C3E50', 'padding-bottom': '10px'}),
html.Table([
html.Tr([html.Td("年化波動率"), html.Td(f"{portfolio_volatility:.2%}" if portfolio_volatility is not None else "N/A")]),
html.Tr([html.Td("年化夏普比率"), html.Td(f"{portfolio_sharpe:.2f}" if portfolio_sharpe is not None else "N/A")]),
html.Tr([html.Td("每日 VaR (95%)"), html.Td(f"{portfolio_var_95:.2%}" if portfolio_var_95 is not None else "N/A")]),
html.Tr([html.Td("每日 VaR (99%)"), html.Td(f"{portfolio_var_99:.2%}" if portfolio_var_99 is not None else "N/A")]),
], style={'width': '100%', 'marginTop': '10px'})
], style={'padding': '20px', 'background': 'linear-gradient(135deg, #a8c0ff 0%, #3f2b96 100%)', 'color': 'white', 'border-radius': '10px', 'margin-bottom': '20px'})
# 圖表：報酬分佈與 VaR
fig = go.Figure()
fig.add_trace(go.Histogram(x=portfolio_returns, nbinsx=50, name='日報酬分佈', marker_color='#3498DB'))
if portfolio_var_95 is not None:
fig.add_vline(x=portfolio_var_95, line_width=2, line_dash="dash", line_color="orange",
annotation_text=f"VaR 95%: {portfolio_var_95:.2%}", annotation_position="top left")
if portfolio_var_99 is not None:
fig.add_vline(x=portfolio_var_99, line_width=2, line_dash="dash", line_color="red",
annotation_text=f"VaR 99%: {portfolio_var_99:.2%}", annotation_position="top right")
fig.update_layout(
title="投資組合日報酬率分佈與 VaR",
xaxis_title="日報酬率",
yaxis_title="頻率",
height=400
)
# 組合最終顯示內容
display_content = html.Div([
portfolio_card,
html.H4("個別資產風險指標", style={'margin-top': '20px', 'color': '#2C3E50'}),
html.Div(individual_metrics_cards)
])
return display_content, fig
# ========================== 新增：風險管理模組回調函數 END ==========================
# 新增：回測結果快取
BACKTEST_CACHE = {}
BACKTEST_CACHE_DURATION = 24 * 60 * 60 # 24小時
# 4. 新增以下回調函數（在其他回調函數之後）：
@app.callback(
[Output('backtest-progress', 'children'),
Output('backtest-summary-card', 'children'),
Output('backtest-chart', 'figure'),
Output('trading-details-table', 'children')],
[Input('run-backtest-button', 'n_clicks')],
[State('backtest-stock-dropdown', 'value'),
State('backtest-period-dropdown', 'value'),
State('initial-capital-input', 'value')]
)
def run_trading_backtest(n_clicks, selected_stock, period, initial_capital_wan):
"""執行交易策略回測"""
if n_clicks == 0:
# 初始狀態
empty_fig = go.Figure()
empty_fig.update_layout(
title="點擊「開始回測」按鈕執行交易策略分析",
height=400
)
empty_fig.add_annotation(
text="等待回測開始...",
x=0.5, y=0.5,
showarrow=False,
font=dict(size=16, color="gray")
)
return (
html.Div("準備就緒，點擊按鈕開始回測", style={'color': '#3498DB', 'text-align': 'center'}),
html.Div(),
empty_fig,
html.Div()
)
try:
# 參數轉換
initial_capital = initial_capital_wan * 10000
# 檢查快取
cache_key = f"backtest_{selected_stock}_{period}_{initial_capital}"
current_time = time.time()
if cache_key in BACKTEST_CACHE:
cached_data = BACKTEST_CACHE[cache_key]
if current_time - cached_data['timestamp'] < BACKTEST_CACHE_DURATION:
results = cached_data['results']
stock_data = cached_data['stock_data']
backtester = cached_data['backtester']
progress_msg = html.Div("✅ 回測完成 (來自快取)",
style={'color': 'green', 'text-align': 'center', 'font-weight': 'bold'})
summary_card = create_backtest_summary_card(results)
chart = backtester.create_backtest_chart(results, stock_data)
details_table = create_trading_details_table(results.get('trades_df', pd.DataFrame()))
return progress_msg, summary_card, chart, details_table
# 獲取資料
stock_data = get_stock_data(selected_stock, period)
if stock_data.empty or len(stock_data) < 100:
raise ValueError("股票資料不足，無法進行回測")
# 執行回測
backtester = TradingBacktester(
initial_capital=initial_capital,
max_position_ratio=0.8,
batch_ratio=0.2
)
results = backtester.run_backtest(stock_data, get_prediction)
# 快取結果
BACKTEST_CACHE[cache_key] = {
'results': results,
'stock_data': stock_data,
'backtester': backtester,
'timestamp': current_time
}
# 生成輸出
progress_msg = html.Div(
f"✅ 回測完成！共執行 {results['total_trades']} 筆交易",
style={'color': 'green', 'text-align': 'center', 'font-weight': 'bold'}
)
summary_card = create_backtest_summary_card(results)
chart = backtester.create_backtest_chart(results, stock_data)
details_table = create_trading_details_table(results.get('trades_df', pd.DataFrame()))
return progress_msg, summary_card, chart, details_table
except Exception as e:
error_fig = go.Figure()
error_fig.add_annotation(text=f"回測執行失敗: {str(e)}", showarrow=False)
error_fig.update_layout(height=400)
return (
html.Div(f"錯誤：{str(e)}", style={'color': 'red', 'text-align': 'center'}),
html.Div("回測執行失敗，請檢查參數或稍後再試"),
error_fig,
html.Div()
)
# 5. 新增輔助函數：
def create_backtest_summary_card(results):
"""創建回測摘要卡片"""
if not results:
return html.Div("回測結果為空", style={'color': 'red'})
return_color = 'red' if results['total_return'] > 0 else 'green'
excess_color = 'red' if results['excess_return'] > 0 else 'green'
return html.Div([
html.H4("交易策略回測摘要", style={'color': '#2C3E50', 'margin-bottom': '20px'}),
# 核心指標
html.Div([
html.Div([
html.H5("總報酬率", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['total_return']:+.2f}%",
style={'margin': '5px 0', 'color': return_color, 'font-weight': 'bold'})
], style={'text-align': 'center'}),
html.Div([
html.H5("vs 買入持有", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['excess_return']:+.2f}%",
style={'margin': '5px 0', 'color': excess_color, 'font-weight': 'bold'})
], style={'text-align': 'center'}),
html.Div([
html.H5("夏普比率", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['sharpe_ratio']:.2f}",
style={'margin': '5px 0', 'color': '#3498DB', 'font-weight': 'bold'})
], style={'text-align': 'center'}),
html.Div([
html.H5("最大回撤", style={'margin': '0', 'color': '#7F8C8D'}),
html.H3(f"{results['max_drawdown']:.2f}%",
style={'margin': '5px 0', 'color': '#E74C3C', 'font-weight': 'bold'})
], style={'text-align': 'center'})
], style={'display': 'flex', 'justify-content': 'space-around', 'margin-bottom': '20px'}),
# 詳細統計表格
html.Hr(),
html.Table([
html.Tr([html.Td("最終投資組合價值"), html.Td(f"${results['final_value']:,.0f}")]),
html.Tr([html.Td("初始資金"), html.Td(f"${results.get('initial_capital', 1000000):,.0f}")]),
html.Tr([html.Td("年化波動率"), html.Td(f"{results['annual_volatility']:.2f}%")]),
html.Tr([html.Td("總交易次數"), html.Td(f"{results['total_trades']}")]),
html.Tr([html.Td("買入次數"), html.Td(f"{results['buy_trades']}")]),
html.Tr([html.Td("賣出次數"), html.Td(f"{results['sell_trades']}")]),
html.Tr([html.Td("交易勝率"), html.Td(f"{results['win_rate']:.1f}%")])
], style={'width': '100%'})
], style={
'padding': '25px',
'background': 'white',
'border-radius': '12px',
'box-shadow': '0 4px 20px rgba(0,0,0,0.08)',
'margin-bottom': '20px'
})
def create_trading_details_table(trades_df):
"""創建交易詳細記錄表格"""
if trades_df.empty:
return html.Div("尚無交易記錄", style={'text-align': 'center', 'color': 'gray'})
# 顯示最近20筆交易
recent_trades = trades_df.tail(20).copy().sort_values('date', ascending=False)
table_rows = []
for _, trade in recent_trades.iterrows():
signal_text = "🔴 買入" if trade['signal'] == 1 else "🟢 賣出"
signal_color = "#E74C3C" if trade['signal'] == 1 else "#27AE60"
table_rows.append(html.Tr([
html.Td(trade['date'].strftime('%Y-%m-%d')),
html.Td(signal_text, style={'color': signal_color, 'font-weight': 'bold'}),
html.Td(f"{abs(trade['shares']):,}"),
html.Td(f"${trade['price']:.2f}"),
html.Td(f"${abs(trade['value']):,.0f}"),
html.Td(f"{trade['strength']:.2f}")
]))
return html.Div([
html.P(f"最近 {len(table_rows)} 筆交易記錄", style={'color': '#7F8C8D'}),
html.Table([
html.Thead(html.Tr([
html.Th("交易日期", style={'background': '#34495E', 'color': 'white', 'padding': '8px'}),
html.Th("方向", style={'background': '#34495E', 'color': 'white', 'padding': '8px'}),
html.Th("股數", style={'background': '#34495E', 'color': 'white', 'padding': '8px'}),
html.Th("價格", style={'background': '#34495E', 'color': 'white', 'padding': '8px'}),
html.Th("金額", style={'background': '#34495E', 'color': 'white', 'padding': '8px'}),
html.Th("信號強度", style={'background': '#34495E', 'color': 'white', 'padding': '8px'})
])),
html.Tbody(table_rows)
], style={'width': '100%', 'border-collapse': 'collapse'})
])
# 6. 完整的TradingBacktester類別需要添加到程式開頭
# (請參考前面提供的完整類別定義)
# 主程式執行
if __name__ == '__main__':
print("=== 應用啟動 ===")
print(f"分析緩存時間: {ANALYSIS_CACHE_DURATION}秒")
print(f"股票數據緩存時間: {CACHE_DURATION_SECONDS}秒")
# 清理緩存確保乾淨啟動
clear_all_cache()
app.run(host="0.0.0.0", port=7860, debug=False)