Skip to content

Core API Reference

The core package provides foundational utilities used across all modules.

Overview

The core package contains:

  • Exceptions - Custom exception hierarchy
  • Configuration - Configuration management
  • Types - Type definitions and aliases
  • Utilities - Common utility functions

Core Package

portfolio_management.core

Core foundation for the portfolio management toolkit.

This package provides the foundational utilities, exceptions, configuration, and types for the entire system. It serves as the stable base layer upon which other components like data loading, factor computation, and portfolio construction are built.

The package exposes its main components at the top level for convenient access.

Example

Importing and using common components from the core package.

from portfolio_management.core import ( ... _run_in_parallel, ... log_duration, ... PortfolioManagementError ... )

def sample_task(x: int) -> int: ... return x * x

with log_duration("Running sample parallel task"): ... # Executes sample_task(i) for i in [1, 2, 3] in parallel ... # The arguments must be wrapped in tuples, e.g., [(1,), (2,), (3,)] ... results = _run_in_parallel(sample_task, [(1,), (2,), (3,)], max_workers=2) ... sorted(results) == [1, 4, 9] True

try: ... raise PortfolioManagementError("Something went wrong.") ... except PortfolioManagementError as e: ... print(f"Caught expected error: {e}") Caught expected error: Something went wrong.

Key Components
  • Exceptions: A comprehensive hierarchy of custom exceptions, with PortfolioManagementError as the base. This allows for granular error handling across the application.
  • Configuration: Global constants and mappings, such as column names (STOOQ_COLUMNS) and region-to-currency maps.
  • Utilities: High-level helper functions, including _run_in_parallel for efficient parallel processing and log_duration for timing code blocks.
  • Types: Common type aliases (SymbolType, DateLike) and TypedDicts for structuring data like prices and returns, enhancing static analysis.
  • Protocols: typing.Protocol definitions that establish contracts for key components like data loaders (DataLoaderProtocol) and factor computation engines (FactorProtocol), enabling a modular architecture.

AssetSelectionError

Bases: PortfolioManagementError

Raised for errors that occur during the asset selection phase.

Source code in src/portfolio_management/core/exceptions.py 
161
162
class AssetSelectionError(PortfolioManagementError):
    """Raised for errors that occur during the asset selection phase."""

BacktestError

Bases: PortfolioManagementError

Base exception for backtest errors.

Source code in src/portfolio_management/core/exceptions.py 
118
119
class BacktestError(PortfolioManagementError):
    """Base exception for backtest errors."""

ClassificationError

Bases: PortfolioManagementError

Raised for errors related to asset classification (e.g., sector, industry).

Source code in src/portfolio_management/core/exceptions.py 
165
166
class ClassificationError(PortfolioManagementError):
    """Raised for errors related to asset classification (e.g., sector, industry)."""

ConfigurationError

Bases: PortfolioManagementError

Configuration validation error.

Source code in src/portfolio_management/core/exceptions.py 
145
146
147
148
149
150
151
152
153
154
155
class ConfigurationError(PortfolioManagementError):
    """Configuration validation error."""

    def __init__(self, config_path: Path | None, reason: str):
        self.config_path = config_path
        self.reason = reason
        msg = "Configuration error"
        if config_path:
            msg += f" in {config_path}"
        msg += f": {reason}"
        super().__init__(msg)

ConstraintViolationError

Bases: PortfolioConstructionError

Portfolio violates constraints.

Source code in src/portfolio_management/core/exceptions.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class ConstraintViolationError(PortfolioConstructionError):
    """Portfolio violates constraints."""

    def __init__(self, constraint_name: str, value: float, limit: float):
        self.constraint_name = constraint_name
        self.value = value
        self.limit = limit
        super().__init__(
            f"Constraint '{constraint_name}' violated: {value:.4f} exceeds limit {limit:.4f}",
        )

DataError

Bases: PortfolioManagementError

Base exception for data-related errors.

