自訂市場物件

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

為什麼需要自訂市場?

不同市場有不同的特性:

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

快速上手:使用內建市場

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('us_price: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()

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

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

Warning

以下價格乘槓桿範例每天重設曝險，不計每口費用、取整、保證金與轉倉。個股期請見第 1 階段口數引擎。

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(dtype=float, index=pd.DatetimeIndex([]))  # 預設為空(不顯示對照)

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 {}

get_delisting_notices()

回傳交易所公告的下市預告。回測會在公告日之後的第一個交易日賣出持股，之後的換股也不再買進。沒有公告的標的，若價格提早結束（至少 20 根 k 棒沒有報價），則在最後一筆報價出場。

@staticmethod
def get_delisting_notices() -> pd.DataFrame:
    """
    Returns:
        pd.DataFrame: 下市預告
            - index: 股票代號（同一代號可以有多筆）
            - announced: 公告日
            - delisted: 終止上市（櫃）日
    """
    return pd.DataFrame(
        {
            "announced": pd.Series(dtype="datetime64[ns]"),
            "delisted": pd.Series(dtype="datetime64[ns]"),
        }
    )

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('us_price: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   請檢查檔案權限設定"
            )

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

        except Exception as e:
            raise RuntimeError(f"❌ 讀取檔案失敗：{file_path}\n   錯誤訊息：{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   請確認檔案內容是否正確")

        # 檢查 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(
                    dtype=float, index=pd.DatetimeIndex([])
                )  # 回傳空 Series
            return df['BTC']
        except Exception as e:
            print(f"⚠️  警告：無法載入對照指數：{e}")
            return pd.Series(
                dtype=float, index=pd.DatetimeIndex([])
            )  # 回傳空 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'

常見格式問題：

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

解決方法：在 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   請確認欄位內容為數字"
                    )

        # 驗證 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   請稍後再試或升級 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   請查看詳細錯誤訊息")

        # 轉換為 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   請確認資料來源一致")

    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

參考資源

• finlab.market API Reference
• 歷史回測詳細說明
• 完整策略開發流程
• 常見問題 FAQ
• TWMarket 原始碼
• USMarket 原始碼

個股期口數回測：第 1 階段

TWStockFuturesMarket 使用獨立的 Python／NumPy 引擎，以新台幣權益與整數口數記帳。標準型一口 2,000 股，小型一口 100 股。position 沿用股票代號，但不會驗證歷史上是否有可交易的股期；請自行選擇適用標的。

import pandas as pd
from finlab import backtest
from finlab.markets.tw_stock_futures import TWStockFuturesMarket

# 可直接離線執行的合成範例；價格須為每日還原價。
dates = pd.bdate_range('2024-01-01', periods=5)
prices = pd.DataFrame({'2330': [100.0, 100.0, 105.0, 110.0, 110.0]}, index=dates)
position = pd.DataFrame({'2330': [1.0, 1.0, 1.0, 0.0, 0.0]}, index=dates)
market = TWStockFuturesMarket(
    contract='standard',
    initial_capital=250_000,
    leverage=1,
    fee_per_lot=20,  # 僅為示例：必須改成自己券商的每口單邊手續費。
    tax_ratio=2e-5,
    margin_ratio=0.135,  # 假設值；也可傳 {'2330': 0.162} 等逐檔對照。
)
report = backtest.sim(position, market=market, trade_at_price=prices, upload=False)
print(report.account)
print(report.transactions)
assert abs(report.account.equity.iloc[-1] - 269_951.6) < 1e-8
report.display()  # 本機報告；可用 report.to_html('futures.html') 匯出。

使用既有選股結果時，可省略 trade_at_price，引擎便讀取 etl:adj_close。月換股用 resample='M'，目標兩倍曝險用市場物件的 leverage=2；position 不用再乘槓桿。第 1 階段只支援每日收盤還原價或自訂每日價格 DataFrame，以及 eager pandas／FinlabDataFrame 部位。

也可用 from finlab import config; config.set_market(market) 設定同一個市場物件，此後省略 sim(market=...) 仍使用期貨口數引擎；顯式傳入的市場優先。config.reset_market() 恢復預設現股市場。這是回測市場設定，與資料集的 data.set_market(...) 分開。

換算、成交與記帳規則

