Portfolio Helpers#

Portfolio Helper Functions Module.

This module provides pure utility functions for portfolio operations. These functions perform data transformations without side effects and do not depend on service state or configuration.

All functions are stateless and can be used independently.

class PortfolioProcessingResult(weights_dict: dict[Timestamp, dict[str, Scalar]], start_date_backtest: Timestamp, end_date_backtest: Timestamp)

Bases: object

Immutable container for portfolio processing pipeline output.

Variables:

  • weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Processed weights dictionary filtered by date range and without zero-weight-only tickers.

  • start_date_backtest (pd.Timestamp) – The earliest rebalancing date in the processed weights.

  • end_date_backtest (pd.Timestamp) – The latest rebalancing date in the processed weights.

weights_dict: dict[Timestamp, dict[str, Scalar]]
start_date_backtest: Timestamp
end_date_backtest: Timestamp
class PortfolioLoadResult(portfolio_entity: PortfolioEntity, portfolio_table: Table | None, portfolio_dict: dict[Timestamp, dict[str, Scalar]] | None, start_date_backtest: Timestamp | None, end_date_backtest: Timestamp | None)

Bases: object

Result object encapsulating portfolio loading and processing outcomes.

This class provides a clean interface for accessing portfolio data through getters instead of tuple unpacking, improving code readability and maintainability.

Variables:

  • _portfolio_entity (PortfolioEntity) – The validated portfolio entity.

  • _portfolio_table (Optional[pa.Table]) – Raw table data (None for dict loading).

  • _portfolio_dict (Optional[dict[pd.Timestamp, dict[str, pa.Scalar]]]) – Processed weights dictionary.

  • _start_date_backtest (Optional[pd.Timestamp]) – Start date for backtesting.

get_portfolio_entity() → PortfolioEntity

Return the validated portfolio entity.

Returns:

The portfolio entity containing tickers,

date columns, and weight data.

Return type:

PortfolioEntity

get_portfolio_table() → Table | None

Return the raw portfolio table.

Returns:

Raw PyArrow table from file loading, or

None when the portfolio was loaded from a dictionary.

Return type:

pa.Table | None

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

Return the processed portfolio weights dictionary.

Returns:

Mapping from

rebalancing dates to ticker-weight dictionaries.

Return type:

dict[pd.Timestamp, dict[str, pa.Scalar]]

get_start_date_backtest() → Timestamp

Return the backtest start date.

Returns:

The earliest rebalancing date after processing.

Return type:

pd.Timestamp

get_end_date_backtest() → Timestamp

Return the backtest end date.

Returns:

The latest rebalancing date after processing.

Return type:

pd.Timestamp

property portfolio_entity: PortfolioEntity

Backwards compatibility property for portfolio entity.

property portfolio_table: Table | None

Backwards compatibility property for portfolio table.

property portfolio_dict: dict[Timestamp, dict[str, Scalar]] | None

Backwards compatibility property for processed weights dictionary.

property start_date_backtest: Timestamp | None

Backwards compatibility property for backtest start date.

property end_date_backtest: Timestamp | None

Backwards compatibility property for backtest end date.

extract_tickers_from_weights_dict(weights_dict: dict[Timestamp, dict[str, Scalar]]) → list[str]

Extract tickers that have a strictly positive weight at least once.

This is a pure function that analyzes the weights dictionary and returns tickers that are active (have positive weight) at any point in time.

Parameters:

weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Mapping from rebalancing date to “ticker -> weight”.

Returns:

Sorted list of active tickers (alphabetical order).

Return type:

list[str]

Notes

Tickers that never appear with a non-zero weight are logged and excluded. Activity is tested on weight != 0, not weight > 0, so a ticker that is only ever shorted stays in the universe.

Example

>>> weights = {
...     pd.Timestamp('2020-01-01'): {'AAPL': 0.6, 'MSFT': 0.0},
...     pd.Timestamp('2020-01-02'): {'AAPL': 0.5, 'MSFT': 0.5}
... }
>>> extract_tickers_from_weights_dict(weights)
['AAPL', 'MSFT']
filter_weights_by_date_range(weights_dict: dict[Timestamp, dict[str, Scalar | float]], start_date: Timestamp | date, end_date: Timestamp | date) → dict[Timestamp, dict[str, Scalar | float]]

Filter weights dictionary to include only dates within the specified range (inclusive).

This is a pure function that filters dates without modifying the original dictionary. All dates are normalized to midnight for consistent comparisons.

Parameters:

  • weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Mapping from rebalancing date to “ticker -> weight”.

  • start_date (pd.Timestamp) – Inclusive lower bound.

  • end_date (pd.Timestamp) – Inclusive upper bound.

Returns:

Filtered mapping with normalized dates.

Return type:

dict[pd.Timestamp, dict[str, pa.Scalar]]

Raises:

PortfolioEntityError – If no dates survive filtering. Includes diagnostic information.

Example