Source code in src/portfolio_management/core/exceptions.py 
31
32
class DataError(PortfolioManagementError):
    """Base exception for data-related errors."""

DataLoadError

Bases: DataError

Error loading data from a file.

Source code in src/portfolio_management/core/exceptions.py 
35
36
37
38
39
40
41
class DataLoadError(DataError):
    """Error loading data from a file."""

    def __init__(self, path: Path, reason: str):
        self.path = path
        self.reason = reason
        super().__init__(f"Failed to load data from {path}: {reason}")

DataQualityError

Bases: DataError

Raised when data quality is insufficient for processing.

Source code in src/portfolio_management/core/exceptions.py 
61
62
class DataQualityError(DataError):
    """Raised when data quality is insufficient for processing."""

DataValidationError

Bases: DataError

Error validating data.

Source code in src/portfolio_management/core/exceptions.py 
44
45
46
47
48
49
class DataValidationError(DataError):
    """Error validating data."""

    def __init__(self, message: str, invalid_rows: int | None = None):
        self.invalid_rows = invalid_rows
        super().__init__(message)

DependencyNotInstalledError

Bases: PortfolioManagementError

Raised when an optional runtime dependency is missing.

Source code in src/portfolio_management/core/exceptions.py 
65
66
67
68
69
70
71
72
73
class DependencyNotInstalledError(PortfolioManagementError):
    """Raised when an optional runtime dependency is missing."""

    def __init__(self, package: str, *, context: str) -> None:
        self.package = package
        self.context = context
        super().__init__(
            f"{package} is required {context}. Please install {package} before continuing.",
        )

InstrumentMatchError

Bases: DataError

Error matching instruments.

Source code in src/portfolio_management/core/exceptions.py 
52
53
54
55
56
57
58
class InstrumentMatchError(DataError):
    """Error matching instruments."""

    def __init__(self, unmatched_symbols: list[str], message: str | None = None):
        self.unmatched_symbols = unmatched_symbols
        msg = message or f"Failed to match {len(unmatched_symbols)} instruments"
        super().__init__(msg)

InsufficientDataError

Bases: PortfolioConstructionError, BacktestError

Not enough data to run backtest.

Source code in src/portfolio_management/core/exceptions.py 
122
123
124
125
126
127
128
129
130
class InsufficientDataError(PortfolioConstructionError, BacktestError):
    """Not enough data to run backtest."""

    def __init__(self, required_periods: int, available_periods: int):
        self.required_periods = required_periods
        self.available_periods = available_periods
        super().__init__(
            f"Insufficient data: need {required_periods} periods, have {available_periods}",
        )

OptimizationError

Bases: PortfolioConstructionError

Raised when the portfolio optimization process fails to converge.

Source code in src/portfolio_management/core/exceptions.py 
104
105
106
107
108
109
110
111
112
class OptimizationError(PortfolioConstructionError):
    """Raised when the portfolio optimization process fails to converge."""

    def __init__(self, *, strategy_name: str, message: str | None = None) -> None:
        self.strategy_name = strategy_name
        final_message = (
            message or f"Optimization failed for strategy: '{strategy_name}'."
        )
        super().__init__(final_message)

PortfolioConstructionError

Bases: PortfolioManagementError

Base exception for portfolio construction errors.

Source code in src/portfolio_management/core/exceptions.py 
79
80
class PortfolioConstructionError(PortfolioManagementError):
    """Base exception for portfolio construction errors."""

PortfolioManagementError

Bases: Exception

Base exception for all errors in the portfolio management toolkit.

Source code in src/portfolio_management/core/exceptions.py 
24
25
class PortfolioManagementError(Exception):
    """Base exception for all errors in the portfolio management toolkit."""

ReturnCalculationError

Bases: PortfolioManagementError

Raised when return calculation fails, often due to missing data.

Source code in src/portfolio_management/core/exceptions.py 
169
170
class ReturnCalculationError(PortfolioManagementError):
    """Raised when return calculation fails, often due to missing data."""

StrategyError

