econometron.Models.dynamicsge.nonlinear_dsge
classin theeconometron.Models.dynamicsgemodule
Overview
Nonlinear DSGE Models
- Nonlinear DSGE models describe the decisions of households, firms, and policymakers under uncertainty, subject to market-clearing conditions, without relying on linear approximations. The nonlinear_dsge class solves the model in its nonlinear form:
where:
: Vector of endogenous variables (states and controls).
: Exogenous shocks.
: Expectation operator.
- Use Case: Models with significant nonlinearities (e.g., large shocks, non-quadratic utility, or complex dynamics).
The nonlinear_dsge class in the econometron.Models.dynamicsge module provides a robust framework for specifying, solving, and analyzing nonlinear Dynamic Stochastic General Equilibrium (DSGE) models using projection methods (collocation, Galerkin, or least squares). It is designed to model complex economic systems with arbitrary state and control variables, user-defined equilibrium conditions, utility and production functions, and stochastic shock processes, while maintaining theoretical rigor.
Key Features
Flexible definition of model parameters and variables with bounds and steady-state values.
Support for custom utility and production functions, with optional marginal derivatives.
Stochastic shock processes (e.g., AR(1)).
Solution of nonlinear equilibrium conditions using projection methods.
Tools for policy function evaluation, validation, simulation, and visualization.
Class Definitions
DSGEVariable
from dataclasses import dataclass
from typing import Tuple, Optional
@dataclass
class DSGEVariable:
name: str
bounds: Tuple[float, float]
steady_state: Optional[float] = None
description: str = ""
is_state: bool = True
is_control: bool = False
is_shock: bool = FalseRepresents a DSGE model variable with properties such as name, bounds, steady-state value, and type (state, control, or shock).
DSGEParameter
@dataclass
class DSGEParameter:
name: str
value: float
description: str = ""
bounds: Optional[Tuple[float, float]] = NoneRepresents a DSGE model parameter with name, value, description, and optional bounds.
nonlinear_dsge Class
from econometron.Models.dynamicsge import nonlinear_dsge
nonlineardsge=nonlinear_dsge(name: str = "DSGE Model")Attributes
| Attribute | Type | Description |
|---|---|---|
name | str | Model name. |
parameters | dict | Dictionary of DSGEParameter objects. |
variables | dict | Dictionary of DSGEVariable objects. |
equations | list | List of callable equilibrium condition functions. |
utility_function | Callable | Callable utility function. |
marginal_utility_function | Callable | Callable marginal utility function (optional; computed numerically if not provided). |
production_function | Callable | Callable production function. |
shock_processes | dict | Dictionary of shock processes with persistence, volatility, and type. |
solver | ProjectionSolver | Instance of ProjectionSolver for solving the model. |
coefficients | np.ndarray | Solution coefficients from projection method. |
solution_info | dict | Dictionary containing solution method, convergence status, and criterion. |
is_solved | bool | Boolean indicating whether the model is solved. |
state_vars | list[str] | List of state variable names. |
control_vars | list[str] | List of control variable names. |
shock_vars | list[str] | List of shock variable names. |
Methods
1 - add_parameter(name: str, value: float, description: str = "", bounds: Optional[Tuple[float, float]] = None)
Purpose:
- Adds a parameter to the DSGE model and stores it in the
parametersdictionary. - Sets the parameter as an attribute of the model instance for convenient access (e.g.,
self.alpha).
- Adds a parameter to the DSGE model and stores it in the
Parameters:
name (str): Parameter name.value (float): Parameter value.description (str, optional): Description of the parameter (default:"").bounds (Tuple[float, float], optional): Parameter bounds (default:None).
Returns:
self: Returns the model instance for method chaining.
Notes:
- Parameters are stored as
DSGEParameterobjects inself.parameters. - The parameter value is accessible as a model attribute (e.g.,
self.name).
- Parameters are stored as
2 - add_variable(name: str, bounds: Tuple[float, float], steady_state: Optional[float] = None, description: str = "", variable_type: str = "state")
Purpose:
- Adds a variable to the DSGE model, classifying it as a state, control, or shock variable.
- Organizes the variable into the appropriate list (
state_vars,control_vars, orshock_vars).
Parameters:
name (str): Variable name.bounds (Tuple[float, float]): Variable bounds.steady_state (float, optional): Steady-state value (default:None).description (str, optional): Description of the variable (default:"").variable_type (str, optional): Type of variable ("state","control","shock", or"both") (default:"state").
Returns:
self: Returns the model instance for method chaining.
Notes:
- Variables are stored as
DSGEVariableobjects inself.variables. - If
variable_typeis"both", the variable is treated as both a state and control variable.
- Variables are stored as
3 - set_utility_function(utility_func: Callable, marginal_utility_func: Optional[Callable] = None)
Purpose:
- Sets the utility function for the DSGE model and optionally its marginal derivative.
- If the marginal utility function is not provided, a numerical derivative is computed using a central difference method.
Parameters:
utility_func (Callable): The utility function.marginal_utility_func (Callable, optional): The marginal utility function (default:None).
Returns:
self: Returns the model instance for method chaining.
Notes:
- Numerical derivative is computed with step size
1e-8ifmarginal_utility_funcis not provided. - Stored in
self.utility_functionandself.marginal_utility_function.
- Numerical derivative is computed with step size
4 - set_production_function(production_func: Callable)
Purpose:
- Sets the production function for the DSGE model.
Parameters:
production_func (Callable): The production function.
Returns:
self: Returns the model instance for method chaining.
Notes:
- Stored in
self.production_function.
- Stored in
5 - add_shock_process(shock_name: str, persistence: float, volatility: float, process_type: str = "AR1")
Purpose:
- Adds a stochastic shock process to the DSGE model, specifying its persistence, volatility, and type (e.g., AR(1)).
Parameters:
shock_name (str): Name of the shock variable.persistence (float): Persistence parameter.volatility (float): Volatility parameter.process_type (str, optional): Type of shock process (default:"AR1").
Returns:
self: Returns the model instance for method chaining.
Notes:
- Stored in
self.shock_processesas a dictionary with keys for persistence, volatility, and type.
- Stored in
6 - add_equilibrium_condition(equation_func: Callable)
Purpose:
- Adds an equilibrium condition (e.g., Euler equation or first-order condition) to the DSGE model.
Parameters:
equation_func (Callable): The equilibrium condition function.
Returns:
self: Returns the model instance for method chaining.
Notes:
- Stored in
self.equationsas a list of callable functions.
- Stored in
7 - compute_steady_state(method: str = "analytical")
Purpose:
- Computes the steady state of the DSGE model using either an analytical or numerical method.
Parameters:
method (str, optional): Method ("analytical"or"numerical") (default:"analytical").
Returns:
self: Returns the model instance for method chaining.
Notes:
- Analytical method requires steady-state values via
add_variable; warns if missing. - Numerical method requires
_solve_steady_state_numerically(not implemented). - Prints a summary of steady-state values.
- Analytical method requires steady-state values via
8 - setup_solver(approximation_orders: Dict[str, int])
Purpose:
- Configures the projection solver with specified polynomial orders for state and shock variables.
Parameters:
approximation_orders (Dict[str, int]): Dictionary mapping variable names to polynomial orders.
Returns:
self: Returns the model instance for method chaining.
Notes:
- Initializes a
ProjectionSolverinstance with polynomial orders, lower bounds, and upper bounds for state and shock variables. - Prints a summary of solver setup, including variables, polynomial orders, and basis function count.
- Initializes a
9 - solve(method: str = "collocation", initial_policy: Optional[Callable] = None, solver_options: Optional[Dict] = None, verbose: bool = True)
Purpose:
- Solves the nonlinear DSGE model using a specified projection method (collocation, Galerkin, or least squares).
Parameters:
method (str, optional): Solution method ("collocation","galerkin", or"least_squares") (default:"collocation").initial_policy (Callable, optional): Function to generate initial policy guess (default:None).solver_options (Dict, optional): Solver options (e.g., max iterations, tolerance) (default:None).verbose (bool, optional): If True, prints solution progress (default:True).
Returns:
self: Returns the model instance for method chaining.
Notes:
- Requires
setup_solverto be called first. - Raises
ValueErrorif no equilibrium conditions exist or solver not initialized. - Uses default solver options (
maxit=5000,stopc=1e-8) unless overridden. - Stores solution coefficients in
self.coefficientsand solution metadata inself.solution_info. - Sets
self.is_solvedtoTrueupon successful solution.
- Requires
10 - evaluate_policy(state_points: np.ndarray)
Purpose:
- Evaluates the solved policy function at specified state points.
Parameters:
state_points (np.ndarray): Array of state points (n_points × n_states).
Returns:
np.ndarray: Policy function values at the specified points.
Notes:
- Raises
ValueErrorif the model is not solved.
- Raises
11 - validate_solution(n_test_points: int = 100, random_seed: Optional[int] = None)
Purpose:
- Validates the solution by computing residuals of equilibrium conditions at random state points.
Parameters:
n_test_points (int, optional): Number of random test points (default: 100).random_seed (int, optional): Random seed for reproducibility (default:None).
Returns:
Dict[str, float]: Dictionary containing validation metrics (mean absolute error, maximum absolute error, RMS error, L∞ norm).
Notes:
- Raises
ValueErrorif the model is not solved. - Generates random test points within variable bounds.
- Prints a summary of validation metrics.
- Raises
12 - analyze_policy(n_plot_points: int = 50, shock_values: Optional[List[float]] = None, figsize: Tuple[int, int] = (15, 10))
Purpose:
- Analyzes and visualizes policy functions for specified shock values.
- Plots policy functions and implied savings rates (if applicable).
Parameters:
n_plot_points (int, optional): Number of points to plot (default: 50).shock_values (List[float], optional): Specific shock values to analyze (default:None).figsize (Tuple[int, int], optional): Figure size for plots (default:(15, 10)).
Returns:
self: Returns the model instance for method chaining.
Notes:
- Raises
ValueErrorif the model is not solved or no state variables exist. - Defaults to bounds and midpoint of the first shock variable if
shock_valuesisNone. - Plots policy functions for the first state variable and, if applicable, the savings rate derived from the production function.
- Raises
13 - simulate(T: int, initial_state: Optional[Dict] = None, random_seed: Optional[int] = None)
Purpose:
- Simulates the DSGE model for T periods, generating time series for state, control, and shock variables.
Parameters:
T (int): Number of simulation periods.initial_state (Dict, optional): Initial state values (default:None, uses steady state or midpoint of bounds).random_seed (int, optional): Random seed for shock simulation (default:None).
Returns:
Dict[str, np.ndarray]: Dictionary with time series for all variables.
Notes:
- Raises
ValueErrorif the model is not solved. - Uses
_update_state_variablesand_update_shocksto advance the simulation. - If
initial_stateis not provided, uses steady-state values or midpoint of variable bounds.
- Raises
14 - summary()
Purpose:
- Prints a comprehensive summary of the DSGE model, including parameters, variables, equilibrium conditions, and solution status.
Parameters:
- None
Returns:
None: Prints the summary to the console.
Notes:
- Displays parameter values, variable bounds and types, equilibrium conditions, and solution metadata (if solved).
- Provides a clear overview of the model’s configuration and status.
Usage Example: Stochastic Optimal Growth Model (Brock-Mirman 1972)
Below is an example of using the nonlinear_dsge class to specify, solve, and analyze a stochastic optimal growth model.
Step 1: Import the Class
from econometron.Models.dynamicsge import nonlinear_dsge
import numpy as np
import matplotlib.pyplot as pltStep 2: Initialize the Model
optimal_growth_model = nonlinear_dsge("Stochastic Optimal Growth Model (Brock-Mirman 1972)")Step 3: Add Parameters
optimal_growth_model.add_parameter('alpha', 0.36, "Capital share")
optimal_growth_model.add_parameter('beta', 0.99, "Discount factor")
optimal_growth_model.add_parameter('sigma', 0.02, "Shock volatility")
optimal_growth_model.add_parameter('rho', 0.95, "Shock persistence")
optimal_growth_model.add_parameter('A', 1.0, "TFP level")
optimal_growth_model.add_parameter('gamma', 2.0, "Risk aversion (CRRA)")Step 4: Compute Steady States
alpha = optimal_growth_model.parameters['alpha'].value
beta = optimal_growth_model.parameters['beta'].value
A = optimal_growth_model.parameters['A'].value
k_ss = ((alpha * beta * A) ** (1 / (1 - alpha)))
c_ss = A * k_ss ** alpha - k_ss
print(f"Computed steady states:")
print(f" Capital (k_ss): {k_ss:.6f}")
print(f" Consumption (c_ss): {c_ss:.6f}")Output:
Computed steady states:
Capital (k_ss): 4.319382
Consumption (c_ss): 0.912049Step 5: Add Variables
optimal_growth_model.add_variable('k', (0.1 * k_ss, 2.0 * k_ss), k_ss, "Capital stock", "state")
optimal_growth_model.add_variable('z', (0.8, 1.2), 1.0, "Productivity shock", "shock")
optimal_growth_model.add_variable('k_next', (0.1 * k_ss, 2.0 * k_ss), k_ss, "Next period capital", "control")Step 6: Add Shock Process
optimal_growth_model.add_shock_process('z', persistence=0.95, volatility=0.02, process_type="AR1")Step 7: Define and Set Functions
def utility(c, gamma):
if gamma == 1:
return np.log(np.maximum(c, 1e-12))
else:
return (np.maximum(c, 1e-12) ** (1 - gamma) - 1) / (1 - gamma)
def marginal_utility(c, gamma):
return np.maximum(c, 1e-12) ** (-gamma)
def production(k, z, alpha, A):
return A * z * k ** alpha
optimal_growth_model.set_utility_function(
lambda state_dict, params: utility(state_dict.get('c', 1), params['gamma'].value),
lambda state_dict, params: marginal_utility(state_dict.get('c', 1), params['gamma'].value)
)
optimal_growth_model.set_production_function(
lambda state_dict, params: production(
state_dict['k'], state_dict['z'], params['alpha'].value, params['A'].value
)
)Step 8: Define Euler Equation
def euler_equation(current_state, params, next_state=None, solver=None, coeffs=None):
if next_state is None or solver is None or coeffs is None:
return 0.0
k_t = current_state['k']
z_t = current_state['z']
k_next = current_state['k_next']
y_t = production(k_t, z_t, params['alpha'].value, params['A'].value)
c_t = y_t - k_next
if c_t <= 0:
return 1e6
lhs = marginal_utility(c_t, params['gamma'].value)
alpha = params['alpha'].value
A = params['A'].value
beta = params['beta'].value
rho = params['rho'].value
sigma = params['sigma'].value
gamma = params['gamma'].value
n_nodes = 5
nodes, weights = solver.cheb_basis.gauss_hermite_nodes(n_nodes, sigma)
rhs = 0.0
for node, weight in zip(nodes, weights):
z_next = np.exp(rho * np.log(z_t) + sigma * np.log(node))
z_next = np.clip(z_next, 0.8, 1.2)
next_point = np.array([[k_next, z_next]])
k_next_next = solver.evaluate_solution(coeffs, next_point)[0]
y_next = production(k_next, z_next, alpha, A)
c_next = y_next - k_next_next
if c_next <= 0:
rhs += weight * 1e6
else:
marginal_product = alpha * A * z_next * k_next ** (alpha - 1)
rhs += weight * marginal_product * marginal_utility(c_next, gamma)
return lhs - beta * rhs
optimal_growth_model.add_equilibrium_condition(euler_equation)Step 9: Compute Steady State
optimal_growth_model.compute_steady_state(method="analytical")Output:
=== STEADY STATE ===
k: 4.319382
z: 1.000000
k_next: 4.319382Step 10: Set Up Solver
optimal_growth_model.setup_solver({
'k': 8,
'z': 4
})Step 11: Solve the Model
def initial_savings_policy(state_dict, params):
k = state_dict['k']
z = state_dict['z']
alpha = params['alpha'].value
A = params['A'].value
output = production(k, z, alpha, A)
savings_rate = 0.2
return savings_rate * output
optimal_growth_model.solve(
method='collocation',
initial_policy=initial_savings_policy,
solver_options={'maxit': 5000, 'stopc': 1e-8},
verbose=True
)Step 12: Validate Solution
validation_metrics = optimal_growth_model.validate_solution(n_test_points=100, random_seed=42)Step 13: Analyze Policy Functions
optimal_growth_model.analyze_policy(
n_plot_points=50,
shock_values=[0.9, 1.0, 1.1],
figsize=(15, 6)
)Step 14: Summarize the Model
optimal_growth_model.summary()Step 15: Simulate and Visualize
simulation_data = optimal_growth_model.simulate(
T=100,
initial_state={'k': k_ss, 'z': 1.0},
random_seed=123
)
fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10))
periods = np.arange(100)
# Capital
ax1.plot(periods, simulation_data['k'], 'b-', linewidth=2, label='Capital')
ax1.axhline(y=k_ss, color='r', linestyle='--', alpha=0.7, label='Steady State')
ax1.set_title('Capital Dynamics')
ax1.legend()
ax1.grid(True, alpha=0.3)
# Productivity
ax2.plot(periods, simulation_data['z'], 'g-', linewidth=2, label='Productivity')
ax2.axhline(y=1.0, color='r', linestyle='--', alpha=0.7, label='Mean')
ax2.set_title('Productivity Shock Process')
ax2.legend()
ax2.grid(True, alpha=0.3)
# Consumption
consumption = np.zeros(100)
for t in range(100):
if t < 99:
k_t = simulation_data['k'][t]
z_t = simulation_data['z'][t]
k_next_t = simulation_data['k_next'][t]
consumption[t] = production(k_t, z_t, alpha, A) - k_next_t
else:
consumption[t] = consumption[t-1]
ax3.plot(periods, consumption, 'm-', linewidth=2, label='Consumption')
ax3.axhline(y=c_ss, color='r', linestyle='--', alpha=0.7, label='Steady State')
ax3.set_title('Consumption Dynamics')
ax3.legend()
ax3.grid(True, alpha=0.3)
# Savings Rate
output = np.array([production(simulation_data['k'][t], simulation_data['z'][t], alpha, A)
for t in range(100)])
savings_rate = simulation_data['k_next'][:] / output[:]
ax4.plot(periods, savings_rate, 'orange', linewidth=2, label='Savings Rate')
ax4.axhline(y=k_ss/(A * k_ss**alpha), color='r', linestyle='--', alpha=0.7, label='SS Savings Rate')
ax4.set_title('Optimal Savings Rate')
ax4.legend()
ax4.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()Notes
- Function Signatures: Equilibrium conditions should accept
current_state,params, and optionallynext_state,solver, andcoeffsfor expectational equations. - Steady-State Values: For analytical steady-state computation, provide values via
add_variable. Numerical steady-state solving requires implementing_solve_steady_state_numerically. - Shock Processes: Currently supports AR(1) processes; extend
_update_shocksfor other types. - Projection Methods: The
ProjectionSolverclass (fromeconometron.utils.projection) handles collocation, Galerkin, or least squares methods. - Numerical Stability: Uses penalties (1e6) for invalid evaluations (e.g., negative consumption).
- Visualization:
analyze_policyand simulation plots assume a single state variable and shock for simplicity; extend for multidimensional cases.
