Data Pipeline Executor#

Data Pipeline Executor Module

This module provides the core data pipeline functionality for the backtesting engine. It centralizes all data validation, loading, and preparation logic to ensure consistent data processing across different entry points (main function, factory methods, etc.).

The pipeline handles:

Portfolio configuration and loading
Market data ingestion and validation
Benchmark construction
Data integrity checking
Field mapping and standardization

All data processing follows a strict validation sequence to ensure data quality before any backtesting operations begin.

REBALANCE_DATE_LOOKAHEAD_DAYS = 10#: Calendar days loaded past an auto end date when rebalancing dates may move to the next trading day.

class DataPipelineResult(portfolio_entity: PortfolioEntity, portfolio_df: DataFrame, start_date_backtest: Timestamp, market_data_bundle: MarketDataBundlePyarrow, data_dict: dict[str, Table], benchmark: Table, portfolio_dict: dict[Timestamp, dict[str, Scalar]], equivalence_dict: dict[str, str], benchmarks: dict[str, Table] | None = None, configuration: Configuration | None = None, forced_exits: Iterable[ForcedExit] | None = None, benchmark_rate_series: RateSeries | None = None)#

Bases: object

Container for data pipeline execution results.

This class encapsulates all outputs from the data pipeline execution, providing a clean interface for accessing validated and processed data. Using a dedicated result class improves type safety and makes the pipeline output contract explicit.

Variables:

portfolio_entity – Entity object representing the portfolio configuration
portfolio_df – DataFrame containing portfolio composition and weights
start_date_backtest – Calculated start date for backtest execution
market_data_bundle – Bundle of PyArrow tables with market data
data_dict – Dictionary mapping price field names to PyArrow tables
benchmark – PyArrow table containing benchmark time series data
portfolio_dict – Dictionary mapping dates to portfolio weight dictionaries
equivalence_dict – Dictionary mapping standard field names to actual column names

get_configuration() → Configuration | None#

Return the effective configuration the pipeline ran with.

Dates are resolved (no "auto") and reflect any adjustment made while preparing the data, such as an auto end date moved to the last rebalancing date after that date was shifted to the next trading day. Callers that build a backtester from this result should prefer it over the configuration they started with.

Returns:: None when the result was built without one.
Return type:: Configuration | None

get_forced_exits() → tuple[ForcedExit, ...]#

Return the forced exits the delisting policy scheduled.

Each entry tells the engine to close a whole position on a date: by selling it at that day’s execution price or by writing it off. Empty with the default fail policy, or when no position outlives its price series.

Returns:: Ordered by date.
Return type:: tuple[ForcedExit, …]

get_benchmark_rate_series() → RateSeries | None#

Return the dated benchmark rate series, when one was configured.

Loaded from the market data directory when financing_benchmark_rate names a rate file; the backtester hands it to the financing model and reports its average over the backtest as the risk-free rate.

Returns:: None when the benchmark is a constant.
Return type:: RateSeries | None

get_portfolio_entity() → PortfolioEntity#

Return the parsed portfolio entity.

The entity is the typed, validated representation of the portfolio definition (file or dict). It carries configuration metadata (name, source path) and the raw weight schedule before any post-processing such as date filtering or zero-weight pruning.

Returns:: Portfolio entity with configuration metadata. Same instance stored on the result; do not mutate.
Return type:: PortfolioEntity

See also

get_portfolio_df: Tabular view of the same data.
get_portfolio_dict: Date-indexed weights derived from the entity.

get_portfolio_df() → DataFrame#

Return the portfolio composition as a DataFrame.

The DataFrame holds the post-processed portfolio (after date-range filtering and zero-weight pruning). It is the version that downstream backtesters consume directly.

Returns:: DataFrame with rebalance dates as the index and ticker columns; values are the target weights (decimals).
Return type:: pandas.DataFrame

See also

get_portfolio_dict: Same data as a nested dict keyed by date.
get_portfolio_entity: Raw entity prior to processing.

get_start_date_backtest() → Timestamp#

Return the resolved start date of the backtest period.

This is the effective start date after applying "auto" resolution against the portfolio rebalance dates. When the user-provided start date is explicit, this returns that value verbatim; when it was "auto", this returns the earliest rebalance date.

Returns:: First date in the resolved backtest window.
Return type:: pandas.Timestamp