Bases: PortfolioConstructionError

Error constructing portfolio with a given strategy.

Source code in src/portfolio_management/core/exceptions.py 
83
84
85
86
87
88
89
class StrategyError(PortfolioConstructionError):
    """Error constructing portfolio with a given strategy."""

    def __init__(self, strategy_name: str, reason: str):
        self.strategy_name = strategy_name
        self.reason = reason
        super().__init__(f"Strategy '{strategy_name}' failed: {reason}")

UniverseLoadError

Bases: PortfolioManagementError

Raised when a universe definition cannot be loaded or parsed.

Source code in src/portfolio_management/core/exceptions.py 
173
174
class UniverseLoadError(PortfolioManagementError):
    """Raised when a universe definition cannot be loaded or parsed."""

CacheProtocol

Bases: Protocol

Protocol for factor and eligibility caching.

Defines the interface that cache implementations must provide for storing and retrieving computed factor scores and eligibility masks.

Source code in src/portfolio_management/core/protocols.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class CacheProtocol(Protocol):
    """Protocol for factor and eligibility caching.

    Defines the interface that cache implementations must provide for
    storing and retrieving computed factor scores and eligibility masks.
    """

    def get(
        self,
        dataset: pd.DataFrame,
        config: dict[str, Any],
        start_date: datetime.date,
        end_date: datetime.date,
    ) -> pd.Series | pd.DataFrame | None:
        """Retrieve cached result if available and valid.

        Args:
            dataset: Input data used for computation
            config: Configuration parameters
            start_date: Start date for the computation
            end_date: End date for the computation

        Returns:
            Cached result if available and valid, None otherwise

        """
        ...

    def put(
        self,
        dataset: pd.DataFrame,
        config: dict[str, Any],
        start_date: datetime.date,
        end_date: datetime.date,
        result: pd.Series | pd.DataFrame,
    ) -> None:
        """Store computed result in cache.

        Args:
            dataset: Input data used for computation
            config: Configuration parameters
            start_date: Start date for the computation
            end_date: End date for the computation
            result: Computed result to cache

        """
        ...

get(dataset, config, start_date, end_date)

Retrieve cached result if available and valid.

Parameters:

Name Type Description Default
dataset DataFrame

Input data used for computation

 required
config dict[str, Any]

Configuration parameters

 required
start_date date

Start date for the computation

 required
end_date date

End date for the computation

 required

Returns:

Type Description
Series | DataFrame | None

Cached result if available and valid, None otherwise
Source code in src/portfolio_management/core/protocols.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get(
    self,
    dataset: pd.DataFrame,
    config: dict[str, Any],
    start_date: datetime.date,
    end_date: datetime.date,
) -> pd.Series | pd.DataFrame | None:
    """Retrieve cached result if available and valid.

    Args:
        dataset: Input data used for computation
        config: Configuration parameters
        start_date: Start date for the computation
        end_date: End date for the computation

    Returns:
        Cached result if available and valid, None otherwise

    """
    ...

put(dataset, config, start_date, end_date, result)

Store computed result in cache.

Parameters:

Name Type Description Default
dataset DataFrame

Input data used for computation

 required
config dict[str, Any]

Configuration parameters

 required
start_date date

Start date for the computation

 required
end_date date

End date for the computation

 required
result Series | DataFrame

Computed result to cache

 required
Source code in src/portfolio_management/core/protocols.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def put(
    self,
    dataset: pd.DataFrame,
    config: dict[str, Any],
    start_date: datetime.date,
    end_date: datetime.date,
    result: pd.Series | pd.DataFrame,
) -> None:
    """Store computed result in cache.

    Args:
        dataset: Input data used for computation
        config: Configuration parameters
        start_date: Start date for the computation
        end_date: End date for the computation
        result: Computed result to cache

    """
    ...

DataLoaderProtocol

Bases: Protocol

Protocol for data loading operations.

Defines the interface for loading price or returns data from various sources (CSV, Parquet, databases, etc.).

