專案網址

深入解析:使用Python構建全面的市場風險監測工具

七月全球同步發生多個利空事件，市場下殺速度快到根本來不及反應，直接被套牢。事後想想，如果有個工具能在風險累積時提早示警，也許不會這麼狼狽。於是花了幾個週末做了這個 VolatilityVigil——用 Python 同時監測多個市場的風險狀態，從波動度計算到 PCA 異常偵測，最後用 K-means 把市場分群，輸出一張能看懂的熱力圖。

1. 程式概述

核心功能分五塊：

透過 yfinance 抓 Yahoo Finance 多市場歷史資料（美股、歐股、亞股、商品、外匯）
計算各市場的年化波動度
用 PCA 重建誤差做異常偵測
整合波動度與異常訊號算出風險評分
用 K-means 把市場狀態分群，並將群號整合進最終判讀輸出

用到的套件：pandas（資料處理）、sklearn（演算法）、yfinance（市場資料）、matplotlib 與 seaborn（視覺化）。

2. 核心功能詳解

2.1 資料獲取

先把整份程式會用到的 import 一次列齊，免得你抄到一半發現少了哪個套件：

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
import yfinance as yf
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns

# 圖表中文不亂碼（Windows 用微軟正黑體，macOS 可換 'PingFang TC'）
plt.rcParams['font.sans-serif'] = ['Microsoft JhengHei']
plt.rcParams['axes.unicode_minus'] = False

監測標的涵蓋美國、歐洲、亞洲主要股市指數，加上商品期貨與外匯市場：

symbols = [
    '^GSPC', '^IXIC', '^DJI',          # 美股：標普 500、納斯達克、道瓊
    '^FTSE', '^GDAXI', '^FCHI',        # 歐股：英國富時、德國 DAX、法國 CAC
    '^N225', '000001.SS', '^HSI', '^TWII',  # 亞股：日經、上證、恆生、台灣加權
    'GC=F', 'CL=F',                    # 商品：黃金、原油期貨
    'EURUSD=X', 'JPYUSD=X', 'GBPUSD=X' # 外匯：歐元、日元、英鎊兌美元
]

接著用 yfinance 抓這些標的過去 90 天的收盤價：

def get_date_range():
    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=90)
    return start_date, end_date

start_date, end_date = get_date_range()
market_data = yf.download(symbols, start=start_date, end=end_date)['Close']

90 天是一個設計選擇：短到能反映近期市場結構，又長到在波動度計算上不至於被單一事件過度主導。如果監測目標是 flash crash 這類極短暫的異動，視窗可以縮短到 30～60 天；若要捕捉季節性循環，則需要拉長。

2.2 風險評估

風險評分由兩個訊號合成：波動度超標 + PCA 異常偵測。

波動度計算：用 20 日滾動標準差乘以 √252 換算成年化數字，把不同長度的取樣窗放到同一個年化基準上比較：

1 2	def calculate_volatility(data, window=20): return data.pct_change().rolling(window=window).std() * np.sqrt(252)

PCA 異常偵測：先用 StandardScaler 標準化，讓不同市場的數值落在相同尺度，再用 PCA 壓縮到保留 95% 變異的維度，把資料反向重建回標準化後的空間（注意是標準化空間、不是原始價格尺度），計算均方重建誤差（MSE）。MSE 高代表該日的市場結構無法被主成分良好重建，通常對應到市場走勢出現不尋常的配置——但這不等同於「MSE 高就是異常」：若多市場同步大跌，PCA 反而能用少數主成分解釋絕大部分的共同走勢，重建誤差可能偏低，訊號會靜音。這個方法對「結構性分歧」（部分市場脫鉤）更敏感，對「全球同步崩跌」的偵測力相對有限。門檻取 threshold 倍的 MSE 標準差，跟著資料的離散程度浮動，比寫死一個固定值不容易誤判：

def detect_anomalies(data, threshold=3):
    if data.empty:
        return pd.Series(dtype=bool)
    data_filled = data.ffill().bfill()
    scaler = StandardScaler()
    normalized_data = scaler.fit_transform(data_filled)
    pca = PCA(n_components=0.95)
    pca_result = pca.fit_transform(normalized_data)
    reconstruction = pca.inverse_transform(pca_result)
    mse = np.mean(np.power(normalized_data - reconstruction, 2), axis=1)
    return pd.Series(mse > threshold * np.std(mse), index=data.index)