See also

get_portfolio_dict: Rebalance schedule used to resolve "auto".

get_market_data_bundle() → MarketDataBundlePyarrow#

Return the bundle of price tables consumed by the backtester.

The bundle groups the columnar price tables (open, close, VWAP, adjusted close, etc.) loaded from the configured input handlers. It is passed directly to PyArrowBacktester and guarantees that all tables share the same date index and ticker set.

Returns:: Container with VWAP and adjusted-close price tables. Same instance stored on the result; do not mutate.
Return type:: MarketDataBundlePyarrow

See also

get_data_dict: Raw price tables before bundling.

get_data_dict() → dict[str, Table]#

Return the raw dictionary of price tables keyed by field name.

Each value is a PyArrow Table with one column per ticker and a date column. This is the unbundled view of the same data exposed via get_market_data_bundle; useful when callers need direct access to a specific price field (e.g. for custom metric computation).

Returns:: Mapping from standardized field name (e.g. "close", "vwap") to its PyArrow table.
Return type:: dict[str, pyarrow.Table]

See also

get_market_data_bundle: Higher-level bundle view of the same tables.

get_benchmark() → Table#

Return the benchmark price table.

The benchmark is loaded only when Configuration.benchmark is configured. If no benchmark was requested, this returns the empty/sentinel benchmark table produced by the pipeline rather than None — callers should not assume an unconditional non-empty table.

Returns:: Benchmark time-series. May be empty if no benchmark was configured.
Return type:: pyarrow.Table

See also

PyArrowBacktester.run_with_benchmark: Primary consumer.

get_benchmarks() → dict[str, Table]#

Return all configured benchmark tables keyed by ticker.

Ordered mapping (primary first) of every benchmark built by the pipeline. When a single benchmark is configured this has one entry equal to get_benchmark(); when none is configured it is empty.

Returns:: Benchmark ticker -> benchmark time-series table.
Return type:: dict[str, pa.Table]

get_portfolio_dict() → dict[Timestamp, dict[str, Scalar]]#

Return the rebalance schedule as a nested dict.

Each key is a rebalance date and each value is a mapping {ticker: weight} for that date. This is the structure consumed directly by PyArrowBacktester via the portfolio_dict parameter.

Returns:: Date-keyed mapping of ticker weights. Weights are PyArrow scalars to preserve precision through the pipeline.
Return type:: dict[pandas.Timestamp, dict[str, pyarrow.Scalar]]

See also

get_portfolio_df: Tabular view of the same data.

get_equivalence_dict() → dict[str, str]#

Return the mapping from standard field names to source columns.

The pipeline normalizes heterogeneous input CSV / Parquet column names to the project’s StandardField vocabulary. This dictionary preserves the original mapping so downstream code (e.g. error messages, debug logs) can refer back to the actual source column name.

Returns:: Mapping from standard field name (e.g. "adjusted_close") to the column name found in the source data.
Return type:: dict[str, str]

See also

get_market_data_bundle: Data normalized via this mapping.

execute_data_pipeline(configuration: Configuration, input_handlers: list[InputHandlerInterface], portfolio_handlers: list[PortfolioInputHandlerInterface]) → DataPipelineResult#

Execute the complete data pipeline with validation and preparation.

This function orchestrates the entire data loading and validation process required before backtesting can begin. It ensures all data is properly formatted, validated, and ready for consumption by the backtesting engine.

The pipeline execution follows these steps:

Load and process portfolio configuration
Extract tickers from portfolio weights
Validate benchmark availability
Load and process market data for all tickers
Build benchmark time series
Perform data integrity checks
Bundle market data into standardized structure

Parameters:

configuration – Configuration object containing all backtest parameters including dates, file paths, execution settings, and commission rates
input_handlers – List of handler objects for reading market data from various sources (CSV, Parquet, Excel, etc.)
portfolio_handlers – List of handler objects for reading portfolio configuration and weight data

Returns:

Container object with all validated and processed: data ready for backtesting, including portfolio configuration, market data, and benchmark information

Return type:

DataPipelineResult

Raises:

MissingData – If data integrity validation fails, indicating missing dates or prices for required tickers
MissingBenchmark – If the benchmark file or data cannot be located in the provided input handlers
ConfigurationError – If the configuration object contains invalid parameters or incompatible settings
ReadingExcelPortfolioError – If portfolio Excel file cannot be read or parsed correctly
ReadingCSVError – If market data CSV files cannot be read

