返回目錄
A
數據驅動的投資分析：從基礎到實戰 - 第 3 章
第 3 章 數據清洗與前處理：確保分析品質
發布於 2026-02-27 08:24
# 第 3 章 數據清洗與前處理：確保分析品質
在前兩章我們已經了解了資料來源與結構，接下來要進行的是「把雜亂的原始資料轉換成乾淨、可直接投入模型的格式」。這一步驟常被稱為 **ETL**（Extract‑Transform‑Load），是投資分析流程中最關鍵也是最容易出錯的環節。資料質量不佳會直接影響模型的預測準確度，甚至導致錯誤的投資決策。以下從概念、技術、實務範例三個層面說明。
## 3.1 什麼是資料清洗？
| 步驟 | 目的 | 典型作業 |
|------|------|----------|
| 缺失值處理 | 消除資料的不完整性，避免模型崩潰 | 0 值填補、前向/後向填充、插值 |
| 重複資料剔除 | 避免統計量被重複計算，保持樣本獨立性 | `drop_duplicates()` |
| 异常值偵測 | 確保統計量不被極端值扭曲 | IQR、Z‑score、DBSCAN |
| 資料型別轉換 | 讓演算法能正確解讀 | 數值→浮點、時間→datetime |
| 時間序列同步 | 同一時間維度，避免資訊不對稱 | 重新取樣、前向填補 |
| 統一單位與尺度 | 方便因子合併與模型比較 | 調整面積、價格、比例 |
> **核心觀點**：資料清洗不是「刪除一切」，而是「保留對分析有價值、可解讀的資訊」。
## 3.2 缺失值處理
### 3.2.1 失效的原因
- 資料來源不完整（API 限制、權限不足）
- 交易停盤、公司停牌、公告缺漏
- 數據抓取過程中的錯誤或衝突
### 3.2.2 常見策略
| 方法 | 適用情境 | 優缺點 |
|------|----------|--------|
| 刪除行/列 | 缺失比例低於 5% | 可能造成樣本量減少 |
| 0 值填補 | 連續數值、指標缺失 | 可能低估波動率 |
| 前向/後向填充 | 時間序列缺失 | 可能保留舊資訊 |
| 均值/中位數填補 | 統計學方法 | 可能忽略時間趨勢 |
| 迴歸插值 | 需保留趨勢 | 計算成本較高 |
| KNN/LOF | 多變量缺失 | 需要參數調整 |
### 3.2.3 範例：Python pandas
python
import pandas as pd
# 讀取範例資料
df = pd.read_csv('stock_prices.csv', parse_dates=['date'])
# 1. 低比例缺失直接刪除
df = df.dropna(thresh=int(0.95*len(df)))
# 2. 連續日期填補（重取樣）
df.set_index('date', inplace=True)
df = df.resample('D').asfreq()
# 3. 前向填充缺失價格
df['close'] = df['close'].fillna(method='ffill')
# 4. 仍缺失的行可使用均值填補
df['volume'] = df['volume'].fillna(df['volume'].median())
> **小提醒**：在填補缺失值前，先確認該欄位為 **時間連續**，否則前向/後向填補可能產生錯誤邏輯。
## 3.3 重複資料剔除
> **情境**：多來源合併、API 重複抓取。
python
# 以 (date, ticker) 為唯一鍵
df = df.drop_duplicates(subset=['date', 'ticker'], keep='last')
### 重複檢查技巧
- **行雜湊**：計算整行哈希值，快速檢測是否全列相同。
- **子集檢查**：僅關注關鍵欄位（如 `open`, `close`, `volume`）作比較。
## 3.4 异常值偵測與處理
| 方法 | 公式 | 適用範例 |
|------|------|----------|
| IQR | Q3‑Q1 | 交易量異常大幅跳變 |
| Z‑score | (x‑μ)/σ | 價格劇烈波動 |
| DBSCAN | eps, min_samples | 連續異常點聚類 |
python
# IQR 方法
Q1 = df['close'].quantile(0.25)
Q3 = df['close'].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
# 篩除異常
df = df[(df['close'] >= lower) & (df['close'] <= upper)]
> **小技巧**：在金融資料中，某些「異常」實際上可能是事件驅動（如重大公告），不應盲目剔除。建議先加上事件標籤，再決定處理方式。
## 3.5 資料型別與單位轉換
| 欄位 | 原型別 | 轉換後型別 | 備註 |
|------|--------|------------|------|
| price | object | float64 | 轉換後可計算收益 |
| date | object | datetime64[ns] | 方便時間序列操作 |
| volume | str | int64 | 去除逗號、空格 |
| currency | object | str | 轉換成統一貨幣（USD） |
> **注意**：統一單位前務必確認來源與匯率來源。若使用多個國家資料，匯率波動可能對模型產生系統性誤差。
## 3.6 時間序列同步
### 3.6.1 重新取樣（Resampling）
| 頻率 | 描述 |
|------|------|
| `D` | 日頻 |
| `W` | 周頻 |
| `M` | 月頻 |
python
# 以日頻為例
# 先將索引設定為日期
df.set_index('date', inplace=True)
# 重新取樣，計算收盤價平均
df_daily = df['close'].resample('D').last()
### 3.6.2 前向填充（Forward Fill）
當交易所停盤或資料缺失時，使用前一個有效值填補。
python
df['close'] = df['close'].fillna(method='ffill')
> **實務提醒**：若前向填補時間跨度超過 5 個工作日，建議重新評估是否應剔除該段資料，避免帶入不實訊息。
## 3.7 統一資料格式（ETL Pipeline 範例）
python
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
# 1. Extract
def extract(**kwargs):
# 讀取 CSV / API
return pd.read_csv('raw_stock_data.csv', parse_dates=['date'])
# 2. Transform
def transform(**kwargs):
df = kwargs['ti'].xcom_pull(task_ids='extract')
# ① 缺失值處理
df = df.dropna(subset=['close', 'volume'])
# ② 重複剔除
df = df.drop_duplicates(subset=['date', 'ticker'], keep='last')
# ③ 异常值檢測
Q1, Q3 = df['close'].quantile([0.25, 0.75])
IQR = Q3 - Q1
df = df[(df['close'] >= Q1-1.5*IQR) & (df['close'] <= Q3+1.5*IQR)]
# ④ 資料型別轉換
df['close'] = df['close'].astype(float)
df['volume'] = df['volume'].astype(int)
df.set_index('date', inplace=True)
df = df.resample('D').last()
df['close'] = df['close'].fillna(method='ffill')
df.reset_index(inplace=True)
return df
# 3. Load
def load(**kwargs):
df = kwargs['ti'].xcom_pull(task_ids='transform')
df.to_sql('clean_stock_prices', con=engine, if_exists='replace', index=False)
# DAG 定義
with DAG(
dag_id='stock_etl',
start_date=datetime(2023, 1, 1),
schedule_interval='@daily',
catchup=False
) as dag:
t_extract = PythonOperator(task_id='extract', python_callable=extract, provide_context=True)
t_transform = PythonOperator(task_id='transform', python_callable=transform, provide_context=True)
t_load = PythonOperator(task_id='load', python_callable=load, provide_context=True)
t_extract >> t_transform >> t_load
> **結語**：資料清洗不是一次性工作，而是資料循環更新、模型再訓練時需要持續重複的作業。建議將清洗流程封裝成自動化 Pipeline，確保每次載入資料都能遵循同樣的規範。
## 3.8 小結
1. **缺失值**：選擇合適填補方法，並根據缺失比例決定刪除或保留。
2. **重複資料**：以關鍵鍵值（日期＋ticker）為基準，避免重複計算。
3. **异常值**：先加上事件標籤，慎重剔除。
4. **型別 & 單位**：統一型別後可直接投入模型，避免演算法崩潰。
5. **時間同步**：重新取樣 + 前向填補，確保每個觀測都有對應的時間戳。
完成這些步驟後，我們得到的「乾淨資料」即可直接輸入回歸、時間序列預測或機器學習模型。下章將把這份資料帶入 **因子構建** 與 **模型訓練**，繼續深化投資策略。