自訂市場物件

FinLab 預設支援台股(TWMarket)、美股(USMarket)、興櫃(ROTCMarket)等市場。若你想回測加密貨幣、期貨、海外股票,或使用自訂的價格資料,可以透過繼承 Market 類別來實現。

為什麼需要自訂市場?

不同市場有不同的特性:

交易時間不同: 美股與台股的開收盤時間不同
價格欄位不同: 加密貨幣可能沒有「開盤價」的概念
資料來源不同: 從 CSV、API、資料庫載入價格
交易規則不同: 期貨有槓桿、加密貨幣沒有漲跌幅限制
對照指數不同: 台股對照加權指數、美股對照 S&P 500

透過自訂 Market 物件,你可以將任何市場的資料套用到 FinLab 的回測引擎。

快速上手:使用內建市場

FinLab 提供 3 個內建市場:

from finlab import data, backtest
from finlab.markets.tw import TWMarket
from finlab.markets.us import USMarket
from finlab.markets.rotc import ROTCMarket

# 台股市場(預設)
close = data.get('price:收盤價')
position = close > close.average(20)
report = backtest.sim(position, resample='M')
# 等同於
report = backtest.sim(position, resample='M', market=TWMarket())

# 美股市場
us_close = data.get('etl:us_adj_close')
us_position = us_close > us_close.average(50)
report = backtest.sim(us_position, resample='W', market=USMarket())

# 興櫃市場
rotc_close = data.get('rotc_price:收盤價')
rotc_position = rotc_close > 10
report = backtest.sim(rotc_position, resample='M', market=ROTCMarket())

內建市場特性對比

特性	TWMarket	USMarket	ROTCMarket
市場名稱	`'tw_stock'`	`'us_stock'`	`'rotc_stock'`
資料頻率	每日(`'1d'`)	每日(`'1d'`)	每日(`'1d'`)
對照指數	加權指數	S&P 500	(無)
時區	Asia/Taipei	US/Eastern	Asia/Taipei
收盤時間	15:00	16:00	14:00
漲跌幅限制	10%	無	無
特殊股票類別	處置股/全額交割	無	無

自訂市場類別

基礎範例:加密貨幣市場

假設你有比特幣(BTC)和以太幣(ETH)的每日收盤價 CSV 檔案:

from finlab.market import Market
import pandas as pd

class CryptoMarket(Market):

    @staticmethod
    def get_name():
        """回傳市場名稱"""
        return 'crypto'

    @staticmethod
    def get_freq():
        """回傳資料頻率"""
        return '1d'  # 每日

    def get_price(self, trade_at_price='close', adj=True):
        """取得價格資料

        Args:
            trade_at_price: 'open', 'close', 'high', 'low' 之一
            adj: 是否使用調整後價格(加密貨幣通常不需要)

        Returns:
            pd.DataFrame: index 為日期,columns 為幣種代號
        """
        # 從 CSV 載入價格
        df = pd.read_csv(f'crypto_{trade_at_price}.csv', index_col=0, parse_dates=True)
        return df

    @staticmethod
    def get_benchmark():
        """對照指數(例如用 BTC 當作對照)"""
        df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
        return df['BTC']

# 使用自訂市場
from finlab.backtest import sim

# 假設 CSV 格式:
# date,BTC,ETH,BNB
# 2020-01-01,7200,130,15
# 2020-01-02,7350,135,16

close = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
position = close > close.rolling(20).mean()

report = sim(position, market=CryptoMarket(), resample='W')
report.display()

進階範例:期貨市場(含槓桿)

期貨有槓桿效果,可透過修改價格來模擬:

class FuturesMarket(Market):

    def __init__(self, leverage=10):
        """
        Args:
            leverage: 槓桿倍數
        """
        self.leverage = leverage
        self.prices = pd.read_csv('futures_price.csv', index_col=0, parse_dates=True)

    @staticmethod
    def get_name():
        return 'futures'

    @staticmethod
    def get_freq():
        return '1d'

    def get_price(self, trade_at_price='close', adj=True):
        """回傳槓桿後的報酬率"""
        price = self.prices[trade_at_price]

        # 計算日報酬率,乘上槓桿倍數
        daily_return = price.pct_change() * self.leverage

        # 還原成價格(從 100 開始累積)
        leveraged_price = (1 + daily_return).fillna(1).cumprod() * 100

        return leveraged_price

    @staticmethod
    def get_benchmark():
        # 使用 1 倍槓桿當對照
        df = pd.read_csv('futures_price.csv', index_col=0, parse_dates=True)
        return df['close'].iloc[:, 0]

# 使用 10 倍槓桿回測
report = sim(position, market=FuturesMarket(leverage=10))

Market 類別完整 API

繼承 Market 類別時,可覆寫以下方法:

必須實作的方法

`get_price(trade_at_price, adj=True)`

