single_step

TemporalScope/src/temporalscope/target_shifters/single_step.py.

This module provides the SingleStepTargetShifter class for shifting target variables in time series data. It works in conjunction with the TimeFrame class to enable consistent target shifting operations across different DataFrame backends through Narwhals. Following the same backend-agnostic design as core_utils.py and temporal_data_loader.py, it ensures consistent behavior across all supported DataFrame types.

Engineering Design:

The SingleStepTargetShifter follows a clear separation between validation and transformation phases, designed to work seamlessly with both TimeFrame and raw DataFrame inputs.

Component	Description
`fit()`	Input validation phase that ensures: - Valid TimeFrame or supported DataFrame type - Target column is set or can be inferred - No Narwhals operations at this stage
`transform()`	Pure Narwhals transformation phase that: - Uses backend-agnostic operations only - Shifts target using Narwhals operations - Preserves TimeFrame metadata if present

Backend-Specific Patterns:

The following table outlines key patterns for working with different DataFrame backends through Narwhals operations:

Backend	Implementation Pattern
LazyFrame (Dask/Polars)	Represents lazy evaluation in Dask and Polars. Use `collect()` for scalar access, avoid direct indexing, and handle lazy evaluation through proper Narwhals operations.
PyArrow	Handles scalar operations differently. Use `nw.Int64` for numeric operations, handle comparisons through Narwhals, and convert types before arithmetic operations.
All Backends	Let `@nw.narwhalify` handle conversions between backends. Use pure Narwhals operations and avoid any backend-specific code to ensure consistent behavior across all supported types.

Examples:

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter
from temporalscope.core.temporal_data_loader import TimeFrame

# With TimeFrame
df = pd.DataFrame({"time": range(10), "target": range(10), "feature": range(10)})
tf = TimeFrame(df=df, time_col="time", target_col="target")
shifter = SingleStepTargetShifter(n_lags=1)
transformed_tf = shifter.fit_transform(tf)

# With DataFrame
df = pd.DataFrame({"target": range(10), "feature": range(10)})
shifter = SingleStepTargetShifter(target_col="target", n_lags=1)
transformed_df = shifter.fit_transform(df)

Notes

Uses a familiar fit/transform pattern for consistency, while implementing all operations through Narwhals' backend-agnostic API
Currently implements single-step prediction only. For multi-step sequence prediction, see the planned MultiStepTargetShifter in temporalscope.target_shifters.multi_step
When validating DataFrames, must get native format first since Narwhals wraps but does not implement actual DataFrame types

CLASS	DESCRIPTION
`SingleStepTargetShifter`	A target shifter for time series data using Narwhals operations.

SingleStepTargetShifter

SingleStepTargetShifter(
    target_col: Optional[str] = None,
    n_lags: int = 1,
    drop_target: bool = True,
    verbose: bool = False,
    mode: str = MODE_SINGLE_TARGET,
)

A target shifter for time series data using Narwhals operations.

This class provides target shifting functionality for single-step prediction tasks, working with both TimeFrame objects and raw DataFrames through Narwhals' backend-agnostic operations.

Engineering Design Assumptions:

Separation of Concerns:
fit: Validates inputs and sets parameters
transform: Pure Narwhals operations for shifting
No mixing of validation and transformation
Single-step Mode:
Each row represents one time step
Target variable is shifted by specified lag
Compatible with traditional ML frameworks
Supports scalar target prediction tasks
Backend Agnostic:
Validation in fit() before any operations
Pure Narwhals operations in transform()
Clean separation of concerns
Input Handling:
TimeFrame: Uses existing metadata
DataFrame: Validates in fit
numpy array: Converts in fit

ATTRIBUTE	DESCRIPTION
`target_col`	Column name to shift (optional, can be inferred from TimeFrame) TYPE: `(str, optional)`
`n_lags`	Number of steps to shift target, must be > 0 TYPE: `int`
`drop_target`	Whether to remove original target column TYPE: `bool`
`verbose`	Enable progress/debug logging TYPE: `bool`
`mode`	Operation mode, defaults to single-step TYPE: `str`

RAISES	DESCRIPTION
`ValueError`	If n_lags ≤ 0

Examples:

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter
from temporalscope.core.temporal_data_loader import TimeFrame

# Create TimeFrame
df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
tf = TimeFrame(df=df, time_col="time", target_col="target")

