第 2 章：資料收集與清洗——從海量資訊中鑿取洞見

# 第 2 章：資料收集與清洗 ## 2.1 資料來源全景 投資資訊並非單一來源，而是一張複雜的網路。\n | 類型 | 典型來源 | 典型數據 | 主要挑戰 | |------|----------|----------|----------| | **市場行情** | Yahoo Finance、Google Finance、Quandl、證券交易所 API | 逐日、逐分鐘價格、成交量、波動率 | 時間戳對齊、缺失值 | | **基本面數據** | Bloomberg、Reuters、證券公司財報 | 企業財報、EPS、PE、ROE | 版權費、更新頻率 | | **新聞與社交** | Bloomberg、Reuters、Twitter、新聞聚合器 | 文本、情感指標 | 自然語言處理、噪音 | | **宏觀經濟** | FRED、國家統計局 | GDP、CPI、利率、失業率 | 週期性、季節調整 | | **非結構化** | 產業報告、白皮書、技術專利 | PDF、HTML、PDF | 文字抽取、格式不規則 | > **思考題**：哪一類資料最容易成為你量化模型的基石？ ## 2.2 抓取工具與策略 | 步驟 | 工具/函式庫 | 說明 | |------|--------------|------| | **HTTP 請求** | `requests`、`httpx` | 取得 RESTful API 資料 | | **資料庫連線** | `pymysql`、`psycopg2` | 從 SQL 取回歷史行情 | | **網頁爬蟲** | `beautifulsoup4`、`scrapy` | 抓取新聞、財報下載頁面 | | **WebSocket** | `websocket-client` | 取得即時行情 | | **資料匯流** | `airflow`、`prefect` | 建立 ETL 管線 | > **實務提醒**：在抓取前確認 API 使用條款，避免被封鎖。若需高頻資料，考慮使用交易所官方 WebSocket。 ## 2.3 資料清洗：把「噪音」轉為「訊息」 ### 2.3.1 缺失值處理 | 方法 | 何時適用 | 優缺點 | |------|----------|--------| | **刪除** | 缺失率低於 5%，且不影響樣本量 | 快速、簡單 | 可能失去資訊 | | **前向/後向填補** | 時序資料、缺失不連續 | 保持時間序列連續 | 可能引入偏差 | | **統計填補**（平均/中位數） | 缺失率中等、資料分布已知 | 平滑處理 | 低波動性 | | **迴歸填補** | 欄位之間關聯強 | 提升精確度 | 複雜、計算量大 | python import pandas as pd df = pd.read_csv('price.csv') # 先前向填補 df['Close'] = df['Close'].fillna(method='ffill') ### 2.3.2 異常值偵測 常用方法：Z‑score、IQR、Isolation Forest。\n python from scipy import stats import numpy as np z_scores = np.abs(stats.zscore(df['Volume'])) df_clean = df[(z_scores < 3)] > **提示**：在金融領域，異常值往往代表重要事件，切勿一概清除。先對異常值做**事件標記**，再決定處理方式。 ### 2.3.3 時間對齊與頻率統一 不同來源的時間戳往往不一致：交易日、工作日、 30 分鐘、 1 小時。\n python # 以日頻為例 ohlc = df.resample('D').agg({ 'Open': 'first', 'High': 'max', 'Low': 'min', 'Close': 'last', 'Volume': 'sum' }) > **實務技巧**：使用 `pd.tseries.offsets` 針對假日調整；若資料跨多時區，統一為 UTC。 ## 2.4 資料品質指標（Data Quality Checklist） | 指標 | 定義 | 評估方法 | |------|------|----------| | **完整性** | 所有必要欄位都有資料 | 欄位缺失率、列缺失率 | | **一致性** | 同一實體的資料不衝突 | 交叉檢查、主鍵唯一性 | | **準確性** | 與真實值相符 | 與官方報表對照、抽樣驗證 | | **時效性** | 資料更新滯後 | 觀測時間戳與事件時間比較 | | **可用性** | 資料易於存取與處理 | API 響應速度、資料格式 | > **案例**：某基金每月結算時更新 NAV，若未檢查「時效性」指標，模型可能使用已過時的 NAV，導致預測偏離。 ## 2.5 儲存與管理：打造可靠的資料湖 | 儲存方案 | 優勢 | 適用場景 | |----------|------|----------| | **本地檔案** (`CSV`, `Parquet`) | 易於快速實驗 | 小規模資料、單機環境 | | **關聯式資料庫** (PostgreSQL, MySQL) | ACID、SQL 查詢 | 交易紀錄、權益管理 | | **分布式檔案系統** (HDFS, S3) | 大規模存取 | 历史行情、行情档案 | | **資料倉儲** (Snowflake, BigQuery) | 雲端即服務、彈性 | 商業智慧、跨團隊共享 | > **最佳實踐**： > 1. 采用 Parquet 以節省空間與提升讀取速度。\n> 2. 為關鍵欄位建立索引，避免全表掃描。\n> 3. 使用資料版本控制（DVC、Delta Lake）以追蹤資料演化。 ## 2.6 自動化流程：ETL 的四步驟 1. **Extract**：從多源抓取，標準化時間戳。\n2. **Transform**：清洗、填補、異常偵測、特徵工程。\n3. **Load**：將資料寫入資料湖或資料倉。\n4. **Monitor**：實時監控缺失率、延遲、錯誤率。 python # Airflow 範例：每日更新 from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def extract(): # 伪代码：抓取 Yahoo Finance pass def transform(): # 伪代码：清洗、填补 pass def load(): # 伪代码：写入 Parquet pass with DAG('daily_etl', start_date=datetime(2024,1,1), schedule_interval='@daily') as dag: t1 = PythonOperator(task_id='extract', python_callable=extract) t2 = PythonOperator(task_id='transform', python_callable=transform) t3 = PythonOperator(task_id='load', python_callable=load) t1 >> t2 >> t3 > **挑戰**：數據品質問題往往在流程早期就已經存在，若無效的清洗策略，後續模型評估將陷入 **Garbage In, Garbage Out**。 ## 2.7 小結 在本章中，我們走過了從資料收集到清洗的完整流程，並結合實務工具與最佳實踐，為後續的量化模型奠定堅實基礎。記住：**資料是投資模型的燃料，只有高品質、乾淨的資料才能讓模型跑得更快、更準確。** > **思考練習**：選擇一支你感興趣的股票，列出你能取得的所有資料來源，並規劃一個簡易的 ETL 流程。把這份計畫寫在筆記本，下一章我們將實際執行並驗證。