Core¶

Model assembly, datasets, calibration and solver support.

Main Model class for equilibria CGE framework.

The Model class is the central component that assembles blocks, manages sets, parameters, variables, and equations, and provides interfaces for calibration and solving.

class equilibria.model.ModelStatistics(*, variables=0, equations=0, degrees_of_freedom=0, blocks=0, sparsity=0.0)[source]¶

Bases: BaseModel

Statistics for a CGE model.

Provides counts of variables, equations, degrees of freedom, and other model metrics.

Variables:

variables (int) – Total number of scalar variables
equations (int) – Total number of scalar equations
degrees_of_freedom (int) – DOF (variables - equations)
blocks (int) – Number of blocks
sparsity (float) – Sparsity ratio (0-1)

Parameters:

variables (int)
equations (int)
degrees_of_freedom (int)
blocks (int)
sparsity (float)

variables: int¶

equations: int¶

degrees_of_freedom: int¶

blocks: int¶

sparsity: float¶

model_config = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class equilibria.model.Model(*, name, description='', set_manager=<factory>, parameter_manager=<factory>, variable_manager=<factory>, equation_manager=<factory>, blocks=<factory>)[source]¶

Bases: BaseModel

CGE Model class.

The Model class assembles equation blocks, manages all model components (sets, parameters, variables, equations), and provides interfaces for calibration and solving.

Variables:

name (str) – Model identifier
description (str) – Model description
set_manager (SetManager) – Manager for all sets
parameter_manager (ParameterManager) – Manager for all parameters
variable_manager (VariableManager) – Manager for all variables
equation_manager (EquationManager) – Manager for all equations
blocks (list[Block]) – List of blocks in the model

Parameters:

name (str)
description (str)
set_manager (SetManager)
parameter_manager (ParameterManager)
variable_manager (VariableManager)
equation_manager (EquationManager)
blocks (list[Block])

Example

>>> model = Model(name="MyCGE")
>>> model.add_sets([
...     Set(name="J", elements=["agr", "mfg", "svc"]),
... ])
>>> model.add_block(CESValueAdded(sigma=0.8))
>>> print(model.statistics)

name: str¶

description: str¶

set_manager: SetManager¶

parameter_manager: ParameterManager¶

variable_manager: VariableManager¶

equation_manager: EquationManager¶

blocks: list[Block]¶

model_config = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

add_set(set_obj)[source]¶

Add a set to the model.

Parameters:: set_obj (Set) – Set to add
Return type:: None

add_sets(sets)[source]¶

Add multiple sets to the model.

Parameters:: sets (list[Set]) – List of sets to add
Return type:: None

add_parameter(param)[source]¶

Add a parameter to the model.

Parameters:: param (Parameter) – Parameter to add
Return type:: None

add_variable(var)[source]¶

Add a variable to the model.

Parameters:: var (Variable) – Variable to add
Return type:: None

add_equation(eq)[source]¶

Add an equation to the model.

Parameters:: eq (Equation) – Equation to add
Return type:: None

add_block(block)[source]¶

Add a block to the model.

This validates that required sets exist, then calls the block’s setup method to add parameters, variables, and equations.

Parameters:: block (Block) – Block to add
Raises:: ValueError – If required sets are missing
Return type:: None

add_blocks(blocks)[source]¶

Add multiple blocks to the model.

Parameters:: blocks (list[Block]) – List of blocks to add
Return type:: None

get_parameter(name)[source]¶

Get a parameter by name.

Parameters:: name (str) – Parameter name
Returns:: Parameter object
Return type:: Parameter

get_variable(name)[source]¶

Get a variable by name.

Parameters:: name (str) – Variable name
Returns:: Variable object
Return type:: Variable

get_equation(name)[source]¶

Get an equation by name.

Parameters:: name (str) – Equation name
Returns:: Equation object
Return type:: Equation

property statistics: ModelStatistics¶

Calculate model statistics.

Returns:: ModelStatistics with counts and metrics

summary()[source]¶

