Skip to content

Mathematical Utilities

Mathematical functions for allocation calculations.

Convergence Solver

find_minimum_convergence_speed

fair_shares.library.utils.math.convergence.find_minimum_convergence_speed

Python
find_minimum_convergence_speed(
    sorted_columns: list,
    start_column: str | int | float,
    year_fraction_of_cumulative_emissions: Series,
    initial_shares: Series,
    target_cumulative_shares: Series,
    diagnostic_params: dict | None = None,
    strict: bool = True,
    max_convergence_speed: float = 0.9,
) -> tuple[
    float, Series, dict[str, float] | None, Series | None
]

Find the minimum convergence_speed that produces valid target cumulative shares.

Uses binary search between 0.001 and max_convergence_speed.

Parameters:

Name Type Description Default
sorted_columns list

Year columns in sorted order.

 required
start_column str | int | float

Column label for first allocation year.

 required
year_fraction_of_cumulative_emissions Series

Fraction of cumulative emissions in each year.

 required
initial_shares Series

Initial emission shares at start year.

 required
target_cumulative_shares Series

Target cumulative shares to achieve.

 required
diagnostic_params dict | None

Parameters for error messages (approach name, years, weights).

 None
strict bool

If True (default), raise AllocationError when exact targets cannot be achieved. If False, accept approximate targets and generate warnings showing deviations.

 True
max_convergence_speed float

Maximum allowed convergence speed (0 to 1.0). Default 0.9. Lower values create smoother pathways. When targets cannot be achieved within this speed, strict=False will generate warnings.

 0.9

Returns:

Type Description
tuple[float, Series, dict[str, float] | None, Series | None]
  • Convergence speed (0.001 to max_convergence_speed)
  • Long-run shares
  • Adjustment warnings dict: Maps iso3c -> (achieved/target) ratio. None if exact targets achieved or strict=True raised error.
  • Achieved cumulative shares (may differ from targets if speed limited). None if exact targets achieved or strict=True raised error.

validate_convergence_speed

fair_shares.library.utils.math.convergence.validate_convergence_speed

Python
validate_convergence_speed(
    convergence_speed: float,
    sorted_columns: list,
    start_column: str | int | float,
    year_fraction_of_cumulative_emissions: Series,
    initial_shares: Series,
    target_cumulative_shares: Series,
) -> tuple[bool, Series | None]

Validate that a convergence_speed produces valid shares hitting cumulative targets.

We need yearly shares to transition from initial_shares (yearly share for starting year) to eventually satisfy target_cumulative_shares (cumulative sum of yearly shares of emissions over all years for each country).

So the constraint is that the cumulative sum of yearly shares of emissions over all years must equal the cumulative target shares for each country:

Text Only
sum_t [year_fraction_of_cumulative_emissions(t) * yearly_share(t)]
    = target_cumulative_shares

where: - year_fraction_of_cumulative_emissions(t): fraction of total cumulative world emissions that occur in year t for each country (sums to 1) - yearly_share(t): each country's yearly allocation share in year t (sums to 1) - target_cumulative_shares: cumulative target (sum of yearly shares of emissions over all years for each country, sums to 1)

Since we want to converge, rather than jump instantly, we model this using an exponential decay where each year's share moves toward a long_run_shares value:

Text Only
yearly_share(t+1) = yearly_share(t) + speed * (long_run_shares
    - yearly_share(t))

where: - speed: convergence rate (0 = no change, 1 = instant jump) - long_run_shares: asymptotic year share that each year converges toward (not cumulative - this is a single asymptotic year value) - initial_shares: yearly share for the starting year (year 0)

By using the exponential decay convergence function we reduce the problem from N unknowns to 1 unknown (long_run_shares). We know initial_shares, convergence_speed (which determines how fast initial_shares decay as (1-speed)^t), and target_cumulative_shares. So, we can calculate how much of the cumulative target shares comes from initial_shares (w = total_initial_contribution, accounting for their decaying contribution to cumulative target shares over time). The rest (1-w) must come from long_run_shares, so we solve: long_run_shares = (target_cumulative_shares - initial_shares*w) / (1-w).