Source code in src/portfolio_management/core/protocols.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class DataLoaderProtocol(Protocol):
    """Protocol for data loading operations.

    Defines the interface for loading price or returns data from
    various sources (CSV, Parquet, databases, etc.).
    """

    def load_prices(
        self,
        symbols: list[str],
        start_date: datetime.date,
        end_date: datetime.date,
    ) -> pd.DataFrame:
        """Load price data for specified symbols and date range.

        Args:
            symbols: List of ticker symbols to load
            start_date: Start date for data
            end_date: End date for data

        Returns:
            DataFrame with prices (index=dates, columns=tickers)

        """
        ...

    def load_returns(
        self,
        symbols: list[str],
        start_date: datetime.date,
        end_date: datetime.date,
    ) -> pd.DataFrame:
        """Load returns data for specified symbols and date range.

        Args:
            symbols: List of ticker symbols to load
            start_date: Start date for data
            end_date: End date for data

        Returns:
            DataFrame with returns (index=dates, columns=tickers)

        """
        ...

load_prices(symbols, start_date, end_date)

Load price data for specified symbols and date range.

Parameters:

Name Type Description Default
symbols list[str]

List of ticker symbols to load

 required
start_date date

Start date for data

 required
end_date date

End date for data

 required

Returns:

Type Description
DataFrame

DataFrame with prices (index=dates, columns=tickers)
Source code in src/portfolio_management/core/protocols.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def load_prices(
    self,
    symbols: list[str],
    start_date: datetime.date,
    end_date: datetime.date,
) -> pd.DataFrame:
    """Load price data for specified symbols and date range.

    Args:
        symbols: List of ticker symbols to load
        start_date: Start date for data
        end_date: End date for data

    Returns:
        DataFrame with prices (index=dates, columns=tickers)

    """
    ...

load_returns(symbols, start_date, end_date)

Load returns data for specified symbols and date range.

Parameters:

Name Type Description Default
symbols list[str]

List of ticker symbols to load

 required
start_date date

Start date for data

 required
end_date date

End date for data

 required

Returns:

Type Description
DataFrame

DataFrame with returns (index=dates, columns=tickers)
Source code in src/portfolio_management/core/protocols.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def load_returns(
    self,
    symbols: list[str],
    start_date: datetime.date,
    end_date: datetime.date,
) -> pd.DataFrame:
    """Load returns data for specified symbols and date range.

    Args:
        symbols: List of ticker symbols to load
        start_date: Start date for data
        end_date: End date for data

    Returns:
        DataFrame with returns (index=dates, columns=tickers)

    """
    ...

EligibilityProtocol

Bases: Protocol

Protocol for eligibility computation.

Defines the interface for determining which assets are eligible for selection at each rebalance date (point-in-time filtering).

Source code in src/portfolio_management/core/protocols.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class EligibilityProtocol(Protocol):
    """Protocol for eligibility computation.

    Defines the interface for determining which assets are eligible
    for selection at each rebalance date (point-in-time filtering).
    """

    def compute_eligibility(
        self,
        returns: pd.DataFrame,
        date: datetime.date,
        min_history: int,
    ) -> pd.Series:
        """Compute eligibility mask for assets at a given date.

        Args:
            returns: Historical returns data
            date: Date to compute eligibility for
            min_history: Minimum history required

        Returns:
            Boolean Series indicating eligible assets (True = eligible)

        """
        ...

compute_eligibility(returns, date, min_history)

Compute eligibility mask for assets at a given date.

Parameters:

Name Type Description Default
returns DataFrame

Historical returns data

 required
date date

Date to compute eligibility for

 required
min_history int

Minimum history required

 required

Returns:

Type Description
Series

Boolean Series indicating eligible assets (True = eligible)
Source code in src/portfolio_management/core/protocols.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def compute_eligibility(
    self,
    returns: pd.DataFrame,
    date: datetime.date,
    min_history: int,
) -> pd.Series:
    """Compute eligibility mask for assets at a given date.

    Args:
        returns: Historical returns data
        date: Date to compute eligibility for
        min_history: Minimum history required

    Returns:
        Boolean Series indicating eligible assets (True = eligible)

    """
    ...