Return comprehensive model summary.

Returns:: Dictionary with model information
Return type:: dict[str, Any]

Bundled reference datasets shipped with equilibria.

load_bundled(category, name) returns a ready-to-use object built from the canonical source files for one of the small datasets that travel with the package (useful for tutorials, tests, and example notebooks). Callers that need a custom aggregation should still go through the underlying loaders directly.

Source-of-truth conventions per category:

"gtap" — native GEMPACK HAR/PRM files. That is what the official convert.cmd flow builds GDX from upstream.
"pep" — Excel workbooks (SAM-V2_0.xlsx, VAL_PAR.xlsx). That is the form in which the PEP-1-1 reference data is published.

equilibria.datasets.dataset_path(category, name)[source]¶

Return the directory holding a bundled dataset’s raw files.

Parameters:

category (Literal['gtap', 'pep'])
name (str)

Return type:

Path

equilibria.datasets.list_bundled(category)[source]¶

List dataset names available for a category.

Parameters:: category (Literal['gtap', 'pep'])
Return type:: list[str]

equilibria.datasets.load_bundled(category, name='default')[source]¶

Load a bundled dataset and return a ready-to-use object.

For category="gtap" this always reads native HAR/PRM files (basedata.har, sets.har, default.prm, plus optional baserate.har for GTAPv7-style aggregations like NUS333) and returns a calibrated GTAPParameters.

For category="pep" this always reads the canonical Excel workbooks (SAM-V2_0.xlsx and VAL_PAR.xlsx), converts the SAM on-the-fly to the 4D GDX layout the PEP calibrator consumes, and returns a PEPModelCalibrator ready to call .calibrate(). The intermediate GDX is written to a tmp directory under ~/.cache/equilibria/pep/ so subsequent calls reuse it.

Parameters:

category (Literal['gtap', 'pep'])
name (str)

Calibration base classes for equilibria CGE framework.

This module provides the foundation for calibrating CGE models from SAM data, including elasticity estimation and parameter computation.

class equilibria.calibration.base.CalibrationResult(*, success=True, parameters=<factory>, statistics=<factory>, messages=<factory>, warnings=<factory>)[source]¶

Bases: BaseModel

Results from a calibration operation.

Stores calibrated parameters, statistics, and diagnostic information from the calibration process.

Variables:

success (bool) – Whether calibration succeeded
parameters (dict[str, np.ndarray]) – Dictionary of calibrated parameter values
statistics (dict[str, Any]) – Calibration statistics
messages (list[str]) – List of diagnostic messages
warnings (list[str]) – List of warning messages

Parameters:

success (bool)
parameters (dict[str, ndarray])
statistics (dict[str, Any])
messages (list[str])
warnings (list[str])

success: bool¶

parameters: dict[str, np.ndarray]¶

statistics: dict[str, Any]¶

messages: list[str]¶

warnings: list[str]¶

model_config = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

add_message(message)[source]¶

Add a diagnostic message.

Parameters:: message (str)
Return type:: None

add_warning(warning)[source]¶

Add a warning message.

Parameters:: warning (str)
Return type:: None

to_dict()[source]¶

Convert result to dictionary.

Return type:: dict[str, Any]

class equilibria.calibration.base.Calibrator(*, name, description='')[source]¶

Bases: ABC, BaseModel

Abstract base class for model calibrators.

Calibrators compute model parameters from SAM data and user-provided elasticities.

Variables:

name (str) – Calibrator name
description (str) – Calibrator description

Parameters:

name (str)
description (str)

name: str¶

description: str¶

model_config = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

abstractmethod calibrate(model, sam, elasticities=None)[source]¶

Calibrate model parameters from SAM data.

Parameters:

model (Model) – Model to calibrate
sam (SAM) – Social Accounting Matrix with base year data
elasticities (dict[str, float] | None) – Optional user-provided elasticities

Returns:

CalibrationResult with calibrated parameters

Return type:

CalibrationResult

validate_sam(sam, tolerance=1e-06)[source]¶

Validate that SAM is balanced.

Parameters:

sam (SAM) – SAM to validate
tolerance (float) – Balance tolerance

Returns:

True if valid

Raises:

ValueError – If SAM is not balanced

Return type:

bool

get_sam_value(sam, row_account, col_account)[source]¶

Get value from SAM matrix.

Parameters:

sam (SAM) – SAM data
row_account (str) – Row account name
col_account (str) – Column account name

Returns:

Matrix value

Return type:

float

compute_io_coefficients(sam, sectors)[source]¶

Compute input-output coefficients.

Parameters:

sam (SAM) – SAM data
sectors (list[str]) – List of sector names

Returns:

Dictionary of (input_sector, output_sector) -> coefficient

Return type:

dict[tuple[str, str], float]

class equilibria.calibration.base.ModelCalibrator[source]¶

Bases: object

Orchestrates calibration of multiple model components.

Manages multiple calibrators and coordinates the full model calibration process.

Variables:: calibrators – List of calibrators to apply

add_calibrator(calibrator)[source]¶

Add a calibrator.

Parameters:: calibrator (Calibrator) – Calibrator to add
Return type:: None

calibrate(model, sam, elasticities=None)[source]¶

Run all calibrators on model.

Parameters:

model (Model) – Model to calibrate
sam (SAM) – SAM data
elasticities (dict[str, float] | None) – Optional elasticities

Returns:

Dictionary of calibrator name -> results

Return type:

dict[str, CalibrationResult]

apply_results(model, results)[source]¶

Apply calibration results to model.

Parameters:

model (Model) – Model to update
results (dict[str, CalibrationResult]) – Calibration results

Return type:

None

CES (Constant Elasticity of Substitution) calibration.

Calibrates CES production function parameters from SAM data.

class equilibria.calibration.ces.CESCalibrator(*, name='CES', description='CES production function calibration', default_sigma=0.8)[source]¶

Bases: Calibrator

Calibrator for CES production functions.

Computes CES share parameters and efficiency parameters from SAM data and user-provided elasticities.

The CES function is: VA[j] = B[j] * (sum_i beta[i,j] * FD[i,j]^(-rho[j]))^(-1/rho[j])

Where: - rho[j] = (sigma[j] - 1) / sigma[j] - beta[i,j] = (WF[i] * FD[i,j]) / PVA[j] / VA[j] - B[j] = VA[j] / (sum_i beta[i,j] * FD[i,j]^(-rho[j]))^(-1/rho[j])

Variables:

default_sigma (float) – Default elasticity if not provided

Parameters:

name (str)
description (str)
default_sigma (Annotated[float, Gt(gt=0)])

name: str¶

description: str¶

default_sigma: float¶

calibrate(model, sam, elasticities=None)[source]¶

Calibrate CES parameters from SAM.

Parameters:

model (Model) – Model with CES blocks
sam (SAM) – SAM data
elasticities (dict[str, float] | None) – Optional sector-specific elasticities

Returns:

CalibrationResult with beta_VA, B_VA, sigma_VA

Return type:

CalibrationResult

get_info()[source]¶

Get calibrator info.

Return type:: dict[str, Any]

model_config = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Leontief calibration for intermediate inputs.

Calibrates input-output coefficients from SAM data.

class equilibria.calibration.leontief.LeontiefCalibrator(*, name='Leontief', description='Leontief intermediate input calibration', min_coefficient=1e-10)[source]¶

Bases: Calibrator

Calibrator for Leontief intermediate inputs.

Computes input-output coefficients from SAM data.

The Leontief function is: XST[i,j] = a_io[i,j] * Z[j]

Where: - XST[i,j] = intermediate demand for commodity i by sector j - Z[j] = total output of sector j - a_io[i,j] = input-output coefficient

The coefficient is computed as: a_io[i,j] = XST[i,j] / Z[j]

Variables:

min_coefficient (float) – Minimum threshold for IO coefficients

Parameters:

name (str)
description (str)
min_coefficient (Annotated[float, Ge(ge=0)])

name: str¶

description: str¶

min_coefficient: float¶

calibrate(model, sam, elasticities=None)[source]¶

Calibrate Leontief IO coefficients from SAM.

Parameters:

model (Model) – Model with Leontief blocks
sam (SAM) – SAM data
elasticities (dict[str, float] | None) – Not used for Leontief (no elasticities)

Returns:

CalibrationResult with a_io coefficients

Return type:

CalibrationResult

get_info()[source]¶

Get calibrator info.

Return type:: dict[str, Any]

model_config = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Canonical post-transform safeguards for solver variables.

equilibria.solver.guards.rebuild_tax_detail_from_rates(vars, sets, params, *, include_tip=True)[source]¶

Rebuild detailed tax-payment variables from ad-valorem policy rates.

This is the canonical reconstruction used after array->variables conversion and in initialization routines that need deterministic tax detail.

Parameters:

vars (PEPModelVariables)
sets (dict[str, list[str]])
params (dict[str, Any])
include_tip (bool)

Return type:

None

Canonical array<->variable transforms for the PEP solver.

equilibria.solver.transforms.pep_array_to_variables(x, sets, *, min_price=1e-06)[source]¶

Convert flat solver vector to PEPModelVariables.

Ordering is intentionally aligned with historical IPOPT packing order.

Parameters:

x (ndarray)
sets (dict[str, list[str]])
min_price (float)

Return type:

PEPModelVariables

equilibria.solver.transforms.pep_variables_to_array(vars, sets)[source]¶

Convert PEPModelVariables to flat solver vector.

Parameters:

vars (PEPModelVariables)
sets (dict[str, list[str]])

Return type:

ndarray

Generic contract/runtime primitives shared across model templates.

equilibria.contracts.base.normalize_string_tuple(value)[source]¶

Normalize string-like inputs into an ordered unique tuple.

Parameters:: value (Any)
Return type:: tuple[str, …]

equilibria.contracts.base.deep_merge_model_dicts(base, updates)[source]¶

Recursively merge nested dictionaries used to build contract/config models.

Parameters:

base (dict[str, Any])
updates (Mapping[str, Any])

Return type:

dict[str, Any]

class equilibria.contracts.base.ModelClosureConfig(*, name='default', numeraire='e', numeraire_mode='fixed_benchmark', capital_mobility='mobile', fixed=<factory>, endogenous=<factory>, label=None)[source]¶

Bases: BaseModel

Generic closure definition shared across model contracts.

Parameters:

name (str)
numeraire (str)
numeraire_mode (str)
capital_mobility (str)
fixed (tuple[str, ...])
endogenous (tuple[str, ...])
label (str | None)

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

numeraire: str¶

numeraire_mode: str¶

capital_mobility: str¶

fixed: tuple[str, ...]¶

endogenous: tuple[str, ...]¶

label: str | None¶

class equilibria.contracts.base.ModelEquationConfig(*, name='full_model', include=<factory>, activation_masks='default')[source]¶

Bases: BaseModel

Generic equation activation policy.

Parameters:

name (str)
include (tuple[str, ...])
activation_masks (str)

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

include: tuple[str, ...]¶

activation_masks: str¶

class equilibria.contracts.base.ModelBoundsConfig(*, name='economic', positive='lower_only', fixed_from_closure=True, free=<factory>)[source]¶

Bases: BaseModel

Generic bounds/domain policy.

Parameters:

name (str)
positive (str)
fixed_from_closure (bool)
free (tuple[str, ...])

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

positive: str¶

fixed_from_closure: bool¶

free: tuple[str, ...]¶

class equilibria.contracts.base.ModelContract(*, name, closure, equations, bounds)[source]¶

Bases: BaseModel

Generic model contract: closure + equations + bounds.

Parameters:

name (str)
closure (ModelClosureConfig)
equations (ModelEquationConfig)
bounds (ModelBoundsConfig)

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