The convergence speed is valid only if long_run_shares are all in [0, 1] (each country's share is valid). Note: if initial_shares and target_cumulative_shares both sum to 1, then long_run_shares automatically sums to 1 (mathematically guaranteed), so we only need to check individual bounds. The intuition for the [0, 1] bound is that if long_run_shares are not in [0, 1], then the speed is invalid because it would require impossible destinations (negative or >100%) to compensate, which would occur over the finite time horizon we care about.

The caller uses this function to find the minimum valid speed: the smoothest transition that still achieves cumulative targets.

Parameters:

Name Type Description Default
convergence_speed float

The convergence speed to validate (0 to 1).

 required
sorted_columns list

Year columns in sorted order.

 required
start_column str | int | float

Column label for first allocation year.

 required
year_fraction_of_cumulative_emissions Series

Fraction of cumulative emissions in each year (sums to 1).

 required
initial_shares Series

Initial emission shares at start year.

 required
target_cumulative_shares Series

Target cumulative shares to achieve.

 required

Returns:

Type Description
tuple[bool, Series | None]

(is_valid, long_run_shares) where: - is_valid: boolean indicating whether the speed is valid - long_run_shares: yearly share value that satisfies the cumulative constraint where long_run_shares is not used anywhere but just reported.

Adjustment Calculations

calculate_responsibility_adjustment_data

fair_shares.library.utils.math.adjustments.calculate_responsibility_adjustment_data

Python
calculate_responsibility_adjustment_data(
    country_actual_emissions_ts: TimeseriesDataFrame,
    population_ts: TimeseriesDataFrame,
    pre_allocation_responsibility_year: int,
    allocation_year: int,
    pre_allocation_responsibility_per_capita: bool,
    group_level: str,
    unit_level: str,
    ur: PlainRegistry,
    historical_discount_rate: float = 0.0,
) -> Series

Calculate historical responsibility data for allocation.

Returns cumulative emissions (or per capita emissions) from pre_allocation_responsibility_year up to (but not including) allocation_year.

Responsibility window: [pre_allocation_responsibility_year, allocation_year - 1].

This function is used by both budget and pathway allocations.

Parameters:

Name Type Description Default
country_actual_emissions_ts TimeseriesDataFrame

Historical emissions timeseries data

 required
population_ts TimeseriesDataFrame

Population timeseries data

 required
pre_allocation_responsibility_year int

Start year of responsibility window (inclusive)

 required
allocation_year int

End year of responsibility window (exclusive). For budgets: the allocation year itself. For pathways: the first allocation year.

 required
pre_allocation_responsibility_per_capita bool

If True, divide cumulative emissions by cumulative population

 required
group_level str

Index level name for country/region grouping

 required
unit_level str

Index level name for units

 required
ur PlainRegistry

Pint unit registry

 required
historical_discount_rate float

Discount rate for historical emissions (0.0 to <1.0). When > 0, earlier emissions are weighted less via (1 - rate)^(reference_year - t), where reference_year = allocation_year - 1. Implements natural CO2 removal rationale (Dekker Eq. 5). Default: 0.0 (no discounting).

 0.0

Returns:

Type Description
Series

Historical responsibility metric by country/region. Units: emissions (or emissions per capita) depending on pre_allocation_responsibility_per_capita.

Raises:

Type Description
AllocationError

If no years found in responsibility window, no country data found, zero population encountered (when per capita), responsibility sums to non-positive, or historical_discount_rate is out of range.
See Also

docs/science/allocations.md : Theoretical basis for historical responsibility

calculate_capability_adjustment_data

fair_shares.library.utils.math.adjustments.calculate_capability_adjustment_data

Python
calculate_capability_adjustment_data(
    population_ts: TimeseriesDataFrame,
    gdp_ts: TimeseriesDataFrame,
    first_allocation_year: int,
    capability_per_capita: bool,
    group_level: str,
    unit_level: str,
    ur: PlainRegistry,
    gini_s: DataFrame | None = None,
    income_floor: float = 0.0,
    max_gini_adjustment: float = 0.8,
) -> Series

Calculate economic capability data (GDP-based proxy for the Ability to Pay Principle).

Returns the raw capability data, NOT an adjustment factor. The caller applies the inverse to reduce allocations for higher capability.

Capability window: from first_allocation_year onwards.

Parameters:

Name Type Description Default
population_ts TimeseriesDataFrame

Population timeseries data

 required
gdp_ts TimeseriesDataFrame

GDP timeseries data

 required
first_allocation_year int

First year of capability window

 required
capability_per_capita bool

If True, divide cumulative GDP by cumulative population

 required
group_level str

Index level name for country/region grouping

 required
unit_level str

Index level name for units

 required
ur PlainRegistry

Pint unit registry

 required
gini_s DataFrame | None

Optional Gini coefficient data for inequality adjustment. When provided, GDP is adjusted to reflect income distribution.

 None
income_floor float

Income floor for Gini adjustment (in USD PPP per capita). Income below this threshold is excluded from capability calculations, implementing the Greenhouse Development Rights development threshold [Baer 2013]. Default: 0.0

 0.0
max_gini_adjustment float

Maximum reduction factor from Gini adjustment (0-1). Default: 0.8

 0.8

Returns:

Type Description
Series

Capability metric by country/region (cumulative GDP or GDP per capita).

Raises:

Type Description
AllocationError

If no common years between population and GDP, or capability sums to non-positive.
See Also

calculate_responsibility_adjustment_data_convergence : For pre-allocation responsibility adjustments docs/science/allocations.md : Theoretical basis for capability adjustment

RCB Pathway Generation

calculate_exponential_decay_pathway

fair_shares.library.utils.calculate_exponential_decay_pathway

Python
calculate_exponential_decay_pathway(
    total_budget: float,
    start_value: float,
    start_year: int,
    end_year: int,
    tolerance: float = 1e-06,
) -> Series

Generate an exponential decay emission pathway that sums to a total budget.

Creates a pathway following normalized shifted exponential decay that reaches exactly zero at the end year:

Text Only
E(t) = start_value * (e^(-k*t) - e^(-k*T)) / (1 - e^(-k*T))

where k is solved such that the discrete sum of annual emissions equals the total budget. The normalization factor (1 - e^(-k*T)) ensures:

  • E(0) = start_value (preserves initial year emissions)
  • E(T) = 0 (reaches exactly zero at end year)
  • Sum over all years equals total_budget

This is more appropriate for RCB-derived pathways than standard exponential decay which asymptotically approaches (but never reaches) zero, ensuring the budget is fully consumed by the end year.

Parameters:

Name Type Description Default
total_budget float

Total cumulative emissions budget to distribute over the pathway. Must be positive.

 required
start_value float

Initial emission rate at start_year (E(t_0)). Must be positive.

 required
start_year int

First year of the pathway (t_0).

 required
end_year int

Last year of the pathway (T), inclusive.

 required
tolerance float

Relative tolerance for budget conservation verification. Default is 1e-6.

 1e-06

Returns:

Type Description
Series

Annual emissions pathway indexed by year (as strings). Values represent emissions for each year.

Raises:

Type Description
AllocationError

If inputs are invalid, no solution exists, or budget constraint cannot be satisfied.

Examples:

Python Console Session
>>> pathway = calculate_exponential_decay_pathway(
...     total_budget=1000.0, start_value=50.0, start_year=2020, end_year=2050
... )
>>> abs(pathway.sum() - 1000.0) < 0.01
True

generate_rcb_pathway_scenarios

fair_shares.library.utils.generate_rcb_pathway_scenarios

Python
generate_rcb_pathway_scenarios(
    rcbs_df: DataFrame,
    world_emissions_df: DataFrame,
    start_year: int,
    end_year: int,
    emission_category: str,
    generator: str = "exponential-decay",
) -> DataFrame

Generate pathway scenarios from RCB data.

For each RCB scenario (combination of source, climate-assessment, quantile), generates a separate pathway starting from historical world emissions at start_year and summing to the RCB budget. This preserves all individual RCB sources rather than averaging across sources.

Parameters:

Name Type Description Default
rcbs_df DataFrame

Processed RCBs with columns: source, climate-assessment, quantile, emission-category, rcb_2020_mt.

 required
world_emissions_df DataFrame

World historical emissions with MultiIndex (iso3c, unit, emission-category) and year columns as strings.

 required
start_year int

First year of pathways (typically 2020).

 required
end_year int

Last year of pathways (typically 2100).

 required
emission_category str

Emission category to process. Available categories depend on the data source; see available_categories in conf/data_sources/data_sources_unified.yaml.

 required
generator str

Name of the pathway generator to use. Default is "exponential-decay". Use list_pathway_generators() to see available options.

 'exponential-decay'

Returns:

Type Description
DataFrame

TimeseriesDataFrame with MultiIndex (climate-assessment, quantile, source, iso3c, unit, emission-category) and year columns containing World pathway emissions. One pathway per RCB source.

Note: 'source' is included as an index level (after quantile) to preserve all individual RCB sources for allocations.

Raises:

Type Description
AllocationError

If required data is missing, generator is unknown, or pathway generation fails.

list_pathway_generators

fair_shares.library.utils.math.pathways.list_pathway_generators

Python
list_pathway_generators() -> list[str]

List available pathway generation methods.

Returns:

Type Description
list[str]

Available generators: []

See Also