返回目錄
A
金融資料科學：從數據到決策的完整流程 - 第 3 章
第 3 章：資料清理與前處理
發布於 2026-03-07 11:47
# 第 3 章：資料清理與前處理
在金融資料科學的實務流程中，資料清理（Data Cleaning）與前處理（Feature Engineering）是連接原始資料與模型輸入的關鍵橋樑。缺失值、重複紀錄、異常數值以及資料類型不一致等問題，若不妥善處理，往往會直接導致模型性能急劇下降，甚至產生誤導性的投資決策。
以下以 **台股日成交資料** 為例，逐步拆解常見的清理與前處理步驟，並提供完整的 Python 程式碼範例與 Airflow 任務示範。
---
## 3.1 欄位缺失與重複資料處理
| 步驟 | 目的 | 常見做法 |
|------|------|----------|
| `detect_missing` | 檢測缺失比例 | `df.isna().sum()` |
| `handle_missing` | 逐欄決策填補 | `fillna`、`interpolate`、`dropna` |
| `detect_duplicates` | 发现重复记录 | `df.duplicated()` |
| `remove_duplicates` | 只保留唯一值 | `df.drop_duplicates()` |
### 程式碼範例
```python
import pandas as pd
def clean_basic(df: pd.DataFrame) -> pd.DataFrame:
"""處理缺失值與重複紀錄。"""
# 1. 欄位缺失檢測
missing = df.isna().sum()
print('Missing values per column:\n', missing)
# 2. 欄位缺失處理（根據業務決策）
# 以成交量為例，若缺失小於 5% 可直接 drop
if missing['Volume'] / len(df) < 0.05:
df = df.dropna(subset=['Volume'])
else:
df['Volume'] = df['Volume'].interpolate(method='linear')
# 3. 重複紀錄處理
dup_count = df.duplicated().sum()
print(f'Duplicate rows detected: {dup_count}')
df = df.drop_duplicates()
return df
```
> **實務建議**：缺失值的處理必須配合業務知識。若缺失屬於極端情況（如公司停牌），直接 drop 可能更安全；若缺失是隨機機率，則使用插補或預測模型補齊。
---
## 3.2 日期時間轉換與時區校正
金融市場資料往往只提供「日期」欄位，而實際分析需要完整的時間戳（timestamp）。若不統一時區，跨市場比較或高頻交易分析將出錯。
```python
# 假設 df['Date'] 為 yyyy-mm-dd
import pytz
from datetime import datetime
def normalize_datetime(df: pd.DataFrame) -> pd.DataFrame:
df['Datetime'] = pd.to_datetime(df['Date'])
# 台灣時區：Asia/Taipei
tz = pytz.timezone('Asia/Taipei')
df['Datetime'] = df['Datetime'].dt.tz_localize(tz)
return df
```
---
## 3.3 數值與類別變數編碼
| 類型 | 編碼方式 | 典型範例 |
|------|----------|----------|
| 數值 | 標準化/正規化 | `StandardScaler`, `MinMaxScaler` |
| 類別 | One-Hot, Target Encoding | `pd.get_dummies`, `category_encoders` |
### 代碼範例
```python
from sklearn.preprocessing import StandardScaler
import category_encoders as ce
def encode_features(df: pd.DataFrame) -> pd.DataFrame:
# 數值欄位標準化
num_cols = ['Close', 'Open', 'High', 'Low']
scaler = StandardScaler()
df[num_cols] = scaler.fit_transform(df[num_cols])
# 類別欄位 One-Hot
cat_cols = ['Market']
df = pd.get_dummies(df, columns=cat_cols, drop_first=True)
return df
```
> **實務建議**：若模型支援稀疏矩陣（如 Logistic Regression），可直接使用 One-Hot；若資料量極大，考慮使用 Target Encoding 或 Embedding 以降低維度。
---
## 3.4 異常值與離群值檢測
金融資料的異常值（Outlier）往往是重要訊號，但若不加控制，會干擾模型學習。常見方法有
* IQR（四分位距）檢測
* Z-Score
* 盒鬚圖（Boxplot）視覺化
* DBSCAN、Isolation Forest
### 範例：IQR
```python
def remove_outliers(df: pd.DataFrame, column: str, factor: float = 1.5) -> pd.DataFrame:
Q1 = df[column].quantile(0.25)
Q3 = df[column].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - factor * IQR
upper = Q3 + factor * IQR
mask = (df[column] >= lower) & (df[column] <= upper)
print(f'Removed {mask.shape[0] - mask.sum()} outliers from {column}')
return df[mask]
```
---
## 3.5 時間序列特徵創造
在金融領域，時間序列特徵極為重要：移動平均、布林帶、RSI、MACD 等技術指標可直接用於模型。
```python
import ta # Technical Analysis Library
def add_time_series_features(df: pd.DataFrame) -> pd.DataFrame:
df['MA_20'] = ta.trend.sma_indicator(df['Close'], window=20)
df['MA_50'] = ta.trend.sma_indicator(df['Close'], window=50)
df['RSI_14'] = ta.momentum.rsi(df['Close'], window=14)
df['MACD'] = ta.trend.macd_diff(df['Close'])
return df
```
> **實務建議**：避免使用未來資訊進行滾動計算，確保「前視窗」的資料窗口僅包含過去資料。
---
## 3.6 完整 ETL 與 Airflow 工作流範例
```python
# dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
# 依賴自前面章節的函式
from data_cleaning import clean_basic, normalize_datetime, encode_features, remove_outliers, add_time_series_features
with DAG('twse_daily_clean', start_date=datetime(2023, 1, 1), schedule_interval='@daily', catchup=False) as dag:
def fetch_and_clean(**context):
df = context['ti'].xcom_pull(task_ids='fetch_data')
df = clean_basic(df)
df = normalize_datetime(df)
df = encode_features(df)
df = remove_outliers(df, 'Close')
df = add_time_series_features(df)
context['ti'].xcom_push(key='cleaned_df', value=df)
clean_task = PythonOperator(
task_id='clean_data',
python_callable=fetch_and_clean,
provide_context=True
)
# 依賴的 fetch_data 任務（已在前章實作）
fetch_task >> clean_task
```
---
## 3.7 資料品質指標
* **完整度（Completeness）**：缺失值比例 < 5%
* **一致性（Consistency）**：資料類型、格式統一
* **準確性（Accuracy）**：數值範圍合理、無極端離群
* **時效性（Timeliness）**：資料延遲 < 1 天
定期生成「資料品質報告」並存於資料倉庫，方便後續模型監控與回溯。
---
## 小結
資料清理與前處理雖然流程繁瑣，但卻是模型成功的關鍵基石。本文以台股日成交資料為例，從缺失值、重複、日期、編碼、異常、特徵到完整的 Airflow 工作流，展示了從資料原始形態到可供模型使用的完整轉化過程。未來在第 4 章，我們將進一步探討 **統計建模** 與 **機器學習** 的實務實作，將前處理好的特徵轉化為投資決策的數學模型。