finlab.market

市場物件(Market)定義了回測系統如何取得價格資料、對照指數、市場特性等資訊。FinLab 內建支援台股、美股、興櫃,也可以透過繼承 Market 類別自訂市場。

市場物件的概念

每個市場有不同的特性:

• 價格資料來源: 從哪裡取得開盤價、收盤價、成交量等資料
• 對照指數: 用於績效比較的基準(如台股用加權指數、美股用 S&P 500)
• 交易時間: 開盤、收盤時間與時區
• 市場規則: 漲跌幅限制、交易成本、特殊股票類別

透過市場物件,回測引擎可以:

1. 取得正確的價格資料
2. 計算正確的報酬率
3. 顯示對照指數的績效比較
4. 支援實盤交易的時間判斷

內建市場對比

特性	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%	無	無
特殊股票類別	處置股/全額交割	無	無
資料來源	finlab.data	finlab.data	finlab.data

使用情境

1. 使用內建市場

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

# 台股回測(預設)
tw_close = data.get('price:收盤價')
tw_position = tw_close > tw_close.average(20)
report = backtest.sim(tw_position, resample='M')
# 等同於
report = backtest.sim(tw_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())

2. 自訂市場 - 加密貨幣

from finlab.market import Market
import pandas as pd

class CryptoMarket(Market):

    @staticmethod
    def get_name():
        return 'crypto'

    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

    @staticmethod
    def get_benchmark():
        df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
        return df['BTC']  # 使用 BTC 當對照

# 使用
report = backtest.sim(position, market=CryptoMarket())

3. 自訂市場 - 期貨(含槓桿)

class FuturesMarket(Market):

    def __init__(self, leverage=10):
        self.leverage = leverage
        self.prices = pd.read_csv('futures_price.csv', index_col=0, parse_dates=True)

    @staticmethod
    def get_name():
        return 'futures'

    def get_price(self, trade_at_price='close', adj=True):
        price = self.prices[trade_at_price]
        daily_return = price.pct_change() * self.leverage
        leveraged_price = (1 + daily_return).fillna(1).cumprod() * 100
        return leveraged_price

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

自訂市場的典型場景

場景	說明	範例
加密貨幣	24 小時交易,無漲跌幅限制	BTC, ETH, BNB
期貨	有槓桿效果,需模擬槓桿報酬	台指期、美股期貨
海外股票	不在 FinLab 資料庫的市場	日股、港股、陸股
自訂資料	從 CSV、API 載入價格	私有資料、模擬資料
混合市場	同時回測多個市場的組合	台股 + 美股 + 加密貨幣

詳細教學

參考 自訂市場物件完整指南，了解:

• Market 類別完整 API 說明
• 必須實作 vs 選用的方法
• 如何處理多時區
• 如何實作市場休市邏輯
• 從 CSV、API、FinLab 資料庫混合載入資料
• 3 個實戰案例(黃金 ETF、全球市場、加密貨幣 4H)

API Reference

finlab.market.Market

Bases: ABC

市場類別 假如希望開發新的交易市場套用到回測系統，可以繼承 finlab.market.Market 來實做新類別。

get_asset_id_to_name staticmethod

get_asset_id_to_name()

設定對標報酬率的時間序列 Returns: (dict): 股號與股名對照表，ex:{'2330':'台積電'}

get_benchmark staticmethod

get_benchmark()

設定對標報酬率的時間序列

這個函數用於設定對標報酬率的時間序列。

RETURNS	DESCRIPTION
Series	pd.Series: 時間序列的報酬率。

RAISES	DESCRIPTION
ExceptionType	Description of conditions under which the exception is raised.

Examples:

date	0050
2007-04-23	100
2007-04-24	100.1
2007-04-25	99
2007-04-26	98.3
2007-04-27	99.55

get_board_lot_size staticmethod

get_board_lot_size()

Returns the board lot size of the market.

RETURNS	DESCRIPTION
int	The board lot size of the market. TYPE: int

get_freq staticmethod

get_freq()

Returns the frequency of the data. Used to determine how to resample the data when the data is not daily. The freq will be saved in finlab.core.report.

Returns: str: The frequency of the data.

get_market_value staticmethod

get_market_value()

取得回測用市值數據

RETURNS	DESCRIPTION
DataFrame	市值數據，其中 index 為日期，而 columns 是股票代號。

get_name staticmethod

get_name()

Returns the name of the market data source.

This function is used to get the name of the market data source.

get_odd_lot staticmethod

get_odd_lot()

Returns the odd lot size of the market.

RETURNS	DESCRIPTION
int	The odd lot size of the market. TYPE: int

get_price abstractmethod

get_price(trade_at_price, adj=True)

取得回測用價格或成交量數據

