feat: add build_period axis to investment once-effect matrices

src/fluxopt/elements.py

@@ -73,8 +73,10 @@ class Investment:
         mandatory: If True, must build exactly once; if False, may build at most once.
         lifetime: Periods active after build; None = forever.
         prior_size: Pre-existing capacity available from period 0.
-        effects_per_size: One-time per-MW costs charged in the build period.
-        effects_fixed: One-time fixed costs charged in the build period.
+        effects_per_size: One-time per-MW costs. Scalar or 1D ``(build_period,)``
+            → diagonal (cost lands in the build period). 2D ``(period, build_period)``
+            → as-is (e.g. installment plans, learning curves).
+        effects_fixed: One-time fixed costs. Same expansion rules as effects_per_size.
         effects_per_size_periodic: Recurring per-MW costs charged every active period.
         effects_fixed_periodic: Recurring fixed costs charged every active period.
     """

src/fluxopt/model.py

@@ -877,17 +877,20 @@ def _create_effects(self) -> None:
 
         once_direct: Any = 0
 
-        # Investment: one-time per-size costs (charged in build period)
+        # Investment: one-time per-size costs — (flow, effect, period, build_period)
+        # Multiply by invest_size_at_build renamed to build_period, sum over both.
         if self.invest_size_at_build is not None and d.flows.invest_effects_per_size is not None:
             eps_once = d.flows.invest_effects_per_size.rename({'invest_flow': 'flow'})
             if (eps_once != 0).any():
-                once_direct = once_direct + (eps_once * self.invest_size_at_build).sum('flow')
+                sab_bp = self.invest_size_at_build.rename({'period': 'build_period'})
+                once_direct = once_direct + (eps_once * sab_bp).sum(['flow', 'build_period'])
 
-        # Investment: one-time fixed costs (charged in build period)
+        # Investment: one-time fixed costs — (flow, effect, period, build_period)
         if self.invest_build is not None and d.flows.invest_effects_fixed is not None:
             ef_once = d.flows.invest_effects_fixed.rename({'invest_flow': 'flow'})
             if (ef_once != 0).any():
-                once_direct = once_direct + (ef_once * self.invest_build).sum('flow')
+                bld_bp = self.invest_build.rename({'period': 'build_period'})
+                once_direct = once_direct + (ef_once * bld_bp).sum(['flow', 'build_period'])
 
         self.m.add_constraints(self.effect_once == once_direct, name='effect_once_eq')
 

src/fluxopt/model_data.py

@@ -13,7 +13,7 @@
 if TYPE_CHECKING:
     from fluxopt.components import Converter, Port
     from fluxopt.elements import Carrier, Effect, Flow, Investment, Sizing, Status, Storage
-    from fluxopt.types import TimeIndex, Timesteps
+    from fluxopt.types import TimeIndex, TimeSeries, Timesteps
 
 
 @dataclass(frozen=True)
@@ -59,6 +59,48 @@ def _effect_template(
     )
 
 
+def _expand_once_effect(value: TimeSeries, period: pd.Index) -> xr.DataArray:
+    """Expand an investment once-effect value to 2D (period, build_period).
+
+    Construction rule:
+        - Scalar → diagonal filled with that constant
+        - 1D ``(build_period,)`` or ``(period,)`` → diagonal
+        - 2D ``(period, build_period)`` → as-is
+
+    Args:
+        value: Scalar, 1D build_period-indexed, or 2D effect value.
+        period: Period index (shared by both axes).
+    """
+    n = len(period)
+    coords: dict[str, Any] = {'period': period, 'build_period': period}
+    dims = ['period', 'build_period']
+
+    if isinstance(value, (int, float)):
+        return xr.DataArray(np.eye(n) * float(value), dims=dims, coords=coords)
+
+    if isinstance(value, xr.DataArray):
+        vdims = {str(d) for d in value.dims}
+        if vdims == {'period', 'build_period'}:
+            return value
+        if vdims <= {'period', 'build_period'} and len(vdims) == 1:
+            vals = value.values
+            if len(vals) != n:
+                dim_name = next(iter(vdims))
+                raise ValueError(
+                    f'Once-effect DataArray with dim {dim_name!r} has length {len(vals)}, '
+                    f'expected {n} (number of periods)'
+                )
+            return xr.DataArray(np.diag(vals), dims=dims, coords=coords)
+        foreign = [str(d) for d in value.dims if d not in ('period', 'build_period')]
+        if foreign:
+            raise ValueError(
+                f'Once-effect DataArray has unexpected dims {foreign}. Expected subset of (period, build_period).'
+            )
+
+    da = as_dataarray(value, {'build_period': period})
+    return xr.DataArray(np.diag(da.values), dims=dims, coords=coords)
+
+
 _NC_GROUPS = {
     'flows': 'model/flows',
     'carriers': 'model/carriers',
@@ -175,8 +217,8 @@ class _InvestmentArrays:
     mandatory: xr.DataArray | None = None  # (invest_dim,)
     lifetime: xr.DataArray | None = None  # (invest_dim,) — NaN = forever
     prior_size: xr.DataArray | None = None  # (invest_dim,)
-    effects_per_size: xr.DataArray | None = None  # (invest_dim, effect, period?) — once
-    effects_fixed: xr.DataArray | None = None  # (invest_dim, effect, period?) — once
+    effects_per_size: xr.DataArray | None = None  # (invest_dim, effect, period, build_period) — once
+    effects_fixed: xr.DataArray | None = None  # (invest_dim, effect, period, build_period) — once
     effects_per_size_periodic: xr.DataArray | None = None  # (invest_dim, effect, period?)
     effects_fixed_periodic: xr.DataArray | None = None  # (invest_dim, effect, period?)
 
@@ -200,7 +242,25 @@ def build(
             return cls()
 
         effect_set = set(effect_ids)
-        tmpl = _effect_template({'effect': effect_ids}, period)
+        periodic_tmpl = _effect_template({'effect': effect_ids}, period)
+
+        # Once-effect template: (effect, period, build_period) when multi-period
+        once_coords: dict[str, Any]
+        once_shape: tuple[int, ...]
+        once_dims: tuple[str, ...]
+        if period is not None:
+            n_p = len(period)
+            once_coords = {
+                'effect': effect_ids,
+                'period': period,
+                'build_period': period,
+            }
+            once_shape = (len(effect_ids), n_p, n_p)
+            once_dims = ('effect', 'period', 'build_period')
+        else:
+            once_coords = {'effect': effect_ids}
+            once_shape = (len(effect_ids),)
+            once_dims = ('effect',)
 
         ids: list[str] = []
         mins: list[float] = []
@@ -230,17 +290,31 @@ def build(
             lifetimes.append(float(inv.lifetime) if inv.lifetime is not None else np.nan)
             prior_sizes.append(inv.prior_size)
 
+            # Once-effects: expand to (effect, period, build_period) via diagonal rule
             for label, src_dict, dest_key in [
                 ('Investment.effects_per_size', inv.effects_per_size, 'eps'),
                 ('Investment.effects_fixed', inv.effects_fixed, 'ef'),
+            ]:
+                arr = xr.DataArray(np.zeros(once_shape), dims=list(once_dims), coords=once_coords)
+                for ek, ev in src_dict.items():
+                    if ek not in effect_set:
+                        raise ValueError(f'Unknown effect {ek!r} in {label} on {item_id!r}')
+                    if period is not None:
+                        arr.loc[ek] = _expand_once_effect(ev, period)
+                    else:
+                        arr.loc[ek] = as_dataarray(ev, {})
+                all_slices[dest_key].append(arr)
+
+            # Periodic effects: (effect, period?) — no build_period axis
+            for label, src_dict, dest_key in [
                 ('Investment.effects_per_size_periodic', inv.effects_per_size_periodic, 'eps_p'),
                 ('Investment.effects_fixed_periodic', inv.effects_fixed_periodic, 'ef_p'),
             ]:
-                arr = tmpl.zeros()
+                arr = periodic_tmpl.zeros()
                 for ek, ev in src_dict.items():
                     if ek not in effect_set:
                         raise ValueError(f'Unknown effect {ek!r} in {label} on {item_id!r}')
-                    arr.loc[ek] = as_dataarray(ev, tmpl.as_da_coords)
+                    arr.loc[ek] = as_dataarray(ev, periodic_tmpl.as_da_coords)
                 all_slices[dest_key].append(arr)
 
         coords = {dim: ids}
@@ -420,8 +494,8 @@ class FlowsData:
     invest_mandatory: xr.DataArray | None = None  # (invest_flow,)
     invest_lifetime: xr.DataArray | None = None  # (invest_flow,) — NaN = forever
     invest_prior_size: xr.DataArray | None = None  # (invest_flow,)
-    invest_effects_per_size: xr.DataArray | None = None  # (invest_flow, effect, period?) — once
-    invest_effects_fixed: xr.DataArray | None = None  # (invest_flow, effect, period?) — once
+    invest_effects_per_size: xr.DataArray | None = None  # (invest_flow, effect, period, build_period) — once
+    invest_effects_fixed: xr.DataArray | None = None  # (invest_flow, effect, period, build_period) — once
     invest_effects_per_size_periodic: xr.DataArray | None = None  # (invest_flow, effect, period?)
     invest_effects_fixed_periodic: xr.DataArray | None = None  # (invest_flow, effect, period?)
 
@@ -1030,8 +1104,8 @@ class StoragesData:
     invest_mandatory: xr.DataArray | None = None  # (invest_storage,)
     invest_lifetime: xr.DataArray | None = None  # (invest_storage,) — NaN = forever
     invest_prior_size: xr.DataArray | None = None  # (invest_storage,)
-    invest_effects_per_size: xr.DataArray | None = None  # (invest_storage, effect, period?) — once
-    invest_effects_fixed: xr.DataArray | None = None  # (invest_storage, effect, period?) — once
+    invest_effects_per_size: xr.DataArray | None = None  # (invest_storage, effect, period, build_period) — once
+    invest_effects_fixed: xr.DataArray | None = None  # (invest_storage, effect, period, build_period) — once
     invest_effects_per_size_periodic: xr.DataArray | None = None  # (invest_storage, effect, period?)
     invest_effects_fixed_periodic: xr.DataArray | None = None  # (invest_storage, effect, period?)
 

tests/math_port/test_multi_period.py

@@ -778,3 +778,156 @@ def test_storage_sizing_effects_per_size_vary_by_period(self, optimize):
             period_weights=[1, 1],
         )
         assert_allclose(result.objective, 40.0, rtol=1e-4)
+
+
+class TestBuildPeriodEffects:
+    def test_capex_per_size_build_period_dim(self, optimize):
+        """Proves: 1D (build_period,) input produces diagonal — cost charged when built.
+
+        CAPEX = 10 if built in 2020, 20 if built in 2025. Mandatory, size=10.
+        Optimizer builds in cheapest period (2020). Once cost = 10*10 = 100.
+        """
+        _xfail_if_validate(optimize)
+        periods = [2020, 2025]
+        capex = xr.DataArray([10.0, 20.0], dims=['build_period'], coords={'build_period': periods})
+        result = optimize(
+            timesteps=ts(3),
+            carriers=[Carrier('Heat')],
+            effects=[Effect('cost', is_objective=True)],
+            ports=[
+                Port(
+                    'Demand',
+                    exports=[
+                        Flow('Heat', size=1, fixed_relative_profile=np.array([10, 10, 10])),
+                    ],
+                ),
+                Port(
+                    'Grid',
+                    imports=[
+                        Flow(
+                            'Heat',
+                            size=Investment(10, 10, effects_per_size={'cost': capex}),
+                        ),
+                    ],
+                ),
+            ],
+            periods=periods,
+            period_weights=[1, 1],
+        )
+        assert_allclose(result.objective, 100.0, rtol=1e-4)
+
+    def test_capex_fixed_build_period_dim(self, optimize):
+        """Proves: 1D (build_period,) fixed cost — charged when built.
+
+        Fixed CAPEX = 50 if built in 2020, 100 if built in 2025. Mandatory.
+        Builds in 2020. Once cost = 50.
+        """
+        _xfail_if_validate(optimize)
+        periods = [2020, 2025]
+        capex = xr.DataArray([50.0, 100.0], dims=['build_period'], coords={'build_period': periods})
+        result = optimize(
+            timesteps=ts(3),
+            carriers=[Carrier('Heat')],
+            effects=[Effect('cost', is_objective=True)],
+            ports=[
+                Port(
+                    'Demand',
+                    exports=[
+                        Flow('Heat', size=1, fixed_relative_profile=np.array([10, 10, 10])),
+                    ],
+                ),
+                Port(
+                    'Grid',
+                    imports=[
+                        Flow(
+                            'Heat',
+                            size=Investment(10, 10, effects_fixed={'cost': capex}),
+                        ),
+                    ],
+                ),
+            ],
+            periods=periods,
+            period_weights=[1, 1],
+        )
+        assert_allclose(result.objective, 50.0, rtol=1e-4)
+
+    def test_capex_2d_spread_across_periods(self, optimize):
+        """Proves: 2D (period, build_period) matrix spreads costs across periods.
+
+        Building in 2020 costs 5/MW in 2020 + 5/MW in 2025 (installment plan).
+        Building in 2025 costs 15/MW in 2025 only.
+        Size=10, mandatory. Optimizer picks 2020: total = 10*(5+5) = 100.
+        vs 2025: total = 10*15 = 150. Objective = 100.
+        """
+        _xfail_if_validate(optimize)
+        periods = [2020, 2025]
+        capex_2d = xr.DataArray(
+            [[5.0, 0.0], [5.0, 15.0]],
+            dims=['period', 'build_period'],
+            coords={'period': periods, 'build_period': periods},
+        )
+        result = optimize(
+            timesteps=ts(3),
+            carriers=[Carrier('Heat')],
+            effects=[Effect('cost', is_objective=True)],
+            ports=[
+                Port(
+                    'Demand',
+                    exports=[
+                        Flow('Heat', size=1, fixed_relative_profile=np.array([10, 10, 10])),
+                    ],
+                ),
+                Port(
+                    'Grid',
+                    imports=[
+                        Flow(
+                            'Heat',
+                            size=Investment(10, 10, effects_per_size={'cost': capex_2d}),
+                        ),
+                    ],
+                ),
+            ],
+            periods=periods,
+            period_weights=[1, 1],
+        )
+        assert_allclose(result.objective, 100.0, rtol=1e-4)
+
+    def test_fixed_2d_installments(self, optimize):
+        """Proves: 2D fixed costs can spread across periods.
+
+        Building in 2020: fixed cost 30 in 2020 + 30 in 2025.
+        Building in 2025: fixed cost 80 in 2025 only.
+        Mandatory. Build in 2020: total = 30+30 = 60. Objective = 60.
+        """
+        _xfail_if_validate(optimize)
+        periods = [2020, 2025]
+        fixed_2d = xr.DataArray(
+            [[30.0, 0.0], [30.0, 80.0]],
+            dims=['period', 'build_period'],
+            coords={'period': periods, 'build_period': periods},
+        )
+        result = optimize(
+            timesteps=ts(3),
+            carriers=[Carrier('Heat')],
+            effects=[Effect('cost', is_objective=True)],
+            ports=[
+                Port(
+                    'Demand',
+                    exports=[
+                        Flow('Heat', size=1, fixed_relative_profile=np.array([10, 10, 10])),
+                    ],
+                ),
+                Port(
+                    'Grid',
+                    imports=[
+                        Flow(
+                            'Heat',
+                            size=Investment(10, 10, effects_fixed={'cost': fixed_2d}),
+                        ),
+                    ],
+                ),
+            ],
+            periods=periods,
+            period_weights=[1, 1],
+        )
+        assert_allclose(result.objective, 60.0, rtol=1e-4)