這裡用 .ffill().bfill() 而不是舊的 fillna(method='ffill')——後者已經被棄用，新版 pandas 會發 FutureWarning 要你改用 .ffill() / .bfill()，趁早換掉。

風險評分整合：異常偵測只跑一次，吃整張多市場資料表，讓 PCA 在所有市場之間找結構性分歧；得到的是「某一天市場結構是否異常」的單一布林訊號。波動度超標則是逐市場各自判斷。兩個訊號相加，得到 0 / 1 / 2 三級評分：

def calculate_risk_score(market_data):
    volatility = calculate_volatility(market_data)
    anomalies = detect_anomalies(market_data)

    common_index = volatility.index.intersection(anomalies.index)
    volatility = volatility.loc[common_index]
    anomalies = anomalies.loc[common_index]

    risk_score = pd.DataFrame(index=common_index)
    for column in market_data.columns:
        vol_threshold = volatility[column].mean() + 2 * volatility[column].std()
        risk_score[column] = anomalies.astype(int) + (volatility[column] > vol_threshold).astype(int)

    return risk_score

anomalies 是跨市場的共同訊號，所以每一欄的異常分量都來自同一個 PCA 判讀——這是刻意的設計。把資料切成單欄各跑一次 PCA 反而會壞掉：單一序列標準化後幾乎能被一個主成分完整重建，重建誤差趨近於零，異常永遠不會觸發。PCA 異常偵測的價值正是在多市場之間找「誰脫鉤了」，餵單欄等於把這個能力關掉。

common_index 那兩行是防禦性寫法，不是必要的對齊：波動度前 20 天是 NaN（注意是 NaN 值、不是缺列，索引仍在），而 detect_anomalies 內部用 .ffill().bfill() 也保留了完整索引，兩邊索引在這份資料上本來就一致，intersection 取完等於沒變。留著它是為了萬一日後改動讓某一邊真的丟了列時不會錯位。另外要記得：前 20 天波動度為 NaN，(NaN > 門檻) 會被判成 False，所以這段期間的評分實際只反映異常分量、不含波動度分量。

2.3 市場狀態聚類

K-means 把每天的多市場風險評分向量分成 5 群，目的是讓工具能辨識不同的市場結構——例如「全球同步緊縮」「單市場異動」「正常低波動」。K-means 不接受 NaN，所以先前向再後向填補缺值：

def cluster_market_states(data, n_clusters=5):
    data_filled = data.ffill().bfill()
    kmeans = KMeans(n_clusters=n_clusters, n_init=10)
    cluster_labels = pd.Series(kmeans.fit_predict(data_filled), index=data.index)
    return cluster_labels

n_init=10 是明確指定的：sklearn 1.4 把預設值從 10 改成 'auto'，而 'auto' 在 init='k-means++'（預設）下等效為 1 次初始化，比舊版少跑 9 次。顯式寫 n_init=10 是讓行為與 1.4 以前的舊版一致，確保跨版本結果穩定。

分群結果會在第 4 節的 explain_risk_score() 一起輸出，讓你同時看到當天的評分等級與所屬的市場狀態群。

3. 視覺化

四種圖表各有用途，程式碼如下：

風險評分熱力圖，一眼看出哪個市場、哪段時間風險集中：

def visualize_risk_scores(risk_score):
    plt.figure(figsize=(12, 6))
    sns.heatmap(risk_score.T, cmap='YlOrRd', cbar_kws={'label': '風險評分'})
    plt.title('各市場風險評分熱力圖')
    plt.xlabel('日期')
    plt.ylabel('市場')
    plt.tight_layout()
    plt.show()

市場趨勢圖，疊加各市場收盤價走勢，方便對照風險訊號與價格變動的時間差：

def visualize_market_trends(market_data):
    plt.figure(figsize=(14, 6))
    for column in market_data.columns:
        normalized = market_data[column] / market_data[column].iloc[0] * 100
        plt.plot(normalized.index, normalized.values, label=column, alpha=0.7)
    plt.title('各市場標準化趨勢（基期=100）')
    plt.xlabel('日期')
    plt.ylabel('相對指數')
    plt.legend(loc='upper left', fontsize=7, ncol=2)
    plt.tight_layout()
    plt.show()

