Scenarios API

API reference for scenario expansion, shock operations, and stress testing.

Core Functions

with_scenarios

Expand ActuarialFrame across scenarios via cross-join.

Creates a new ActuarialFrame with len(af) x len(scenario_ids) rows, preserving all original columns and adding a scenario_id column.

This is the fundamental operation for running actuarial models across multiple economic scenarios in a single vectorized execution.

Parameters:

Name	Type	Description	Default
af	ActuarialFrame	Input ActuarialFrame to expand	required
scenario_ids	list[str] | list[int]	List of scenario identifiers (strings or integers)	required
scenario_column	str	Name for the scenario ID column (default: "scenario_id")	'scenario_id'
categorical	bool	If True and scenario_ids are strings, use Categorical dtype for better join/groupby performance (default: False)	False

Returns:

Type	Description
ActuarialFrame	ActuarialFrame with expanded rows and scenario_column added.

Examples:

Basic scenario expansion:

from gaspatchio_core import ActuarialFrame
from gaspatchio_core.scenarios import with_scenarios

# 2 policies
af = ActuarialFrame({"policy_id": [1, 2], "premium": [100.0, 200.0]})

# Expand to 3 scenarios → 6 rows
af = with_scenarios(af, ["BASE", "UP", "DOWN"])
print(af.collect())

Single deterministic scenario (scenario-ready-by-default):

# Even single-scenario models should use with_scenarios
af = with_scenarios(af, ["DETERMINISTIC"])

Integer scenarios for stochastic runs:

# For 10K stochastic scenarios, use integers for performance
af = with_scenarios(af, list(range(1, 10001)))

Categorical encoding for named scenarios:

# Use categorical=True for better groupby/join performance with string IDs
af = with_scenarios(af, ["BASE", "UP", "DOWN"], categorical=True)

batch_scenarios

Yield batches of scenario IDs for memory-efficient processing.

For large stochastic runs (e.g., 10,000+ scenarios), processing all scenarios at once may exceed available memory. This generator yields chunks of scenario IDs that can be processed iteratively and aggregated.

Parameters:

Name	Type	Description	Default
scenario_ids	list[T]	Complete list of scenario identifiers to batch	required
batch_size	int	Number of scenarios per batch (default: 1000)	1000

Yields:

Type	Description
list[T]	Lists of scenario IDs, each containing up to batch_size elements.
list[T]	The last batch may contain fewer elements.

Raises:

Type	Description
ValueError	If batch_size is not positive

When to use

• Processing 1000+ stochastic scenarios
• Memory-constrained environments
• Parallel processing with worker pools
• Progress reporting during long runs

Examples:

Basic batching for large stochastic runs:

```python no_output_check from gaspatchio_core import ActuarialFrame from gaspatchio_core.scenarios import batch_scenarios, with_scenarios

model_points = ActuarialFrame({"policy_id": [1, 2], "premium": [100.0, 200.0]}) scenario_ids = list(range(1, 10001)) # 10K scenarios

all_results = [] for batch in batch_scenarios(scenario_ids, batch_size=1000): af = with_scenarios(model_points, batch) # Run model and collect results all_results.append(af.collect())

**With progress reporting:**

```python no_output_check
total_scenarios = 5000
scenario_ids = list(range(1, total_scenarios + 1))

for i, batch in enumerate(batch_scenarios(scenario_ids, batch_size=500)):
    print(f"Processing batch {i + 1}/10: scenarios {batch[0]}-{batch[-1]}")
    # Process batch...

describe_scenarios

Generate human-readable descriptions of scenario configurations.

Creates audit-trail-ready documentation of what shocks are applied in each scenario. Useful for governance, compliance, and model documentation requirements.

When to use

• Generating audit trails for regulatory compliance
• Documenting scenario configurations in model reports
• Creating change logs for assumption modifications
• Building scenario comparison reports

Parameters:

Name	Type	Description	Default
scenarios	dict[str, list[Shock]]	Mapping of scenario ID to list of shocks	required
output_format	Literal['text', 'markdown', 'dict']	Output format - "markdown" (default), "text", or "dict"	'markdown'

Returns:

Type	Description
str | dict[str, list[str]]	Formatted description string, or dict for programmatic access

Examples:

Generate markdown documentation:

```python no_output_check from gaspatchio_core.scenarios import describe_scenarios from gaspatchio_core.scenarios.shocks import MultiplicativeShock

scenarios = { "BASE": [], "STRESSED": [MultiplicativeShock(factor=1.2, table="mortality")], }

print(describe_scenarios(scenarios))

**Get as dictionary for programmatic access:**

```python no_output_check
result = describe_scenarios(scenarios, output_format="dict")
for scenario_id, shocks in result.items():
    print(f"{scenario_id}: {len(shocks)} shocks")

sensitivity_analysis

Generate shock configurations for sensitivity analysis across a range of values.

Creates a dictionary of scenarios, each with a single shock applied at different values. Useful for parameter sweeps and sensitivity testing.

When to use

• Sensitivity testing across parameter ranges
• Generating stress scenarios for regulatory reporting
• Parameter calibration and validation
• Creating scenario grids for analysis

Parameters:

Name	Type	Description	Default
table	str	The table name the shocks target	required
shock_type	Literal['multiplicative', 'additive', 'override']	Type of shock - "multiplicative", "additive", or "override"	required
values	list[float]	List of values to sweep (factors, deltas, or constants)	required
column	str | None	Optional column name within the table	None
scenario_format	str | None	Format string for scenario IDs (default: "{table}_{value}")	None
include_base	bool	If True, include a "BASE" scenario with no shocks	False

Returns:

Type	Description
dict[str, list[Shock]]	Dictionary mapping scenario IDs to lists of shocks

Raises:

Type	Description
ValueError	If values is empty or shock_type is invalid

Examples:

Mortality sensitivity sweep:

```python no_output_check from gaspatchio_core.scenarios import sensitivity_analysis

scenarios = sensitivity_analysis( table="mortality", shock_type="multiplicative", values=[0.8, 0.9, 1.0, 1.1, 1.2], )

Returns:

**Interest rate parallel shifts:**

```python no_output_check
scenarios = sensitivity_analysis(
    table="discount_rates",
    shock_type="additive",
    values=[-0.01, -0.005, 0.0, 0.005, 0.01],
)

With custom scenario naming:

```python no_output_check scenarios = sensitivity_analysis( table="mortality", shock_type="multiplicative", values=[0.9, 1.1], scenario_format="mort_shock_{value}", )

Returns:

**Include base case:**

```python no_output_check
scenarios = sensitivity_analysis(
    table="mortality",
    shock_type="multiplicative",
    values=[0.9, 1.1],
    include_base=True,
)
# Returns: {"BASE": [], "mortality_0.9": [...], "mortality_1.1": [...]}

Config Parsing

parse_shock_config

Parse a single shock configuration dict into a Shock object.

Converts LLM-friendly dict format into the appropriate Shock subclass. This enables LLMs to generate JSON/dict configs that the framework can execute without writing Python code.

Supports the unified schema with: - Basic operations: multiply, add, set - Value constraints: clip - Composable operations: pipeline, max, min - Dimension filters: where - Time conditions: when

When to use

• Parsing LLM-generated shock specifications
• Loading shock configs from JSON/YAML files
• Building shocks from API payloads

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Dict with keys: - table (required): Target table name - column (optional): Target column within table - One operation or pipeline (required) - where (optional): Dimension filter conditions - when (optional): Time condition	required

Returns:

Type	Description
Shock | ParameterShock	Appropriate Shock subclass instance

Raises:

Type	Description
ValueError	If config is missing required fields or has invalid structure

Examples:

Simple multiplicative shock:

from gaspatchio_core.scenarios import parse_shock_config

config = {"table": "mortality", "multiply": 1.2}
shock = parse_shock_config(config)
# Returns: MultiplicativeShock(factor=1.2, table="mortality")

Solvency II lapse up (multiply then cap):

config = {"table": "lapse", "pipeline": [{"multiply": 1.5}, {"clip": {"max": 1.0}}]}
shock = parse_shock_config(config)
# Returns: PipelineShock(...)

Syntactic sugar with clip:

config = {"table": "lapse", "multiply": 1.5, "clip": [None, 1.0]}
shock = parse_shock_config(config)
# Returns: PipelineShock([MultiplicativeShock, ClipShock])

Solvency II lapse down (max of two options):

config = {"table": "lapse", "max": [{"multiply": 0.5}, {"add": -0.2}]}
shock = parse_shock_config(config)
# Returns: MaxShock(...)

Dimension-filtered shock:

config = {"table": "lapse", "multiply": 1.25, "where": {"duration": {"lte": 3}}}
shock = parse_shock_config(config)
# Returns: FilteredShock(...)

Time-conditional shock:

config = {"table": "lapse", "add": 0.40, "when": {"t": {"eq": 0}}}
shock = parse_shock_config(config)
# Returns: TimeConditionalShock(...)

parse_scenario_config

Parse a full scenario configuration into shock mappings.

Converts LLM-friendly list of scenario specs into the dict[str, list[Shock]] format used by describe_scenarios() and Table.from_shocks(). This is the main entry point for LLM-generated scenario configurations.

When to use

• Parsing LLM-generated scenario configurations
• Loading scenario configs from JSON/YAML files
• Building scenarios from natural language queries

Parameters:

Name	Type	Description	Default
config	list[str | dict[str, Any]]	List of scenario specifications. Each element can be: - str: Scenario ID with no shocks (e.g., "BASE") - dict: Scenario with optional shocks: - id (required): Scenario identifier - shocks (optional): List of shock configs	required

Returns:

Type	Description
dict[str, list[Shock | ParameterShock]]	Dictionary mapping scenario IDs to lists of Shock objects

Raises:

Type	Description
ValueError	If config is empty, has duplicates, or invalid structure

Examples:

Simple config with strings:

from gaspatchio_core.scenarios import parse_scenario_config

config = ["BASE", "STRESS"]
scenarios = parse_scenario_config(config)
# Returns: {"BASE": [], "STRESS": []}

LLM-generated config with shocks:

config = [
    {"id": "BASE"},
    {
        "id": "RATES_UP_50BPS",
        "shocks": [{"table": "discount_rates", "add": 0.005}],
    },
]
scenarios = parse_scenario_config(config)
# Returns: {
#     "BASE": [],
#     "RATES_UP_50BPS": [AdditiveShock(delta=0.005, table="discount_rates")],
# }

Complex multi-shock scenario:

config = [
    {"id": "BASE"},
    {
        "id": "ADVERSE",
        "shocks": [
            {"table": "mortality", "multiply": 1.2},
            {"table": "lapse", "multiply": 0.8},
            {"table": "interest", "add": -0.01},
        ],
    },
]
scenarios = parse_scenario_config(config)

Basic Shock Classes

MultiplicativeShock

Bases: Shock

A shock that multiplies values by a factor.

Used for scenarios like "increase mortality by 20%" (factor=1.2) or "decrease lapse by 10%" (factor=0.9).

Parameters:

Name	Type	Description	Default
factor	float	The multiplicative factor to apply	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Stress testing with percentage changes
• Sensitivity analysis on rates
• Regulatory capital scenarios (e.g., SCR shocks)

Examples:

20% increase in mortality:

```python no_output_check from gaspatchio_core.scenarios.shocks import MultiplicativeShock

shock = MultiplicativeShock(factor=1.2, table="mortality")

**10% decrease in lapse rates:**

```python no_output_check
shock = MultiplicativeShock(factor=0.9, table="lapse")

describe() -> str

Return description of this shock.

AdditiveShock

Bases: Shock

A shock that adds a constant delta to values.

Used for scenarios like "increase discount rate by 50bps" (delta=0.005) or "decrease expense loading by 1%" (delta=-0.01).

Parameters:

Name	Type	Description	Default
delta	float	The additive constant to apply	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Interest rate shocks (parallel shifts)
• Expense loading adjustments
• Basis point changes to rates

Examples:

Add 50bps to discount rates:

```python no_output_check from gaspatchio_core.scenarios.shocks import AdditiveShock

shock = AdditiveShock(delta=0.005, table="discount_rates")

**Subtract 1% from expense loading:**

```python no_output_check
shock = AdditiveShock(delta=-0.01, table="expenses")

describe() -> str

Return description of this shock.

OverrideShock

Bases: Shock

A shock that replaces all values with a constant.

Used for scenarios like "set lapse to zero" (value=0.0) or "assume 100% mortality" (value=1.0).

Parameters:

Name	Type	Description	Default
value	Any	The constant value to set	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Extreme stress scenarios
• Disabling a decrement entirely
• Testing boundary conditions

Examples:

Set lapse rates to zero:

```python no_output_check from gaspatchio_core.scenarios.shocks import OverrideShock

shock = OverrideShock(value=0.0, table="lapse")

**Override discount rate to flat 5%:**

```python no_output_check
shock = OverrideShock(value=0.05, table="discount_rates")