>>> weights = {
...     pd.Timestamp('2020-01-01'): {'AAPL': 0.6},
...     pd.Timestamp('2020-06-01'): {'AAPL': 0.5}
... }
>>> filtered = filter_weights_by_date_range(
...     weights,
...     pd.Timestamp('2020-01-01'),
...     pd.Timestamp('2020-03-31')
... )
>>> len(filtered)
1
filter_zero_weight_tickers(weights_dict: dict[Timestamp, dict[str, Scalar]]) → dict[Timestamp, dict[str, Scalar]]

Filter out tickers that have zero weight across all rebalancing dates.

This is a pure function that does not mutate the original mapping and returns a new dictionary containing only tickers that are active (i.e., strictly positive weight at least once).

Parameters:

weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Mapping from rebalancing date to an inner mapping ticker -> weight (pa.Scalar).

Returns:

New dictionary that excludes tickers that consistently have weight 0.

Return type:

dict[pd.Timestamp, dict[str, pa.Scalar]]

Notes

  • A ticker is considered active if it has a non-zero weight for at least one date. The test is weight != 0, not weight > 0, so a ticker that is only ever shorted stays in the universe.

  • Logs the removed tickers for traceability.

Example

>>> weights = {
...     pd.Timestamp('2020-01-01'): {'AAPL': 0.6, 'MSFT': 0.0, 'GOOGL': 0.4},
...     pd.Timestamp('2020-01-02'): {'AAPL': 0.5, 'MSFT': 0.0, 'GOOGL': 0.5}
... }
>>> filtered = filter_zero_weight_tickers(weights)
>>> 'MSFT' in filtered[pd.Timestamp('2020-01-01')]
False
get_start_date_backtest(weights_dict: dict[Timestamp, dict[str, Scalar]]) → Timestamp

Determine the earliest rebalancing date from a weights dictionary.

This is a pure function that finds the minimum date without side effects.

Parameters:

weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar]]) – Mapping from rebalancing date to “ticker -> weight”.

Returns:

The earliest date available.

Return type:

pd.Timestamp

Raises:

PortfolioEntityError – If the mapping is empty or None.

Examples

>>> weights = {
...     pd.Timestamp('2020-06-01'): {'AAPL': 0.5},
...     pd.Timestamp('2020-01-01'): {'AAPL': 0.6}
... }
>>> get_start_date_backtest(weights)
Timestamp('2020-01-01 00:00:00')
get_end_date_backtest(weights_dict: dict[Timestamp, dict[str, Scalar]]) → Timestamp

Determine the latest rebalancing date from a weights dictionary.

Pure helper that returns max(weights_dict.keys()) after a non-empty check. Used by the pipeline to derive the effective end date of the backtest window when the user-provided end_date is "auto".

Parameters:

weights_dict (dict[pandas.Timestamp, dict[str, pyarrow.Scalar]]) – Mapping from rebalancing date to {ticker: weight}. Must be non-empty.

Returns:

The latest rebalancing date present in the mapping.

Return type:

pandas.Timestamp

Raises:

PortfolioEntityError – If the mapping is empty or None.

Examples

>>> weights = {
...     pd.Timestamp('2020-06-01'): {'AAPL': 0.5},
...     pd.Timestamp('2020-01-01'): {'AAPL': 0.6},
... }
>>> get_end_date_backtest(weights)
Timestamp('2020-06-01 00:00:00')

Notes

The function does not assume the mapping is ordered; it relies on max over the keys so insertion order does not matter.

See also

get_start_date_backtest

Companion helper returning the earliest date.

align_rebalance_dates_to_trading_days(weights_dict: dict[Timestamp, dict[str, Scalar | float]], trading_dates: Iterable[Timestamp | date], handling: RebalanceDateHandling) → tuple[dict[Timestamp, dict[str, Scalar | float]], list[tuple[Timestamp, Timestamp]]]

Move rebalancing dates that are not trading days to the next trading day.

Pure function: the input mapping is never mutated and the weight dictionaries are carried over untouched. With RebalanceDateHandling.EXACT the mapping is returned as is and the data integrity check later rejects any date without market data.

Earlier trading days are never used. Weights on a date are decided with information available up to that date, so executing them before it would be look-ahead.

Parameters:

  • weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar | float]]) – Mapping from rebalancing date to “ticker -> weight”.

  • trading_dates (Iterable[pd.Timestamp | date]) – Every trading day present in the market data (the union over all tickers), in any order.

  • handling (RebalanceDateHandling) – EXACT leaves the dates alone; NEXT_TRADING_DAY shifts.

Returns:

The aligned mapping, ordered by date with normalized pd.Timestamp keys, and the list of (original, new) pairs for every date that moved.

Return type:

tuple[dict, list[tuple[pd.Timestamp, pd.Timestamp]]]

Raises:

DataIntegrityError – If a date has no trading day after it in the market data, or if two rebalancing dates land on the same trading day after the shift (an exact date and a shifted one, or two shifted ones).

Examples

