17580505858

Committed 09 Sep 2025 11:03AM UTC coverage: 96.722% (-0.6%) from 97.336%

Build # 17580505858

Build Type

Pull #48

github

Committed by

web-flow

Commit Message

Merge acef97386 into 8b571c6fc

Pull Request Pull Request #48: Autograd: gradient passthrough

Run Details

239 of 241 branches covered (99.17%)

Branch coverage included in aggregate %.

41 of 50 new or added lines in 3 files covered. (82.0%)

4 existing lines in 1 file now uncovered.

1561 of 1620 relevant lines covered (96.36%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.91

/src/macrostat/core/model.py

# -*- coding: utf-8 -*-
"""
Generic model class as a wrapper to specific implementations
"""

__author__ = ["Karl Naumann-Woleske"]
__credits__ = ["Karl Naumann-Woleske"]
__license__ = "MIT"
__version__ = "0.1.0"
__maintainer__ = ["Karl Naumann-Woleske"]

import logging
import os
import pickle

import torch

from macrostat.core.behavior import Behavior
from macrostat.core.parameters import Parameters
from macrostat.core.scenarios import Scenarios
from macrostat.core.variables import Variables

logger = logging.getLogger(__name__)


class Model:
    """A general class to represent a macroeconomic model.

    This class provides a wrapper for users to write their underlying model
    behavior while maintaining a uniformly accessible interface.

    Attributes
    ----------
    parameters : macrostat.core.parameters.Parameters
        The parameters of the model.
    scenarios : macrostat.core.scenarios.Scenarios
        The scenarios of the model.
    variables : macrostat.core.variables.Variables
        The variables of the model.
    behavior : macrostat.core.behavior.Behavior
        The behavior class of the model.
    name : str
        The name of the model.

    Example
    -------
    A general workflow for a model might look like:

    >>> model = Model()
    >>> output = model.simulate()
    >>> model.save()

    """

    def __init__(
        self,
        parameters: Parameters | dict | None = None,
        hyperparameters: dict | None = None,
        scenarios: Scenarios | dict = None,
        variables: Variables | dict = None,
        behavior: Behavior = Behavior,
        name: str = "model",
        log_level: int = logging.INFO,
        log_file: str = "macrostat_model.log",
    ):
        """Initialization of the model class.


        Parameters
        ----------
        parameters: macrostat.core.parameters.Parameters | dict
            The parameters of the model.
        hyperparameters: dict (optional)
            The hyperparameters of the model.
        scenarios: macrostat.core.scenarios.Scenarios | dict (optional)
            The scenarios of the model.
        variables: macrostat.core.variables.Variables | dict (optional)
            The variables of the model.
        behavior: macrostat.core.behavior.Behavior (optional)
            The behavior of the model.
        name: str (optional)
            The name of the model.
        log_level: int (optional)
            The log level, defaults to logging.INFO but can be set to logging.DEBUG
            for more verbose output.
        log_file: str (optional)
            The log file, defaults to "macrostat_model.log" in the current working
            directory.
        """
        # Essential attributes
        if isinstance(parameters, dict):
            self.parameters = Parameters(
                parameters=parameters, hyperparameters=hyperparameters
            )
        elif isinstance(parameters, Parameters):
            self.parameters = parameters
            if hyperparameters is not None:
                self.parameters.hyper.update(hyperparameters)
        else:
            logger.warning("No parameters provided, using default parameters")
            self.parameters = Parameters()

        if isinstance(scenarios, Scenarios):
            self.scenarios = scenarios
        else:
            logger.warning("No scenarios provided, using default scenarios")
            self.scenarios = Scenarios(parameters=self.parameters)

        if isinstance(variables, Variables):
            self.variables = variables
        else:
            logger.warning("No variables provided, using default variables")
            self.variables = Variables(parameters=self.parameters)

        if behavior is not None and issubclass(behavior, Behavior):
            self.behavior = behavior
        else:
            logger.warning("No behavior provided, using default behavior")
            self.behavior = Behavior

        self.name = name

        logging.basicConfig(level=log_level, filename=log_file)

    @classmethod
    def from_json(
        cls,
        parameter_file: str,
        scenario_file: str,
        variable_file: str,
        *args,
        **kwargs,
    ):
        """Initialize the model from a JSON file."""
        parameters = Parameters.from_json(parameter_file)
        scenarios = Scenarios.from_json(scenario_file, parameters=parameters)
        variables = Variables.from_json(variable_file, parameters=parameters)
        return cls(parameters=parameters, scenarios=scenarios, variables=variables)

    @classmethod
    def load(cls, path: os.PathLike):
        """Class method to load a model instance from a pickled file.

        Parameters
        ----------
        path: os.PathLike
            path to the targeted file containing the model.

        Notes
        -----
        .. note:: This implementation is dependent on your pickling version

        """
        with open(path, "rb") as f:
            model = pickle.load(f)
        return model

    def save(self, path: os.PathLike):
        """Save the model object as a pickled file

        Parameters
        ----------
        path: os.PathLike
            path where the model will be stored. If it is None then
            the model's name will be used and the file stored in the
            working directory.

        Notes
        -----
        .. note:: This implementation is dependent on your pickling version
        """
        with open(path, "wb") as f:
            pickle.dump(self, f)

    def simulate(self, scenario: int | str = 0, *args, **kwargs):
        """Simulate the model.

        Parameters
        ----------
        scenario: int (optional)
            The scenario to use for the model run, defaults to 0, which
            represents the default scenario (no shocks).
        """
        if isinstance(scenario, str):
            scenario = self.scenarios.get_scenario_index(scenario)

        logging.debug(f"Starting simulation. Scenario: {scenario}")
        behavior = self.behavior(
            self.parameters,
            self.scenarios,
            self.variables,
            scenario=scenario,
            *args,
            **kwargs,
        )
        behavior = behavior.to(self.parameters["device"])
        with torch.no_grad():
            return behavior.forward(*args, **kwargs)

    def get_model_training_instance(self, scenario: int | str = 0, *args, **kwargs):
        """Simulate the model.

        Parameters
        ----------
        scenario: int (optional)
            The scenario to use for the model run, defaults to 0, which
            represents the default scenario (no shocks).
        """
        if isinstance(scenario, str):
            scenario = self.scenarios.get_scenario_index(scenario)

        logging.debug(f"Starting simulation. Scenario: {scenario}")
        behavior = self.behavior(
            self.parameters,
            self.scenarios,
            self.variables,
            scenario=scenario,
            *args,
            **kwargs,
        )
        behavior = behavior.to(self.parameters["device"])
        behavior.train()
        return behavior

    def compute_theoretical_steady_state(
        self, scenario: int | str = 0, *args, **kwargs
    ):
        """Compute the theoretical steady state of the model.

        This process generally follows the structure of the forward() function,
        but instead of simulating the model, the steady state is computed at
        each timestep. Therefore, (1) the model is initialized, and (2) for
        each timestep the parameter and scenario information is passed to the
        compute_theoretical_steady_state_per_step() function that computes the
        steady state at that timestep.

        Parameters
        ----------
        scenario: int (optional)
            The scenario to use for the model run, defaults to 0, which
            represents the default scenario (no shocks).
        """
        if isinstance(scenario, str):
            scenario = self.scenarios.get_scenario_index(scenario)

        logging.info(f"Computing theoretical steady state. Scenario: {scenario}")
        behavior = self.behavior(
            self.parameters,
            self.scenarios,
            self.variables,
            scenario=scenario,
            *args,
            **kwargs,
        )
        with torch.no_grad():
            return behavior.compute_theoretical_steady_state(*args, **kwargs)

    def to_json(self, file_path: os.PathLike, *args, **kwargs):
        """Convert the model to a JSON file split into parameters, scenarios,
        and variables.

        Parameters
        ----------
        file_path: os.PathLike
            The path to the file to save the model to.
        """
        self.parameters.to_json(f"{file_path}_params.json")
        self.scenarios.to_json(f"{file_path}_scenarios.json")
        self.variables.to_json(f"{file_path}_variables.json")

1	# -- coding: utf-8 --
2	"""
3	Generic model class as a wrapper to specific implementations
4	"""
5
6	__author__ = ["Karl Naumann-Woleske"]	1✔
7	__credits__ = ["Karl Naumann-Woleske"]	1✔
8	__license__ = "MIT"	1✔
9	__version__ = "0.1.0"	1✔
10	__maintainer__ = ["Karl Naumann-Woleske"]	1✔
11
12	import logging	1✔
13	import os	1✔
14	import pickle	1✔
15
16	import torch	1✔
17
18	from macrostat.core.behavior import Behavior	1✔
19	from macrostat.core.parameters import Parameters	1✔
20	from macrostat.core.scenarios import Scenarios	1✔
21	from macrostat.core.variables import Variables	1✔
22
23	logger = logging.getLogger(__name__)	1✔
24
25
26	class Model:	1✔
27	"""A general class to represent a macroeconomic model.
28
29	This class provides a wrapper for users to write their underlying model
30	behavior while maintaining a uniformly accessible interface.
31
32	Attributes
33	----------
34	parameters : macrostat.core.parameters.Parameters
35	The parameters of the model.
36	scenarios : macrostat.core.scenarios.Scenarios
37	The scenarios of the model.
38	variables : macrostat.core.variables.Variables
39	The variables of the model.
40	behavior : macrostat.core.behavior.Behavior
41	The behavior class of the model.
42	name : str
43	The name of the model.
44
45	Example
46	-------
47	A general workflow for a model might look like:
48
49	>>> model = Model()
50	>>> output = model.simulate()
51	>>> model.save()
52
53	"""
54
55	def __init__(	1✔
56	self,
57	parameters: Parameters \| dict \| None = None,
58	hyperparameters: dict \| None = None,
59	scenarios: Scenarios \| dict = None,
60	variables: Variables \| dict = None,
61	behavior: Behavior = Behavior,
62	name: str = "model",
63	log_level: int = logging.INFO,
64	log_file: str = "macrostat_model.log",
65	):
66	"""Initialization of the model class.
67
68
69	Parameters
70	----------
71	parameters: macrostat.core.parameters.Parameters \| dict
72	The parameters of the model.
73	hyperparameters: dict (optional)
74	The hyperparameters of the model.
75	scenarios: macrostat.core.scenarios.Scenarios \| dict (optional)
76	The scenarios of the model.
77	variables: macrostat.core.variables.Variables \| dict (optional)
78	The variables of the model.
79	behavior: macrostat.core.behavior.Behavior (optional)
80	The behavior of the model.
81	name: str (optional)
82	The name of the model.
83	log_level: int (optional)
84	The log level, defaults to logging.INFO but can be set to logging.DEBUG
85	for more verbose output.
86	log_file: str (optional)
87	The log file, defaults to "macrostat_model.log" in the current working
88	directory.
89	"""
90	# Essential attributes
91	if isinstance(parameters, dict):	1✔
92	self.parameters = Parameters(	1✔
93	parameters=parameters, hyperparameters=hyperparameters
94	)
95	elif isinstance(parameters, Parameters):	1✔
96	self.parameters = parameters	1✔
97	if hyperparameters is not None:	1✔
98	self.parameters.hyper.update(hyperparameters)	1✔
99	else:
100	logger.warning("No parameters provided, using default parameters")	1✔
101	self.parameters = Parameters()	1✔
102
103	if isinstance(scenarios, Scenarios):	1✔
104	self.scenarios = scenarios	1✔
105	else:
106	logger.warning("No scenarios provided, using default scenarios")	1✔
107	self.scenarios = Scenarios(parameters=self.parameters)	1✔
108
109	if isinstance(variables, Variables):	1✔
110	self.variables = variables	1✔
111	else:
112	logger.warning("No variables provided, using default variables")	1✔
113	self.variables = Variables(parameters=self.parameters)	1✔
114
115	if behavior is not None and issubclass(behavior, Behavior):	1✔
116	self.behavior = behavior	1✔
117	else:
118	logger.warning("No behavior provided, using default behavior")	1✔
119	self.behavior = Behavior	1✔
120
121	self.name = name	1✔
122
123	logging.basicConfig(level=log_level, filename=log_file)	1✔
124
125	@classmethod	1✔
126	def from_json(	1✔
127	cls,
128	parameter_file: str,
129	scenario_file: str,
130	variable_file: str,
131	*args,
132	**kwargs,
133	):
134	"""Initialize the model from a JSON file."""
135	parameters = Parameters.from_json(parameter_file)	1✔
136	scenarios = Scenarios.from_json(scenario_file, parameters=parameters)	1✔
137	variables = Variables.from_json(variable_file, parameters=parameters)	1✔
138	return cls(parameters=parameters, scenarios=scenarios, variables=variables)	1✔
139
140	@classmethod	1✔
141	def load(cls, path: os.PathLike):	1✔
142	"""Class method to load a model instance from a pickled file.
143
144	Parameters
145	----------
146	path: os.PathLike
147	path to the targeted file containing the model.
148
149	Notes
150	-----
151	.. note:: This implementation is dependent on your pickling version
152
153	"""
154	with open(path, "rb") as f:	1✔
155	model = pickle.load(f)	1✔
156	return model	1✔
157
158	def save(self, path: os.PathLike):	1✔
159	"""Save the model object as a pickled file
160
161	Parameters
162	----------
163	path: os.PathLike
164	path where the model will be stored. If it is None then
165	the model's name will be used and the file stored in the
166	working directory.
167
168	Notes
169	-----
170	.. note:: This implementation is dependent on your pickling version
171	"""
172	with open(path, "wb") as f:	1✔
173	pickle.dump(self, f)	1✔
174
175	def simulate(self, scenario: int \| str = 0, args, *kwargs):	1✔
176	"""Simulate the model.
177
178	Parameters
179	----------
180	scenario: int (optional)
181	The scenario to use for the model run, defaults to 0, which
182	represents the default scenario (no shocks).
183	"""
184	if isinstance(scenario, str):	1✔
185	scenario = self.scenarios.get_scenario_index(scenario)	1✔
186
187	logging.debug(f"Starting simulation. Scenario: {scenario}")	1✔
188	behavior = self.behavior(	1✔
189	self.parameters,
190	self.scenarios,
191	self.variables,
192	scenario=scenario,
193	*args,
194	**kwargs,
195	)
196	behavior = behavior.to(self.parameters["device"])	1✔
197	with torch.no_grad():	1✔
198	return behavior.forward(args, *kwargs)	1✔
199
200	def get_model_training_instance(self, scenario: int \| str = 0, args, *kwargs):	1✔
201	"""Simulate the model.
202
203	Parameters
204	----------
205	scenario: int (optional)
206	The scenario to use for the model run, defaults to 0, which
207	represents the default scenario (no shocks).
208	"""
NEW 209	if isinstance(scenario, str):	×
NEW 210	scenario = self.scenarios.get_scenario_index(scenario)	×
211
NEW 212	logging.debug(f"Starting simulation. Scenario: {scenario}")	×
NEW 213	behavior = self.behavior(	×
214	self.parameters,
215	self.scenarios,
216	self.variables,
217	scenario=scenario,
218	*args,
219	**kwargs,
220	)
NEW 221	behavior = behavior.to(self.parameters["device"])	×
NEW 222	behavior.train()	×
NEW 223	return behavior	×
224
225	def compute_theoretical_steady_state(	1✔
226	self, scenario: int \| str = 0, args, *kwargs
227	):
228	"""Compute the theoretical steady state of the model.
229
230	This process generally follows the structure of the forward() function,
231	but instead of simulating the model, the steady state is computed at
232	each timestep. Therefore, (1) the model is initialized, and (2) for
233	each timestep the parameter and scenario information is passed to the
234	compute_theoretical_steady_state_per_step() function that computes the
235	steady state at that timestep.
236
237	Parameters
238	----------
239	scenario: int (optional)
240	The scenario to use for the model run, defaults to 0, which
241	represents the default scenario (no shocks).
242	"""
243	if isinstance(scenario, str):	1✔
244	scenario = self.scenarios.get_scenario_index(scenario)	1✔
245
246	logging.info(f"Computing theoretical steady state. Scenario: {scenario}")	1✔
247	behavior = self.behavior(	1✔
248	self.parameters,
249	self.scenarios,
250	self.variables,
251	scenario=scenario,
252	*args,
253	**kwargs,
254	)
255	with torch.no_grad():	1✔
256	return behavior.compute_theoretical_steady_state(args, *kwargs)	1✔
257
258	def to_json(self, file_path: os.PathLike, args, *kwargs):	1✔
259	"""Convert the model to a JSON file split into parameters, scenarios,
260	and variables.
261
262	Parameters
263	----------
264	file_path: os.PathLike
265	The path to the file to save the model to.
266	"""
267	self.parameters.to_json(f"{file_path}_params.json")	1✔
268	self.scenarios.to_json(f"{file_path}_scenarios.json")	1✔
269	self.variables.to_json(f"{file_path}_variables.json")	1✔

KarlNaumann / MacroStat / 17580505858

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous