Custom Market Objects

FinLab supports Taiwan stocks (TWMarket), US stocks (USMarket), and Emerging Market (ROTCMarket) by default. If you want to backtest cryptocurrencies, futures, international stocks, or use custom price data, you can do so by inheriting the Market class.

Why Custom Markets?

Different markets have different characteristics:

• Different trading hours: US and Taiwan markets have different open/close times
• Different price fields: Cryptocurrencies may not have a traditional "open price" concept
• Different data sources: Load prices from CSV, API, or database
• Different trading rules: Futures have leverage; cryptocurrencies have no price limits
• Different benchmarks: Taiwan stocks benchmark against TAIEX; US stocks against S&P 500

By creating custom Market objects, you can apply any market's data to FinLab's backtesting engine.

Quick Start: Using Built-in Markets

FinLab provides 3 built-in markets:

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

# Taiwan market (default)
close = data.get('price:收盤價')
position = close > close.average(20)
report = backtest.sim(position, resample='M')
# Equivalent to:
report = backtest.sim(position, resample='M', market=TWMarket())

# US market
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())

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

Built-in Market Comparison

Feature	TWMarket	USMarket	ROTCMarket
Market name	'tw_stock'	'us_stock'	'rotc_stock'
Data frequency	Daily ('1d')	Daily ('1d')	Daily ('1d')
Benchmark index	TAIEX	S&P 500	(None)
Timezone	Asia/Taipei	US/Eastern	Asia/Taipei
Market close	15:00	16:00	14:00
Price limits	10%	None	None
Special stock types	Disposition/Full-delivery	None	None

---

Custom Market Class

Basic Example: Cryptocurrency Market

Suppose you have daily close prices for Bitcoin (BTC) and Ethereum (ETH) in CSV files:

from finlab.market import Market
import pandas as pd

class CryptoMarket(Market):

    @staticmethod
    def get_name():
        """Return market name"""
        return 'crypto'

    @staticmethod
    def get_freq():
        """Return data frequency"""
        return '1d'  # Daily

    def get_price(self, trade_at_price='close', adj=True):
        """Get price data

        Args:
            trade_at_price: One of 'open', 'close', 'high', 'low'
            adj: Whether to use adjusted prices (usually not needed for crypto)

        Returns:
            pd.DataFrame: index is dates, columns are coin symbols
        """
        # Load prices from CSV
        df = pd.read_csv(f'crypto_{trade_at_price}.csv', index_col=0, parse_dates=True)
        return df

    @staticmethod
    def get_benchmark():
        """Benchmark index (e.g., use BTC as benchmark)"""
        df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
        return df['BTC']

# Usage
from finlab.backtest import sim

# Assuming CSV format:
# 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()

Advanced Example: Futures Market (with Leverage)

Futures have leverage, which can be simulated by modifying prices:

class FuturesMarket(Market):

    def __init__(self, leverage=10):
        """
        Args:
            leverage: Leverage multiplier
        """
        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):
        """Return leveraged returns"""
        price = self.prices[trade_at_price]

        # Calculate daily returns multiplied by leverage
        daily_return = price.pct_change() * self.leverage

        # Convert back to price (cumulative from 100)
        leveraged_price = (1 + daily_return).fillna(1).cumprod() * 100

        return leveraged_price

    @staticmethod
    def get_benchmark():
        # Use 1x leverage as benchmark
        df = pd.read_csv('futures_price.csv', index_col=0, parse_dates=True)
        return df['close'].iloc[:, 0]

# Backtest with 10x leverage
report = sim(position, market=FuturesMarket(leverage=10))

---

Market Class Complete API

When inheriting the Market class, you can override the following methods:

Required Methods

get_price(trade_at_price, adj=True)

The most important method -- returns price data.

def get_price(self, trade_at_price='close', adj=True) -> pd.DataFrame:
    """
    Args:
        trade_at_price (str): One of 'open', 'close', 'high', 'low', 'volume'
        adj (bool): Whether to return adjusted prices (accounting for splits, dividends, etc.)

    Returns:
        pd.DataFrame:
            - index: Dates (DatetimeIndex)
            - columns: Stock/asset symbols
            - values: Prices

    Examples:
        Return format example:

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

Optional Static Methods

get_name()

Returns the market name, used for identifying the market type.

@staticmethod
def get_name() -> str:
    return 'crypto'  # Default is 'auto'

get_freq()

Returns data frequency, affecting return calculations.

@staticmethod
def get_freq() -> str:
    return '1d'   # Daily
    # Other common: '1h' (hourly), '4h', '1w' (weekly)

get_benchmark()

Returns the benchmark index for performance comparison.

@staticmethod
def get_benchmark() -> pd.Series:
    """
    Returns:
        pd.Series: Benchmark index price series
            - index: Dates
            - values: Index prices
    """
    return pd.Series([])  # Default is empty (no benchmark displayed)

get_asset_id_to_name()

Returns a mapping from asset ID to name, used in report display.

@staticmethod
def get_asset_id_to_name() -> dict:
    """
    Returns:
        dict: asset_id -> asset_name mapping

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

