FinancialData.Net adapter for Backtesting.py (daily and intraday data)

[financialdatanet]

This script allows you to instantly import data for stocks, ETFs, indices, forex, crypto, commodities, and OTC markets directly into a backtesting.py environment. It supports both EOD (daily) and intraday (minute) data. API key for FinancialData.Net is required, get one at https://financialdata.net/.

The Script

import datetime as dt
from dataclasses import dataclass, asdict
import pandas as pd

try:
    from fdnpy import FinancialDataClient
except ImportError as exc:
    FinancialDataClient = None
    _FDNPY_IMPORT_ERROR = exc
else:
    _FDNPY_IMPORT_ERROR = None


@dataclass(frozen=True)
class Bar:
    dt: dt.datetime
    Open: float
    High: float
    Low: float
    Close: float
    Volume: float
    OpenInterest: float = 0.0


class FDNBacktestingAdapter:
    DAILY_METHODS = {
        'stock': 'get_stock_prices',
        'international_stock': 'get_international_stock_prices',
        'etf': 'get_etf_prices',
        'commodity': 'get_commodity_prices',
        'otc': 'get_otc_prices',
        'crypto': 'get_crypto_prices',
        'forex': 'get_forex_prices',
    }

    MINUTE_METHODS = {
        'stock': 'get_minute_prices',
        'etf': 'get_minute_prices',
        'crypto': 'get_crypto_minute_prices',
        'forex': 'get_forex_minute_prices',
    }

    def __init__(self, api_key):
        if FinancialDataClient is None:
            raise ImportError("fdnpy is required. Install it with 'pip install fdnpy'.")
        self.client = FinancialDataClient(api_key=api_key)

    def get_data(self, symbol, asset_class='stock', timeframe='days', fromdate=None, todate=None):
        return self.load_bars(
            identifier=symbol,
            asset_class=asset_class,
            timeframe=timeframe,
            fromdate=fromdate,
            todate=todate,
        )

    def load_bars(self, identifier, asset_class, timeframe, fromdate=None, todate=None):
        asset_class = asset_class.lower().strip()
        fromdate = self._normalize_dt(fromdate)
        todate = self._normalize_dt(todate)

        # timeframe handles 'days' or 'minutes' as strings for kernc compatibility
        if timeframe.lower() == 'days':
            method_name = self.DAILY_METHODS.get(asset_class)
            if method_name is None:
                raise ValueError('Unsupported daily asset_class: %r' % asset_class)
            rows = getattr(self.client, method_name)(identifier=identifier)
            bars = [self._parse_daily_row(row) for row in rows]

        elif timeframe.lower() == 'minutes':
            method_name = self.MINUTE_METHODS.get(asset_class)
            if method_name is None:
                raise ValueError('Unsupported minute asset_class: %r' % asset_class)
            if fromdate is None and todate is None:
                raise ValueError('Historical minute data requires fromdate and/or todate')
            
            start = fromdate or todate
            end = todate or fromdate
            start_date, end_date = min(start.date(), end.date()), max(start.date(), end.date())
            
            bars = []
            current_date = start_date
            one_day = dt.timedelta(days=1)
            fetch = getattr(self.client, method_name)
            while current_date <= end_date:
                rows = fetch(identifier=identifier, date=current_date.isoformat())
                bars.extend(self._parse_minute_row(row) for row in rows)
                current_date += one_day
        else:
            raise ValueError('Only "days" and "minutes" timeframes are supported.')

        return self._finalize_bars(bars, fromdate=fromdate, todate=todate)

    @staticmethod
    def _parse_daily_row(row):
        return Bar(
            dt=dt.datetime.strptime(row['date'], '%Y-%m-%d'),
            Open=row['open'],
            High=row['high'],
            Low=row['low'],
            Close=row['close'],
            Volume=row['volume'],
        )

    @staticmethod
    def _parse_minute_row(row):
        return Bar(
            dt=dt.datetime.strptime(row['time'], '%Y-%m-%d %H:%M:%S'),
            Open=row['open'],
            High=row['high'],
            Low=row['low'],
            Close=row['close'],
            Volume=row['volume'],
        )

    @staticmethod
    def _normalize_dt(value):
        if value is None or value.tzinfo is None:
            return value
        return value.astimezone(dt.timezone.utc).replace(tzinfo=None)

    @classmethod
    def _finalize_bars(cls, bars, fromdate=None, todate=None):
        unique_bars = {}
        for bar in bars:
            n_dt = cls._normalize_dt(bar.dt) or bar.dt
            unique_bars[n_dt] = Bar(
                dt=n_dt, Open=bar.Open, High=bar.High, Low=bar.Low, 
                Close=bar.Close, Volume=bar.Volume, OpenInterest=bar.OpenInterest
            )

        ordered_bars = [unique_bars[k] for k in sorted(unique_bars)]
        if fromdate: ordered_bars = [b for b in ordered_bars if b.dt >= fromdate]
        if todate: ordered_bars = [b for b in ordered_bars if b.dt <= todate]

        df = pd.DataFrame([asdict(b) for b in ordered_bars])
        if not df.empty:
            df = df.set_index('dt')
            df.index.name = None
        return df

Minimal Example

from backtesting import Backtest, Strategy
from backtesting.lib import crossover
from backtesting.test import SMA

class SmaCross(Strategy):
    def init(self):
        # Define 10 and 20 period moving averages
        self.sma1 = self.I(SMA, self.data.Close, 10)
        self.sma2 = self.I(SMA, self.data.Close, 20)

    def next(self):
        if crossover(self.sma1, self.sma2):
            self.buy()
        elif crossover(self.sma2, self.sma1):
            self.position.close()

# 1. Fetch data from FinancialDataNet
adapter = FDNBacktestingAdapter(api_key='YOUR_API_KEY')
data = adapter.get_data(symbol='AAPL', asset_class='stock')

# 2. Execute Backtest
bt = Backtest(data, SmaCross, cash=10000, commission=.002)
stats = bt.run()
print(stats)
bt.plot()