`equilibria.sam_tools`¶

High-level SAM API focused on Sam as the user-facing object.

This module provides the main entry points for converting Input-Output matrices (MIP) to Social Accounting Matrices (SAM) compatible with CGE models like PEP.

Key functions: - run_mip_to_sam: Convert MIP to SAM with flexible balancing options - run_ieem_to_pep: Convert IEEM SAM format to PEP format

class equilibria.sam_tools.api.IEEMToPEPResult(sam, steps, output_path, report_path)[source]¶

Bases: NamedTuple

Result object returned by run_ieem_to_pep().

Parameters:

sam (Sam)
steps (list[dict[str, Any]])
output_path (Path | None)
report_path (Path | None)

sam: Sam¶: Alias for field number 0

steps: list[dict[str, Any]]¶: Alias for field number 1

output_path: Path | None¶: Alias for field number 2

report_path: Path | None¶: Alias for field number 3

class equilibria.sam_tools.api.MIPToSAMResult(sam, steps, output_path, report_path)[source]¶

Bases: NamedTuple

Result object returned by run_mip_to_sam().

Parameters:

sam (Sam)
steps (list[dict[str, Any]])
output_path (Path | None)
report_path (Path | None)

sam: Sam¶: Alias for field number 0

steps: list[dict[str, Any]]¶: Alias for field number 1

output_path: Path | None¶: Alias for field number 2

report_path: Path | None¶: Alias for field number 3

equilibria.sam_tools.api.run_ieem_to_pep(input_path, mapping_path, *, sheet_name='MCS2016', commodity_to_sector=None, margin_commodity='ser', ras_type='arithmetic', ras_tol=1e-09, ras_max_iter=200, output_path=None, report_path=None)[source]¶

Run a linear IEEM->PEP conversion and return the transformed Sam.

Parameters:

input_path (Path | str)
mapping_path (Path | str)
sheet_name (str)
commodity_to_sector (Mapping[str, str] | None)
margin_commodity (str)
ras_type (str)
ras_tol (float)
ras_max_iter (int)
output_path (Path | str | None)
report_path (Path | str | None)

Return type:

IEEMToPEPResult

equilibria.sam_tools.api.run_mip_to_sam(input_path, *, va_factor_shares=None, factor_to_household_shares=None, tax_rates=None, product_tax_rates=None, sheet_name='MIP', va_row_label='Valor Agregado', import_row_label='Importaciones', balancing_method='gras', ras_type='arithmetic', ras_tol=1e-09, ras_max_iter=200, output_path=None, report_path=None)[source]¶

Convert MIP (Input-Output Matrix) to SAM compatible with PEP CGE models.

This function transforms a standard national accounts MIP with aggregated Value Added into a complete SAM with explicit factors (L, K) and institutions (households, government, firms, ROW).

System of Equations (MIP Balancing):

Product balance: X = Z_d @ 1 + F_d
Import balance: M = Z_m @ 1 + F_m
Industry balance: X = Z_d.T @ 1 + Z_m.T @ 1 + VA
Aggregate identity: sum(F_d) = sum(VA) + sum(Z_m)

Note: Supply-Demand balance (per product) is mathematically incompatible with the above constraints when IMP_F > 0. The balancing methods prioritize PIB identity and Z balance, accepting S-D imbalance. CGE models equilibrate S-D endogenously via prices.

Parameters:

input_path (Path | str) – Path to MIP Excel file
va_factor_shares (dict[str, float] | None) – Factor shares of VA. Default: {“L”: 0.65, “K”: 0.35}
factor_to_household_shares (dict[str, dict[str, float]] | None) – Income distribution from factors to institutions. Default: simple 1-household distribution
tax_rates (dict[str, float] | None) – Tax rates for production_tax, import_tariff, direct_tax. Default: {“production_tax”: 0.10, “import_tariff”: 0.05, “direct_tax”: 0.15}
product_tax_rates (dict[str, float] | None) – Optional tax rates on products by sector. If provided, adds AG.ti and AG.tm accounts to reach PIB at market prices. Default: None (work with VAB at basic prices)
sheet_name (str) – Name of Excel sheet containing MIP
va_row_label (str) – Label to identify Value Added row in MIP
import_row_label (str) – Label to identify imports row in MIP
balancing_method (Literal['ras', 'gras', 'sut_ras', 'entropy', 'none']) – Method for MIP balancing. Options: - “gras” (default): GRAS algorithm, handles negatives - “ras”: Classic RAS for non-negative matrices - “sut_ras”: SUT-RAS simultaneous balancing - “entropy”: Cross-entropy minimization - “none”: No MIP balancing (use pre-balanced input)
ras_type (str) – SAM balancing algorithm type (“arithmetic” or “multiplicative”)
ras_tol (float) – Tolerance for RAS convergence
ras_max_iter (int) – Maximum iterations for RAS balancing
output_path (Path | str | None) – Optional path to save resulting SAM as Excel
report_path (Path | str | None) – Optional path to save transformation report as JSON

Returns:

MIPToSAMResult with transformed SAM and transformation history

Return type:

MIPToSAMResult

Example

>>> # Basic usage with GRAS balancing
>>> result = run_mip_to_sam(
...     "data/mip_bolivia.xlsx",
...     balancing_method="gras",
...     va_factor_shares={"L": 0.39, "K": 0.61},
...     output_path="output/sam_bolivia.xlsx"
... )

>>> # With product taxes (market prices)
>>> result = run_mip_to_sam(
...     "data/mip_ecuador.xlsx",
...     product_tax_rates={"effective_rate": 0.08},
...     output_path="output/sam_ecuador_market.xlsx"
... )

Transformation primitives¶

MIP to SAM structural transformations.

equilibria.sam_tools.mip_to_sam_transforms.normalize_mip_accounts(sam, op)[source]¶

Convert RAW MIP labels to initial PEP structure (J, I, VA, FD).

The MIP raw matrix is square-padded so FD labels (HH/GOV/INV/EXP) appear on the row axis with zero data. We classify columns first using the FD closed vocabulary; everything else in the columns is a sector (J), and rows that match an FD label are dropped as spurious zero-rows.

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_make_matrix(sam, op)[source]¶

Add diagonal make-matrix entries so each sector J has income matching its cost.

A standard MIP records sector costs in J columns (intermediate + VA + IMP) and commodity supply in I rows, but it does not record sector output per commodity. SAM closure requires J row income to equal J column cost. We close the loop with a diagonal make matrix: each sector j sells exactly its column total to commodity j, i.e. df.loc[(“J”, j), (“I”, j)] = sum of (“J”, j) column.

This assumes one-to-one sector-commodity correspondence, which holds for MIPs where sectors and commodities share the same labels.

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.disaggregate_va_to_factors(sam, op)[source]¶

Split VA aggregate row into L (labor) and K (capital) using shares.

Before:: (“VA”, “aggregate”) → (“J”, “agr”) = 100
After:: (“L”, “labor”) → (“J”, “agr”) = 65 # 100 * 0.65 (“K”, “capital”) → (“J”, “agr”) = 35 # 100 * 0.35

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_factor_income_distribution(sam, op)[source]¶

Route factor income to institutions (households, firms, government).

Before:: (“L”, “labor”) → (“J”, “agr”) = 65 # Only production cost
After:: (“L”, “labor”) → (“J”, “agr”) = 65 # Keep production cost (“AG”, “hh”) → (“L”, “labor”) = 61.75 # 65 * 0.95 to households (“AG”, “gvt”) → (“L”, “labor”) = 3.25 # 65 * 0.05 taxes

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_household_expenditure(sam, op)[source]¶

Convert final demand “HH” from MIP to AG.hh → I flows.

Before:: (“I”, “agr”) → (“FD”, “HH”) = 20 # Final demand temporary
After:: (“AG”, “hh”) → (“I”, “agr”) = 20 # Household consumption

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_government_flows(sam, op)[source]¶

Consolidate government flows (fiscal revenues + expenditure).

