返回目錄
A
自由數據：用資料科學解鎖個人財務自由 - 第 2 章
第二章：財務數據的收集與清理
發布於 2026-02-24 23:59
# 第二章：財務數據的收集與清理
> 在量化投資與個人財務管理中，**資料的質量往往是決定成功與否的關鍵**。本章將從三個層面帶領讀者完成財務數據的收集、清理與儲存，為後續的特徵工程與模型構建奠定堅實基礎。
## 2.1 金融市場數據來源
| 類型 | 主要資料 | 代表性 API | 特色說明 |
|------|----------|-----------|-----------|
| 股票 | 交易價格、成交量、公司公告 | `yfinance`, `Alpha Vantage`, `Polygon.io` | 免費與付費結合，覆蓋全球主要交易所 |
| 匯率 | 即時匯率、歷史匯率 | `exchangerate.host`, `OANDA` | 低延遲、精確對應交易日 |
| 基金 | NAV、基金淨值、分紅 | `Morningstar`, `ETF.com` | 基金特定資訊、分類資料 |
| 債券 | 票息、到期日、信用評級 | `Bloomberg`, `FactSet` | 需付費，資訊深度高 |
| 行情指標 | MACD、RSI、布林帶 | 自行計算 | 基於原始價格資料 |
> **選擇資料來源時需考慮：**
>
> - **可獲取性**：免費 vs 付費、API 金鑰限制。
> - **資料完整度**：歷史長度、更新頻率。
> - **資料品質**：精度、缺失值、日誌一致性。
## 2.2 API 連結實務
### 2.2.1 使用 `yfinance` 下載股票歷史價格
python
import yfinance as yf
import pandas as pd
# 下載台積電（TSE:2330）過去 2 年的日行情
ticker = "2330.TW"
data = yf.download(ticker, period="2y", interval="1d")
# 基本資料檢查
print(data.head())
> `yfinance` 直接從 Yahoo Finance 抓取資料，使用方便但需要注意：
> - **資料延遲**：Yahoo 的資料大約 5 分鐘延遲。
> - **資料完整度**：周末與節假日缺失；若需完整日曆，可自行補齊。
### 2.2.2 Alpha Vantage 取得 5 分鐘 K 線
python
import requests
import json
import pandas as pd
API_KEY = "YOUR_ALPHA_VANTAGE_KEY"
symbol = "AAPL"
function = "TIME_SERIES_INTRADAY"
interval = "5min"
url = f"https://www.alphavantage.co/query?function={function}&symbol={symbol}&interval={interval}&apikey={API_KEY}&outputsize=compact"
res = requests.get(url)
raw = json.loads(res.text)
# 轉成 DataFrame
key = f"Time Series ({interval})"
df = pd.DataFrame(raw[key]).T
# 列名轉為小寫
cols = {c.split(' ')[-1]: c for c in df.columns}
df.rename(columns=cols, inplace=True)
# 資料型態轉換
for col in df.columns:
df[col] = pd.to_numeric(df[col])
print(df.head())
> Alpha Vantage 提供 **免費配額**（每分鐘 5 次請求），對於小型投資者或初學者足夠。
### 2.2.3 匯率資料抓取（exchangerate.host）
python
import requests
import pandas as pd
base = "USD"
symbols = "TWD,EUR"
url = f"https://api.exchangerate.host/timeseries?start_date=2023-01-01&end_date=2023-12-31&base={base}&symbols={symbols}"
res = requests.get(url)
raw = res.json()
rates = raw["rates"]
# 轉成 DataFrame
df = pd.DataFrame(rates).T
print(df.head())
> `exchangerate.host` 為免費且開源的匯率 API，沒有金鑰限制，適合日常使用。
## 2.3 資料庫操作
### 2.3.1 選擇資料庫
| 資料庫 | 優點 | 缺點 |
|--------|------|------|
| SQLite | 嵌入式、輕量、易於部署 | 不支援併發、容量有限 |
| PostgreSQL | 開源、擴展性好、複雜查詢 | 需要伺服器、配置複雜 |
| MySQL | 廣泛使用、易於學習 | 某些高階功能需付費版 |
| MongoDB | 文件導向、彈性結構 | 不適合大量關聯型資料 |
> **建議**：在個人財務分析中，SQLite 足以滿足日常需求；若需要多使用者協作或更高併發，可考慮 PostgreSQL。
### 2.3.2 建立資料表結構（以 SQLite 為例）
sql
CREATE TABLE IF NOT EXISTS stock_prices (
symbol TEXT NOT NULL,
date TEXT NOT NULL,
open REAL,
high REAL,
low REAL,
close REAL,
volume INTEGER,
PRIMARY KEY (symbol, date)
);
CREATE TABLE IF NOT EXISTS forex_rates (
base TEXT NOT NULL,
symbol TEXT NOT NULL,
date TEXT NOT NULL,
rate REAL,
PRIMARY KEY (base, symbol, date)
);
### 2.3.3 將 `pandas` DataFrame 直接寫入 SQLite
python
import sqlite3
conn = sqlite3.connect("financial_data.db")
# 假設 df 為股票歷史價格 DataFrame
# columns: ['Open','High','Low','Close','Volume']
# 添加 symbol 與 date
symbol = "2330.TW"
# 日期列必須為字串
df['date'] = df.index.strftime("%Y-%m-%d")
# 重新排序列
cols = ['symbol', 'date', 'Open', 'High', 'Low', 'Close', 'Volume']
stock_df = df[cols].rename(columns={
'Open': 'open',
'High': 'high',
'Low': 'low',
'Close': 'close',
'Volume': 'volume'
})
stock_df.insert(0, 'symbol', symbol)
# 寫入資料庫
stock_df.to_sql("stock_prices", conn, if_exists="append", index=False)
> **備註**：在寫入前可先使用 `df.drop_duplicates(subset=['symbol','date'])` 以避免重複資料。
## 2.4 資料清理流程
1. **缺失值處理**
- `NaN` → 直接丟棄、前向/後向填補或插值（`df.interpolate()`）。
- 對於金融資料，缺失通常代表交易所休市，直接丟棄即可。
2. **重複值處理**
python
df.drop_duplicates(subset=['symbol','date'], keep='first', inplace=True)
3. **資料型態轉換**
- 日期轉為 `datetime`：`pd.to_datetime(df['date'])`。
- 數值轉為 `float`：`df['close'] = df['close'].astype(float)`。
4. **貨幣統一**
- 若同時處理多幣種，需先取得匯率，進行兌換。示例：將 TWD 兌換為 USD。
python
df['close_usd'] = df['close'] * df['twd_to_usd_rate']
5. **時間對齊**
- **補齊日曆**：
python
all_dates = pd.date_range(start=df['date'].min(), end=df['date'].max(), freq='D')
df = df.set_index('date').reindex(all_dates).reset_index()
df.rename(columns={'index':'date'}, inplace=True)
- **去除節假日**：可使用 `holidays` 套件判斷是否為交易日。
6. **異常值檢測**
- **價格跳變**：`df['pct_change'] = df['close'].pct_change()`。
- 若跳變幅度超過 10%（或自訂門檻），可標記為 **潛在異常**。
### 2.4.1 清理範例：台積電日行情
python
import yfinance as yf
import pandas as pd
symbol = "2330.TW"
raw = yf.download(symbol, period="2y", interval="1d")
# 基本清理
raw.dropna(inplace=True) # 台股週末已缺失
raw.drop_duplicates(subset=['Date'], keep='first', inplace=True)
raw.reset_index(inplace=True)
raw['date'] = pd.to_datetime(raw['Date']).dt.date
# 列名統一
raw.rename(columns={
'Open':'open', 'High':'high', 'Low':'low',
'Close':'close', 'Volume':'volume'
}, inplace=True)
raw = raw[['symbol', 'date', 'open', 'high', 'low', 'close', 'volume']]
raw.insert(0, 'symbol', symbol)
# 檢查是否有負數量化
print(raw.describe())
## 2.5 實戰小結
| 步驟 | 目的 | 常見陷阱 |
|------|------|-----------|
| 1. 設定 API 限額與金鑰管理 | 避免因超額而被封鎖 | 忘記檢查 `requests` 狀態碼、JSON 結構變動 |
| 2. 將資料寫入 SQLite 時保持 **主鍵唯一** | 防止重複 | 沒設定主鍵時會產生多筆相同資料 |
| 3. 資料型態統一 | 方便後續計算 | 字串數值混用會導致 `astype` 失敗 |
| 4. 缺失值只在 **非交易日** 允許 | 不影響趨勢 | 低頻資料若出現 `NaN` 代表資料錯誤，需進一步檢查 |
| 5. 匯率兌換時使用「當日」匯率 | 使不同幣種可比 | 若使用「前一天」匯率，將產生 **時間失真** |
> **實務建議**：將清理邏輯打包成 **函式庫**（例如 `clean_stock_prices(df)`），並在 ETL 腳本中重複使用，能顯著提升可維護性與可重現性。
## 2.6 章節小結
本章介紹了金融市場數據的主要來源、各種 API 的使用範例、資料庫的選型與操作，以及一套完整的資料清理流程。以下是關鍵 Take‑away：
1. **資料來源要先確定**：選擇能滿足頻率、歷史長度與品質的 API。
2. **資料寫入前先做好資料型態與重複值處理**，避免在資料庫層面出現難以調查的錯誤。
3. **清理工作是持續迴圈**，隨著資料源更新、API 版本變動，需要定期重跑 ETL 流程。
4. **把資料庫設計當成「靜態」結構**，配合版本控制 (`sql` 腳本) 能確保分析環境的可追蹤性。
> 接下來的章節將進一步探討如何利用已清理好的財務資料，結合技術指標、基本面數據與外部宏觀訊息，進行 **特徵工程** 與 **機器學習模型** 的開發。祝你在資料收集與清理的道路上一路順風！