# Initialize and transform
shifter = SingleStepTargetShifter(n_lags=1)
transformed_tf = shifter.fit_transform(tf)

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter

# Create DataFrame
df = pd.DataFrame({"target": range(5), "feature": range(5)})

# Initialize and transform
shifter = SingleStepTargetShifter(target_col="target")
transformed_df = shifter.fit_transform(df)

Notes

Backend-Specific Patterns: - Use collect() for scalar access (LazyFrame) - Use nw.Int64 for scalar operations (PyArrow) - Let @nw.narwhalify handle conversions

METHOD	DESCRIPTION
`fit`	Validate inputs and prepare for transformation.
`fit_transform`	Fit the transformer and transform the input data.
`transform`	Transform input data using Narwhals operations.

ATTRIBUTE	DESCRIPTION
`drop_target`
`mode`
`n_lags`
`target_col`
`verbose`

Source code in src/temporalscope/target_shifters/single_step.py

def __init__(
    self,
    target_col: Optional[str] = None,
    n_lags: int = 1,
    drop_target: bool = True,
    verbose: bool = False,
    mode: str = MODE_SINGLE_TARGET,
):
    """Initialize the shifter with target column and lag settings."""
    if n_lags <= 0:
        raise ValueError("`n_lags` must be greater than 0")

    self.target_col = target_col
    self.n_lags = n_lags
    self.drop_target = drop_target
    self.verbose = verbose
    self.mode = mode

    if verbose:
        print(f"Initialized SingleStepTargetShifter with target_col={target_col}, n_lags={n_lags}")

drop_target

drop_target = drop_target

mode

mode = mode

n_lags

n_lags = n_lags

target_col

target_col = target_col

verbose

verbose = verbose

fit

fit(
    X: Union[TimeFrame, FrameT, ndarray], y=None
) -> SingleStepTargetShifter

Validate inputs and prepare for transformation.

This method handles all input validation before any Narwhals operations: - TimeFrame: Uses existing target_col - DataFrame: Validates using Narwhals operations - numpy array: Converts to DataFrame first

PARAMETER	DESCRIPTION
`X`	Input data to validate TYPE: `Union[TimeFrame, FrameT, ndarray]`
`y`	Ignored, exists for scikit-learn compatibility TYPE: `Any` DEFAULT: `None`

RETURNS	DESCRIPTION
`SingleStepTargetShifter`	self

RAISES	DESCRIPTION
`ValueError`	If target_col not set and cannot be inferred If target_col does not exist in DataFrame
`TypeError`	If input type is not supported

Examples:

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter
from temporalscope.core.temporal_data_loader import TimeFrame

# Create TimeFrame
df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
tf = TimeFrame(df=df, time_col="time", target_col="target")

# Initialize and fit
shifter = SingleStepTargetShifter(n_lags=1)
shifter.fit(tf)

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter

# Create DataFrame
df = pd.DataFrame({"target": range(5), "feature": range(5)})

# Initialize and fit
shifter = SingleStepTargetShifter(target_col="target")
shifter.fit(df)

Notes

Input Validation: - No Narwhals operations in fit() - Validates before any transformations - Handles all input types consistently

Source code in src/temporalscope/target_shifters/single_step.py

def fit(self, X: Union[TimeFrame, FrameT, np.ndarray], y=None) -> "SingleStepTargetShifter":
    """Validate inputs and prepare for transformation.

    This method handles all input validation before any Narwhals operations:
    - TimeFrame: Uses existing target_col
    - DataFrame: Validates using Narwhals operations
    - numpy array: Converts to DataFrame first

    Parameters
    ----------
    X : Union[TimeFrame, FrameT, np.ndarray]
        Input data to validate
    y : Any, optional
        Ignored, exists for scikit-learn compatibility

    Returns
    -------
    SingleStepTargetShifter
        self

    Raises
    ------
    ValueError
        If target_col not set and cannot be inferred
        If target_col does not exist in DataFrame
    TypeError
        If input type is not supported

    Examples
    --------
    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter
    from temporalscope.core.temporal_data_loader import TimeFrame

    # Create TimeFrame
    df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
    tf = TimeFrame(df=df, time_col="time", target_col="target")

    # Initialize and fit
    shifter = SingleStepTargetShifter(n_lags=1)
    shifter.fit(tf)
    ```

    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter

    # Create DataFrame
    df = pd.DataFrame({"target": range(5), "feature": range(5)})

    # Initialize and fit
    shifter = SingleStepTargetShifter(target_col="target")
    shifter.fit(df)
    ```

    Notes
    -----
    Input Validation:
    - No Narwhals operations in fit()
    - Validates before any transformations
    - Handles all input types consistently
    """
    if isinstance(X, TimeFrame):
        self.target_col = X._target_col
    elif isinstance(X, np.ndarray):
        # Convert numpy array to DataFrame first
        cols = [f"feature_{i}" for i in range(X.shape[1] - 1)]
        target_col = self.target_col if self.target_col is not None else f"feature_{X.shape[1] - 1}"
        cols.append(target_col)
        X = pd.DataFrame(X, columns=cols)
        if self.target_col is None:
            self.target_col = target_col
    else:
        # Convert to Narwhals DataFrame
        df = nw.from_native(X)
        # Validate target column exists
        if self.target_col not in df.columns:
            raise ValueError(f"Column '{self.target_col}' does not exist in DataFrame")

    if self.target_col is None:
        raise ValueError("target_col must be set before transform (call fit first)")

    return self