PARAMETER	DESCRIPTION
trade_at_price	價格類型或自訂資料 - str: 可選 'open', 'close', 'high', 'low', 'volume' - pd.Series: 自訂價格序列（單一股票或時間序列） - pd.DataFrame: 自訂價格資料（多檔股票） TYPE: Union[str, Series, DataFrame]
adj	是否使用還原股價計算。僅對價格資料有效，volume 會忽略此參數。 TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
DataFrame	價格或成交量數據，index 為日期，columns 為股票代號。

Note

必須實作的 trade_at_price 值：

• 'open', 'close', 'high', 'low': 基本價格資料（必須實作）
• 用於 get_trading_price() 計算組合價格

• 用於回測的交易價格

• 'volume': 成交量資料（強烈建議實作）

• LiquidityAnalysis: 流動性分析需要計算成交金額
• ml/qlib.py: 機器學習特徵生成
• portfolio_sync_manager.py: 實時 portfolio 管理

• 若不實作，相關分析功能會失效

• pd.Series, pd.DataFrame: 自訂資料（建議支援）

• 用於彈性回測，允許傳入自訂價格資料
• Series 會自動轉換為單欄 DataFrame

Examples:

回傳格式範例：

date	0015	0050	0051	0052
2007-04-23	9.54	57.85	32.83	38.4
2007-04-24	9.54	58.1	32.99	38.65
2007-04-25	9.52	57.6	32.8	38.59
2007-04-26	9.59	57.7	32.8	38.6
2007-04-27	9.55	57.5	32.72	38.4

實作範例：

def get_price(self, trade_at_price, adj=True):
    # 處理自訂資料
    if isinstance(trade_at_price, pd.Series):
        return trade_at_price.to_frame()
    if isinstance(trade_at_price, pd.DataFrame):
        return trade_at_price

    # 處理字串
    if isinstance(trade_at_price, str):
        if trade_at_price == 'volume':
            return data.get('your_volume_data')  # 必須實作

        # 處理 open/close/high/low
        if adj:
            return data.get(f'adj_{trade_at_price}')
        else:
            return data.get(trade_at_price)

    raise ValueError(f'Unsupported trade_at_price: {trade_at_price}')

詳細實作可參考 finlab.markets.tw.TWMarket.get_price()

get_reference_price

get_reference_price()

Returns the most recent reference price of the market.

RETURNS	DESCRIPTION
Dict[str, float]	pandas.Series: The most recent reference price of the market.

get_trading_price

get_trading_price(name, adj=True)

取得回測用價格數據

PARAMETER	DESCRIPTION
name	選擇回測之還原股價以收盤價或開盤價計算，預設為'close'。可選 'open'、'close'、'high'、'low'、'open_close_avg'、'high_low_avg'、或 'price_avg'。 TYPE: str

Returns: (pd.DataFrame): 價格數據

market_close_at_timestamp

market_close_at_timestamp(timestamp=None)

Returns the timestamp of the market close of the given timestamp.

PARAMETER	DESCRIPTION
timestamp	The timestamp to find the market close to. TYPE: datetime DEFAULT: None

RETURNS	DESCRIPTION
datetime	The timestamp of the closest market close.

market_open_time staticmethod

market_open_time()

Returns (hour, minute) of market open in local timezone.

tzinfo staticmethod

tzinfo()

Returns the timezone of the market.

RETURNS	DESCRIPTION
Union[timezone, None]	datetime.timezone: The timezone of the market.

快速開發技巧

最簡單的自訂市場只需實作兩個方法:

1. get_name(): 回傳市場名稱
2. get_price(): 回傳價格 DataFrame

其他方法都是選用的,可根據需求實作。

常見錯誤

• get_price() 回傳的 DataFrame 必須 index 為 DatetimeIndex
• columns 為股票代號(str)
• 缺失值建議使用 forward fill 處理

常見問題

Q: 如何處理調整後價格(adj=True)?

def get_price(self, trade_at_price='close', adj=True):
    if adj:
        # 回傳考慮除權息的調整後價格
        return pd.read_csv('adj_close.csv', index_col=0, parse_dates=True)
    else:
        # 回傳原始收盤價
        return pd.read_csv('raw_close.csv', index_col=0, parse_dates=True)

Q: 為什麼需要 get_benchmark()?

對照指數用於績效比較,在 report.display() 時會顯示策略 vs 對照指數的報酬曲線。如果不需要對照,回傳空 Series 即可:

@staticmethod
def get_benchmark():
    return pd.Series([])  # 不顯示對照指數

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

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

from finlab import data

class MixedMarket(Market):
    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']]
        return pd.concat([tw_close, us_close], axis=1)

Q: 如何模擬交易成本?

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

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

參考資源

• 自訂市場物件完整指南
• 歷史回測詳細說明
• 完整策略開發流程
• TWMarket 原始碼
• USMarket 原始碼