get_market_value()

Returns market capitalization data for market-cap-weighted backtesting.

@staticmethod
def get_market_value() -> pd.DataFrame:
    """
    Returns:
        pd.DataFrame: Market cap data
            - index: Dates
            - columns: Stock symbols
            - values: Market cap values
    """
    return pd.DataFrame()

get_industry()

Returns industry classification for industry analysis.

@staticmethod
def get_industry() -> dict:
    """
    Returns:
        dict: asset_id -> industry mapping

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

market_close_at_timestamp(timestamp)

Returns the market close time for a given timestamp, used in live trading.

def market_close_at_timestamp(self, timestamp=None) -> pd.Timestamp:
    """
    Args:
        timestamp: Query time point (defaults to latest)

    Returns:
        pd.Timestamp: Most recent market close time
    """
    # Crypto trades 24 hours; use 23:59 as close
    if timestamp is None:
        timestamp = pd.Timestamp.now()
    return pd.Timestamp(timestamp.date()) + pd.Timedelta('23:59:00')

---

Using Custom Data Sources

1. Loading from CSV

The simplest approach, suitable for one-off data:

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

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

2. Dynamic Loading from API

Suitable for data that needs real-time updates:

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):
        # Get JSON data from API
        response = requests.get(f'{self.api_url}/{trade_at_price}')
        data = response.json()

        # Convert to DataFrame
        df = pd.DataFrame(data)
        df['date'] = pd.to_datetime(df['date'])
        df = df.set_index('date')

        return df

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

3. Hybrid Loading from FinLab Database

Combine FinLab data with custom data:

from finlab import data

class HybridMarket(Market):

    @staticmethod
    def get_name():
        return 'hybrid'

    def get_price(self, trade_at_price='close', adj=True):
        # Get Taiwan stock prices from FinLab
        tw_close = data.get('price:收盤價')

        # Load custom cryptocurrency prices
        crypto_close = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)

        # Merge both DataFrames (outer join)
        combined = pd.concat([tw_close, crypto_close], axis=1)

        return combined

# Usage (backtest Taiwan stocks + crypto together)
report = sim(position, market=HybridMarket())

---

Advanced Features

1. Custom Trading Price Calculation

Override get_trading_price() to customize execution prices:

class CustomPriceMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        # ... return basic prices

    def get_trading_price(self, name, adj=True):
        """Custom execution price calculation"""
        if name == 'vwap':  # Volume-weighted average price
            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:
            # Use default logic for other cases
            return super().get_trading_price(name, adj=adj)

# Use VWAP as execution price
report = sim(position, market=CustomPriceMarket(), trade_at='vwap')

2. Multi-timezone Support

Handle cross-timezone markets:

import datetime

class MultiTimezoneMarket(Market):

    @staticmethod
    def get_name():
        return 'multi_tz'

    def market_close_at_timestamp(self, timestamp=None):
        """Different markets have different close times"""
        if timestamp is None:
            timestamp = pd.Timestamp.now()

        # For a global 24-hour market, use 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. Implementing Market Holidays

Exclude weekends and holidays:

class WithHolidaysMarket(Market):

    def __init__(self):
        # Define holidays (e.g., Taiwan national holidays)
        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)

        # Remove weekends
        df = df[df.index.dayofweek < 5]

        # Remove holidays
        df = df[~df.index.isin(self.holidays)]

        return df

---

Testing Custom Markets

After building a market object, always test it:

# 1. Check price data format
market = CryptoMarket()
close = market.get_price('close')
print(close.head())
print(close.index)   # Should be DatetimeIndex
print(close.shape)   # (num_dates, num_stocks)

# 2. Check benchmark index
benchmark = market.get_benchmark()
print(benchmark.head())

# 3. Run a simple backtest
position = close > close.average(20)
report = sim(position, market=market, resample='M', upload=False)
report.display()

# 4. Check report's market name
print(report.market.get_name())  # Should be 'crypto'

---

FAQ

Q1: Why is the adj=True parameter needed?

adj=True means using adjusted prices (accounting for stock splits, dividends, etc.):

• Taiwan stocks: Historical prices are adjusted after ex-rights/ex-dividends to ensure correct return calculations
• US stocks: Also require price adjustment
• Cryptocurrency: No ex-rights/dividends; the adj parameter can be ignored

# Taiwan stock example
tw_market = TWMarket()
adj_close = tw_market.get_price('close', adj=True)    # Adjusted price
raw_close = tw_market.get_price('close', adj=False)   # Raw closing price

Q2: How to handle missing values (NaN)?

FinLab automatically forward-fills missing values, but it is recommended to handle them in get_price():

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

    # Method 1: Forward fill
    df = df.ffill()

    # Method 2: Remove stocks with missing values
    df = df.dropna(axis=1, how='any')

    # Method 3: Fill with 0 (not recommended)
    df = df.fillna(0)

    return df

Q3: Can I use finlab.data.get() inside get_price()?

Yes! This is the standard approach for mixing FinLab data:

from finlab import data

class MixedMarket(Market):

    def get_price(self, trade_at_price='close', adj=True):
        # Use FinLab Taiwan stock data
        tw_close = data.get('price:收盤價')

        # Filter specific stocks
        tw_close = tw_close[['2330', '2317', '2454']]

        return tw_close

Q4: How to simulate transaction costs?

Transaction costs are set in the sim() function, not in the Market object:

# Crypto typically has lower fees and no tax
report = sim(
    position,
    market=CryptoMarket(),
    fee_ratio=0.001,  # 0.1% commission
    tax_ratio=0       # No transaction tax
)

# Taiwan stock defaults: fee_ratio=0.001425, tax_ratio=0.003

Q5: Why is get_benchmark() a static method?

Because the benchmark index is usually fixed and does not depend on instance variables. But if you need a dynamic benchmark:

class DynamicBenchmarkMarket(Market):

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

    def get_benchmark(self):  # Changed to instance method (removed @staticmethod)
        df = pd.read_csv('benchmarks.csv', index_col=0, parse_dates=True)
        return df[self.benchmark_ticker]

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

---

Practical Examples

Example 1: Gold ETF Backtest

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):
        # Load Gold ETF (GLD) prices from Yahoo Finance
        import yfinance as yf
        gold = yf.download('GLD', start='2010-01-01')
        return gold[[trade_at_price.capitalize()]]

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

# Usage
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')

Example 2: Taiwan + US Mixed Backtest

class GlobalMarket(Market):

    @staticmethod
    def get_name():
        return 'global'

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

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

        # Merge (outer join, auto-aligns dates)
        combined = pd.concat([tw_close, us_close], axis=1)

        return combined

    @staticmethod
    def get_benchmark():
        # Use MSCI World Index or 60/40 TW/US combination
        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

# Usage
position = ...  # Position including TW + US stocks
report = sim(position, market=GlobalMarket())

Example 3: Cryptocurrency 4-Hour Candle Backtest

class Crypto4HMarket(Market):

    @staticmethod
    def get_name():
        return 'crypto_4h'

    @staticmethod
    def get_freq():
        return '4h'  # 4-hour frequency

    def get_price(self, trade_at_price='close', adj=True):
        # Load 4-hour candle data from Binance API
        df = pd.read_csv('binance_4h_close.csv', index_col=0, parse_dates=True)
        return df

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

# Note: resample parameter should match the frequency
report = sim(position, market=Crypto4HMarket(), resample='1d')  # Daily rebalancing

---

Debugging Tips

1. Step-by-Step Market Object Testing

# Step 1: Test basic methods
market = CryptoMarket()
print(f"Market name: {market.get_name()}")
print(f"Data frequency: {market.get_freq()}")

# Step 2: Test price loading
close = market.get_price('close')
print(f"\nClose price data:\n{close.head()}")

# Step 3: Test benchmark index
benchmark = market.get_benchmark()
print(f"\nBenchmark index:\n{benchmark.head()}")

# Step 4: Run simple strategy
position = close > close.rolling(20).mean()
print(f"\nPosition signals:\n{position.tail()}")

# Step 5: Small-scale backtest
position_small = position.iloc[-100:]  # Only last 100 days
report = sim(position_small, market=market, resample='M', upload=False)
report.display()

2. Wrap Key Steps with try-except

class SafeMarket(Market):

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

        except FileNotFoundError as e:
            print(f"File not found: {e}")
            raise

        except Exception as e:
            print(f"Unexpected error: {e}")
            print(f"   Error type: {type(e).__name__}")
            raise

3. Detailed Logging

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"Loading price data: {trade_at_price}")

        try:
            df = pd.read_csv(f'data_{trade_at_price}.csv', index_col=0, parse_dates=True)
            logger.info(f"Data loaded successfully: {df.shape}")
            return df

        except Exception as e:
            logger.error(f"Data loading failed: {e}", exc_info=True)
            raise

Reference Resources

• finlab.market API Reference
• Backtesting Details
• Complete Strategy Development Workflow
• FAQ
• TWMarket Source Code
• USMarket Source Code