fit_transform

fit_transform(
    X: Union[TimeFrame, FrameT, ndarray], y=None
) -> Union[TimeFrame, FrameT, ndarray]

Fit the transformer and transform the input data.

This method combines input validation (fit) with Narwhals transformations (transform) in a single operation.

PARAMETER	DESCRIPTION
`X`	Input data to transform TYPE: `Union[TimeFrame, FrameT, ndarray]`
`y`	Ignored, exists for scikit-learn compatibility TYPE: `Any` DEFAULT: `None`

RETURNS	DESCRIPTION
`Union[TimeFrame, FrameT, ndarray]`	Transformed data with shifted target column

Examples:

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter
from temporalscope.core.temporal_data_loader import TimeFrame

# Create TimeFrame
df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
tf = TimeFrame(df=df, time_col="time", target_col="target")

# Initialize and transform
shifter = SingleStepTargetShifter(n_lags=1)
transformed_tf = shifter.fit_transform(tf)

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter

# Create DataFrame
df = pd.DataFrame({"target": range(5), "feature": range(5)})

# Initialize and transform
shifter = SingleStepTargetShifter(target_col="target", n_lags=1)
transformed_df = shifter.fit_transform(df)

Notes

Operation Flow: 1. fit(): Validates inputs 2. transform(): Pure Narwhals operations 3. Handles all backend types consistently

Source code in src/temporalscope/target_shifters/single_step.py

def fit_transform(self, X: Union[TimeFrame, FrameT, np.ndarray], y=None) -> Union[TimeFrame, FrameT, np.ndarray]:
    """Fit the transformer and transform the input data.

    This method combines input validation (fit) with Narwhals transformations
    (transform) in a single operation.

    Parameters
    ----------
    X : Union[TimeFrame, FrameT, np.ndarray]
        Input data to transform
    y : Any, optional
        Ignored, exists for scikit-learn compatibility

    Returns
    -------
    Union[TimeFrame, FrameT, np.ndarray]
        Transformed data with shifted target column

    Examples
    --------
    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter
    from temporalscope.core.temporal_data_loader import TimeFrame

    # Create TimeFrame
    df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
    tf = TimeFrame(df=df, time_col="time", target_col="target")

    # Initialize and transform
    shifter = SingleStepTargetShifter(n_lags=1)
    transformed_tf = shifter.fit_transform(tf)
    ```

    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter

    # Create DataFrame
    df = pd.DataFrame({"target": range(5), "feature": range(5)})

    # Initialize and transform
    shifter = SingleStepTargetShifter(target_col="target", n_lags=1)
    transformed_df = shifter.fit_transform(df)
    ```

    Notes
    -----
    Operation Flow:
    1. fit(): Validates inputs
    2. transform(): Pure Narwhals operations
    3. Handles all backend types consistently
    """
    return self.fit(X).transform(X)

transform

transform(
    X: Union[TimeFrame, FrameT, ndarray], y=None
) -> Union[TimeFrame, FrameT, ndarray]

Transform input data using Narwhals operations.

This method assumes inputs are already validated by fit() and uses pure Narwhals operations for all transformations.

PARAMETER	DESCRIPTION
`X`	Input data to transform TYPE: `Union[TimeFrame, FrameT, ndarray]`
`y`	Ignored, exists for scikit-learn compatibility TYPE: `Any` DEFAULT: `None`