市場相關性矩陣，用相關係數顯示各市場的同步性，高度相關的市場在壓力測試時很難分散風險：

def visualize_correlation_matrix(market_data):
    corr = market_data.pct_change().corr()
    plt.figure(figsize=(10, 8))
    sns.heatmap(corr, annot=True, fmt='.2f', cmap='coolwarm',
                vmin=-1, vmax=1, center=0,
                cbar_kws={'label': '相關係數'})
    plt.title('市場相關性矩陣（日報酬率）')
    plt.tight_layout()
    plt.show()

總體風險評分分佈圖，把各市場的每日風險評分加總，畫出時序分佈，快速抓出系統性壓力最高的時間點：

def visualize_overall_risk(risk_score):
    overall = risk_score.sum(axis=1)
    plt.figure(figsize=(12, 4))
    plt.fill_between(overall.index, overall.values, alpha=0.4, color='tomato')
    plt.plot(overall.index, overall.values, color='firebrick', linewidth=1.2)
    plt.title('總體市場風險評分（各市場加總）')
    plt.xlabel('日期')
    plt.ylabel('風險評分總和')
    plt.tight_layout()
    plt.show()

4. 結果解釋

explain_risk_score() 接收風險評分與聚類結果，同時輸出評分描述和當天所屬的市場狀態群號。這樣就能回答「今天評分高，但這是哪一類型的高風險？」——全球同步飆升（群 0）和單一市場孤立異動（群 3）的處理策略不同：

def explain_risk_score(risk_score, cluster_labels=None):
    explanations = {
        0: "低風險：兩個訊號都沒觸發（PCA 結構正常、波動度未超門檻）。",
        1: "中等風險：兩個訊號觸發其中一個（市場結構異常，或波動度超過門檻）。建議密切追蹤。",
        2: "高風險：兩個訊號同時觸發（結構異常且波動度超門檻）。建議檢視部位並考慮風險管理措施。"
    }

    latest = risk_score.iloc[-1]
    high_risk_markets = latest[latest == 2].index.tolist()

    print("=== 最新市場風險狀態 ===")
    for market, score in latest.items():
        print(f"{market}: {explanations[score]}")

    if high_risk_markets:
        print(f"\n⚠ 高風險市場警示：{', '.join(high_risk_markets)}")
    else:
        print("\n目前無高風險市場。")

    if cluster_labels is not None:
        today_cluster = cluster_labels.iloc[-1]
        print(f"\n當前市場狀態群：Cluster {today_cluster}")
        print("（可對照歷史群分佈，判斷此狀態在過去 90 天屬於哪類市場結構）")

呼叫方式：

1
2
3

risk_score = calculate_risk_score(market_data)
cluster_labels = cluster_market_states(risk_score)
explain_risk_score(risk_score, cluster_labels=cluster_labels)

5. 使用注意事項

資料時效性：若最新交易日距今超過數天，通常是遇到連假或 Yahoo Finance API 暫時異常，建議手動確認資料的最新日期再解讀輸出結果。
缺值處理：前向填補優先（用前一個交易日的收盤價），再後向填補。程式不會印出缺值統計，若需確認哪些欄位有缺值，請自行對 market_data 執行 market_data.isna().sum()。
市場休市：不同市場的交易日曆不同，例如日本市場黃金週連休，這段期間的訊號應降低權重解讀。
yfinance 抓取可靠性：yf.download 多標的時，部分指數（如 ^TWII、000001.SS）或匯率代碼偶爾抓取失敗，會靜默掉欄或整欄回傳 NaN。執行後建議用 print(market_data.shape, market_data.columns.tolist()) 確認欄數與欄名是否如預期，再繼續後續計算。
PCA 異常偵測的方法論限制：detect_anomalies 是對「標準化後的收盤價 level」做 PCA 重建，而波動度計算用的是報酬率。收盤價 level 本身非定態，長期趨勢會影響 PCA 重建誤差的基線，因此異常訊號偏向捕捉結構性分歧（各市場走勢脫鉤），較不適合作為精準的進出場警報，應視為輔助判讀的參考訊號。
KMeans 收斂警告：cluster_market_states 預設 n_clusters=5。若 90 天視窗內 risk_score 多數為 0（市場長期低波動），相異樣本點可能少於 5 個，KMeans 會拋出 ConvergenceWarning，屬正常現象，可將 n_clusters 調小（如 3）或直接忽略此警告。