Data pipelines

How do you turn raw downloads into data that is ready for analysis and stays current? VBT keeps every symbol and feature in one Data object that aligns symbols with different histories, normalizes timezones, updates incrementally, and runs indicators directly, so a market data pipeline in Python is a few method calls.

symbols = [
    "AAPL", "MSFT", "AMZN", "GOOGL", "META", "NVDA", "JPM", "V", "MA", "UNH",
    "HD", "PG", "JNJ", "XOM", "CVX", "KO", "PEP", "MRK", "ABBV", "LLY",
    "COST", "WMT", "BAC", "WFC", "CSCO", "ORCL", "INTC", "AMD", "QCOM", "TXN",
    "ADBE", "CRM", "NFLX", "DIS", "NKE", "MCD", "SBUX", "BA", "CAT", "DE",
    "GE", "HON", "IBM", "MMM", "UPS", "LOW", "T", "VZ", "PFE", "TMO",
]
data = vbt.YFData.pull(symbols, start="2024-01-01", end="2024-06-01")
data.close.shape, data.index[-1].date()

((105, 50), datetime.date(2024, 5, 31))

data = data.update(end="2024-07-01")  
data.close.shape, data.index[-1].date()

((124, 50), datetime.date(2024, 6, 28))

rsi = data.run("rsi", window=14).rsi  
rsi.iloc[-1].sort_values(ascending=False).head(5).round(1)

rsi_window  symbol
14          LLY       75.8
            T         75.4
            ADBE      74.5
            ORCL      66.5
            HON       63.6
Name: 2024-06-28 00:00:00-04:00, dtype: float64

Data containers

A Data object maps each symbol to its features, such as open, high, low, close, and volume. Features are available by name, as in data.close or data.get("Volume"), and derived prices such as data.hlc3 are computed on demand. Each feature comes back as a DataFrame with one column per symbol, ready for indicators and backtests. The same container can also be organized by feature, which suits data that arrives as one table per field, as the Feature-oriented data highlight below shows. Symbols and features can be selected, renamed, added, and removed without leaving the object.

Your own data becomes a data object with vbt.Data.from_data. It takes one DataFrame per symbol, so a long table with a symbol column, as many exports and databases produce, is split first:

long = pd.DataFrame({
    "date": pd.to_datetime(["2024-01-02", "2024-01-03", "2024-01-02", "2024-01-03"], utc=True),
    "symbol": ["AAPL", "AAPL", "MSFT", "MSFT"],
    "close": [185.64, 184.25, 370.87, 370.60],
    "volume": [82488700, 58414500, 25258600, 23083500],
})
data = vbt.Data.from_data({
    symbol: df.set_index("date").drop(columns="symbol")
    for symbol, df in long.groupby("symbol")
})
data.symbols, data.features

(['AAPL', 'MSFT'], ['close', 'volume'])

data.close

symbol                       AAPL    MSFT
date
2024-01-02 00:00:00+00:00  185.64  370.87
2024-01-03 00:00:00+00:00  184.25  370.60

A dictionary of wide DataFrames, one per feature with symbols as columns, works too with columns_are_symbols=True. This is useful for prices, fundamentals, model scores, or portfolio time series that already arrive as separate tables. A custom provider can also fetch and update one feature across many symbols by implementing fetch_feature and update_feature.

Alignment

Symbols rarely share the same history. When they are pulled together, their indexes are combined and the missing bars are filled with NaN, or dropped with missing_index="drop". Either way, a warning reports that the indexes did not match:

data = vbt.YFData.pull(["BTC-USD", "SOL-USD"], start="2020-01-01", end="2021-01-01")
data.close.shape

(366, 2)

data.close.apply(lambda x: x.first_valid_index().date()).tolist()

[datetime.date(2020, 1, 1), datetime.date(2020, 4, 10)]

dropped = vbt.YFData.pull(
    ["BTC-USD", "SOL-USD"],
    start="2020-01-01",
    end="2021-01-01",
    missing_index="drop",
)
dropped.close.shape, dropped.index[0].date()

((266, 2), datetime.date(2020, 4, 10))

Timezones are handled the same way: data can be converted to one timezone on load, and symbols that come back in different timezones are reported instead of silently misaligned. Indexes are sorted on load, and for duplicate timestamps the last value is kept.

Keep missing observations when you want to study the full history of each asset, or use the shared period when comparing assets over the same dates. Alignment does not invent prices for the missing bars. For daily and intraday inputs that become available at different times, use multi-timeframe analysis to align them by availability.