FactorProtocol

Bases: Protocol

Protocol for factor computation.

Defines the interface for computing factor scores (momentum, volatility, value, quality, etc.) from returns or other data.

Source code in src/portfolio_management/core/protocols.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class FactorProtocol(Protocol):
    """Protocol for factor computation.

    Defines the interface for computing factor scores (momentum,
    volatility, value, quality, etc.) from returns or other data.
    """

    def compute_scores(
        self,
        returns: pd.DataFrame,
        date: datetime.date,
        lookback: int,
    ) -> pd.Series:
        """Compute factor scores for assets at a given date.

        Args:
            returns: Historical returns data
            date: Date to compute scores for (uses data up to this date)
            lookback: Number of periods to look back

        Returns:
            Series of factor scores (index=tickers, higher is better)

        """
        ...

    @property
    def name(self) -> str:
        """Return the factor name (e.g., 'momentum', 'low_vol').

        Returns:
            Factor name string

        """
        ...

name property

Return the factor name (e.g., 'momentum', 'low_vol').

Returns:

Type Description
str

Factor name string

compute_scores(returns, date, lookback)

Compute factor scores for assets at a given date.

Parameters:

Name Type Description Default
returns DataFrame

Historical returns data

 required
date date

Date to compute scores for (uses data up to this date)

 required
lookback int

Number of periods to look back

 required

Returns:

Type Description
Series

Series of factor scores (index=tickers, higher is better)
Source code in src/portfolio_management/core/protocols.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def compute_scores(
    self,
    returns: pd.DataFrame,
    date: datetime.date,
    lookback: int,
) -> pd.Series:
    """Compute factor scores for assets at a given date.

    Args:
        returns: Historical returns data
        date: Date to compute scores for (uses data up to this date)
        lookback: Number of periods to look back

    Returns:
        Series of factor scores (index=tickers, higher is better)

    """
    ...

log_duration(step)

Context manager to log the duration of an operation.

This is useful for performance monitoring and identifying bottlenecks. It logs the completion time upon exiting the context.

Parameters:

Name Type Description Default
step str

A description of the operation being timed.

 required

Yields:

Type Description
Generator[None]

None.
Example

import time import logging import sys

Configure a logger to capture output for the doctest

logger = logging.getLogger("portfolio_management.core.utils") handler = logging.StreamHandler(sys.stdout) logger.addHandler(handler) logger.setLevel(logging.INFO)

with log_duration("data processing"): # doctest: +ELLIPSIS ... time.sleep(0.01) data processing completed in ...s

Clean up the handler to avoid affecting other tests

logger.removeHandler(handler)

Source code in src/portfolio_management/core/utils.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@contextmanager
def log_duration(step: str) -> Generator[None]:
    """Context manager to log the duration of an operation.

    This is useful for performance monitoring and identifying bottlenecks.
    It logs the completion time upon exiting the context.

    Args:
        step: A description of the operation being timed.

    Yields:
        None.

    Example:
        >>> import time
        >>> import logging
        >>> import sys
        >>>
        >>> # Configure a logger to capture output for the doctest
        >>> logger = logging.getLogger("portfolio_management.core.utils")
        >>> handler = logging.StreamHandler(sys.stdout)
        >>> logger.addHandler(handler)
        >>> logger.setLevel(logging.INFO)
        >>>
        >>> with log_duration("data processing"): # doctest: +ELLIPSIS
        ...     time.sleep(0.01)
        data processing completed in ...s
        >>>
        >>> # Clean up the handler to avoid affecting other tests
        >>> logger.removeHandler(handler)

    """
    start_time = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start_time
        LOGGER.info("%s completed in %.2fs", step, elapsed)

options: show_root_heading: true show_source: false members_order: source group_by_category: true show_category_heading: true