FinancialData.Net adapter for vectorbt (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 vectorbt 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)

        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:
            return value
        if isinstance(value, str):
            value = pd.to_datetime(value)
        if 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['dt'] = pd.to_datetime(df['dt'])
            df = df.set_index('dt')
            df.index.name = None
            
            if len(df) > 1:
                df.index = df.index.tz_localize(None)
                df = df.asfreq(pd.infer_freq(df.index)) 
                df = df.ffill()
        
        return df[['Open', 'High', 'Low', 'Close', 'Volume']]

Minimal example

adapter = FDNBacktestingAdapter(api_key='YOUR_API_KEY')
df = adapter.get_data('AAPL', asset_class='stock', timeframe='days')

import vectorbt as vbt
sma = vbt.MA.run(df['Close'], 20).ma
entries = df['Close'] > sma
exits = df['Close'] < sma

pf = vbt.Portfolio.from_signals(df['Close'], entries, exits, init_cash=10000)
print(f"Strategy Profit: {pf.total_profit()}")