RETURNS	DESCRIPTION
`Union[TimeFrame, FrameT, ndarray]`	Transformed data with shifted target column

Examples:

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter
from temporalscope.core.temporal_data_loader import TimeFrame

# Create TimeFrame
df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
tf = TimeFrame(df=df, time_col="time", target_col="target")

# Initialize and transform
shifter = SingleStepTargetShifter(n_lags=1)
shifter.fit(tf)
transformed_tf = shifter.transform(tf)

import pandas as pd
from temporalscope.target_shifters.single_step import SingleStepTargetShifter

# Create DataFrame
df = pd.DataFrame({"target": range(5), "feature": range(5)})

# Initialize and transform
shifter = SingleStepTargetShifter(target_col="target")
shifter.fit(df)
transformed_df = shifter.transform(df)

Notes

Pure Narwhals implementation: - _get_row_count() for counting - _shift_target() for shifting - Backend-agnostic operations - Handles LazyFrame and PyArrow scalars

Source code in src/temporalscope/target_shifters/single_step.py

@nw.narwhalify
def transform(self, X: Union[TimeFrame, FrameT, np.ndarray], y=None) -> Union[TimeFrame, FrameT, np.ndarray]:
    """Transform input data using Narwhals operations.

    This method assumes inputs are already validated by fit() and uses pure
    Narwhals operations for all transformations.

    Parameters
    ----------
    X : Union[TimeFrame, FrameT, np.ndarray]
        Input data to transform
    y : Any, optional
        Ignored, exists for scikit-learn compatibility

    Returns
    -------
    Union[TimeFrame, FrameT, np.ndarray]
        Transformed data with shifted target column

    Examples
    --------
    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter
    from temporalscope.core.temporal_data_loader import TimeFrame

    # Create TimeFrame
    df = pd.DataFrame({"time": range(5), "target": range(5), "feature": range(5)})
    tf = TimeFrame(df=df, time_col="time", target_col="target")

    # Initialize and transform
    shifter = SingleStepTargetShifter(n_lags=1)
    shifter.fit(tf)
    transformed_tf = shifter.transform(tf)
    ```

    ```python
    import pandas as pd
    from temporalscope.target_shifters.single_step import SingleStepTargetShifter

    # Create DataFrame
    df = pd.DataFrame({"target": range(5), "feature": range(5)})

    # Initialize and transform
    shifter = SingleStepTargetShifter(target_col="target")
    shifter.fit(df)
    transformed_df = shifter.transform(df)
    ```

    Notes
    -----
    Pure Narwhals implementation:
    - _get_row_count() for counting
    - _shift_target() for shifting
    - Backend-agnostic operations
    - Handles LazyFrame and PyArrow scalars
    """
    was_numpy = isinstance(X, np.ndarray)
    if was_numpy:
        # Type cast X to numpy array to satisfy mypy
        X_array: np.ndarray = X
        # Convert numpy array to DataFrame first
        cols = [f"feature_{i}" for i in range(X_array.shape[1] - 1)]
        cols.append(str(self.target_col))  # Ensure it's a string
        X = pd.DataFrame(X_array, columns=cols)

    # Get DataFrame to transform
    df = X.df if isinstance(X, TimeFrame) else X

    # Get row count before transformation
    rows_before = self._get_row_count(df)
    if rows_before == 0:
        raise ValueError("Cannot transform empty DataFrame")

    # Transform DataFrame
    transformed = self._shift_target(df)

    # Get row count after transformation
    rows_after = self._get_row_count(transformed, check_empty=False)
    if rows_after == 0:
        raise ValueError("All rows were dropped during transformation")

    if self.verbose:
        print(f"Rows before: {rows_before}; Rows after: {rows_after}; Dropped: {rows_before - rows_after}")

    # Handle TimeFrame output
    if isinstance(X, TimeFrame):
        return TimeFrame(
            transformed,
            time_col=X._time_col,
            target_col=f"{self.target_col}_shift_{self.n_lags}",
            mode=self.mode,
            ascending=X.ascending,
        )

    # Convert back to numpy array if input was numpy array
    if was_numpy:
        if hasattr(transformed, "collect"):  # For LazyFrame
            transformed = transformed.collect()
        return transformed.to_numpy()

    return transformed