• 權重絕對值總和大於 1 時按比例縮至 1；小於 1 時保留未配置額度，再套用 position_limit。布林等權訊號亦適用，負權重為空單。目標名目本金 = 再平衡前權益 × leverage × 權重。
• 目標口數 = 目標名目本金 ÷（成交還原價 × 契約股數），向零取整。買賣訊號一律在訊號日之後的第一個價格交易日成交，避免使用同日收盤訊號提前成交。訊號與價格索引均依其時區的交易日期正規化為午夜，13:30 等日頻收盤時間標記不改變 T 日訊號、T+1 成交規則；同一天多筆訊號或價格明確拒絕。resample=None 只在部位變更時調整；字串頻率（如 'M'、'D'）在日曆排定日期取最新已知訊號，下一個交易日成交。
• 兩次再平衡之間固定口數，逐日損益 = 持有口數 × 契約股數 × 價格變動，不會每天重設槓桿。保證金僅占用權益，沒有把契約名目本金當成買入支出，也沒有另外扣除保證金。
• 每次成交（包括進場、減碼、出場、反向）手續費 = 絕對成交口數 × fee_per_lot；期交稅 = 絕對成交口數 × 契約股數 × 成交價 × tax_ratio，買賣兩側都收。fee_per_lot 沒有預設費率，未提供會報錯，因為各券商每口單邊費率不同。預設期交稅十萬分之二的出處是財政部稅務入口網；期交所費用表說明券商實收費用由雙方議定。
• 保證金不足時，按比例縮減並取整，計入成交費用與稅後仍須能負擔保證金，並發出 FuturesSimulationWarning。縮減採保守的最大換手成本上界，不將取整後的零頭資金再配到某檔股票，可能保留更多現金。目標資金不足一口時為 0 口並警告。
• 費用與稅以浮點新台幣金額保存，不模擬券商的元以下取整規則。範例進場一口於 100 元：費用 20、稅 4；110 元出場：費用 20、稅 4.4；毛損益 20,000，淨損益 19,951.6。
• end_date 包含指定日期，開放部位不會為了結束回測而假設平倉。日期、股票代號須唯一且由舊到新排列；價格須正數且無缺值。不能估值的缺價會報錯，不以前值虛構成交。

報告與限制

沿用既有 Report 的回報統計（get_stats()）、報酬指標（report.metrics）及淨值序列；回傳的是其子類別 FuturesReport。沒有提供股票指數時，基準為持有現金。get_metrics() 只回傳可用統計、契約設定與缺少分析的清單，不套用現股的完整儀表板欄位。

權益一旦降至零或以下，即使後來恢復，整段回測的百分比報酬統計不再有效。get_stats()、report.metrics 的 Sharpe／CAGR／風險等比例指標、metrics_only=True 的同類指標，以及相關分析會拋出含首次日期的 ValueError；get_metrics()／to_json() 回傳空 stats、performance.valid=False、首次日期與原因，並把 percentage_return_statistics 列為不可用，HTML 也會標示原因。帳本與口數仍完整保留，不以截短期間隱藏虧損，也不假設已有追繳或斷頭模型；FIFO 的損益與進場名目本金報酬仍可核算。

to_html(path, title=None) 保持既有 Report 呼叫契約，也接受 path=... 與 title=...。本機 to_json() 的交易日期為 ISO 字串、未平倉日期為 null，非有限統計值亦為 null，可用既有 JSON serializer 或標準 json.dumps(..., allow_nan=False) 序列化；此摘要仍不包含完整股票儀表板或雲端 payload 欄位。

欄位	內容
creturn / daily_creturn	權益 ÷ 起始資金，包含成本；遇首筆訊號為假日，會保留成交前的現金觀察點
account	每日 NTD 權益、損益、手續費、稅、保證金、可用權益、名目本金與 margin_excess
lot_positions	每日逐檔實際口數（多單正、空單負）
transactions	每次成交日期、訊號日、股票、成交口數、價格、名目金額、費用與稅
trades	FIFO 已平倉／未平倉明細；未平倉以最後價格估值，exit_date 為 NaT；return 是扣除已收成本後損益 ÷ 進場名目本金，不是保證金報酬
weights / position	實際名目曝險 ÷ 權益，可能大於 1；權益非正時非零部位的權重為 NaN

第 1 階段用標的現股還原價代表期貨價格，忽略基差、換月跳空、到期轉倉與其成本，也沒有真實結算價、歷史股期標的名單／保證金、除權息契約調整、盤後交易、滑價、漲跌停與流動性模型。margin_ratio 是使用者假設的原始保證金比例，預設 0.135 不是歷史級距或維持保證金。持有期間保證金超過權益的日期記錄於 account.margin_excess 並警告，但不模擬追繳、斷頭或強制平倉；極端損失可能使權益非正。

沒有 MAE／MFE、現股流動性與實單同步欄位；get_mae_mfe()、相關分析與實單 execution 介面明確報錯。公開 ReportExecution.from_report(report)／建構子以及 Position.from_report(report, ...) 也會在價格讀取前拒絕，不會產生可下現股單的 CASH 部位；市場的現股交易時間與整張股數介面同樣拒絕，契約乘數請讀 contract_multiplier。display()／to_html() 提供本機淨值、口數與成本表，雲端上傳尚未支援。stop_loss、take_profit、trail_stop、trail_stop_activation、touched_exit、MAE／MFE 視窗、成本保留、通知等非預設設定會報錯，不能默默略過。sim(fee_ratio=...)／sim(tax_ratio=...) 亦報錯，請在市場物件設定費用與期交稅。

後續階段依序補到期日曆與轉倉成本／價差、真實近遠月價格、逐檔歷史保證金與契約調整；本階段不做這些假設的真實驗證。