closure: ModelClosureConfig¶

equations: ModelEquationConfig¶

bounds: ModelBoundsConfig¶

class equilibria.contracts.base.ModelReferenceConfig(*, enabled=False, source='none', model_type=None, solver=None, slice=None, levels_tol=1e-08, params_tol=1e-08)[source]¶

Bases: BaseModel

Optional parity/reference settings kept outside the economic contract.

Parameters:

enabled (bool)
source (str)
model_type (str | None)
solver (str | None)
slice (str | None)
levels_tol (float)
params_tol (float)

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

enabled: bool¶

source: str¶

model_type: str | None¶

solver: str | None¶

slice: str | None¶

levels_tol: float¶

params_tol: float¶

class equilibria.contracts.base.ModelRuntimeConfig(*, name='default', problem_type='nlp', solver='ipopt', tolerance=1e-08, max_iterations=300, require_solver_success=True, accept_square_feasible=True, reference=<factory>)[source]¶

Bases: BaseModel

Generic runtime configuration for executing a model contract.

Parameters:

name (str)
problem_type (str)
solver (str)
tolerance (float)
max_iterations (int)
require_solver_success (bool)
accept_square_feasible (bool)
reference (ModelReferenceConfig)

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

problem_type: str¶

solver: str¶

tolerance: float¶

max_iterations: int¶

require_solver_success: bool¶

accept_square_feasible: bool¶

reference: ModelReferenceConfig¶

class equilibria.contracts.base.ModelClosureValidationReport(*, is_valid, system_shape, active_equation_count, free_endogenous_variable_count, fixed_by_closure_count, fixed_by_bounds_only_count, equation_variable_gap, numeraire=None, numeraire_is_fixed=None, duplicate_equations=<factory>, duplicate_free_variables=<factory>, unsupported_fixed_symbols=<factory>, unsupported_endogenous_symbols=<factory>, messages=<factory>)[source]¶

Bases: BaseModel

Generic structural validation report for closure/system shape.

Parameters:

is_valid (bool)
system_shape (Literal['square', 'overdetermined', 'underdetermined'])
active_equation_count (int)
free_endogenous_variable_count (int)
fixed_by_closure_count (int)
fixed_by_bounds_only_count (int)
equation_variable_gap (int)
numeraire (str | None)
numeraire_is_fixed (bool | None)
duplicate_equations (tuple[str, ...])
duplicate_free_variables (tuple[str, ...])
unsupported_fixed_symbols (tuple[str, ...])
unsupported_endogenous_symbols (tuple[str, ...])
messages (tuple[str, ...])

model_config = {'extra': 'forbid', 'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

is_valid: bool¶

system_shape: Literal['square', 'overdetermined', 'underdetermined']¶

active_equation_count: int¶

free_endogenous_variable_count: int¶

fixed_by_closure_count: int¶

fixed_by_bounds_only_count: int¶

equation_variable_gap: int¶

numeraire: str | None¶

numeraire_is_fixed: bool | None¶

duplicate_equations: tuple[str, ...]¶

duplicate_free_variables: tuple[str, ...]¶

unsupported_fixed_symbols: tuple[str, ...]¶

unsupported_endogenous_symbols: tuple[str, ...]¶

messages: tuple[str, ...]¶

equilibria.contracts.base.validate_closure_structure(*, active_equations, free_endogenous_variables, fixed_by_closure=None, fixed_by_bounds_only=None, numeraire=None, unsupported_fixed_symbols=None, unsupported_endogenous_symbols=None)[source]¶

Validate the structural shape implied by the current closure.

Parameters:

active_equations (Iterable[str])
free_endogenous_variables (Iterable[str])
fixed_by_closure (Iterable[str] | None)
fixed_by_bounds_only (Iterable[str] | None)
numeraire (str | None)
unsupported_fixed_symbols (Iterable[str] | None)
unsupported_endogenous_symbols (Iterable[str] | None)

Return type:

ModelClosureValidationReport