建立量化交易儀表板
在本課程的最後一章，我們要整合所學的全部知識，建立一個完整的量化交易儀表板 (Dashboard)。這個儀表板可以：
- 監控即時股價與技術指標
- 顯示最新的交易信號
- 回測報告與績效指標
- 投資組合風險分析
使用 Plotly 建立互動式儀表板
Plotly 是一個可以建立互動式圖表的 Python 函式庫。與 Matplotlib 不同，Plotly 的圖表可以縮放、平移、懸停查看數據。
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import yfinance as yf
import pandas as pd
import numpy as np
# 下載資料
df = yf.download("2330.TW", start="2023-01-01", end="2024-12-31")
# 計算技術指標
df['MA10'] = df['Close'].rolling(10).mean()
df['MA30'] = df['Close'].rolling(30).mean()
# 建立子圖
fig = make_subplots(
rows=3, cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
row_heights=[0.5, 0.25, 0.25],
subplot_titles=('股價與移動平均線', 'RSI', '成交量')
)
# 上圖：K 線圖 + 移動平均線
fig.add_trace(go.Candlestick(
x=df.index,
open=df['Open'],
high=df['High'],
low=df['Low'],
close=df['Close'],
name='K 線'
), row=1, col=1)
fig.add_trace(go.Scatter(
x=df.index, y=df['MA10'],
line=dict(color='orange', width=1),
name='MA10'
), row=1, col=1)
fig.add_trace(go.Scatter(
x=df.index, y=df['MA30'],
line=dict(color='purple', width=1),
name='MA30'
), row=1, col=1)
# 中圖：RSI
# 計算 RSI
from ta.momentum import RSIIndicator
rsi = RSIIndicator(close=df['Close'], window=14).rsi()
fig.add_trace(go.Scatter(
x=df.index, y=rsi,
line=dict(color='purple', width=1),
name='RSI (14)'
), row=2, col=1)
# RSI 超買超賣線
fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
# 下圖：成交量
fig.add_trace(go.Bar(
x=df.index, y=df['Volume'],
name='成交量',
marker_color='lightblue'
), row=3, col=1)
# 更新版面
fig.update_layout(
title='台積電 (2330) 技術分析儀表板',
xaxis_title='日期',
yaxis_title='價格',
template='plotly_dark',
height=900,
showlegend=True,
xaxis_rangeslider_visible=False
)
fig.show()
建立交易信號儀表板
# === 信號產生器 ===
def generate_signals(df):
"""
根據技術指標產生綜合交易信號
回傳：-1 = 賣出, 0 = 中立, 1 = 買進
"""
signals = pd.DataFrame(index=df.index)
# 1. 移動平均線信號
ma10 = df['Close'].rolling(10).mean()
ma30 = df['Close'].rolling(30).mean()
signals['MA_Signal'] = 0
signals.loc[ma10 > ma30, 'MA_Signal'] = 1
signals.loc[ma10 < ma30, 'MA_Signal'] = -1
# 2. RSI 信號
rsi = RSIIndicator(close=df['Close'], window=14).rsi()
signals['RSI_Signal'] = 0
signals.loc[rsi < 30, 'RSI_Signal'] = 1 # 超賣 → 買進
signals.loc[rsi > 70, 'RSI_Signal'] = -1 # 超買 → 賣出
# 3. 綜合信號
signals['Overall'] = (
signals['MA_Signal'] + signals['RSI_Signal']
)
return signals
signals = generate_signals(df)
# 顯示最新信號
latest = signals.iloc[-1]
print("=== 最新交易信號 ===")
print(f"移動平均線信號: {'買進' if latest['MA_Signal'] == 1 else '賣出' if latest['MA_Signal'] == -1 else '中立'}")
print(f"RSI 信號: {'買進' if latest['RSI_Signal'] == 1 else '賣出' if latest['RSI_Signal'] == -1 else '中立'}")
print(f"綜合信號: {'強烈買進 🔥' if latest['Overall'] >= 2 else '買進 📈' if latest['Overall'] == 1 else '中立 ➖' if latest['Overall'] == 0 else '賣出 📉' if latest['Overall'] == -1 else '強烈賣出 🚨'}")
建立回測報告
# === 回測報告產生器 ===
def generate_backtest_report(results):
"""從 Backtrader 結果產生報告"""
strategy = results[0]
# 取得分析數據
sharpe = strategy.analyzers.sharpe.get_analysis()
drawdown = strategy.analyzers.drawdown.get_analysis()
returns = strategy.analyzers.returns.get_analysis()
trades = strategy.analyzers.trades.get_analysis()
report = {
'績效指標': {
'總報酬率': f"{returns.get('rtot100', 0):.2f}%",
'年化報酬率': f"{returns.get('rnorm100', 0):.2f}%",
'夏普比率': f"{sharpe.get('sharperatio', 'N/A'):.2f}" if sharpe.get('sharperatio') else 'N/A',
'最大回撤': f"{drawdown['max']['drawdown']:.2f}%",
'最大回撤期間': f"{drawdown['max']['len']} 天",
},
'交易統計': {}
}
if 'total' in trades:
report['交易統計'] = {
'總交易次數': trades['total']['total'],
'獲利交易': trades['won']['total'],
'虧損交易': trades['lost']['total'],
'勝率': f"{trades['won']['total'] / trades['total']['total'] * 100:.1f}%",
'平均獲利': f"${trades['won']['pnl']['average']:.2f}" if trades['won']['total'] > 0 else 'N/A',
'平均虧損': f"${trades['lost']['pnl']['average']:.2f}" if trades['lost']['total'] > 0 else 'N/A',
}
return report
# 使用方式
report = generate_backtest_report(results)
for category, metrics in report.items():
print(f"\n=== {category} ===")
for key, value in metrics.items():
print(f"{key}: {value}")
使用 FastAPI 建立量化交易 API
from fastapi import FastAPI, Query
from pydantic import BaseModel
import yfinance as yf
import pandas as pd
app = FastAPI(title="量化交易 API")
class StockAnalysis(BaseModel):
ticker: str
current_price: float
ma10: float
ma30: float
rsi: float
signal: str
recommendation: str
@app.get("/analyze/{ticker}", response_model=StockAnalysis)
def analyze_stock(ticker: str):
"""分析單一股票並產生交易建議"""
data = yf.download(ticker, period="6mo")
close = data['Close']
ma10 = close.rolling(10).mean().iloc[-1]
ma30 = close.rolling(30).mean().iloc[-1]
from ta.momentum import RSIIndicator
rsi = RSIIndicator(close=close, window=14).rsi().iloc[-1]
# 決定信號
if ma10 > ma30 and rsi < 70:
signal = "bullish"
recommendation = "買進 📈"
elif ma10 < ma30 and rsi > 30:
signal = "bearish"
recommendation = "賣出 📉"
else:
signal = "neutral"
recommendation = "觀望 ⏸️"
return StockAnalysis(
ticker=ticker,
current_price=round(float(close.iloc[-1]), 2),
ma10=round(float(ma10), 2),
ma30=round(float(ma30), 2),
rsi=round(float(rsi), 2),
signal=signal,
recommendation=recommendation
)
@app.get("/watchlist")
def scan_watchlist(tickers: str = Query("2330.TW,2317.TW,TSLA,AAPL")):
"""掃描多檔股票"""
ticker_list = tickers.split(",")
results = []
for ticker in ticker_list:
try:
result = analyze_stock(ticker.strip())
results.append(result.dict())
except:
results.append({"ticker": ticker, "error": "讀取失敗"})
return {"results": results}
使用 Vibe Coding 建立儀表板
🔥 【量化儀表板詠唱範例】
「請幫我建立一個 Streamlit 量化交易儀表板：1. 使用者可以輸入股票代號（支援台股美股）。2. 顯示 K 線圖、MA10/MA30、RSI、MACD、成交量。3. 顯示最新的綜合交易信號（買進／賣出／中立）。4. 顯示一個回測報告表格（總報酬、夏普比率、最大回撤）。5. 支援多股比較功能。6. 使用深色主題，專業的金融風格。」
本日總結
在本章中，你學到了：
- ✅ Plotly 互動式圖表：建立專業級的技術分析圖表
- ✅ 多信號綜合判斷：整合 MA 與 RSI 產生買賣信號
- ✅ 回測報告產生器：自動化生成績效報告
- ✅ FastAPI 量化 API：將分析能力包裝成 API
- ✅ 完整的儀表板架構：從資料到信號到報告的完整流程
恭喜你完成了整個 量化交易回測系統 課程！
你現在已經具備了：
- 📊 用程式抓取與分析股票資料的能力
- 📈 計算與解讀技術指標的能力
- 🔄 建立回測系統驗證策略的能力
- 🛡️ 風險管理與投資組合最佳化的能力
- 🚀 將量化交易系統部署為 API 的能力
這些技能無論是拿來打造自己的被動收入系統，或是作為量化交易工程師求職／接案，都能讓你擁有巨大的競爭優勢！