最重要的方法,回傳價格資料。

def get_price(self, trade_at_price='close', adj=True) -> pd.DataFrame:
    """
    Args:
        trade_at_price (str): 'open', 'close', 'high', 'low', 'volume' 之一
        adj (bool): 是否回傳調整後價格(考慮股票分割、配息等)

    Returns:
        pd.DataFrame:
            - index: 日期(DatetimeIndex)
            - columns: 股票代號
            - values: 價格

    Examples:
        回傳格式範例:

        | date       |   BTC  |   ETH  |   BNB  |
        |:-----------|-------:|-------:|-------:|
        | 2020-01-01 |  7200  |   130  |   15   |
        | 2020-01-02 |  7350  |   135  |   16   |
        | 2020-01-03 |  7100  |   128  |   14.5 |
    """
    # 必須實作
    raise NotImplementedError()

選用的靜態方法

`get_name()`

回傳市場名稱,用於識別市場類型。

@staticmethod
def get_name() -> str:
    return 'crypto'  # 預設為 'auto'

`get_freq()`

回傳資料頻率,影響報酬率計算。

@staticmethod
def get_freq() -> str:
    return '1d'   # 每日
    # 其他常見: '1h'(每小時), '4h', '1w'(每週)

`get_benchmark()`

回傳對照指數,用於績效比較。

@staticmethod
def get_benchmark() -> pd.Series:
    """
    Returns:
        pd.Series: 對照指數的價格序列
            - index: 日期
            - values: 指數價格
    """
    return pd.Series([])  # 預設為空(不顯示對照)

`get_asset_id_to_name()`

回傳股號與股名的對照表,用於報告顯示。

@staticmethod
def get_asset_id_to_name() -> dict:
    """
    Returns:
        dict: 股號 -> 股名的對照表

    Examples:
        {'BTC': 'Bitcoin', 'ETH': 'Ethereum'}
    """
    return {}

`get_market_value()`

回傳市值資料,用於市值加權回測。

@staticmethod
def get_market_value() -> pd.DataFrame:
    """
    Returns:
        pd.DataFrame: 市值資料
            - index: 日期
            - columns: 股票代號
            - values: 市值
    """
    return pd.DataFrame()

`get_industry()`

回傳產業分類,用於產業分析。

@staticmethod
def get_industry() -> dict:
    """
    Returns:
        dict: 股號 -> 產業的對照表

    Examples:
        {'BTC': 'Cryptocurrency', 'ETH': 'Cryptocurrency'}
    """
    return {}

`market_close_at_timestamp(timestamp)`

回傳特定時間的市場收盤時間,用於實盤交易。

def market_close_at_timestamp(self, timestamp=None) -> pd.Timestamp:
    """
    Args:
        timestamp: 查詢的時間點(預設為最新)

    Returns:
        pd.Timestamp: 最近的市場收盤時間
    """
    # 加密貨幣 24 小時交易,可用當天 23:59 當收盤
    if timestamp is None:
        timestamp = pd.Timestamp.now()
    return pd.Timestamp(timestamp.date()) + pd.Timedelta('23:59:00')

使用自訂資料源

1. 從 CSV 載入

最簡單的方式,適合一次性資料:

class CSVMarket(Market):

    def __init__(self, csv_path):
        self.csv_path = csv_path

    @staticmethod
    def get_name():
        return 'csv_market'

    def get_price(self, trade_at_price='close', adj=True):
        df = pd.read_csv(self.csv_path, index_col=0, parse_dates=True)
        return df

# 使用
report = sim(position, market=CSVMarket('my_prices.csv'))

2. 從 API 動態載入

適合需要即時更新的資料:

import requests

class APIMarket(Market):

    def __init__(self, api_url):
        self.api_url = api_url

    @staticmethod
    def get_name():
        return 'api_market'

    def get_price(self, trade_at_price='close', adj=True):
        # 從 API 取得 JSON 資料
        response = requests.get(f'{self.api_url}/{trade_at_price}')
        data = response.json()

        # 轉換成 DataFrame
        df = pd.DataFrame(data)
        df['date'] = pd.to_datetime(df['date'])
        df = df.set_index('date')

        return df

# 使用
report = sim(position, market=APIMarket('https://api.example.com/prices'))

3. 從 FinLab 資料庫混合載入

結合 FinLab 資料與自訂資料:

from finlab import data

class HybridMarket(Market):

    @staticmethod
    def get_name():
        return 'hybrid'

    def get_price(self, trade_at_price='close', adj=True):
        # 從 FinLab 取得台股價格
        tw_close = data.get('price:收盤價')

        # 載入自訂的加密貨幣價格
        crypto_close = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)

        # 合併兩個 DataFrame(外連接)
        combined = pd.concat([tw_close, crypto_close], axis=1)

        return combined