describe() -> str

Return description of this shock.

Value Constraint Shocks

ClipShock

Bases: Shock

A shock that clips (caps/floors) values to a range.

Used for scenarios like "lapse rate cannot exceed 100%" (max=1.0) or "mortality floor of 0.1%" (min=0.001). Can also combine both.

This is essential for regulatory scenarios like Solvency II SCR lapse up, where shocked values must be capped at actuarial limits.

Parameters:

Name	Type	Description	Default
min_value	float | None	Optional floor value (values below are set to this)	None
max_value	float | None	Optional ceiling value (values above are set to this)	None
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Post-shock value constraints (e.g., lapse ≤ 100%)
• Regulatory scenarios with actuarial limits
• Preventing unrealistic shocked values
• Combining with other shocks in a pipeline

Examples:

Cap lapse rates at 100%:

```python no_output_check from gaspatchio_core.scenarios.shocks import ClipShock

shock = ClipShock(max_value=1.0, table="lapse")

**Floor mortality at 0.1%:**

```python no_output_check
shock = ClipShock(min_value=0.001, table="mortality")

Clip to a range:

python no_output_check shock = ClipShock(min_value=0.0, max_value=1.0, table="rates")

describe() -> str

Return description of this shock.

Composable Shocks

PipelineShock

Bases: Shock

A shock that chains multiple operations in sequence.

Used for complex scenarios like "multiply by 1.5 then cap at 100%" which requires composing multiple shock operations.

The operations are applied left-to-right: the output of each shock becomes the input to the next.

Parameters:

Name	Type	Description	Default
shocks	tuple[Shock, ...]	Sequence of shocks to apply in order	tuple()
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Solvency II lapse up: multiply then cap
• Complex stress scenarios with multiple transformations
• Building reusable shock combinations

Examples:

Solvency II lapse up (multiply by 1.5, cap at 100%):

```python no_output_check from gaspatchio_core.scenarios.shocks import ( PipelineShock, MultiplicativeShock, ClipShock, )

shock = PipelineShock( shocks=[ MultiplicativeShock(factor=1.5), ClipShock(max_value=1.0), ], table="lapse", ) ```

Lapse down with floor (multiply by 0.5, floor at original - 0.2):

This would need a custom approach for relative floors.

describe() -> str

Return description of this shock.

MaxShock

Bases: Shock

A shock that takes the maximum of two shock expressions.

Used for scenarios like Solvency II lapse down: "max(lapse × 0.5, lapse - 0.2)" where the result is the larger of two transformations.

Parameters:

Name	Type	Description	Default
shock_a	Shock	First shock option	required
shock_b	Shock	Second shock option	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Solvency II lapse down: max(×0.5, -0.2)
• Taking the less severe of two shocks
• Complex regulatory scenarios

Examples:

Solvency II lapse down:

```python no_output_check from gaspatchio_core.scenarios.shocks import ( MaxShock, MultiplicativeShock, AdditiveShock, )

shock = MaxShock( shock_a=MultiplicativeShock(factor=0.5), shock_b=AdditiveShock(delta=-0.2), table="lapse", )

Result: max(lapse × 0.5, lapse - 0.2)

```

describe() -> str

Return description of this shock.

MinShock

Bases: Shock

A shock that takes the minimum of two shock expressions.

Used for scenarios where the result should be the smaller of two transformations.

Parameters:

Name	Type	Description	Default
shock_a	Shock	First shock option	required
shock_b	Shock	Second shock option	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Taking the more severe of two shocks
• Cap scenarios (similar to ClipShock but based on transformations)
• Complex regulatory scenarios

Examples:

Take the lower of two mortality assumptions:

```python no_output_check from gaspatchio_core.scenarios.shocks import ( MinShock, MultiplicativeShock, OverrideShock, )

shock = MinShock( shock_a=MultiplicativeShock(factor=1.5), shock_b=OverrideShock(value=0.1), # Cap at 10% mortality table="mortality", ) ```

describe() -> str

Return description of this shock.

Filter Shocks

FilteredShock

Bases: Shock

A shock that applies only to rows matching a filter condition (WHERE clause).

Used for dimension-filtered shocks like "increase early-duration lapse by 25%" where only rows matching the filter are modified.

This implements GSP-65: Dimension-filtered shocks.

Parameters:

Name	Type	Description	Default
shock	Shock	The shock to apply to matching rows	required
where	FilterCondition	Filter condition dictionary	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None

When to use

• Apply shocks to specific segments (e.g., early durations)
• Age-specific mortality adjustments
• Product-specific lapse stress
• Regulatory scenarios with conditional shocks

Examples:

Increase early-duration lapse by 25%:

```python no_output_check from gaspatchio_core.scenarios.shocks import FilteredShock, MultiplicativeShock

shock = FilteredShock( shock=MultiplicativeShock(factor=1.25), where={"duration": {"lte": 3}}, table="lapse", )

**Mortality shock for elderly lives:**

```python no_output_check
shock = FilteredShock(
    shock=MultiplicativeShock(factor=1.15),
    where={"attained_age": {"gte": 65}},
    table="mortality",
)

Complex filter with multiple conditions:

python no_output_check shock = FilteredShock( shock=AdditiveShock(delta=0.02), where={"sex": "F", "smoker_status": "S"}, table="mortality", )

describe() -> str

Return description of this shock.

TimeConditionalShock

Bases: Shock

A shock that applies only at specific projection times (WHEN clause).

Used for time-conditional shocks like "40% mass lapse at t=0 only" where the shock is applied based on projection period.

This implements GSP-74: Time-conditional shocks.

Parameters:

Name	Type	Description	Default
shock	Shock	The shock to apply at matching times	required
when	FilterCondition	Time condition dictionary (uses 't' column by default)	required
table	str | None	Optional table name this shock targets	None
column	str | None	Optional column name this shock targets	None
time_column	str	Column name for time (default: "t")	't'

When to use

• Mass lapse at policy inception (t=0)
• First-year expense shocks
• Time-limited stress scenarios
• Shock only during specific periods

Examples:

Mass lapse at t=0:

```python no_output_check from gaspatchio_core.scenarios.shocks import TimeConditionalShock, AdditiveShock

shock = TimeConditionalShock( shock=AdditiveShock(delta=0.40), # Add 40% lapse when={"t": {"eq": 0}}, table="lapse", )

**Expense shock for first 5 years:**

```python no_output_check
shock = TimeConditionalShock(
    shock=MultiplicativeShock(factor=1.10),
    when={"t": {"lte": 5}},
    table="expenses",
)

describe() -> str

Return description of this shock.

Parameter Shocks

ParameterShock

A shock specification for scalar model parameters (not table values).

Used for scenarios like "increase expense inflation by 1%" where the target is a scalar model input rather than an assumption table.

Unlike table shocks, parameter shocks store the transformation specification and are applied at model setup time by the model code.

Parameters:

Name	Type	Description	Default
param	str	Name of the parameter to shock	required
operation	str	Type of operation ("multiply", "add", or "set")	required
value	float	Value for the operation (factor, delta, or constant)	required

When to use

• Shocking scalar model inputs (expense inflation, discount rate spread)
• Parameters that aren't stored in assumption tables
• Model-level sensitivity analysis

Not a Shock subclass

ParameterShock is NOT a Shock subclass because it doesn't operate on Polars expressions. It stores the shock specification for the model code to apply.

Examples:

Add 1% to expense inflation:

```python no_output_check from gaspatchio_core.scenarios.shocks import ParameterShock

shock = ParameterShock(param="expense_inflation", operation="add", value=0.01)

Apply in model code:

base_inflation = 0.02 shocked_inflation = shock.apply(base_inflation) # 0.03

**Multiply discount spread:**

```python no_output_check
shock = ParameterShock(param="discount_spread", operation="multiply", value=1.5)

apply(base_value: float) -> float

Apply this shock to a base parameter value.

Parameters:

Name	Type	Description	Default
base_value	float	The original parameter value	required

Returns:

Type	Description
float	The shocked parameter value

describe() -> str

Return a human-readable description of this shock.

Related Table Methods

The following methods on the Table class work with scenarios:

• Table.from_scenario_files() - Load separate assumption files per scenario
• Table.from_scenario_template() - Load scenario files using a path template
• Table.from_shocks() - Create multiple shocked tables from shock specifications
• Table.with_shock() - Apply a single shock to create a modified table

See the Actuarial Frame API for details on these methods.

See Also

• Shock Operations - Conceptual guide with progressive examples
• What-If Analysis - Natural language to config translation
• Scenarios Overview - High-level scenario concepts