Side Effects:

Logs progress information including:

Data loading completion times
Number of tickers being processed
Validation results
Performance metrics

Note

This function performs comprehensive validation and will raise exceptions rather than proceed with invalid data. All exceptions should be caught and handled by the caller.

Example

>>> config = Configuration(...)
>>> handlers = [CSVInputHandler(), ParquetInputHandler()]
>>> portfolio_handlers = [ExcelPortfolioHandler()]
>>> result = execute_data_pipeline(config, handlers, portfolio_handlers)
>>> backtester = PyArrowBacktester.create_from_configuration(...)

See also

Quick Start Guide: End-to-end walkthrough for setting up and running a backtest from scratch.
Running from Python: Run a backtest from a Python script, including this pipeline as part of the flow.
Data Formats: File formats expected by the input and portfolio handlers used here.

trading_dates_in(data_dict: dict[str, Table]) → list[Timestamp]#

Every trading day present in the wide market-data tables.

All tables in data_dict share the same date axis (the union over every ticker), so the first one is enough.

Parameters:: data_dict (dict[str, pa.Table]) – Wide tables keyed by the user’s column names.
Returns:: Sorted, normalized trading days; empty when there are no tables.
Return type:: list[pd.Timestamp]

execution_cost_columns_mapped(configuration: Configuration) → bool#

Whether any column the execution-cost metrics derive from is mapped.

This is the gate for the whole feature: with none of them mapped nothing is loaded early, nothing is computed and nothing is trimmed, which is the default and costs exactly what it did before.

Parameters:: configuration (Configuration) – Settings to inspect.
Returns:: True when at least one of the volume, high, low, close or spread columns is mapped.
Return type:: bool

warmup_configuration(configuration: Configuration) → Configuration#

The configuration to load market data with, start date moved back.

A rolling window needs history behind the first day of the backtest. The input handlers filter each file to [start_date, end_date] as they read it, so without this the first execution_cost_window_days of every backtest would have no metric at all and every ticker would fall back to its constant, however much history the file holds.

Parameters:: configuration (Configuration) – The real configuration.
Returns:: The same object when no execution-cost column is mapped or the start date is "auto" (in which case the handlers already load the whole file); otherwise a copy whose start_date is early enough to fill the windows.
Return type:: Configuration

Notes

The extra rows are dropped again by trim_to_backtest_window() before anything else in the pipeline sees them, so the schedule, the integrity check, the delisting scan and the engines all work on exactly the rows they worked on before.

trim_to_backtest_window(data_dict: dict[str, Table], configuration: Configuration) → dict[str, Table]#

Drop the warm-up rows once the execution-cost metrics are computed.

Parameters:

data_dict (dict[str, pa.Table]) – Wide tables, possibly carrying rows before the backtest starts.
configuration (Configuration) – The real configuration, i.e. the one warmup_configuration() was given.

Returns:

The same dictionary when there was no warm-up, otherwise one whose tables start on configuration.start_date, exactly as they did before the warm-up existed.

Return type:

dict[str, pa.Table]

apply_rebalance_date_handling(*, configuration: Configuration, portfolio_dict: dict[Timestamp, dict[str, Scalar]], data_dict: dict[str, Table], end_date_was_auto: bool) → tuple[Configuration, dict[Timestamp, dict[str, Scalar]], dict[str, Table]]#

Apply the configured rebalancing-date handling to a loaded schedule.

Shared by the single-backtest pipeline and the batch data pool. With exact nothing changes and the integrity check later rejects any rebalancing date without market data. With next_trading_day every date that is not a trading day moves to the next one, each move is logged as a warning, and when the end date was "auto" the window is tightened to the last aligned rebalancing date: the configuration’s end date is set to it and the tables are trimmed, so auto keeps meaning “the backtest ends on the last rebalancing date”.

Parameters:

configuration (Configuration) – Configuration with dates already resolved.
portfolio_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Rebalancing schedule as loaded from the portfolio file.
data_dict (dict[str, pa.Table]) – Wide market-data tables; their date axis is the trading calendar.
end_date_was_auto (bool) – Whether the user left end_date as "auto".

Returns:

The (possibly updated) configuration, the aligned schedule and the (possibly trimmed) tables.