# 使用(可同時回測台股 + 加密貨幣組合)
report = sim(position, market=HybridMarket())

進階功能

1. 客製化交易價格計算

覆寫 get_trading_price() 方法自訂成交價:

class CustomPriceMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        # ... 回傳基本價格

    def get_trading_price(self, name, adj=True):
        """自訂成交價計算"""
        if name == 'vwap':  # 成交量加權均價
            open_price = self.get_price('open', adj=adj)
            close_price = self.get_price('close', adj=adj)
            high_price = self.get_price('high', adj=adj)
            low_price = self.get_price('low', adj=adj)
            volume = self.get_price('volume', adj=False)

            vwap = (open_price + close_price + high_price + low_price) / 4
            return vwap
        else:
            # 其他情況使用預設邏輯
            return super().get_trading_price(name, adj=adj)

# 使用 VWAP 當成交價
report = sim(position, market=CustomPriceMarket(), trade_at='vwap')

2. 支援多時區

處理跨時區的市場:

import datetime

class MultiTimezoneMarket(Market):

    @staticmethod
    def get_name():
        return 'multi_tz'

    def market_close_at_timestamp(self, timestamp=None):
        """不同市場有不同收盤時間"""
        if timestamp is None:
            timestamp = pd.Timestamp.now()

        # 假設是全球 24 小時市場,用 UTC 時間
        timestamp_utc = timestamp.tz_convert('UTC')
        market_close = pd.Timestamp(timestamp_utc.date()) + pd.Timedelta('23:59:59')
        return market_close.tz_localize('UTC')

3. 實作市場休市邏輯

排除週末或假日:

class WithHolidaysMarket(Market):

    def __init__(self):
        # 定義休市日(例如台灣國定假日)
        self.holidays = ['2024-01-01', '2024-02-08', '2024-02-09']
        self.holidays = [pd.Timestamp(d) for d in self.holidays]

    def get_price(self, trade_at_price='close', adj=True):
        df = pd.read_csv('prices.csv', index_col=0, parse_dates=True)

        # 移除週末
        df = df[df.index.dayofweek < 5]

        # 移除假日
        df = df[~df.index.isin(self.holidays)]

        return df

測試自訂市場

建立市場物件後,務必進行測試:

# 1. 檢查價格資料格式
market = CryptoMarket()
close = market.get_price('close')
print(close.head())
print(close.index)   # 應為 DatetimeIndex
print(close.shape)   # (日期數, 股票數)

# 2. 檢查對照指數
benchmark = market.get_benchmark()
print(benchmark.head())

# 3. 執行簡單回測
position = close > close.average(20)
report = sim(position, market=market, resample='M', upload=False)
report.display()

# 4. 檢查報告的市場名稱
print(report.market.get_name())  # 應為 'crypto'

常見問題

Q1: 為什麼需要 `adj=True` 參數?

adj=True 代表使用調整後價格(考慮股票分割、配息、配股等):

台股: 除權息後會調整歷史價格,確保報酬率計算正確
美股: 同樣需要調整價格
加密貨幣: 沒有除權息,adj 參數可忽略

# 台股範例
tw_market = TWMarket()
adj_close = tw_market.get_price('close', adj=True)    # 還原股價
raw_close = tw_market.get_price('close', adj=False)   # 原始收盤價

Q2: 如何處理缺失值(NaN)?

FinLab 會自動 forward fill 缺失值,但建議在 get_price() 中處理:

def get_price(self, trade_at_price='close', adj=True):
    df = pd.read_csv('prices.csv', index_col=0, parse_dates=True)

    # 方法 1: Forward fill
    df = df.ffill()

    # 方法 2: 移除有缺失值的股票
    df = df.dropna(axis=1, how='any')

    # 方法 3: 填充 0(不推薦)
    df = df.fillna(0)

    return df

Q3: 可以在 `get_price()` 中使用 `finlab.data.get()` 嗎?

可以!這是混合使用 FinLab 資料的標準做法:

from finlab import data

class MixedMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        # 使用 FinLab 台股資料
        tw_close = data.get('price:收盤價')

        # 篩選特定股票
        tw_close = tw_close[['2330', '2317', '2454']]

        return tw_close

Q4: 如何模擬交易成本?

交易成本在 sim() 函數中設定,不在 Market 物件:

# 加密貨幣通常手續費較低,稅率為 0
report = sim(
    position,
    market=CryptoMarket(),
    fee_ratio=0.001,  # 0.1% 手續費
    tax_ratio=0       # 無交易稅
)

# 台股預設 fee_ratio=0.001425, tax_ratio=0.003

Q5: 為什麼 `get_benchmark()` 是靜態方法?

因為對照指數通常是固定的,不依賴實例變數。但若需要動態對照指數:

class DynamicBenchmarkMarket(Market):

    def __init__(self, benchmark_ticker):
        self.benchmark_ticker = benchmark_ticker

    def get_benchmark(self):  # 改為實例方法(移除 @staticmethod)
        df = pd.read_csv('benchmarks.csv', index_col=0, parse_dates=True)
        return df[self.benchmark_ticker]

# 使用
report = sim(position, market=DynamicBenchmarkMarket('BTC'))

實戰案例

案例 1: 黃金 ETF 回測

class GoldMarket(Market):

    @staticmethod
    def get_name():
        return 'gold'

    @staticmethod
    def get_freq():
        return '1d'

    def get_price(self, trade_at_price='close', adj=True):
        # 從 Yahoo Finance 載入黃金 ETF (GLD) 價格
        import yfinance as yf
        gold = yf.download('GLD', start='2010-01-01')
        return gold[[trade_at_price.capitalize()]]  # 'Close' -> 'close'

    @staticmethod
    def get_benchmark():
        import yfinance as yf
        sp500 = yf.download('^GSPC', start='2010-01-01')
        return sp500['Close'].squeeze()

# 使用
gold_close = pd.read_csv('gold_close.csv', index_col=0, parse_dates=True)
position = gold_close > gold_close.average(50)
report = sim(position, market=GoldMarket(), resample='W')

案例 2: 台股 + 美股混合回測

class GlobalMarket(Market):

    @staticmethod
    def get_name():
        return 'global'

    def get_price(self, trade_at_price='close', adj=True):
        # 台股
        tw_close = data.get('price:收盤價')[['2330', '2317']]

        # 美股
        us_close = data.get('etl:us_adj_close')[['AAPL', 'TSLA']]

        # 合併(外連接,自動對齊日期)
        combined = pd.concat([tw_close, us_close], axis=1)

        return combined

    @staticmethod
    def get_benchmark():
        # 使用 MSCI 世界指數或 60/40 台美股組合
        tw_benchmark = data.get('benchmark_return:發行量加權股價報酬指數').squeeze()
        us_benchmark = data.get('world_index:adj_close')['^GSPC']

        combined = pd.concat([tw_benchmark * 0.6, us_benchmark * 0.4], axis=1).sum(axis=1)
        return combined

# 使用
position = ...  # 包含台股 + 美股的持倉
report = sim(position, market=GlobalMarket())

案例 3: 加密貨幣 4 小時 K 線回測

class Crypto4HMarket(Market):

    @staticmethod
    def get_name():
        return 'crypto_4h'

    @staticmethod
    def get_freq():
        return '4h'  # 4 小時頻率

    def get_price(self, trade_at_price='close', adj=True):
        # 從 Binance API 載入 4 小時 K 線
        df = pd.read_csv('binance_4h_close.csv', index_col=0, parse_dates=True)
        return df

# 使用
close = pd.read_csv('binance_4h_close.csv', index_col=0, parse_dates=True)
position = close > close.rolling(50).mean()

# 注意: resample 參數要對應頻率
report = sim(position, market=Crypto4HMarket(), resample='1d')  # 每日調整

參考資源

常見錯誤與解決方法

錯誤 1：檔案讀取失敗

現象：執行自訂市場時拋出 FileNotFoundError

class CryptoMarket(Market):
    def get_price(self, trade_at_price='close', adj=True):
        df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
        return df

market = CryptoMarket()
close = market.get_price('close')
# FileNotFoundError: [Errno 2] No such file or directory: 'crypto_close.csv'

原因： - CSV 檔案不存在或路徑錯誤 - 使用相對路徑但工作目錄不正確 - 檔名拼寫錯誤或大小寫不符

解決方法：

from finlab.market import Market
import pandas as pd
import os

