Custom markets

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()

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

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

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

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

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

@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()

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

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

@staticmethod
def get_delisting_notices() -> pd.DataFrame:
    """
    Returns:
        pd.DataFrame: delisting notices
            - index: symbol (a symbol may have several rows)
            - announced: announcement date
            - delisted: delisting date
    """
    return pd.DataFrame(
        {
            "announced": pd.Series(dtype="datetime64[ns]"),
            "delisted": pd.Series(dtype="datetime64[ns]"),
        }
    )

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

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

class CryptoMarket(Market):
    def __init__(self, data_dir='./data'):
        """
        Args:
            data_dir: Data directory path (default: ./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):
        """Fetch price data (with error handling)."""

        # Build the full path
        file_path = os.path.join(self.data_dir, f'crypto_{trade_at_price}.csv')

        # Check whether the file exists
        if not os.path.exists(file_path):
            raise FileNotFoundError(
                f"Data file not found: {file_path}\n"
                f"   Please verify:\n"
                f"   1. The file exists in the {self.data_dir} directory\n"
                f"   2. The filename is correct (crypto_{trade_at_price}.csv)\n"
                f"   3. The working directory is correct (current: {os.getcwd()})\n"
                f"   Tip: sample data can be downloaded from https://example.com/data"
            )

        # Try to read the file
        try:
            df = pd.read_csv(file_path, index_col=0, parse_dates=True)

        except PermissionError:
            raise PermissionError(
                f"No permission to read file: {file_path}\n"
                f"   Please check the file permission settings"
            )

        except pd.errors.EmptyDataError:
            raise ValueError(
                f"Data file is empty: {file_path}\n   Please verify the file contents"
            )

        except Exception as e:
            raise RuntimeError(
                f"Failed to read file: {file_path}\n   Error message: {e}"
            )

        # Validate data integrity (next step, see Error 2)
        df = self._validate_dataframe(df, file_path)

        return df

    def _validate_dataframe(self, df, file_path):
        """Validate DataFrame format."""

        # Check whether it is empty
        if df.empty:
            raise ValueError(
                f"Data is empty: {file_path}\n   Please verify the file contents"
            )

        # Check whether the index is a DatetimeIndex
        if not isinstance(df.index, pd.DatetimeIndex):
            raise TypeError(
                f"Index must be a DatetimeIndex (date format)\n"
                f"   Current index type: {type(df.index)}\n"
                f"   Please ensure the first CSV column is a date (e.g. 2020-01-01)\n"
                f"   and use the parse_dates=True argument"
            )

        # Check whether there is any numeric data
        if df.shape[1] == 0:
            raise ValueError(
                f"No stock columns found: {file_path}\n"
                f"   Please ensure the CSV format is correct (first column dates, other columns ticker symbols)"
            )

        print(f"Data loaded successfully: {file_path}")
        print(f"   Date range: {df.index[0]} ~ {df.index[-1]}")
        print(f"   Number of stocks: {df.shape[1]}")
        print(f"   Number of trading days: {df.shape[0]}")

        return df

    @staticmethod
    def get_benchmark():
        """Benchmark index (with error handling)."""
        try:
            df = pd.read_csv('crypto_close.csv', index_col=0, parse_dates=True)
            if 'BTC' not in df.columns:
                print("Warning: BTC column not found, benchmark index cannot be set")
                return pd.Series(
                    dtype=float, index=pd.DatetimeIndex([])
                )  # Return an empty Series
            return df['BTC']
        except Exception as e:
            print(f"Warning: failed to load benchmark index: {e}")
            return pd.Series(
                []
            )  # Return an empty Series so the main workflow is unaffected

# Usage: specify the data directory
try:
    market = CryptoMarket(data_dir='/path/to/crypto/data')
    close = market.get_price('close')
    print("Market object created successfully")

except FileNotFoundError as e:
    print(f"\n{e}")
    print("\nPlease download sample data or fix the file path")

except Exception as e:
    print(f"\nFailed to create market object: {e}")

import numpy as np
import pandas as pd
from finlab.backtest import sim
from finlab.market import Market

dates = pd.bdate_range('2024-01-01', periods=60)
prices = pd.DataFrame(
    {'BTC': np.linspace(100, 160, 60), 'ETH': np.linspace(50, 40, 60)}, index=dates
)

class StringDateMarket(Market):
    def get_price(self, trade_at_price='close', adj=True):
        df = prices.copy()
        df.index = df.index.strftime('%Y-%m-%d')  # dates as strings
        return df

position = prices > prices.rolling(5).mean()
try:
    sim(position, market=StringDateMarket(), upload=False)
except TypeError as e:
    print(e)  # operation 'sub' not supported for dtype 'str' with dtype 'str'

class RobustMarket(Market):
    def get_price(self, trade_at_price='close', adj=True):
        """Fetch price data (with full validation)."""

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

        # Validation 1: convert index to DatetimeIndex
        if not isinstance(df.index, pd.DatetimeIndex):
            try:
                df.index = pd.to_datetime(df.index)
                print("Index converted to DatetimeIndex")
            except Exception as e:
                raise TypeError(
                    f"Cannot convert index to DatetimeIndex\n"
                    f"   Index sample: {df.index[:3].tolist()}\n"
                    f"   Error message: {e}\n"
                    f"   Please ensure the first column is a date (e.g. 2020-01-01 or 2020/01/01)"
                )

        # Validation 2: check data types
        non_numeric_cols = df.select_dtypes(exclude=['number']).columns.tolist()
        if non_numeric_cols:
            print(
                f"Warning: the following columns are non-numeric and will be converted: {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"Cannot convert column {col} to numeric\n"
                        f"   Please ensure the column contains numbers"
                    )

        # Validation 3: check missing-value ratio
        missing_ratio = df.isna().sum() / len(df)
        high_missing_cols = missing_ratio[missing_ratio > 0.3].index.tolist()

        if high_missing_cols:
            print(f"Warning: the following columns have > 30% missing values:")
            for col in high_missing_cols:
                print(f"     {col}: {missing_ratio[col]:.1%} missing")
            print("   Suggestions:")
            print("   1. Drop these columns (df.drop(columns=high_missing_cols))")
            print("   2. Forward-fill the gaps (df.ffill())")

            # Auto-handle: forward fill
            df = df.ffill()
            print("   Missing values have been forward-filled automatically")

        # Validation 4: check data size
        if len(df) < 100:
            print(f"Warning: too few data points ({len(df)} rows)")
            print("   At least 252 trading days (about 1 year) is recommended")

        if df.shape[1] < 2:
            print(f"Warning: too few stocks ({df.shape[1]})")
            print("   At least 10 stocks are recommended for diversification")

        # Validation 5: sort the index
        if not df.index.is_monotonic_increasing:
            print("Warning: dates are unsorted, sorting automatically")
            df = df.sort_index()

        # Validation 6: remove duplicate dates
        if df.index.duplicated().any():
            dup_dates = df.index[df.index.duplicated()].tolist()
            print(f"Warning: duplicate dates found: {dup_dates[:5]}...")
            df = df[~df.index.duplicated(keep='last')]
            print("   Keeping the last record only")

        print(f"\nData validation complete")
        print(f"   Date range: {df.index[0].date()} ~ {df.index[-1].date()}")
        print(f"   Trading days: {len(df)}")
        print(f"   Number of stocks: {df.shape[1]}")
        print(
            f"   Missing values: {df.isna().sum().sum()} ({df.isna().sum().sum() / df.size:.2%})"
        )

        return df

# Usage
try:
    market = RobustMarket()
    close = market.get_price('close')

    # Run backtest
    position = close > close.rolling(20).mean()
    report = sim(position, market=market, resample='M')
    print("\nBacktest succeeded")

except (TypeError, ValueError) as e:
    print(f"\n{e}")
    print("\nPlease fix the data format and retry")

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 base URL
            api_key: API key (optional)
            max_retries: Maximum number of retries
        """
        self.api_url = api_url
        self.api_key = api_key
        self.max_retries = max_retries

        # Build a Session with retry logic
        self.session = self._create_session()

    def _create_session(self):
        """Create a requests Session with retry logic."""
        session = requests.Session()

        # Configure retry strategy
        retry_strategy = Retry(
            total=self.max_retries,
            backoff_factor=1,  # Exponential backoff: 1s, 2s, 4s
            status_forcelist=[429, 500, 502, 503, 504],  # HTTP status codes to retry on
            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):
        """Fetch price data from an API (with error handling)."""

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

        # Set request headers
        headers = {}
        if self.api_key:
            headers['Authorization'] = f'Bearer {self.api_key}'

        print(f"Downloading data from API: {url}")

        try:
            # Send the request (with a timeout)
            response = self.session.get(
                url,
                headers=headers,
                timeout=30,  # 30-second timeout
            )

            # Check the HTTP status code
            if response.status_code == 401:
                raise PermissionError(
                    f"API authentication failed (HTTP 401)\n"
                    f"   Please verify the API key is correct and not expired\n"
                    f"   Re-issue a key from the API provider if needed"
                )

            elif response.status_code == 429:
                raise RuntimeError(
                    f"API rate limit exceeded (HTTP 429)\n"
                    f"   Please retry later or upgrade your API plan"
                )

            elif response.status_code >= 500:
                raise RuntimeError(
                    f"API server error (HTTP {response.status_code})\n"
                    f"   The server is temporarily unavailable, please retry later"
                )

            response.raise_for_status()  # Raise any other HTTP errors

            # Parse JSON
            data = response.json()

        except requests.exceptions.Timeout:
            raise TimeoutError(
                f"API request timed out (> 30 seconds)\n"
                f"   Network is too slow or the API server is overloaded\n"
                f"   Suggestions:\n"
                f"   1. Check network connectivity\n"
                f"   2. Reduce the data range\n"
                f"   3. Retry later"
            )

        except requests.exceptions.ConnectionError:
            raise ConnectionError(
                f"Network connection failed\n"
                f"   Unable to connect to {self.api_url}\n"
                f"   Please check:\n"
                f"   1. Network connectivity\n"
                f"   2. API URL is correct\n"
                f"   3. Firewall is not blocking the connection"
            )

        except requests.exceptions.JSONDecodeError:
            raise ValueError(
                f"API returned an invalid format\n"
                f"   Expected JSON, but received: {response.text[:200]}...\n"
                f"   Please contact the API provider to confirm the documentation"
            )

        except Exception as e:
            raise RuntimeError(
                f"API request failed: {e}\n   Please inspect the detailed error message"
            )

        # Convert to DataFrame
        try:
            df = pd.DataFrame(data)

            # Handle common formats
            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 data downloaded successfully")
            print(f"   Date range: {df.index[0]} ~ {df.index[-1]}")
            print(f"   Number of stocks: {df.shape[1]}")

            return df

        except Exception as e:
            raise ValueError(
                f"Data conversion failed: {e}\n"
                f"   API response: {data}\n"
                f"   Please verify the data format documented by the API"
            )

