Refactor piecewise polynomial to use explicit interval notation (GEP-08)

Replace numbered dict format with ordered list of interval dicts:
- Add interval_utils.py: validate_intervals(),
  extend_intervals_to_real_line()
- Support partial domains (NaN for out-of-domain values)
- Transpose rates shape from (n_coefficients, n_intervals) to
  (n_intervals, n_coefficients)
- Rename keys: rate_linear->slope, rate_quadratic->quadratic,
  rate_cubic->cubic, intercept_at_lower_threshold->intercept
- Handle list-based YAML specs and updates_previous merging
- Add portion dependency for interval parsing/validation
- Update mettsim YAML files and tests

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: hmgaudecker

pixi.lock

@@ -28,6 +28,7 @@ dependencies = [
   "optree>=0.16.0",
   "pandas",
   "plotly>=6.2.0",
+  "portion",
   "pygments",
   "pygraphviz",
   "pytest",
@@ -101,6 +102,7 @@ numpy_groupies = "*"
 numpydoc = "*"
 openpyxl = "*"
 pandas = ">=2.3"
+portion = "*"
 prek = "*"
 pygments = "*"
 pygraphviz = "*"

pyproject.toml

@@ -829,8 +829,10 @@ def _param_with_active_periods(
 ) -> list[_ParamWithActivePeriod]:
     """Return parameter with active periods."""
 
-    def _remove_note_and_reference(entry: dict[str | int, Any]) -> dict[str | int, Any]:
+    def _remove_note_and_reference(entry: dict[str | int, Any] | list) -> dict[str | int, Any] | list:
         """Remove note and reference from a parameter specification."""
+        if isinstance(entry, list):
+            return entry
         entry.pop("note", None)
         entry.pop("reference", None)
         return entry

src/ttsim/interface_dag_elements/fail_if.py

@@ -28,6 +28,7 @@
 from ttsim.tt.column_objects_param_function import (
     DEFAULT_END_DATE,
 )
+from ttsim.tt.interval_utils import extend_intervals_to_real_line
 from ttsim.tt.piecewise_polynomial import get_piecewise_parameters
 
 if TYPE_CHECKING:
@@ -188,7 +189,7 @@ def _get_one_param(
         cleaned_spec["value"] = get_piecewise_parameters(
             leaf_name=leaf_name,
             func_type=param_type,  # ty: ignore[invalid-argument-type]
-            parameter_dict=cleaned_spec["value"],
+            parameter_list=cleaned_spec["value"],
             xnp=xnp,
         )
         return PiecewisePolynomialParam(**cleaned_spec)
@@ -242,17 +243,24 @@ def _clean_one_param_spec(
     out["reference_period"] = spec.get("reference_period", None)
     out["name"] = spec["name"]
     out["description"] = spec["description"]
-    current_spec: dict[str | int, Any] = copy.deepcopy(spec[policy_dates[idx - 1]])
+    raw_current = copy.deepcopy(spec[policy_dates[idx - 1]])
+    if isinstance(raw_current, list):
+        # List-based spec (no note/reference at this level)
+        out["note"] = None
+        out["reference"] = None
+        out["value"] = _get_param_value([spec[d] for d in policy_dates[:idx]])
+        return out
+
+    current_spec: dict[str | int, Any] = raw_current
     out["note"] = current_spec.pop("note", None)
     out["reference"] = current_spec.pop("reference", None)
-    if len(current_spec) == 0:
+    if not current_spec:
         return None
     if len(current_spec) == 1 and "updates_previous" in current_spec:
         raise ValueError(
             "'updates_previous' cannot be specified as the only element, found:\n\n"
             f"{spec}\n\n",
         )
-        # Parameter ceased to exist
     if spec["type"] == "scalar":
         if "updates_previous" in current_spec:
             raise ValueError(
@@ -265,14 +273,24 @@ def _clean_one_param_spec(
 
 
 def _get_param_value(
-    relevant_specs: list[dict[str | int, Any]],
-) -> dict[str | int, Any]:
+    relevant_specs: list[dict[str | int, Any] | list[dict[str, Any]]],
+) -> dict[str | int, Any] | list[dict[str, Any]]:
     """Get the value of a parameter.
 
     Implementation is a recursion in order to handle the 'updates_previous' machinery.
 
+    Supports both dict-based and list-based (piecewise) specs. When the raw spec
+    is a list (no reference/note fields), it's used directly. When it's a dict
+    with integer keys (has reference/note alongside), the integer-keyed entries
+    are converted to a list.
+
     """
-    current_spec = relevant_specs[-1].copy()
+    raw_spec = relevant_specs[-1]
+    if isinstance(raw_spec, list):
+        # Already a list (YAML date entry was a plain list)
+        return raw_spec
+
+    current_spec = raw_spec.copy()
     updates_previous = current_spec.pop("updates_previous", False)
     current_spec.pop("note", None)
     current_spec.pop("reference", None)
@@ -282,8 +300,33 @@ def _get_param_value(
                 "'updates_previous' cannot be missing in the initial spec, found "
                 f"{relevant_specs}"
             )
+        base = _get_param_value(relevant_specs=relevant_specs[:-1])
+        if isinstance(base, list):
+            # List-based spec: convert list to dict with integer keys for merging
+            base_dict = dict(enumerate(base))
+            merged = upsert_tree(
+                base=base_dict,
+                to_upsert=current_spec,  # ty: ignore[invalid-argument-type]
+            )
+            result = [
+                merged[i] for i in sorted(k for k in merged if isinstance(k, int))
+            ]
+            return extend_intervals_to_real_line(result)
         return upsert_tree(
-            base=_get_param_value(relevant_specs=relevant_specs[:-1]),  # ty: ignore[invalid-argument-type]
+            base=base,  # ty: ignore[invalid-argument-type]
             to_upsert=current_spec,  # ty: ignore[invalid-argument-type]
         )
+
+    # Convert integer-keyed dict to list when keys are consecutive ints 0..n-1
+    # and values are dicts (piecewise polynomial interval specs).
+    # Do not convert when values are scalars (consecutive_int_lookup_table)
+    # or keys are non-consecutive (partial overlay specs like {3: {...}}).
+    if (
+        current_spec
+        and all(isinstance(k, int) for k in current_spec)
+        and all(isinstance(v, dict) for v in current_spec.values())
+        and sorted(current_spec) == list(range(len(current_spec)))
+    ):
+        return [current_spec[i] for i in range(len(current_spec))]
+
     return current_spec

src/ttsim/interface_dag_elements/policy_environment.py

@@ -17,6 +17,7 @@
     policy_function,
     policy_input,
 )
+from ttsim.tt.interval_utils import extend_intervals_to_real_line
 from ttsim.tt.param_objects import (
     ConsecutiveIntLookupTableParam,
     ConsecutiveIntLookupTableParamValue,
@@ -63,6 +64,7 @@
     "agg_by_group_function",
     "agg_by_p_id_function",
     "convert_sparse_to_consecutive_int_lookup_table",
+    "extend_intervals_to_real_line",
     "get_consecutive_int_lookup_table_param_value",
     "get_month_based_phase_inout_of_age_thresholds_param_value",
     "get_piecewise_parameters",

src/ttsim/tt/__init__.py

@@ -0,0 +1,93 @@
+"""Utilities for validating interval notation in piecewise specs."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import numpy
+import portion
+
+if TYPE_CHECKING:
+    from types import ModuleType
+
+    from jaxtyping import Array, Float
+
+
+def validate_intervals(intervals: list[portion.Interval], leaf_name: str) -> None:
+    """Validate that intervals are ascending, non-overlapping, and contiguous."""
+    if not intervals:
+        raise ValueError(f"No intervals provided for {leaf_name}.")
+
+    for i in range(1, len(intervals)):
+        prev, curr = intervals[i - 1], intervals[i]
+        if curr.lower <= prev.lower:
+            raise ValueError(
+                f"Intervals for {leaf_name} are not in ascending order: "
+                f"interval {i - 1} has lower bound {prev.lower}, "
+                f"interval {i} has lower bound {curr.lower}."
+            )
+        if not (prev & curr).empty:
+            raise ValueError(
+                f"Overlapping intervals for {leaf_name}: "
+                f"interval {i - 1} = {prev} and interval {i} = {curr}."
+            )
+        if prev.upper != curr.lower:
+            raise ValueError(
+                f"Gap between intervals for {leaf_name}: "
+                f"interval {i - 1} upper = {prev.upper}, "
+                f"interval {i} lower = {curr.lower}."
+            )
+
+
+def _bound_to_float(v: object) -> float:
+    """Convert a portion bound (including portion.inf) to a Python float."""
+    if v == -portion.inf:
+        return float("-inf")
+    if v == portion.inf:
+        return float("inf")
+    return float(v)
+
+
+def intervals_to_thresholds(
+    intervals: list[portion.Interval], xnp: ModuleType
+) -> tuple[
+    Float[Array, " n"],
+    Float[Array, " n"],
+    Float[Array, " n_plus_1"],
+]:
+    """Extract threshold arrays from parsed intervals."""
+    lower = numpy.array([_bound_to_float(iv.lower) for iv in intervals])
+    upper = numpy.array([_bound_to_float(iv.upper) for iv in intervals])
+    all_bounds = sorted(set(lower) | set(upper))
+    return xnp.array(lower), xnp.array(upper), xnp.array(all_bounds)
+
+
+def extend_intervals_to_real_line(
+    items: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Extend intervals so adjacent intervals are contiguous.
+
+    After merging specs (e.g., via updates_previous), changing one interval's
+    bounds can leave adjacent intervals with stale boundaries. This propagates
+    each interval's upper bound as the next interval's lower bound.
+    """
+    if not items or not any("interval" in item for item in items):
+        return items
+
+    result = [item.copy() for item in items]
+    for i in range(len(result) - 1):
+        if "interval" not in result[i] or "interval" not in result[i + 1]:
+            continue
+        curr = portion.from_string(result[i]["interval"], conv=float)
+        next_ = portion.from_string(result[i + 1]["interval"], conv=float)
+        if curr.upper != next_.lower:
+            complement = portion.CLOSED if curr.right == portion.OPEN else portion.OPEN
+            fixed = portion.Interval.from_atomic(
+                complement, curr.upper, next_.upper, next_.right
+            )
+            result[i + 1] = {
+                **result[i + 1],
+                "interval": portion.to_string(fixed),
+            }
+
+    return result

src/ttsim/tt/interval_utils.py

@@ -181,16 +181,18 @@ class PiecewisePolynomialParamValue:
     """The parameters expected by `piecewise_polynomial`.
 
     thresholds:
-        Thresholds defining the pieces / different segments on the real line.
+        Boundary points defining the pieces / different segments.
     intercepts:
-        Intercepts of the polynomial on each segment.
+        Intercepts of the polynomial on each segment (one per interval).
     rates:
-        Slope and higher-order coefficients of the polynomial on each segment.
+        Coefficients of the polynomial on each segment, shape
+        (n_intervals, n_coefficients). For piecewise_constant, this is
+        (n_intervals, 1) with all zeros.
     """
 
-    thresholds: Float[Array, " n_segments"]
-    intercepts: Float[Array, " n_segments"]
-    rates: Float[Array, " n_segments"]
+    thresholds: Float[Array, " n_thresholds"]
+    intercepts: Float[Array, " n_intervals"]
+    rates: Float[Array, "n_intervals n_coefficients"]
 
 
 def get_consecutive_int_lookup_table_param_value(