Return type:

tuple[Configuration, dict, dict]

Raises:

DataIntegrityError – If a date cannot be moved (no later trading day in the data) or two dates land on the same trading day.

last_tradable_dates(data_dict: dict[str, Table], *, columns: Iterable[str]) → dict[str, date | None]#

Last date on which each ticker has a value in every listed price column.

A forced sale must fill at the execution price and size its commission at the commission price, so a ticker counts as tradable on a date only when every price column has a value; the earliest column end wins.

Parameters:

data_dict (dict[str, pa.Table]) – Wide tables keyed by the user’s column names.
columns (Iterable[str]) – The price columns to consider.

Returns:

Ticker -> last fully priced date, or None when some column never has a value for the ticker.

Return type:

dict[str, datetime.date | None]

apply_delisted_position_handling(*, configuration: Configuration, portfolio_dict: dict[Timestamp, dict[str, Scalar]], data_dict: dict[str, Table], equivalence_dict: dict[str, str]) → tuple[dict[Timestamp, dict[str, Scalar]], tuple[ForcedExit, ...]]#

Apply the configured delisted-position handling to a loaded schedule.

Shared by the single-backtest pipeline and the batch data pool. With fail nothing changes and the integrity check later rejects any position held past its last price. Otherwise the schedule goes through plan_delisting_exits(): the weights of a delisted ticker are set aside from its exit date on (each one logged as a warning), and the forced exits the engines must execute are returned for the integrity check and the backtester.

Parameters:

configuration (Configuration) – Configuration with dates already resolved and aligned.
portfolio_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Rebalancing schedule, aligned to trading days.
data_dict (dict[str, pa.Table]) – Wide market-data tables; their date axis is the trading calendar.
equivalence_dict (dict[str, str]) – Standard field name -> user column name.

Returns:

The (possibly adjusted) schedule and the forced exits, ordered by date.

Return type:

tuple[dict, tuple[ForcedExit, …]]

resolve_backtest_end_date(value: object) → date | None#

Coerce a configuration end_date to a plain date for the integrity check.

Parameters:: value (object) – Configuration.end_date after auto-date resolution: a date, datetime/Timestamp, an ISO string, "auto" or None.
Returns:: The end date, or None when it is not an explicit date ("auto" resolves to the last rebalancing date, which the per-date validation already covers).
Return type:: datetime.date | None

check_and_bundle_market_data(*, data_dict: dict[str, Table], equivalence_dict: dict[str, str], portfolio_dict: dict[Timestamp, dict[str, Scalar]], backtest_end_date: date | None, forced_exits: Iterable[ForcedExit] = ()) → MarketDataBundlePyarrow#

Validate the market data against the rebalancing schedule and assemble the bundle.

Shared by the single-backtest pipeline and the batch data pool so both apply exactly the same rules:

The integrity check rejects any rebalancing date that is not a trading day in the data, any ticker the engine would trade without a valid positive price, and any position held past the end of its price series before backtest_end_date (a position closed by a forced exit is not held past it).
The mark-to-market table gets its interior gaps carried forward (last valid price of the same ticker), so a held position is never valued at zero on a day the ticker simply has no row. Execution and commission tables are never filled: a carried price cannot be traded at. data_dict itself is left untouched.
When a per-ticker borrow fee column is mapped, its wide table joins the bundle as is (nulls mean “use the global rate”) after a range check; it is not a price, so the integrity check ignores it.
Any execution-cost metric the service computed (average daily volume, volatility, half spread) is moved out of data_dict and into the bundle’s metric_tables. They are derived, never mapped, so they are not prices either.

Parameters:

data_dict (dict[str, pa.Table]) – Wide tables keyed by the user’s column names.
equivalence_dict (dict[str, str]) – Standard field name -> user column name.
portfolio_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Rebalancing schedule.
backtest_end_date (datetime.date | None) – Last date of the backtest window, or None to skip the end-of-series check.
forced_exits (Iterable[ForcedExit], optional) – Forced exits scheduled by the delisting policy. Empty by default.

Returns:

Bundle with the three price tables, plus the optional borrow-fee table and any execution-cost metric tables that were computed.

Return type:

MarketDataBundlePyarrow

Raises:

DataIntegrityError – From the checker, with the full list of problems.
MissingDataError – If the checker returns errors without raising (defensive).