數據層:行情資料的取得與清洗

安裝必要套件

確認你的 freqtrade 虛擬環境已啟動(第 1 課建立的),或另建一個:

# 啟動虛擬環境(macOS / Linux)
source freqtrade-env/bin/activate
# Windows 用: freqtrade-env\Scripts\activate

# 安裝本課需要的套件
pip install ccxt yfinance pandas pyarrow

# 驗證安裝
python -c "import ccxt, yfinance, pandas, pyarrow; print('全部安裝 OK')"

pyarrow 是 pandas 讀寫 Parquet 格式的後端,必須裝。裝完建工作目錄:

mkdir -p ~/quant-data/raw ~/quant-data/clean

用 ccxt 抓 BTC/USDT 一年 OHLCV

新建 fetch_btc.py,完整可跑:

# fetch_btc.py
import ccxt
import pandas as pd
from datetime import datetime, timedelta, timezone
from pathlib import Path

DATA_DIR = Path.home() / 'quant-data'

exchange = ccxt.binance({
    'enableRateLimit': True,   # 自動限速,絕對要開
})

def fetch_ohlcv_full(symbol: str, timeframe: str = '1h', days: int = 365) -> pd.DataFrame:
    """
    從 Binance 抓指定天數的 OHLCV 資料。
    Binance 每次最多回傳 1000 根 K 線,自動分批抓完。
    不需要 API key。
    """
    since_dt = datetime.now(timezone.utc) - timedelta(days=days)
    since_ms = int(since_dt.timestamp() * 1000)

    all_candles = []
    print(f"開始抓 {symbol} {timeframe},從 {since_dt.strftime('%Y-%m-%d')} 起...")

    while True:
        candles = exchange.fetch_ohlcv(symbol, timeframe, since=since_ms, limit=1000)
        if not candles:
            break
        all_candles.extend(candles)
        last_ts = candles[-1][0]
        since_ms = last_ts + 1  # 下一批從最後一根的下一毫秒開始

        print(
            f"  已抓 {len(all_candles)} 筆,"
            f"最新:{pd.to_datetime(last_ts, unit='ms', utc=True)}"
        )

        if len(candles) < 1000:
            # 回傳不足 1000 筆代表已到最新資料
            break

    df = pd.DataFrame(
        all_candles,
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
    df = df.set_index('datetime').drop(columns=['timestamp'])
    df = df.sort_index()
    return df


if __name__ == '__main__':
    df_btc = fetch_ohlcv_full('BTC/USDT', '1h', days=365)
    print(f"\n下載完成:{len(df_btc)} 筆")
    print(f"時間範圍:{df_btc.index[0]} ~ {df_btc.index[-1]}")
    print(df_btc.tail(3))

    # 先存原始資料(不做任何修改)
    out_path = DATA_DIR / 'raw' / 'btc_usdt_1h_raw.parquet'
    df_btc.to_parquet(out_path)
    print(f"原始資料已存至 {out_path}")

執行:

python fetch_btc.py

正常輸出看起來像這樣:

開始抓 BTC/USDT 1h,從 2025-07-04 起...
  已抓 1000 筆,最新:2025-07-12 07:00:00+00:00
  已抓 2000 筆,最新:2025-07-23 15:00:00+00:00
  ...
  已抓 8760 筆,最新:2026-07-04 23:00:00+00:00

下載完成:8760 筆
時間範圍:2025-07-05 00:00:00+00:00 ~ 2026-07-04 23:00:00+00:00

一年 365 天 × 24 小時 = 8,760 根 K 線。實際筆數通常少一些,因為交易所會有短暫維護。如果低於 8,500,代表有明顯缺口——記錄一下,清洗步驟會偵測出來在哪段時間。

用 yfinance 抓台積電日 K 線

新建 fetch_twstock.py:

# fetch_twstock.py
import yfinance as yf
import pandas as pd
from pathlib import Path

DATA_DIR = Path.home() / 'quant-data'

def fetch_tw_stock(symbol: str, period: str = '1y') -> pd.DataFrame:
    """
    用 yfinance 抓台股歷史日 K 線。
    symbol 格式:股票代號 + .TW,例如台積電 = '2330.TW'
    period: '1y'=一年, '2y'=兩年, '5y'=五年
    """
    ticker = yf.Ticker(symbol)
    # auto_adjust=True 會做除權除息還原,但可能讓 OHLCV 邏輯矛盾(見常見坑)
    # 入門階段先用 True,注意後續驗證
    df = ticker.history(period=period, interval='1d', auto_adjust=True)

    if df.empty:
        raise ValueError(
            f"抓不到 {symbol} 的資料。確認:\n"
            f"  1. Symbol 格式是否正確(台股加 .TW)\n"
            f"  2. Yahoo Finance 是否暫時異常"
        )

    # yfinance 回傳的 index 帶有時區,但各版本行為不一致
    # 統一做法:先確認有時區,再轉成 UTC 存儲
    if df.index.tz is None:
        df.index = df.index.tz_localize('Asia/Taipei')
    df.index = df.index.tz_convert('UTC')
    df.index.name = 'datetime'

    # 只保留 OHLCV 欄位,欄位名稱統一小寫
    df = df[['Open', 'High', 'Low', 'Close', 'Volume']].rename(columns=str.lower)
    return df


if __name__ == '__main__':
    # 台積電
    df_tsmc = fetch_tw_stock('2330.TW', period='1y')
    print(f"台積電 (2330.TW):{len(df_tsmc)} 筆")
    print(f"時間範圍:{df_tsmc.index[0]} ~ {df_tsmc.index[-1]}")
    print(df_tsmc.tail(5))

    out_path = DATA_DIR / 'raw' / 'tsmc_1d_raw.parquet'
    df_tsmc.to_parquet(out_path)
    print(f"台積電原始資料已存至 {out_path}")

執行:

python fetch_twstock.py

注意:台股日 K 線一年約 248 個交易日(扣掉週末、國定假日)。如果你看到 Volume 欄位的數字很小(幾千到幾萬),這是「張」(1 張 = 1,000 股),不是「股數」。和加密貨幣的 Volume 單位完全不同,策略中使用時要特別標注單位,避免和其他資料混用。

pandas 清洗:缺值、重複、時區、異常值

新建 clean_data.py。這是本課最核心的程式碼,封裝成可重用的函式:

# clean_data.py
import pandas as pd
import numpy as np
from pathlib import Path

DATA_DIR = Path.home() / 'quant-data'

def clean_ohlcv(df: pd.DataFrame, timeframe_minutes: int, symbol: str = '') -> pd.DataFrame:
    """
    標準 OHLCV 清洗流程,適用加密貨幣和股票資料。
    
    參數:
      df: 輸入 DataFrame,index 為 DatetimeIndex(含時區)
      timeframe_minutes: K 線間隔分鐘數 (1h=60, 1d=1440)
      symbol: 用於 log 顯示
    
    回傳:清洗後的 DataFrame
    """
    print(f"\n{'='*55}")
    print(f"清洗 {symbol or 'DataFrame'},原始筆數:{len(df)}")

    # --- 1. 時間戳記截斷到 timeframe 粒度,再去重 ---
    # 有時資料回傳的時間戳記會差幾毫秒到幾秒
    freq = f'{timeframe_minutes}min'
    df.index = df.index.floor(freq)
    dup_count = df.index.duplicated().sum()
    if dup_count > 0:
        print(f"[!] 重複時間戳記 {dup_count} 筆(截斷後),保留第一筆")
        df = df[~df.index.duplicated(keep='first')]

    # --- 2. 確保按時間遞增排列 ---
    df = df.sort_index()

    # --- 3. 偵測並處理缺值 ---
    null_counts = df.isnull().sum()
    if null_counts.any():
        print(f"[!] 缺值統計:\n{null_counts[null_counts > 0]}")
        # 前向填充(ffill):用前一根已知資料填補,最多連填 3 根
        # 注意:不用 bfill(後向填充),bfill 會用未來資料 → 前視偏差
        df = df.ffill(limit=3)
        remaining = df.isnull().sum().sum()
        if remaining > 0:
            print(f"[!] ffill 後仍有 {remaining} 個缺值,丟棄對應列")
            df = df.dropna()

    # --- 4. 偵測異常 K 線(OHLCV 的基本邏輯約束) ---
    anomaly = (
        (df['high'] < df['low'])    |  # 最高 < 最低,不可能
        (df['high'] < df['open'])   |  # 最高 < 開盤,不可能
        (df['high'] < df['close'])  |  # 最高 < 收盤,不可能
        (df['open'] <= 0)           |  # 價格不可為零或負
        (df['close'] <= 0)          |
        (df['volume'] < 0)             # 成交量不可為負
    )
    if anomaly.sum() > 0:
        print(f"[!] 發現 {anomaly.sum()} 根異常 K 線,已移除")
        print(df[anomaly])
        df = df[~anomaly]

    # --- 5. 偵測時間缺口 ---
    expected_gap = pd.Timedelta(minutes=timeframe_minutes)
    time_diffs = df.index.to_series().diff().dropna()
    large_gaps = time_diffs[time_diffs > expected_gap * 2]
    if len(large_gaps) > 0:
        print(f"[!] 發現 {len(large_gaps)} 個時間缺口(超過 2 倍 timeframe):")
        for ts, gap in large_gaps.items():
            print(f"    {ts}  前有 {gap} 的空白")
    else:
        print("[OK] 無明顯時間缺口")

    print(f"清洗完成,剩餘筆數:{len(df)}")
    return df


if __name__ == '__main__':
    # 清洗 BTC/USDT 1h
    df_btc_raw = pd.read_parquet(DATA_DIR / 'raw' / 'btc_usdt_1h_raw.parquet')
    df_btc_clean = clean_ohlcv(df_btc_raw, timeframe_minutes=60, symbol='BTC/USDT 1h')
    df_btc_clean.to_parquet(DATA_DIR / 'clean' / 'btc_usdt_1h_clean.parquet')

    # 清洗台積電 1d
    df_tsmc_raw = pd.read_parquet(DATA_DIR / 'raw' / 'tsmc_1d_raw.parquet')
    df_tsmc_clean = clean_ohlcv(df_tsmc_raw, timeframe_minutes=1440, symbol='2330.TW 1d')
    df_tsmc_clean.to_parquet(DATA_DIR / 'clean' / 'tsmc_1d_clean.parquet')

    print("\n全部清洗完成。")

執行:

python clean_data.py

這個 clean_ohlcv 函式其實是一條五道關卡的流水線,每根 K 線都要依序通過:

為什麼用 ffill 不用 fillna(method='bfill')?

前向填充(ffill)是「用你當下能知道的最後一個值填補」,符合真實交易情境——你不知道下一根 K 線是多少。後向填充(bfill)是「用未來的值填補現在」,這正是前視偏差的典型來源,後面的章節會深度拆解。

存成 Parquet 和 SQLite,建立可查詢的資料庫

清洗完的資料已經是 Parquet 格式了。這一步補充 SQLite 的用法,以及兩種格式的選擇原則:

# db_utils.py
import pandas as pd
import sqlite3
from pathlib import Path

DATA_DIR = Path.home() / 'quant-data'
DB_PATH = DATA_DIR / 'market.db'

def save_to_sqlite(df: pd.DataFrame, table: str) -> None:
    """存入 SQLite,資料表已存在就新增(不覆蓋)"""
    conn = sqlite3.connect(DB_PATH)
    # if_exists='append':新增資料到現有資料表
    # if_exists='replace':清空後重新存(小心用)
    df.to_sql(table, conn, if_exists='append', index=True)
    conn.close()
    print(f"已存入 {DB_PATH}  資料表:{table}")


def query_db(sql: str) -> pd.DataFrame:
    """執行 SQL 查詢並回傳 DataFrame"""
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql(
        sql, conn,
        parse_dates=['datetime'],
        index_col='datetime'
    )
    conn.close()
    return df


if __name__ == '__main__':
    # 把 Parquet 的清洗資料一併存進 SQLite
    df_btc = pd.read_parquet(DATA_DIR / 'clean' / 'btc_usdt_1h_clean.parquet')
    save_to_sqlite(df_btc, 'btc_usdt_1h')

    df_tsmc = pd.read_parquet(DATA_DIR / 'clean' / 'tsmc_1d_clean.parquet')
    save_to_sqlite(df_tsmc, 'tsmc_1d')

    # 用 SQL 查某段時間
    df_q1 = query_db(
        "SELECT * FROM btc_usdt_1h "
        "WHERE datetime >= '2026-01-01' AND datetime < '2026-04-01'"
    )
    print(f"2026 Q1 BTC 1h 資料:{len(df_q1)} 筆")

    # Parquet 也可以直接用 loc 篩選,更快
    df_btc_all = pd.read_parquet(DATA_DIR / 'clean' / 'btc_usdt_1h_clean.parquet')
    df_q1_parquet = df_btc_all.loc['2026-01-01':'2026-03-31']
    print(f"Parquet loc 篩選結果:{len(df_q1_parquet)} 筆")

Parquet vs SQLite:怎麼選

入門階段兩個都建:Parquet 當主要存儲、SQLite 當可以 SQL 查詢的二次索引。