Updates

update fetches only what is new. Pass return_meta=True to see which bars were appended and which overlapping bars the provider revised, which matters when the last bar was still forming at the previous update. For a fixed schedule, vbt.DataUpdater runs updates at an interval and calls your code after each one, which is the basis of a live pipeline, as described on the Live simulation page.

Assign the returned object, as in data = data.update(). Keeping the original history and extending it gives repeated indicator calculations the same starting point. This matters for calculations such as exponential averages, whose values can change if each run starts from a different recent window.

If you only need the updated range, use concat=False. That suits a collector that writes each batch to disk through a data saver. For calculations that retain indicator state between observations, see streaming indicators.

Already receive bars from an external feed? Your integration can wrap each batch as a data object and merge it with the previous data, or provide its own update method. Later data replaces overlapping observations when merging, so a revised bar can replace the earlier version. The feed connection and recovery of missed bars stay with your integration.

Transform and merge

transform applies any function to the data as one DataFrame and returns a new data object, which covers cleaning, filtering, and adding features. merge combines data objects, for example from two providers, and concatenation joins consecutive periods. Raw files and DataFrames become data objects through parsing, with timeframes such as "15 minutes" and "1h" understood everywhere. The Data parsing, Data transformation, and Data merging highlights below show each step.

Running functions on data

data.run() takes an indicator name, a function, or an expression and fills in the features it needs by name, so data.run("rsi") reads the close and data.run("atr") reads high, low, and close. Arguments can differ per function, and many functions can run in one call, in parallel if needed, as the Runnable data and Parallel data highlights show.

Keep research features beside prices

You can attach calculated features to the same data object. This makes it easier to keep prices, indicator outputs, and model scores together for the next stage of research. The calculation can be a regular Python function whose argument names match the data fields:

research_close = pd.DataFrame(
    {"AAA": [100.0, 102.0, 104.0], "BBB": [50.0, 49.0, 51.0]},
    index=pd.date_range("2024-01-01", periods=3, freq="D", tz="UTC"),
)
research_data = vbt.Data.from_data(
    vbt.feature_dict(
        {
            "Close": research_close,
            "High": research_close + 2,
            "Low": research_close - 2,
        }
    )
)

def range_pct(high, low, close):
    return 100 * (high - low) / close

range_values = research_data.run(range_pct)
research_data = research_data.add_feature("RangePct", range_values)
research_data.features

['Close', 'High', 'Low', 'RangePct']

research_data.get("RangePct").round(2)

                            AAA   BBB
2024-01-01 00:00:00+00:00  4.00  8.00
2024-01-02 00:00:00+00:00  3.92  8.16
2024-01-03 00:00:00+00:00  3.85  7.84

One calculation adds the high-to-low range as a percentage of the close for both sample symbols at once. You can select it later by name or save it with the other features using the storage methods.

Universes and futures

Symbols can carry classes such as sector or exchange, which become column levels for grouping. For research on index members, build a membership table with the dates each symbol was in the index and use it as a mask, so the universe does not include stocks that only joined later. Continuous futures series should be rolled and back-adjusted before they are loaded, so that roll gaps do not appear as price moves.

Errors

Keep a large download moving with skip_on_error=True: VBT collects the available symbols and reports those that failed. Retry those symbols later and merge them into the dataset.

✅ Keep your data organized the way you use it. VBT supports both symbol-oriented and feature-oriented storage: one table per instrument, or one table per field such as open, high, low, and close. Feature-oriented data makes those price tables directly available across symbols, ready for indicators and backtests.

data = vbt.YFData.pull(["AAPL", "MSFT", "GOOG"], tz_convert="UTC")
pf = data.run("from_random_signals", n=[10, 20, 30], seed=42)

pf_data = vbt.Data.from_data(
    vbt.feature_dict({
        "cash": pf.cash,
        "assets": pf.assets,
        "asset_value": pf.asset_value,
        "value": pf.value
    })
)
pf_data.get(feature="cash", symbol=(10, "AAPL"))

Date
1980-12-12 05:00:00+00:00      100.000000
1980-12-15 05:00:00+00:00      100.000000
1980-12-16 05:00:00+00:00      100.000000
1980-12-17 05:00:00+00:00      100.000000
1980-12-18 05:00:00+00:00      100.000000
                                      ...