>>> weights = {pd.Timestamp("2024-01-06"): {"AAPL": 1.0}}   # a Saturday
>>> calendar = [pd.Timestamp("2024-01-05"), pd.Timestamp("2024-01-08")]
>>> aligned, shifts = align_rebalance_dates_to_trading_days(
...     weights, calendar, RebalanceDateHandling.NEXT_TRADING_DAY
... )
>>> list(aligned)
[Timestamp('2024-01-08 00:00:00')]
>>> shifts
[(Timestamp('2024-01-06 00:00:00'), Timestamp('2024-01-08 00:00:00'))]
class IgnoredWeight(ticker: str, date: Timestamp, weight: float, last_valid_date: date, exit_date: Timestamp)

Bases: object

A non-zero weight the delisting policy set aside.

Variables:

  • ticker (str) – Ticker the weight referred to.

  • date (pd.Timestamp) – Rebalancing date carrying the weight.

  • weight (float) – The weight in the portfolio file.

  • last_valid_date (date) – Last date on which the ticker has a price.

  • exit_date (pd.Timestamp) – Trading day from which the ticker is no longer held.

ticker: str
date: Timestamp
weight: float
last_valid_date: date
exit_date: Timestamp
class DelistingPlan(weights: dict[Timestamp, dict[str, Scalar | float]], exits: tuple[ForcedExit, ...], ignored_weights: tuple[IgnoredWeight, ...])

Bases: object

Output of plan_delisting_exits().

Variables:

  • weights (dict[pd.Timestamp, dict[str, pa.Scalar | float]]) – The schedule with the weights of delisted tickers set aside from their exit date on. Ordered by date with normalized keys when the policy is not fail; the input mapping itself otherwise.

  • exits (tuple[ForcedExit, ...]) – The forced exits the engines must execute, ordered by date.

  • ignored_weights (tuple[IgnoredWeight, ...]) – Every non-zero weight that was set aside, for the log.

weights: dict[Timestamp, dict[str, Scalar | float]]
exits: tuple[ForcedExit, ...]
ignored_weights: tuple[IgnoredWeight, ...]
plan_delisting_exits(weights_dict: dict[Timestamp, dict[str, Scalar | float]], *, trading_dates: Iterable[Timestamp | date], last_valid_dates: Mapping[str, date | None], policy: DelistingPolicy, backtest_end_date: date | Timestamp) → DelistingPlan

Decide when each delisted position leaves the book and what the file’s weights mean after that.

Pure function: the input mapping and its weight dictionaries are never mutated. With the fail policy the schedule is returned untouched and the data integrity check later rejects any position held past its last price.

A ticker is delisted inside the backtest when its last price date is earlier than the effective end of the backtest (the configured end date capped at the last trading day in the data). For such a ticker:

  • sell_at_last_price: the exit date is the last price date moved back by policy.exit_lead_days trading days (never before the first trading day). From the exit date on, every non-zero weight of the ticker in the file is set aside (replaced by zero) and reported: the ticker cannot be held any more. If the position is open when the exit date arrives, a ForcedExit closes it, unless the exit date is itself a rebalancing date that lists the ticker, in which case the scheduled rebalance liquidates it and no extra event is needed.

  • mark_to_zero: the exit date is the first trading day after the last price date. The ticker is removed from every rebalancing date from then on (a weight there could neither be bought nor sold) and, if the position is open, a ForcedExit writes it off on that day. exit_lead_days plays no part.

The other weights of a date are never renormalized: the capital the ticker would have taken stays in cash until the next scheduled rebalance. A ticker without any last valid date (never priced) is left alone for the integrity check to report.

Parameters:

  • weights_dict (dict[pd.Timestamp, dict[str, pa.Scalar | float]]) – Rebalancing schedule, already aligned to trading days.

  • trading_dates (Iterable[pd.Timestamp | date]) – Every trading day present in the market data, in any order.

  • last_valid_dates (Mapping[str, date | None]) – Ticker -> last date with a price in every price column, or None when the ticker never has one.

  • policy (DelistingPolicy) – Handling and exit lead.

  • backtest_end_date (date | pd.Timestamp) – Last date the backtest values the portfolio on.

Returns:

The adjusted schedule, the forced exits and the weights set aside.

Return type:

DelistingPlan

Examples

>>> calendar = [pd.Timestamp(f"2024-01-{day:02d}") for day in (2, 3, 4, 5, 8)]
>>> weights = {calendar[0]: {"AAA": 0.5, "BBB": 0.5}}
>>> plan = plan_delisting_exits(
...     weights,
...     trading_dates=calendar,
...     last_valid_dates={"AAA": calendar[2].date(), "BBB": calendar[-1].date()},
...     policy=DelistingPolicy(DelistedPositionHandling.SELL_AT_LAST_PRICE),
...     backtest_end_date=calendar[-1].date(),
... )
>>> [(exit.ticker, str(exit.date.date())) for exit in plan.exits]
[('AAA', '2024-01-04')]