# Usage
import os

try:
    # Read the API key from an environment variable (recommended)
    api_key = os.environ.get('CRYPTO_API_KEY')
    if not api_key:
        print("Warning: CRYPTO_API_KEY environment variable is not set")
        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("Market data fetched successfully")

except (ConnectionError, TimeoutError, PermissionError) as e:
    print(f"\n{e}")
    print("\nPlease fix the issue and retry")

except Exception as e:
    print(f"\nUnexpected error: {e}")

def validate_before_backtest(position, market):
    """Validate compatibility between position and market before backtesting."""

    print("=== Pre-backtest validation ===\n")

    # 1. Fetch market prices
    try:
        close = market.get_price('close')
    except Exception as e:
        raise RuntimeError(f"Unable to fetch market prices: {e}")

    # 2. Check date ranges
    position_dates = position.index
    price_dates = close.index

    print(f"Position date range: {position_dates[0]} ~ {position_dates[-1]}")
    print(f"Price date range: {price_dates[0]} ~ {price_dates[-1]}")

    # Intersection
    common_dates = position_dates.intersection(price_dates)
    if len(common_dates) == 0:
        raise ValueError(
            f"Position and Price share no overlapping dates\n"
            f"   Please ensure the data sources are consistent"
        )

    if len(common_dates) < len(position_dates) * 0.8:
        print(
            f"Warning: only {len(common_dates) / len(position_dates):.1%} of position dates have matching prices"
        )
        print("   Some trading days may be ignored")

    # 3. Check ticker symbols
    position_stocks = position.columns.tolist()
    price_stocks = close.columns.tolist()

    print(f"\nPosition stock count: {len(position_stocks)}")
    print(f"Price stock count: {len(price_stocks)}")

    # Intersection
    common_stocks = set(position_stocks).intersection(set(price_stocks))
    if len(common_stocks) == 0:
        raise ValueError(
            f"Position and Price share no overlapping tickers\n"
            f"   Position sample: {position_stocks[:5]}\n"
            f"   Price sample: {price_stocks[:5]}\n"
            f"   Please ensure the ticker symbol formats match"
        )

    missing_stocks = set(position_stocks) - set(price_stocks)
    if missing_stocks:
        print(f"Warning: {len(missing_stocks)} tickers are not present in Price:")
        print(f"   {list(missing_stocks)[:10]}...")
        print("   These tickers will be ignored")

    # 4. Check market object methods
    required_methods = ['get_name', 'get_freq', 'get_price']
    for method in required_methods:
        if not hasattr(market, method):
            raise AttributeError(
                f"Market object is missing a required method: {method}\n"
                f"   Please ensure the Market class is subclassed and all methods are implemented"
            )

    print(
        f"\nValidation passed: {len(common_stocks)} stocks, {len(common_dates)} trading days"
    )
    print("=" * 50)

# Usage: check a U.S. strategy against USMarket (or your own market object)
from finlab.markets.us import USMarket

try:
    market = USMarket()
    data.set_market('us')
    with data.us_universe(index='SP500'):
        close = data.get('price:adj_close')
    position = close > close.rolling(20).mean()

    # Run validation
    validate_before_backtest(position, market)

    # Run backtest
    report = sim(position, market=market, resample='M', upload=False)
    print("\nBacktest succeeded")
    report.display()

except Exception as e:
    print(f"\nBacktest failed: {e}")
    print("Please fix the issue and retry")

# CryptoMarket is the CSV market from the first example; it reads crypto_close.csv

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

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

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