Creates: - (“AG”, “ti”) → (“AG”, “gvt”): Indirect taxes - (“AG”, “tm”) → (“AG”, “gvt”): Import tariffs - (“AG”, “gvt”) → (“I”, *): Government consumption

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_row_account(sam, op)[source]¶

Create rest-of-world (ROW) account for foreign trade.

Before:: (“IMP”, “total”) → (“I”, “agr”) = 15 # Imports (“I”, “agr”) → (“FD”, “EXP”) = 10 # Exports
After:: (“AG”, “row”) → (“I”, “agr”) = 15 # Import supply (“I”, “agr”) → (“AG”, “row”) = 10 # Export demand (moved to X later)

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

equilibria.sam_tools.mip_to_sam_transforms.create_investment_account(sam, op)[source]¶

Create savings-investment closure.

Creates: - (“AG”, *) → (“OTH”, “inv”): Savings from institutions - (“I”, *) → (“OTH”, “inv”): Investment demand

Parameters:

sam (Sam)
op (dict[str, Any])

Return type:

dict[str, Any]

Models¶

Data models for SAM workflow execution.

class equilibria.sam_tools.models.Sam(*, dataframe)[source]¶

Bases: BaseModel

Contenedor base para una SAM, asegura matriz cuadrada con cuentas consistentes.

Parameters:: dataframe (DataFrame)

dataframe: pd.DataFrame¶

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

classmethod from_matrix(matrix, row_keys, col_keys)[source]¶

Parameters:

matrix (ndarray)
row_keys (Sequence[tuple[str, str]])
col_keys (Sequence[tuple[str, str]])

Return type:

Sam

property row_keys: list[tuple[str, ...]]¶

property col_keys: list[tuple[str, ...]]¶

property matrix: ndarray¶

update_matrix(matrix)[source]¶

Parameters:: matrix (ndarray)
Return type:: None

to_dataframe()[source]¶

Return type:: DataFrame

replace_dataframe(frame)[source]¶

Parameters:: frame (DataFrame)
Return type:: None

aggregate(mapping_path)[source]¶

Parameters:: mapping_path (Path)
Return type:: Sam

balance_ras(*, ras_type='arithmetic', tolerance=1e-09, max_iterations=200)[source]¶

Parameters:

ras_type (str)
tolerance (float)
max_iterations (int)

Return type:

RASBalanceResult

balance_status(*, tolerance=1e-09)[source]¶

Return balance diagnostics for row vs column totals.

Parameters:: tolerance (float)
Return type:: dict[str, Any]

class equilibria.sam_tools.models.SamTable(*, sam, source_path, source_format, raw_df=None, data_start_row=None, data_start_col=None)[source]¶

Bases: BaseModel

Table-level SAM object with source metadata and editable matrix access.

Parameters:

sam (Sam)
source_path (Path)
source_format (str)
raw_df (DataFrame | None)
data_start_row (int | None)
data_start_col (int | None)

sam: Sam¶

source_path: Path¶

source_format: str¶

raw_df: pd.DataFrame | None¶

data_start_row: int | None¶

data_start_col: int | None¶

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

property row_keys: list[tuple[str, ...]]¶

property col_keys: list[tuple[str, ...]]¶

property matrix: ndarray¶

to_dataframe()[source]¶

Return type:: DataFrame

class equilibria.sam_tools.models.SAMWorkflowConfig(*, name, country, input_path, input_format, output_path, output_format, input_options=<factory>, transforms=<factory>, report_path, output_symbol)[source]¶

Bases: BaseModel

Resolved workflow config from YAML.

Parameters:

name (str)
country (str | None)
input_path (Path)
input_format (SAMFormat)
output_path (Path)
output_format (SAMFormat)
input_options (dict[str, Any])
transforms (list[dict[str, Any]])
report_path (Path | None)
output_symbol (str)

name: str¶

country: str | None¶

input_path: Path¶

input_format: SAMFormat¶

output_path: Path¶

output_format: SAMFormat¶

input_options: dict[str, Any]¶

transforms: list[dict[str, Any]]¶

report_path: Path | None¶

output_symbol: str¶

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

equilibria.sam_tools¶

Transformation primitives¶

Models¶

`equilibria.sam_tools`¶