2023-08-25 04:00:00+00:00     1837.648268
2023-08-28 04:00:00+00:00     1837.648268
2023-08-29 04:00:00+00:00     1837.648268
2023-08-30 04:00:00+00:00     1837.648268
2023-08-31 04:00:00+00:00     1837.648268
Name: (10, AAPL), Length: 10770, dtype: float64

✅ Data fetching and updating can be easily parallelized.

symbols = ["SPY", "TLT", "XLF", "XLE", "XLU", "XLK", "XLB", "XLP", "XLY", "XLI", "XLV"]

with vbt.Timer() as timer:
    data = vbt.YFData.pull(symbols)
print(timer.elapsed())

4.52 seconds

with vbt.Timer() as timer:
    data = vbt.YFData.pull(symbols, execute_kwargs=dict(engine="threadpool"))
print(timer.elapsed())

918.54 milliseconds

✅ Tired of figuring out which arguments are required by an indicator? Data instances can now recognize the arguments of any indicator or function, map them to column names, and run the function by passing in the required columns. You can also change the mapping, override indicator parameters, and query indicators by name. The data instance will search all integrated indicator packages and return the first (and best) match it finds.

data = vbt.YFData.pull("BTC-USD")
stochrsi = data.run("stochrsi")
stochrsi.fastd

Date
2014-09-17 00:00:00+00:00          NaN
2014-09-18 00:00:00+00:00          NaN
2014-09-19 00:00:00+00:00          NaN
2014-09-20 00:00:00+00:00          NaN
2014-09-21 00:00:00+00:00          NaN
                                   ...
2023-01-15 00:00:00+00:00    96.168788
2023-01-16 00:00:00+00:00    91.733393
2023-01-17 00:00:00+00:00    78.295255
2023-01-18 00:00:00+00:00    48.793133
2023-01-20 00:00:00+00:00    26.242474
Name: Close, Length: 3047, dtype: float64

✅ Tired of passing open, high, low, and close as separate time series? Portfolio class methods now accept a data instance instead of just close and automatically extract the contained OHLC data. This small but handy feature saves you time!

data = vbt.YFData.pull("BTC-USD", start="2020-01", end="2020-03")
pf = vbt.PF.from_random_signals(data, n=10, seed=42)

✅ After fetching data, how do you change it? There is a new method that puts all symbols into a single DataFrame and passes it to a UDF for transformation.

data = vbt.YFData.pull(["BTC-USD", "ETH-USD"], start="2020-01-01", end="2020-01-14")
new_data = data.transform(lambda df: df[~df.index.weekday.isin([5, 6])])
new_data.close

symbol                         BTC-USD     ETH-USD
Date
2020-01-01 00:00:00+00:00  7200.174316  130.802002
2020-01-02 00:00:00+00:00  6985.470215  127.410179
2020-01-03 00:00:00+00:00  7344.884277  134.171707
2020-01-06 00:00:00+00:00  7769.219238  144.304153
2020-01-07 00:00:00+00:00  8163.692383  143.543991
2020-01-08 00:00:00+00:00  8079.862793  141.258133
2020-01-09 00:00:00+00:00  7879.071289  138.979202
2020-01-10 00:00:00+00:00  8166.554199  143.963776
2020-01-13 00:00:00+00:00  8144.194336  144.226593

✅ Often, there is a need to backtest symbols from different exchanges by placing them in the same basket. For this purpose, VBT offers a class method that can merge multiple data instances into a single one. You can not only combine multiple symbols, but also merge datasets for a single symbol, all done automatically!

binance_data = vbt.CCXTData.pull("BTCUSDT", exchange="binance")
bybit_data = vbt.CCXTData.pull("BTCUSDT", exchange="bybit")
bitfinex_data = vbt.CCXTData.pull("BTC/USDT", exchange="bitfinex")
kucoin_data = vbt.CCXTData.pull("BTC-USDT", exchange="kucoin")

data = vbt.Data.merge([
    binance_data.rename({"BTCUSDT": "Binance"}),
    bybit_data.rename({"BTCUSDT": "Bybit"}),
    bitfinex_data.rename({"BTC/USDT": "Bitfinex"}),
    kucoin_data.rename({"BTC-USDT": "KuCoin"}),
], missing_index="drop", silence_warnings=True)

@njit
def rescale_nb(x):
    return (x - x.mean()) / x.mean()

rescaled_close = data.close.vbt.row_apply(rescale_nb)
rescaled_close = rescaled_close.vbt.rolling_mean(30)
last_full_year = rescaled_close.index[-1].year - 1
rescaled_close.loc[str(last_full_year)].vbt.plot().show()