Skip to content

Macro API Reference

The macro package provides macroeconomic signal infrastructure and regime detection.

Overview

The macro package contains:

  • Models - Data models for macro signals
  • Provider - Macro data provider interfaces
  • Regime - Regime detection and gating

Macro Package

portfolio_management.macro

Macroeconomic signal provider and regime gating.

This package provides infrastructure for loading macroeconomic time series and defining regime configurations that can gate asset selection decisions. The current implementation uses local Stooq data files and features a NoOp regime gate that serves as a placeholder for future logic.

Key Classes: - MacroSignalProvider: Locates and loads macro series from a local Stooq data directory. - RegimeGate: A NoOp class that applies placeholder regime gating rules to an asset selection. It always returns neutral signals. - MacroSeries: A data model for a single macro time series. - RegimeConfig: Configuration for regime detection rules.

Usage Example

from portfolio_management.macro import RegimeConfig, RegimeGate

The RegimeGate is a NoOp and always returns a neutral regime.

config = RegimeConfig(enable_gating=True) gate = RegimeGate(config) current_regime = gate.get_current_regime() print(current_regime)

Data Sources: - This module is designed to work with local data files downloaded from Stooq (https://stooq.com/). It expects a directory structure where macroeconomic data is organized, typically by region and category.

MacroSeries dataclass

Represents metadata for a macroeconomic time series.

This immutable dataclass captures key information about a macro series as located within a Stooq data directory. It does not contain the series data itself but rather points to its location and describes it.

Attributes:

Name Type Description
ticker str

The unique identifier for the macro series (e.g., "gdp.us").
rel_path str

The relative path to the series data file within the Stooq directory.
start_date str

The first available date in the series (YYYY-MM-DD). Note: This is often populated after loading the data.
end_date str

The last available date in the series (YYYY-MM-DD). Note: This is often populated after loading the data.
region str

The geographic region of the series (e.g., "us", "de").
category str

The category of the series (e.g., "economic", "pmi").
Source code in src/portfolio_management/macro/models.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class MacroSeries:
    """Represents metadata for a macroeconomic time series.

    This immutable dataclass captures key information about a macro series
    as located within a Stooq data directory. It does not contain the series
    data itself but rather points to its location and describes it.

    Attributes:
        ticker (str): The unique identifier for the macro series (e.g., "gdp.us").
        rel_path (str): The relative path to the series data file within the
                        Stooq directory.
        start_date (str): The first available date in the series (YYYY-MM-DD).
                          Note: This is often populated after loading the data.
        end_date (str): The last available date in the series (YYYY-MM-DD).
                        Note: This is often populated after loading the data.
        region (str): The geographic region of the series (e.g., "us", "de").
        category (str): The category of the series (e.g., "economic", "pmi").

    """

    ticker: str
    rel_path: str
    start_date: str
    end_date: str
    region: str = ""
    category: str = ""

RegimeConfig dataclass

Configuration for regime detection and gating rules.

This dataclass defines the parameters for detecting market regimes (e.g., recession, risk-off) and specifies whether this gating logic should be applied to an asset selection.

Note

The logic that uses this configuration is currently a NoOp (No Operation) placeholder. The system is designed to accept this configuration, but the gating itself is not yet implemented.

Attributes:

Name Type Description
recession_indicator str

The ticker for the series used to indicate a recession. Defaults to None.
risk_off_threshold float

A threshold for detecting a risk-off environment. Defaults to None.
enable_gating bool

If True, enables the regime gating step in the selection process. Defaults to False.
custom_rules dict

A dictionary for any custom or experimental rules. Reserved for future use. Defaults to None.
Source code in src/portfolio_management/macro/models.py 
 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
 72
 73
 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
@dataclass
class RegimeConfig:
    """Configuration for regime detection and gating rules.

    This dataclass defines the parameters for detecting market regimes (e.g.,
    recession, risk-off) and specifies whether this gating logic should be
    applied to an asset selection.

    Note:
        The logic that uses this configuration is currently a NoOp (No Operation)
        placeholder. The system is designed to accept this configuration, but
        the gating itself is not yet implemented.

    Attributes:
        recession_indicator (str, optional): The ticker for the series used to
            indicate a recession. Defaults to `None`.
        risk_off_threshold (float, optional): A threshold for detecting a
            risk-off environment. Defaults to `None`.
        enable_gating (bool): If `True`, enables the regime gating step in the
            selection process. Defaults to `False`.
        custom_rules (dict, optional): A dictionary for any custom or
            experimental rules. Reserved for future use. Defaults to `None`.

    """

    recession_indicator: str | None = None
    risk_off_threshold: float | None = None
    enable_gating: bool = False
    custom_rules: dict[str, Any] | None = None

    def is_enabled(self) -> bool:
        """Check if regime gating is enabled in the configuration.

        Returns:
            `True` if gating is enabled, `False` otherwise.

        """
        return self.enable_gating

    def validate(self) -> None:
        """Validate the regime configuration parameters.

        This method ensures that the configuration values are logical and
        within acceptable bounds.

        Raises:
            ValueError: If `risk_off_threshold` is negative.

        Example:
            >>> config = RegimeConfig(risk_off_threshold=-0.5)
            >>> try:
            ...     config.validate()
            ... except ValueError as e:
            ...     print(e)
            risk_off_threshold must be non-negative, got -0.5

        """
        if self.risk_off_threshold is not None and self.risk_off_threshold < 0:
            raise ValueError(
                f"risk_off_threshold must be non-negative, got {self.risk_off_threshold}",
            )

is_enabled()

Check if regime gating is enabled in the configuration.

Returns:

Type Description
bool

True if gating is enabled, False otherwise.
Source code in src/portfolio_management/macro/models.py 
72
73
74
75
76
77
78
79
def is_enabled(self) -> bool:
    """Check if regime gating is enabled in the configuration.

    Returns:
        `True` if gating is enabled, `False` otherwise.

    """
    return self.enable_gating

validate()

Validate the regime configuration parameters.

This method ensures that the configuration values are logical and within acceptable bounds.

Raises:

Type Description
ValueError

If risk_off_threshold is negative.
Example

config = RegimeConfig(risk_off_threshold=-0.5) try: ... config.validate() ... except ValueError as e: ... print(e) risk_off_threshold must be non-negative, got -0.5

Source code in src/portfolio_management/macro/models.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def validate(self) -> None:
    """Validate the regime configuration parameters.

    This method ensures that the configuration values are logical and
    within acceptable bounds.

    Raises:
        ValueError: If `risk_off_threshold` is negative.

    Example:
        >>> config = RegimeConfig(risk_off_threshold=-0.5)
        >>> try:
        ...     config.validate()
        ... except ValueError as e:
        ...     print(e)
        risk_off_threshold must be non-negative, got -0.5

    """
    if self.risk_off_threshold is not None and self.risk_off_threshold < 0:
        raise ValueError(
            f"risk_off_threshold must be non-negative, got {self.risk_off_threshold}",
        )

MacroSignalProvider

Load macroeconomic time series from local Stooq data directories.

This class provides methods to locate and load macro series (e.g., GDP, PMI, yields) from a structured local directory containing data from Stooq. It supports locating single or multiple series and loading their data into pandas DataFrames.

Attributes:

Name Type Description
data_dir Path

The root directory where the Stooq data is stored.
Data Source

This provider expects data files to be in the format provided by Stooq (https://stooq.com/) and organized in a searchable directory structure.

Source code in src/portfolio_management/macro/provider.py 
 22
 23
 24
 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
 72
 73
 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
118
119
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
155
156
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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class MacroSignalProvider:
    """Load macroeconomic time series from local Stooq data directories.

    This class provides methods to locate and load macro series (e.g., GDP,
    PMI, yields) from a structured local directory containing data from Stooq.
    It supports locating single or multiple series and loading their data into
    pandas DataFrames.

    Attributes:
        data_dir (Path): The root directory where the Stooq data is stored.

    Data Source:
        This provider expects data files to be in the format provided by
        Stooq (https://stooq.com/) and organized in a searchable directory
        structure.

    """

    def __init__(self, data_dir: Path | str) -> None:
        """Initialize the MacroSignalProvider.

        Args:
            data_dir: The path to the Stooq data directory root.

        Raises:
            DataLoadError: If the specified `data_dir` does not exist.

        """
        self.data_dir = Path(data_dir)
        if not self.data_dir.exists():
            raise DataLoadError(self.data_dir, "Macro data directory not found")

        LOGGER.info("Initialized MacroSignalProvider with data_dir=%s", self.data_dir)

    def locate_series(self, ticker: str) -> MacroSeries | None:
        """Locate a macro series file in the Stooq data directory.

        This method searches for a series file by its ticker, following a set
        of predefined potential directory structures common to Stooq data.

        Args:
            ticker: The ticker symbol for the macro series (e.g., "gdp.us").

        Returns:
            A `MacroSeries` object with metadata if the file is found,
            otherwise `None`.

        """
        potential_paths = self._generate_search_paths(ticker)

        for potential_path in potential_paths:
            full_path = self.data_dir / potential_path
            if full_path.exists():
                LOGGER.debug("Located series %s at %s", ticker, potential_path)
                region, category = self._parse_path_metadata(potential_path)
                return MacroSeries(
                    ticker=ticker,
                    rel_path=str(potential_path),
                    start_date="",  # To be filled when data is loaded
                    end_date="",  # To be filled when data is loaded
                    region=region,
                    category=category,
                )

        LOGGER.warning("Could not locate series for ticker: %s", ticker)
        return None

    def locate_multiple_series(self, tickers: list[str]) -> dict[str, MacroSeries]:
        """Locate multiple macro series files.

        Args:
            tickers: A list of ticker symbols to locate.

        Returns:
            A dictionary mapping the tickers that were found to their
            `MacroSeries` metadata. Tickers that are not found are excluded.

        """
        result = {}
        for ticker in tickers:
            series = self.locate_series(ticker)
            if series is not None:
                result[ticker] = series
            else:
                LOGGER.info("Series not found: %s", ticker)

        LOGGER.info(
            "Located %d out of %d requested series",
            len(result),
            len(tickers),
        )
        return result

    def load_series_data(
        self,
        ticker: str,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> pd.DataFrame | None:
        """Load data for a macro series from its file.

        This method first locates the series file and then loads its contents
        into a pandas DataFrame. It can optionally filter the data by a
        date range.

        Args:
            ticker: The ticker symbol for the macro series.
            start_date: An optional start date to filter the data (inclusive),
                        in "YYYY-MM-DD" format.
            end_date: An optional end date to filter the data (inclusive),
                      in "YYYY-MM-DD" format.

        Returns:
            A pandas DataFrame with the series data, indexed by date, if the
            file is found and valid. Otherwise, returns `None`.

        Raises:
            DataValidationError: If the series file exists but cannot be read
                                 or parsed as a valid CSV.

        """
        series = self.locate_series(ticker)
        if series is None:
            return None

        full_path = self.data_dir / series.rel_path

        try:
            df = pd.read_csv(full_path, parse_dates=["date"])

            if start_date:
                df = df[df["date"] >= pd.to_datetime(start_date)]
            if end_date:
                df = df[df["date"] <= pd.to_datetime(end_date)]

            df = df.set_index("date").sort_index()

            LOGGER.debug(
                "Loaded %d rows for series %s (date range: %s to %s)",
                len(df),
                ticker,
                df.index[0].date() if not df.empty else "N/A",
                df.index[-1].date() if not df.empty else "N/A",
            )

            return df

        except Exception as e:
            raise DataValidationError(
                f"Failed to load series {ticker} from {full_path}: {e}",
            ) from e

    def _generate_search_paths(self, ticker: str) -> list[str]:
        """Generate potential file paths for a given ticker.

        This helper method constructs a list of likely file paths based on
        common Stooq directory structures.

        Args:
            ticker: The ticker symbol (e.g., "gdp.us").

        Returns:
            A list of potential relative file paths to search for.

        """
        parts = ticker.lower().split(".")
        if len(parts) >= 2:
            base_ticker = parts[0]
            region = parts[1]
        else:
            base_ticker = ticker.lower()
            region = "us"  # Default to US if no region specified

        return [
            f"data/daily/{region}/economic/{base_ticker}.txt",
            f"data/daily/{region}/indicators/{base_ticker}.txt",
            f"data/daily/{region}/macro/{base_ticker}.txt",
            f"data/daily/{region}/{base_ticker}.txt",
            f"{region}/economic/{base_ticker}.txt",
            f"{region}/indicators/{base_ticker}.txt",
            f"{region}/{base_ticker}.txt",
            f"{base_ticker}.txt",
        ]

    def _parse_path_metadata(self, rel_path: str) -> tuple[str, str]:
        """Parse the region and category from a relative file path.

        Args:
            rel_path: The relative path to the series file.

        Returns:
            A tuple containing the extracted region and category as strings.

        Example:
            >>> provider = MacroSignalProvider("data/stooq")
            >>> region, category = provider._parse_path_metadata(
            ...     "data/daily/us/economic/gdp.txt"
            ... )
            >>> print(f"Region: {region}, Category: {category}")
            Region: us, Category: economic

        """
        parts = Path(rel_path).parts
        region = ""
        category = ""

        # Try to extract region and category from path structure
        if "daily" in parts:
            idx = parts.index("daily")
            if idx + 1 < len(parts):
                region = parts[idx + 1]
            if idx + 2 < len(parts) - 1:  # -1 to exclude filename
                category = parts[idx + 2]
        elif len(parts) >= 2:
            region = parts[0]
            if len(parts) >= 3:
                category = parts[1]

        return region, category

locate_series(ticker)

Locate a macro series file in the Stooq data directory.

This method searches for a series file by its ticker, following a set of predefined potential directory structures common to Stooq data.

Parameters:

Name Type Description Default
ticker str

The ticker symbol for the macro series (e.g., "gdp.us").

 required

Returns:

Type Description
MacroSeries | None

A MacroSeries object with metadata if the file is found,
MacroSeries | None

otherwise None.
Source code in src/portfolio_management/macro/provider.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def locate_series(self, ticker: str) -> MacroSeries | None:
    """Locate a macro series file in the Stooq data directory.

    This method searches for a series file by its ticker, following a set
    of predefined potential directory structures common to Stooq data.

    Args:
        ticker: The ticker symbol for the macro series (e.g., "gdp.us").

    Returns:
        A `MacroSeries` object with metadata if the file is found,
        otherwise `None`.

    """
    potential_paths = self._generate_search_paths(ticker)

    for potential_path in potential_paths:
        full_path = self.data_dir / potential_path
        if full_path.exists():
            LOGGER.debug("Located series %s at %s", ticker, potential_path)
            region, category = self._parse_path_metadata(potential_path)
            return MacroSeries(
                ticker=ticker,
                rel_path=str(potential_path),
                start_date="",  # To be filled when data is loaded
                end_date="",  # To be filled when data is loaded
                region=region,
                category=category,
            )

    LOGGER.warning("Could not locate series for ticker: %s", ticker)
    return None

locate_multiple_series(tickers)

Locate multiple macro series files.

Parameters:

Name Type Description Default
tickers list[str]

A list of ticker symbols to locate.

 required

Returns:

Type Description
dict[str, MacroSeries]

A dictionary mapping the tickers that were found to their
dict[str, MacroSeries]

MacroSeries metadata. Tickers that are not found are excluded.
Source code in src/portfolio_management/macro/provider.py 
 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
def locate_multiple_series(self, tickers: list[str]) -> dict[str, MacroSeries]:
    """Locate multiple macro series files.

    Args:
        tickers: A list of ticker symbols to locate.

    Returns:
        A dictionary mapping the tickers that were found to their
        `MacroSeries` metadata. Tickers that are not found are excluded.

    """
    result = {}
    for ticker in tickers:
        series = self.locate_series(ticker)
        if series is not None:
            result[ticker] = series
        else:
            LOGGER.info("Series not found: %s", ticker)

    LOGGER.info(
        "Located %d out of %d requested series",
        len(result),
        len(tickers),
    )
    return result

load_series_data(ticker, start_date=None, end_date=None)

Load data for a macro series from its file.

This method first locates the series file and then loads its contents into a pandas DataFrame. It can optionally filter the data by a date range.

Parameters:

Name Type Description Default
ticker str

The ticker symbol for the macro series.

 required
start_date str | None

An optional start date to filter the data (inclusive), in "YYYY-MM-DD" format.

 None
end_date str | None

An optional end date to filter the data (inclusive), in "YYYY-MM-DD" format.

 None

Returns:

Type Description
DataFrame | None

A pandas DataFrame with the series data, indexed by date, if the
DataFrame | None

file is found and valid. Otherwise, returns None.

Raises:

Type Description
DataValidationError

If the series file exists but cannot be read or parsed as a valid CSV.
Source code in src/portfolio_management/macro/provider.py 
115
116
117
118
119
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def load_series_data(
    self,
    ticker: str,
    start_date: str | None = None,
    end_date: str | None = None,
) -> pd.DataFrame | None:
    """Load data for a macro series from its file.

    This method first locates the series file and then loads its contents
    into a pandas DataFrame. It can optionally filter the data by a
    date range.

    Args:
        ticker: The ticker symbol for the macro series.
        start_date: An optional start date to filter the data (inclusive),
                    in "YYYY-MM-DD" format.
        end_date: An optional end date to filter the data (inclusive),
                  in "YYYY-MM-DD" format.

    Returns:
        A pandas DataFrame with the series data, indexed by date, if the
        file is found and valid. Otherwise, returns `None`.

    Raises:
        DataValidationError: If the series file exists but cannot be read
                             or parsed as a valid CSV.

    """
    series = self.locate_series(ticker)
    if series is None:
        return None

    full_path = self.data_dir / series.rel_path

    try:
        df = pd.read_csv(full_path, parse_dates=["date"])

        if start_date:
            df = df[df["date"] >= pd.to_datetime(start_date)]
        if end_date:
            df = df[df["date"] <= pd.to_datetime(end_date)]

        df = df.set_index("date").sort_index()

        LOGGER.debug(
            "Loaded %d rows for series %s (date range: %s to %s)",
            len(df),
            ticker,
            df.index[0].date() if not df.empty else "N/A",
            df.index[-1].date() if not df.empty else "N/A",
        )

        return df

    except Exception as e:
        raise DataValidationError(
            f"Failed to load series {ticker} from {full_path}: {e}",
        ) from e

RegimeGate

Apply regime-based gating to asset selection (currently NoOp).

This class is designed to apply rules to an asset selection based on macroeconomic conditions. However, the current implementation is a NoOp stub. It always returns neutral signals and leaves the selection unchanged, irrespective of the provided configuration. This ensures that the system is ready for future regime logic without affecting current workflows.

Attributes:

Name Type Description
config RegimeConfig

A configuration object that defines the rules for regime detection. Currently, this configuration is logged but not used for any logic.
Example

from portfolio_management.macro.models import RegimeConfig from portfolio_management.macro.regime import RegimeGate

Mock asset class for the example

class MockAsset: ... def init(self, symbol): ... self.symbol = symbol

assets = [MockAsset("AAPL"), MockAsset("GOOG")] config = RegimeConfig(enable_gating=True) gate = RegimeGate(config)

The apply_gating method returns the original assets without changes.

filtered_assets = gate.apply_gating(assets) assert filtered_assets is assets

Source code in src/portfolio_management/macro/regime.py 
 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
 72
 73
 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
118
119
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
155
156
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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class RegimeGate:
    """Apply regime-based gating to asset selection (currently NoOp).

    This class is designed to apply rules to an asset selection based on
    macroeconomic conditions. However, the current implementation is a NoOp
    stub. It always returns neutral signals and leaves the selection unchanged,
    irrespective of the provided configuration. This ensures that the system
    is ready for future regime logic without affecting current workflows.

    Attributes:
        config (RegimeConfig): A configuration object that defines the rules for
                               regime detection. Currently, this configuration
                               is logged but not used for any logic.

    Example:
        >>> from portfolio_management.macro.models import RegimeConfig
        >>> from portfolio_management.macro.regime import RegimeGate
        >>>
        >>> # Mock asset class for the example
        >>> class MockAsset:
        ...     def __init__(self, symbol):
        ...         self.symbol = symbol
        >>>
        >>> assets = [MockAsset("AAPL"), MockAsset("GOOG")]
        >>> config = RegimeConfig(enable_gating=True)
        >>> gate = RegimeGate(config)
        >>>
        >>> # The apply_gating method returns the original assets without changes.
        >>> filtered_assets = gate.apply_gating(assets)
        >>> assert filtered_assets is assets

    """

    def __init__(self, config: RegimeConfig) -> None:
        """Initialize the RegimeGate.

        Args:
            config: A `RegimeConfig` object that defines the gating rules.
                    This is stored for future use but does not affect the
                    current NoOp behavior.

        """
        self.config = config
        LOGGER.info(
            "Initialized RegimeGate (gating_enabled=%s, mode=NoOp)",
            config.is_enabled(),
        )

    def apply_gating(
        self,
        assets: list[SelectedAsset],
        date: str | None = None,
    ) -> list[SelectedAsset]:
        """Apply regime gating to selected assets (currently a NoOp).

        This method is intended to filter an asset list based on the prevailing
        macroeconomic regime. In its current implementation, it acts as a
        pass-through, returning the original list of assets without any
        modifications.

        Args:
            assets: A list of `SelectedAsset` objects to potentially filter.
            date: An optional date for regime evaluation (e.g., "YYYY-MM-DD").
                  This parameter is currently ignored.

        Returns:
            The original list of assets, unchanged.

        """
        if not self.config.is_enabled():
            LOGGER.debug(
                "Regime gating disabled, passing through %d assets unchanged",
                len(assets),
            )
            return assets

        # Even if enabled, current implementation is NoOp
        LOGGER.debug(
            "Regime gating enabled but NoOp implementation active, "
            "passing through %d assets unchanged",
            len(assets),
        )
        return assets

    def get_current_regime(self, date: str | None = None) -> dict[str, str]:
        """Get the current regime classification (always returns 'neutral').

        This method is designed to return the current market regime. As a NoOp,
        it consistently returns a dictionary indicating a 'neutral' state for
        all regime types.

        Args:
            date: An optional date for which to determine the regime. This
                  parameter is currently ignored.

        Returns:
            A dictionary with predefined neutral regime classifications.

        Example:
            >>> from portfolio_management.macro.models import RegimeConfig
            >>> from portfolio_management.macro.regime import RegimeGate
            >>>
            >>> config = RegimeConfig()
            >>> gate = RegimeGate(config)
            >>> regime = gate.get_current_regime()
            >>> print(regime)
            {'recession': 'neutral', 'risk_sentiment': 'neutral', 'mode': 'noop'}

        """
        # NoOp: always return neutral regime
        regime = {
            "recession": "neutral",
            "risk_sentiment": "neutral",
            "mode": "noop",
        }

        LOGGER.debug(
            "Current regime (NoOp): %s (date=%s)",
            regime,
            date if date else "current",
        )
        return regime

    def filter_by_asset_class(
        self,
        assets: list[SelectedAsset],
        allowed_classes: list[str] | None = None,
    ) -> list[SelectedAsset]:
        """Filter assets by asset class based on regime (currently a NoOp).

        In a future implementation, this method could filter assets by their
        class (e.g., exclude equities during a risk-off regime). Currently,
        it returns the original list of assets without any filtering.

        Args:
            assets: A list of `SelectedAsset` objects to filter.
            allowed_classes: An optional list of asset classes to permit. This
                             parameter is currently ignored.

        Returns:
            The original list of assets, unchanged.

        """
        # NoOp: always pass through unchanged
        LOGGER.debug(
            "Asset class filtering (NoOp): passing through %d assets unchanged",
            len(assets),
        )
        return assets

    def adjust_selection_scores(
        self,
        assets: list[SelectedAsset],
        date: str | None = None,
    ) -> list[tuple[SelectedAsset, float]]:
        """Adjust selection scores based on regime (currently a NoOp).

        This method is intended to adjust asset scores based on regime
        conditions. In its current NoOp implementation, it returns all assets
        with a neutral score of 1.0.

        Args:
            assets: A list of `SelectedAsset` objects.
            date: An optional date for regime evaluation. This parameter is
                  currently ignored.

        Returns:
            A list of tuples, where each tuple contains the original asset and
            a neutral score of 1.0.

        """
        # NoOp: return all assets with neutral score
        scored_assets = [(asset, 1.0) for asset in assets]
        LOGGER.debug(
            "Score adjustment (NoOp): returning %d assets with neutral score 1.0",
            len(scored_assets),
        )
        return scored_assets

apply_gating(assets, date=None)

Apply regime gating to selected assets (currently a NoOp).

This method is intended to filter an asset list based on the prevailing macroeconomic regime. In its current implementation, it acts as a pass-through, returning the original list of assets without any modifications.

Parameters:

Name Type Description Default
assets list[SelectedAsset]

A list of SelectedAsset objects to potentially filter.

 required
date str | None

An optional date for regime evaluation (e.g., "YYYY-MM-DD"). This parameter is currently ignored.

 None

Returns:

Type Description
list[SelectedAsset]

The original list of assets, unchanged.
Source code in src/portfolio_management/macro/regime.py 
 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
def apply_gating(
    self,
    assets: list[SelectedAsset],
    date: str | None = None,
) -> list[SelectedAsset]:
    """Apply regime gating to selected assets (currently a NoOp).

    This method is intended to filter an asset list based on the prevailing
    macroeconomic regime. In its current implementation, it acts as a
    pass-through, returning the original list of assets without any
    modifications.

    Args:
        assets: A list of `SelectedAsset` objects to potentially filter.
        date: An optional date for regime evaluation (e.g., "YYYY-MM-DD").
              This parameter is currently ignored.

    Returns:
        The original list of assets, unchanged.

    """
    if not self.config.is_enabled():
        LOGGER.debug(
            "Regime gating disabled, passing through %d assets unchanged",
            len(assets),
        )
        return assets

    # Even if enabled, current implementation is NoOp
    LOGGER.debug(
        "Regime gating enabled but NoOp implementation active, "
        "passing through %d assets unchanged",
        len(assets),
    )
    return assets

get_current_regime(date=None)

Get the current regime classification (always returns 'neutral').

This method is designed to return the current market regime. As a NoOp, it consistently returns a dictionary indicating a 'neutral' state for all regime types.

Parameters:

Name Type Description Default
date str | None

An optional date for which to determine the regime. This parameter is currently ignored.

 None

Returns:

Type Description
dict[str, str]

A dictionary with predefined neutral regime classifications.
Example

from portfolio_management.macro.models import RegimeConfig from portfolio_management.macro.regime import RegimeGate

config = RegimeConfig() gate = RegimeGate(config) regime = gate.get_current_regime() print(regime)

Source code in src/portfolio_management/macro/regime.py 
114
115
116
117
118
119
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
def get_current_regime(self, date: str | None = None) -> dict[str, str]:
    """Get the current regime classification (always returns 'neutral').

    This method is designed to return the current market regime. As a NoOp,
    it consistently returns a dictionary indicating a 'neutral' state for
    all regime types.

    Args:
        date: An optional date for which to determine the regime. This
              parameter is currently ignored.

    Returns:
        A dictionary with predefined neutral regime classifications.

    Example:
        >>> from portfolio_management.macro.models import RegimeConfig
        >>> from portfolio_management.macro.regime import RegimeGate
        >>>
        >>> config = RegimeConfig()
        >>> gate = RegimeGate(config)
        >>> regime = gate.get_current_regime()
        >>> print(regime)
        {'recession': 'neutral', 'risk_sentiment': 'neutral', 'mode': 'noop'}

    """
    # NoOp: always return neutral regime
    regime = {
        "recession": "neutral",
        "risk_sentiment": "neutral",
        "mode": "noop",
    }

    LOGGER.debug(
        "Current regime (NoOp): %s (date=%s)",
        regime,
        date if date else "current",
    )
    return regime

filter_by_asset_class(assets, allowed_classes=None)

Filter assets by asset class based on regime (currently a NoOp).

In a future implementation, this method could filter assets by their class (e.g., exclude equities during a risk-off regime). Currently, it returns the original list of assets without any filtering.

Parameters:

Name Type Description Default
assets list[SelectedAsset]

A list of SelectedAsset objects to filter.

 required
allowed_classes list[str] | None

An optional list of asset classes to permit. This parameter is currently ignored.

 None

Returns:

Type Description
list[SelectedAsset]

The original list of assets, unchanged.
Source code in src/portfolio_management/macro/regime.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def filter_by_asset_class(
    self,
    assets: list[SelectedAsset],
    allowed_classes: list[str] | None = None,
) -> list[SelectedAsset]:
    """Filter assets by asset class based on regime (currently a NoOp).

    In a future implementation, this method could filter assets by their
    class (e.g., exclude equities during a risk-off regime). Currently,
    it returns the original list of assets without any filtering.

    Args:
        assets: A list of `SelectedAsset` objects to filter.
        allowed_classes: An optional list of asset classes to permit. This
                         parameter is currently ignored.

    Returns:
        The original list of assets, unchanged.

    """
    # NoOp: always pass through unchanged
    LOGGER.debug(
        "Asset class filtering (NoOp): passing through %d assets unchanged",
        len(assets),
    )
    return assets

adjust_selection_scores(assets, date=None)

Adjust selection scores based on regime (currently a NoOp).

This method is intended to adjust asset scores based on regime conditions. In its current NoOp implementation, it returns all assets with a neutral score of 1.0.

Parameters:

Name Type Description Default
assets list[SelectedAsset]

A list of SelectedAsset objects.

 required
date str | None

An optional date for regime evaluation. This parameter is currently ignored.

 None

Returns:

Type Description
list[tuple[SelectedAsset, float]]

A list of tuples, where each tuple contains the original asset and
list[tuple[SelectedAsset, float]]

a neutral score of 1.0.
Source code in src/portfolio_management/macro/regime.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def adjust_selection_scores(
    self,
    assets: list[SelectedAsset],
    date: str | None = None,
) -> list[tuple[SelectedAsset, float]]:
    """Adjust selection scores based on regime (currently a NoOp).

    This method is intended to adjust asset scores based on regime
    conditions. In its current NoOp implementation, it returns all assets
    with a neutral score of 1.0.

    Args:
        assets: A list of `SelectedAsset` objects.
        date: An optional date for regime evaluation. This parameter is
              currently ignored.

    Returns:
        A list of tuples, where each tuple contains the original asset and
        a neutral score of 1.0.

    """
    # NoOp: return all assets with neutral score
    scored_assets = [(asset, 1.0) for asset in assets]
    LOGGER.debug(
        "Score adjustment (NoOp): returning %d assets with neutral score 1.0",
        len(scored_assets),
    )
    return scored_assets

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