`chronax.data`

Data pipeline and scaling utilities for the Chronax Autoformer model.

`RobustScaler`

chronax.data.RobustScaler

Median + MAD scaler with a std-based fallback when MAD == 0.

Stateless: stats computes (shift, scale) so the same statistics can be reused to inverse-transform predictions.

`stats(self, x, axis=1) -> Tuple[jnp.ndarray, jnp.ndarray]`

Return (shift, scale) computed along axis with keepdims.

Parameter	Type	Default	Description
`x`	`jnp.ndarray`	-	-
`axis`	`int`	`1`	-

Returns: Tuple[jnp.ndarray, jnp.ndarray] (shift, scale)

`transform(self, x, shift, scale) -> jnp.ndarray`

Parameter	Type	Default	Description
`x`	`jnp.ndarray`	-	-
`shift`	`jnp.ndarray`	-	-
`scale`	`jnp.ndarray`	-	-

Returns: jnp.ndarray

`inverse(self, z, shift, scale) -> jnp.ndarray`

Parameter	Type	Default	Description
`z`	`jnp.ndarray`	-	-
`shift`	`jnp.ndarray`	-	-
`scale`	`jnp.ndarray`	-	-

Returns: jnp.ndarray

`build_windows(y, input_size, h) -> Tuple[np.ndarray, np.ndarray]`

NF-style rolling windows with right-padding (ConstantPad1d((0, h))).

Right-pads y with h zeros before unfolding, yielding len(y) - input_size windows of length input_size + h — including partial-horizon windows whose context reaches the end of the series. Returns (windows, mask) where mask is 1 on real points and 0 on the padded tail so the loss can drop padded horizon steps. The insample (first input_size) of every window is fully real.

Parameter	Type	Default	Description
`y`	`np.ndarray`	-	-
`input_size`	`int`	-	-
`h`	`int`	-	-

Returns: Tuple[np.ndarray, np.ndarray] (windows, mask) Raises: ValueError: When T < input_size + 1 (no window with a real outsample).

`split_train_val_windows(windows, masks, *, val_fraction=0.1) -> Tuple[Tuple[np.ndarray, np.ndarray], Tuple[np.ndarray, np.ndarray]]`

Chronological split: earliest windows train, latest validate.

When val_fraction <= 0 all windows train (empty validation set). Returns ((train_windows, train_masks), (val_windows, val_masks)).

Parameter	Type	Default	Description
`windows`	`np.ndarray`	-	-
`masks`	`np.ndarray`	-	-
`val_fraction`	`float`	`0.1`	-

Returns: Tuple[Tuple[np.ndarray, np.ndarray], Tuple[np.ndarray, np.ndarray]]

`pad_sequence(y, target_len, pad_value=0.0) -> Tuple[np.ndarray, np.ndarray]`

Left-pad / left-truncate a 1-D series to target_len.

Returns (padded, mask) where mask is 1 on real observations.

Parameter	Type	Default	Description
`y`	`np.ndarray`	-	-
`target_len`	`int`	-	-
`pad_value`	`float`	`0.0`	-

Returns: Tuple[np.ndarray, np.ndarray] (padded, mask)

`create_batch(y_series, input_size, h, sample_mask_list=None) -> Dict[str, Optional[jnp.ndarray]]`

Build a single JAX batch from a list of complete time series.

Each series is split into (history, horizon); history is padded/truncated to input_size (yielding available_mask) and horizon to h (yielding sample_mask).

Parameter	Type	Default	Description
`y_series`	`List[np.ndarray]`	-	-
`input_size`	`int`	-	-
`h`	`int`	-	-
`sample_mask_list`	`Optional[List[Optional[np.ndarray]]]`	`None`	-

Returns: Dict[str, Optional[jnp.ndarray]] Keys: * insample_y: [B, input_size, 1] * outsample_y: [B, h, 1] * available_mask: [B, input_size, 1] (1 = real, 0 = padded) * sample_mask: [B, h, 1] (1 = compute loss, 0 = ignore)

`batch_generator(y_series, input_size, h, batch_size, sample_mask_list=None, shuffle=True, seed=42) -> Iterator[Dict[str, Optional[jnp.ndarray]]]`

Iterate over y_series yielding fully-prepared JAX batches.

Parameter	Type	Default	Description
`y_series`	`List[np.ndarray]`	-	-
`input_size`	`int`	-	-
`h`	`int`	-	-
`batch_size`	`int`	-	-
`sample_mask_list`	`Optional[List[Optional[np.ndarray]]]`	`None`	-
`shuffle`	`bool`	`True`	-
`seed`	`int`	`42`	-

Returns: Iterator[Dict[str, Optional[jnp.ndarray]]]