class CryptoMarket(Market):

    def __init__(self, data_dir='./data'):
        """
        Args:
            data_dir: 資料目錄路徑（預設為 ./data）
        """
        self.data_dir = data_dir

    @staticmethod
    def get_name():
        return 'crypto'

    @staticmethod
    def get_freq():
        return '1d'

    def get_price(self, trade_at_price='close', adj=True):
        """取得價格資料（含錯誤處理）"""

        # 構建完整路徑
        file_path = os.path.join(self.data_dir, f'crypto_{trade_at_price}.csv')

        # 檢查檔案是否存在
        if not os.path.exists(file_path):
            raise FileNotFoundError(
                f"❌ 找不到資料檔案：{file_path}\n"
                f"   請確認：\n"
                f"   1. 檔案是否存在於 {self.data_dir} 目錄\n"
                f"   2. 檔案名稱是否正確（crypto_{trade_at_price}.csv）\n"
                f"   3. 工作目錄是否正確（當前：{os.getcwd()}）\n"
                f"   提示：可至 https://example.com/data 下載範例資料"
            )

        # 嘗試讀取檔案
        try:
            df = pd.read_csv(file_path, index_col=0, parse_dates=True)

        except PermissionError:
            raise PermissionError(
                f"❌ 無權限讀取檔案：{file_path}\n"
                f"   請檢查檔案權限設定"
            )

        except pd.errors.EmptyDataError:
            raise ValueError(
                f"❌ 資料檔案為空：{file_path}\n"
                f"   請確認檔案內容是否正確"
            )

        except Exception as e:
            raise RuntimeError(
                f"❌ 讀取檔案失敗：{file_path}\n"
                f"   錯誤訊息：{e}"
            )

        # 檢查資料完整性（後續步驟，見錯誤 2）
        df = self._validate_dataframe(df, file_path)

        return df

    def _validate_dataframe(self, df, file_path):
        """驗證 DataFrame 格式"""

        # 檢查是否為空
        if df.empty:
            raise ValueError(
                f"❌ 資料為空：{file_path}\n"
                f"   請確認檔案內容是否正確"
            )

        # 檢查 index 是否為 DatetimeIndex
        if not isinstance(df.index, pd.DatetimeIndex):
            raise TypeError(
                f"❌ Index 必須為 DatetimeIndex（日期格式）\n"
                f"   當前 Index 類型：{type(df.index)}\n"
                f"   請確認 CSV 第一欄為日期格式（如 2020-01-01）\n"
                f"   並使用 parse_dates=True 參數"
            )

        # 檢查是否有數值資料
        if df.shape[1] == 0:
            raise ValueError(
                f"❌ 無任何股票欄位：{file_path}\n"
                f"   請確認 CSV 格式正確（第一欄為日期，其他欄為股票代號）"
            )

        print(f"✅ 資料載入成功：{file_path}")
        print(f"   資料範圍：{df.index[0]} ~ {df.index[-1]}")
        print(f"   股票數量：{df.shape[1]} 檔")
        print(f"   交易日數：{df.shape[0]} 日")

        return df

    @staticmethod
    def get_benchmark():
        """對照指數（含錯誤處理）"""
        try:
            df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
            if 'BTC' not in df.columns:
                print("⚠️  警告：找不到 BTC 欄位，無法設定對照指數")
                return pd.Series([])  # 回傳空 Series
            return df['BTC']
        except Exception as e:
            print(f"⚠️  警告：無法載入對照指數：{e}")
            return pd.Series([])  # 回傳空 Series，不影響主要功能


# 使用範例：指定資料目錄
try:
    market = CryptoMarket(data_dir='/path/to/crypto/data')
    close = market.get_price('close')
    print("✅ 市場物件建立成功")

except FileNotFoundError as e:
    print(f"\n{e}")
    print("\n請下載範例資料或修正檔案路徑")

except Exception as e:
    print(f"\n❌ 建立市場物件失敗：{e}")

錯誤 2：資料格式錯誤

現象：DataFrame 格式不符合預期，導致回測失敗

close = market.get_price('close')
report = sim(position, market=market)
# TypeError: Index must be DatetimeIndex
# 或
# KeyError: 'BTC'

常見格式問題：

Index 不是 DatetimeIndex
缺少必要欄位
資料型態錯誤（字串而非數值）
包含過多缺失值

解決方法：在 get_price() 中強化驗證

class RobustMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        """取得價格資料（完整驗證）"""

        df = pd.read_csv(f'data_{trade_at_price}.csv', index_col=0)

        # 驗證 1：轉換 index 為 DatetimeIndex
        if not isinstance(df.index, pd.DatetimeIndex):
            try:
                df.index = pd.to_datetime(df.index)
                print("✅ Index 已轉換為 DatetimeIndex")
            except Exception as e:
                raise TypeError(
                    f"❌ 無法將 Index 轉換為 DatetimeIndex\n"
                    f"   Index 範例：{df.index[:3].tolist()}\n"
                    f"   錯誤訊息：{e}\n"
                    f"   請確認第一欄格式為日期（如 2020-01-01 或 2020/01/01）"
                )

        # 驗證 2：檢查資料型態
        non_numeric_cols = df.select_dtypes(exclude=['number']).columns.tolist()
        if non_numeric_cols:
            print(f"⚠️  警告：以下欄位非數值型態，將嘗試轉換：{non_numeric_cols}")
            for col in non_numeric_cols:
                try:
                    df[col] = pd.to_numeric(df[col], errors='coerce')
                except Exception as e:
                    raise TypeError(
                        f"❌ 無法將欄位 {col} 轉換為數值\n"
                        f"   請確認欄位內容為數字"
                    )

        # 驗證 3：檢查缺失值比例
        missing_ratio = df.isna().sum() / len(df)
        high_missing_cols = missing_ratio[missing_ratio > 0.3].index.tolist()

        if high_missing_cols:
            print(f"⚠️  警告：以下欄位缺失值 > 30%：")
            for col in high_missing_cols:
                print(f"     {col}: {missing_ratio[col]:.1%} 缺失")
            print("   建議：")
            print("   1. 移除這些欄位（df.drop(columns=high_missing_cols)）")
            print("   2. 使用 forward fill 填充（df.ffill()）")

            # 自動處理：使用 forward fill
            df = df.ffill()
            print("   已自動使用 forward fill 填充缺失值")

        # 驗證 4：檢查資料範圍
        if len(df) < 100:
            print(f"⚠️  警告：資料筆數過少（{len(df)} 筆）")
            print("   建議至少 252 個交易日（約 1 年）以上的資料")

        if df.shape[1] < 2:
            print(f"⚠️  警告：股票數量過少（{df.shape[1]} 檔）")
            print("   建議至少 10 檔以上的股票以分散風險")

        # 驗證 5：排序 index
        if not df.index.is_monotonic_increasing:
            print("⚠️  警告：日期未排序，已自動排序")
            df = df.sort_index()

        # 驗證 6：移除重複日期
        if df.index.duplicated().any():
            dup_dates = df.index[df.index.duplicated()].tolist()
            print(f"⚠️  警告：發現重複日期：{dup_dates[:5]}...")
            df = df[~df.index.duplicated(keep='last')]
            print("   已保留最後一筆記錄")

        print(f"\n✅ 資料驗證完成")
        print(f"   資料範圍：{df.index[0].date()} ~ {df.index[-1].date()}")
        print(f"   交易日數：{len(df)} 日")
        print(f"   股票數量：{df.shape[1]} 檔")
        print(f"   缺失值：{df.isna().sum().sum()} 個（{df.isna().sum().sum() / df.size:.2%}）")

        return df


# 使用範例
try:
    market = RobustMarket()
    close = market.get_price('close')

    # 執行回測
    position = close > close.rolling(20).mean()
    report = sim(position, market=market, resample='M')
    print("\n✅ 回測成功")

except (TypeError, ValueError) as e:
    print(f"\n{e}")
    print("\n請修正資料格式後重試")

錯誤 3：API 請求失敗

現象：使用 API 載入資料時拋出網路錯誤

class APIMarket(Market):
    def get_price(self, trade_at_price='close', adj=True):
        response = requests.get(f'https://api.example.com/{trade_at_price}')
        return pd.DataFrame(response.json())

market = APIMarket()
close = market.get_price('close')
# requests.exceptions.ConnectionError: Failed to establish a new connection

原因： - 網路連線問題 - API 金鑰無效或過期 - API 請求次數超過限制 - API 回傳格式錯誤

解決方法：實施重試機制與錯誤處理

import requests
import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class APIMarket(Market):

    def __init__(self, api_url, api_key=None, max_retries=3):
        """
        Args:
            api_url: API 基礎網址
            api_key: API 金鑰（選填）
            max_retries: 最大重試次數
        """
        self.api_url = api_url
        self.api_key = api_key
        self.max_retries = max_retries

        # 建立帶重試機制的 Session
        self.session = self._create_session()

    def _create_session(self):
        """建立帶重試機制的 requests Session"""
        session = requests.Session()

        # 設定重試策略
        retry_strategy = Retry(
            total=self.max_retries,
            backoff_factor=1,  # 指數退避：1s, 2s, 4s
            status_forcelist=[429, 500, 502, 503, 504],  # 重試的 HTTP 狀態碼
            allowed_methods=["HEAD", "GET", "OPTIONS"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)

        return session

    @staticmethod
    def get_name():
        return 'api_market'

    def get_price(self, trade_at_price='close', adj=True):
        """從 API 取得價格資料（含錯誤處理）"""

        url = f'{self.api_url}/prices/{trade_at_price}'

        # 設定請求標頭
        headers = {}
        if self.api_key:
            headers['Authorization'] = f'Bearer {self.api_key}'

        print(f"正在從 API 下載資料：{url}")

        try:
            # 發送請求（帶超時設定）
            response = self.session.get(
                url,
                headers=headers,
                timeout=30  # 30 秒超時
            )

            # 檢查 HTTP 狀態碼
            if response.status_code == 401:
                raise PermissionError(
                    f"❌ API 認證失敗（HTTP 401）\n"
                    f"   請檢查 API 金鑰是否正確或已過期\n"
                    f"   至 API 服務商網站重新取得金鑰"
                )

            elif response.status_code == 429:
                raise RuntimeError(
                    f"❌ API 請求次數超過限制（HTTP 429）\n"
                    f"   請稍後再試或升級 API 方案"
                )

            elif response.status_code >= 500:
                raise RuntimeError(
                    f"❌ API 伺服器錯誤（HTTP {response.status_code}）\n"
                    f"   伺服器暫時無法回應，請稍後再試"
                )

            response.raise_for_status()  # 拋出其他 HTTP 錯誤

            # 解析 JSON
            data = response.json()

        except requests.exceptions.Timeout:
            raise TimeoutError(
                f"❌ API 請求超時（> 30 秒）\n"
                f"   網路連線過慢或 API 伺服器負載過高\n"
                f"   建議：\n"
                f"   1. 檢查網路連線\n"
                f"   2. 縮短資料範圍\n"
                f"   3. 稍後再試"
            )

        except requests.exceptions.ConnectionError:
            raise ConnectionError(
                f"❌ 網路連線失敗\n"
                f"   無法連線至 {self.api_url}\n"
                f"   請檢查：\n"
                f"   1. 網路連線是否正常\n"
                f"   2. API 網址是否正確\n"
                f"   3. 防火牆是否阻擋連線"
            )

        except requests.exceptions.JSONDecodeError:
            raise ValueError(
                f"❌ API 回傳格式錯誤\n"
                f"   預期 JSON 格式，但收到：{response.text[:200]}...\n"
                f"   請聯繫 API 服務商確認文件"
            )

        except Exception as e:
            raise RuntimeError(
                f"❌ API 請求失敗：{e}\n"
                f"   請查看詳細錯誤訊息"
            )

        # 轉換為 DataFrame
        try:
            df = pd.DataFrame(data)

            # 處理常見格式
            if 'date' in df.columns:
                df['date'] = pd.to_datetime(df['date'])
                df = df.set_index('date')
            elif 'timestamp' in df.columns:
                df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
                df = df.set_index('timestamp')

            print(f"✅ API 資料下載成功")
            print(f"   資料範圍：{df.index[0]} ~ {df.index[-1]}")
            print(f"   股票數量：{df.shape[1]} 檔")

            return df

        except Exception as e:
            raise ValueError(
                f"❌ 資料轉換失敗：{e}\n"
                f"   API 回傳格式：{data}\n"
                f"   請確認 API 文件中的資料格式"
            )


# 使用範例
import os

try:
    # 從環境變數讀取 API 金鑰（推薦方式）
    api_key = os.environ.get('CRYPTO_API_KEY')
    if not api_key:
        print("⚠️  警告：未設定 CRYPTO_API_KEY 環境變數")
        print("   export CRYPTO_API_KEY='your_api_key'")

    market = APIMarket(
        api_url='https://api.example.com',
        api_key=api_key,
        max_retries=3
    )

    close = market.get_price('close')
    print("✅ 市場資料取得成功")

except (ConnectionError, TimeoutError, PermissionError) as e:
    print(f"\n{e}")
    print("\n請修正問題後重試")

except Exception as e:
    print(f"\n❌ 未預期的錯誤：{e}")

錯誤 4：回測執行失敗

現象：市場物件建立成功，但回測時拋出錯誤

market = CryptoMarket()
position = close > close.average(20)
report = sim(position, market=market, resample='M')
# ValueError: Position index does not match price index

原因： - position 與 market.get_price() 的日期範圍不一致 - position 與價格資料的股票代號不一致 - 市場物件的方法未正確實作

解決方法：在執行回測前驗證相容性

def validate_before_backtest(position, market):
    """回測前驗證 position 與 market 相容性"""

    print("=== 回測前驗證 ===\n")

    # 1. 取得市場價格
    try:
        close = market.get_price('close')
    except Exception as e:
        raise RuntimeError(f"❌ 無法取得市場價格：{e}")

    # 2. 檢查日期範圍
    position_dates = position.index
    price_dates = close.index

    print(f"Position 日期範圍：{position_dates[0]} ~ {position_dates[-1]}")
    print(f"Price 日期範圍：{price_dates[0]} ~ {price_dates[-1]}")

    # 計算交集
    common_dates = position_dates.intersection(price_dates)
    if len(common_dates) == 0:
        raise ValueError(
            f"❌ Position 與 Price 日期範圍無交集\n"
            f"   請確認資料來源一致"
        )

    if len(common_dates) < len(position_dates) * 0.8:
        print(f"⚠️  警告：僅 {len(common_dates) / len(position_dates):.1%} 的日期有對應價格")
        print("   部分交易日可能被忽略")

    # 3. 檢查股票代號
    position_stocks = position.columns.tolist()
    price_stocks = close.columns.tolist()

    print(f"\nPosition 股票數量：{len(position_stocks)}")
    print(f"Price 股票數量：{len(price_stocks)}")

    # 計算交集
    common_stocks = set(position_stocks).intersection(set(price_stocks))
    if len(common_stocks) == 0:
        raise ValueError(
            f"❌ Position 與 Price 股票代號無交集\n"
            f"   Position 範例：{position_stocks[:5]}\n"
            f"   Price 範例：{price_stocks[:5]}\n"
            f"   請確認股票代號格式一致"
        )

    missing_stocks = set(position_stocks) - set(price_stocks)
    if missing_stocks:
        print(f"⚠️  警告：{len(missing_stocks)} 檔股票在 Price 中不存在：")
        print(f"   {list(missing_stocks)[:10]}...")
        print("   這些股票將被忽略")

    # 4. 檢查市場物件方法
    required_methods = ['get_name', 'get_freq', 'get_price']
    for method in required_methods:
        if not hasattr(market, method):
            raise AttributeError(
                f"❌ Market 物件缺少必要方法：{method}\n"
                f"   請確認已正確繼承 Market 類別並實作所有方法"
            )

    print(f"\n✅ 驗證通過，共 {len(common_stocks)} 檔股票、{len(common_dates)} 個交易日")
    print("=" * 50)


# 使用範例
try:
    market = CryptoMarket()
    close = market.get_price('close')
    position = close > close.rolling(20).mean()

    # 執行驗證
    validate_before_backtest(position, market)

    # 執行回測
    report = sim(position, market=market, resample='M', upload=False)
    print("\n✅ 回測成功")
    report.display()

except Exception as e:
    print(f"\n❌ 回測失敗：{e}")
    print("請修正問題後重試")

除錯技巧

1. 逐步測試市場物件

# Step 1: 測試基本方法
market = CryptoMarket()
print(f"市場名稱：{market.get_name()}")
print(f"資料頻率：{market.get_freq()}")

# Step 2: 測試價格載入
close = market.get_price('close')
print(f"\n收盤價資料：\n{close.head()}")

# Step 3: 測試對照指數
benchmark = market.get_benchmark()
print(f"\n對照指數：\n{benchmark.head()}")

# Step 4: 執行簡單策略
position = close > close.rolling(20).mean()
print(f"\n持倉訊號：\n{position.tail()}")

# Step 5: 小範圍回測
position_small = position.iloc[-100:]  # 僅最近 100 天
report = sim(position_small, market=market, resample='M', upload=False)
report.display()

2. 使用 try-except 包裝關鍵步驟

class SafeMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        try:
            # 主要邏輯
            df = pd.read_csv(f'data_{trade_at_price}.csv', index_col=0, parse_dates=True)
            return df

        except FileNotFoundError as e:
            print(f"❌ 檔案不存在：{e}")
            raise

        except Exception as e:
            print(f"❌ 未預期的錯誤：{e}")
            print(f"   錯誤類型：{type(e).__name__}")
            print(f"   錯誤位置：{e.__traceback__.tb_lineno}")
            raise

3. 記錄詳細日誌

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LoggedMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        logger.info(f"開始載入價格資料：{trade_at_price}")

        try:
            df = pd.read_csv(f'data_{trade_at_price}.csv', index_col=0, parse_dates=True)
            logger.info(f"資料載入成功：{df.shape}")
            return df

        except Exception as e:
            logger.error(f"資料載入失敗：{e}", exc_info=True)
            raise

自訂市場物件

為什麼需要自訂市場?

快速上手:使用內建市場

內建市場特性對比

自訂市場類別

基礎範例:加密貨幣市場

進階範例:期貨市場(含槓桿)

Market 類別完整 API

必須實作的方法

get_price(trade_at_price, adj=True)

選用的靜態方法

get_name()

get_freq()

get_benchmark()

get_asset_id_to_name()

get_market_value()

get_industry()

market_close_at_timestamp(timestamp)

使用自訂資料源

1. 從 CSV 載入

2. 從 API 動態載入

3. 從 FinLab 資料庫混合載入

進階功能

1. 客製化交易價格計算

2. 支援多時區

3. 實作市場休市邏輯

測試自訂市場

常見問題

Q1: 為什麼需要 adj=True 參數?

Q2: 如何處理缺失值(NaN)?

Q3: 可以在 get_price() 中使用 finlab.data.get() 嗎?

Q4: 如何模擬交易成本?

Q5: 為什麼 get_benchmark() 是靜態方法?

實戰案例

案例 1: 黃金 ETF 回測

案例 2: 台股 + 美股混合回測

案例 3: 加密貨幣 4 小時 K 線回測

參考資源

常見錯誤與解決方法

錯誤 1：檔案讀取失敗

錯誤 2：資料格式錯誤

錯誤 3：API 請求失敗

錯誤 4：回測執行失敗

除錯技巧

1. 逐步測試市場物件

2. 使用 try-except 包裝關鍵步驟

3. 記錄詳細日誌

參考資源

`get_price(trade_at_price, adj=True)`

`get_name()`

`get_freq()`

`get_benchmark()`

`get_asset_id_to_name()`

`get_market_value()`

`get_industry()`

`market_close_at_timestamp(timestamp)`

Q1: 為什麼需要 `adj=True` 參數?

Q3: 可以在 `get_price()` 中使用 `finlab.data.get()` 嗎?

Q5: 為什麼 `get_benchmark()` 是靜態方法?