from typing import Any

import numpy as np
import pandas as pd
from sientia_do.operations.df_preprocessor import create_features, limit_dataset, treat_nan
from sientia_do.timeseries.analyzer import TimeSeriesDiscontinuityAnalyzer
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import PolynomialFeatures, StandardScaler

DISCONTINUITY_TREATMENT = 'Discontinuity Treatment'
LAG_SELECTION = 'Lag Selection'
RANGE_SELECTION = 'Range Selection & Data Removal'
STATIC_WINDOW_REMOVAL = 'Static Window Removal'
DEFINE_VARIABLES_LIMITS = 'Define Variables Limits'
NORMALIZATION = 'Normalization'
FEATURE_CREATION = 'Feature Creation'
LAG_CREATION = 'Lag Creation'


def _frontend_date_format_to_strftime(fmt: str | None) -> str | None:
    """Convert front-end date format (e.g. dd/MM/yyyy HH:mm:ss) to Python strftime."""
    if not fmt:
        return None
    out = fmt.replace('yyyy', '%Y').replace('MM', '%m').replace('dd', '%d')
    out = out.replace('HH', '%H').replace('mm', '%M').replace('ss', '%S')
    return out


class LinearRegressionModel(BaseEstimator, TransformerMixin):
    """
    Linear Regression Model for Time Series Analysis.

    Supports both simple linear regression and polynomial regression.

    Thread-safety: This class is NOT thread-safe during fit() operations.
    Do not call fit() on the same instance from multiple threads simultaneously.
    After fitting, predict() is thread-safe for read-only operations.

    For multi-threaded environments:
    - Fit the model in a single thread
    - Share the fitted instance across threads for prediction only
    - Or create separate instances per thread
    """

    def __init__(
        self,
        target_variable: str = '',
        variable_columns: list[str] | None = None,
        model_params: dict[str, Any] | None = None,
        clipping: dict[str, float] | None = None,
        weights: dict[str, float] | None = None,
        degree: int = 1,
        interaction_only: bool = False,
    ):
        """
        Linear Regression Model for Time Series Analysis

        Args:
            target_variable (str): The target variable name
            variable_columns (list): The input columns names in a list
            model_params (dict): The parameters used for training the model \\
            clipping (dict): The lower and upper limits for the target variable to be clipped \\
                *Format: {'min': min_value, 'max': max_value}*
            weights (dict): The weights for the Linear Regression model \\
                *Format: {'variable_name': weight}*
            degree (int): The degree of the polynomial features (1 = linear, >1 = polynomial)
            interaction_only (bool): If True, only interaction features are produced

        Returns:
            LinearRegressionModel: The prediction model object
        """
        self.target_variable: str = target_variable
        self.variable_columns: list[str] | None = variable_columns
        self.model_params: dict[str, Any] | None = model_params
        self.clipping: dict[str, float] | None = clipping
        self.regr = LinearRegression()
        self.q1_target: float | None = None
        self.q3_target: float | None = None
        self.weights: dict[str, float] | None = weights
        self.degree: int = degree
        self.interaction_only: bool = interaction_only
        self.poly: PolynomialFeatures | None = None
        self.poly_feature_names: list[str] | None = None

    def create_poly_features(self, input_data: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
        """
        Create polynomial features from input data.

        Args:
            input_data (pd.DataFrame): Input data with feature columns
            fit (bool): If True, fit the PolynomialFeatures transformer

        Returns:
            pd.DataFrame: DataFrame with polynomial features
        """
        if self.degree <= 1:
            return input_data

        if fit:
            self.poly = PolynomialFeatures(
                degree=self.degree,
                interaction_only=self.interaction_only,
                include_bias=False,
            )
            poly_features = self.poly.fit_transform(input_data)
            self.poly_feature_names = list(self.poly.get_feature_names_out(input_data.columns))
        else:
            if self.poly is None:
                raise ValueError('PolynomialFeatures not fitted. Call fit() first.')
            poly_features = self.poly.transform(input_data)

        return pd.DataFrame(
            poly_features,
            columns=self.poly_feature_names,
            index=input_data.index,
        )

    def fit(self, input_data: pd.DataFrame) -> 'LinearRegressionModel':
        """
        Function to fit the model

        Args:
            input_data (pandas.DataFrame): The data used to fit the Linear Regression model

        Returns:
            LinearRegressionModel: The prediction model object
        """
        if not self.target_variable:
            raise ValueError('target_variable must be set before fitting')

        # Infer variable_columns if not provided
        if self.variable_columns is None:
            self.variable_columns = [
                col for col in input_data.columns if col != self.target_variable
            ]

        # Validate columns exist
        missing_cols = [col for col in self.variable_columns if col not in input_data.columns]
        if missing_cols:
            raise ValueError(f'Columns not found in input data: {missing_cols}')

        if self.target_variable not in input_data.columns:
            raise ValueError(f'Target variable {self.target_variable} not found in input data')

        X_train = input_data[self.variable_columns].copy()
        y_train = input_data[self.target_variable].copy()

        # Handle infinite values
        X_train = X_train.replace([np.inf, -np.inf], np.nan)
        y_train = y_train.replace([np.inf, -np.inf], np.nan)

        # Remove rows with NaN
        valid_mask = ~(X_train.isna().any(axis=1) | y_train.isna())
        X_train = X_train[valid_mask]
        y_train = y_train[valid_mask]

        # Remove columns with all NaN values
        cols_to_drop = X_train.columns[X_train.isna().all()].tolist()
        if cols_to_drop:  # pragma: no cover
            X_train = X_train.drop(columns=cols_to_drop)
            self.variable_columns = [c for c in self.variable_columns if c not in cols_to_drop]

        self.q1_target = y_train.quantile(0.25)
        self.q3_target = y_train.quantile(0.75)

        # Apply polynomial features if degree > 1
        if self.degree > 1:
            X_train = self.create_poly_features(X_train, fit=True)

        # Fit the model
        self.regr.fit(X_train, y_train)

        # Get the weights
        round_coef = np.round(self.regr.coef_, 3)
        round_intercept = np.round(self.regr.intercept_, 3)

        # Save the weights
        feature_names = self.poly_feature_names if self.degree > 1 else self.variable_columns
        assert feature_names is not None, 'feature_names should be set at this point'
        weights = dict(zip(feature_names, [float(c) for c in round_coef], strict=True))
        weights = dict(sorted(weights.items(), key=lambda item: abs(item[1]), reverse=True))
        weights = {'Bias': float(round_intercept), **weights}
        self.weights = weights

        return self

    def predict(self, input_data: pd.DataFrame) -> np.ndarray:
        """
        Function to predict the target variable.
        If clipping is True, the predictions are clipped based on the target variable quartiles.

        Args:
            input_data (pandas.DataFrame): The data used to predict the target variable

        Returns:
            numpy.ndarray: The predicted target variable
        """
        assert self.variable_columns is not None, 'variable_columns must be set before predict'
        X_test: pd.DataFrame = input_data[self.variable_columns].copy()

        # Handle infinite values
        X_test = X_test.replace([np.inf, -np.inf], np.nan)

        # Apply polynomial features if degree > 1
        if self.degree > 1:
            X_test = self.create_poly_features(X_test, fit=False)

        y_pred = self.regr.predict(X_test)

        if self.clipping:
            for i in range(len(y_pred)):
                if y_pred[i] > self.clipping['max']:
                    y_pred[i] = self.q3_target
                elif y_pred[i] < self.clipping['min']:
                    y_pred[i] = self.q1_target

        return y_pred

    def get_regressor(self) -> LinearRegression:
        """
        Get the underlying LinearRegression model.

        Returns:
            LinearRegression: The sklearn LinearRegression model
        """
        return self.regr


class DataPreprocessor(BaseEstimator, TransformerMixin):
    """
    Data Preprocessor for Time Series Analysis.

    Thread-safety: This class is NOT thread-safe during fit() operations.
    Do not call fit() on the same instance from multiple threads simultaneously.
    After fitting, transform() is thread-safe for read-only operations IF the
    input DataFrames are not shared between threads.

    For multi-threaded environments:
    - Fit the preprocessor in a single thread
    - Share the fitted instance across threads for transform() only
    - Ensure each thread passes its own DataFrame copy to transform()
    - Or create separate instances per thread
    """

    def __init__(
        self,
        date_column: str = '',
        target_variable: str = '',
        input_columns: list[str] | None = None,
        nan_treatment: str | None = None,
        lag_train: dict[str, int] | None = None,
        lag_transform: dict[str, int] | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
        date_format: str | None = None,
        removed_intervals: list[tuple[str, str]] | None = None,
        static_threshold: int | None = None,
        low_lim: dict[str, float] | None = None,
        upp_lim: dict[str, float] | None = None,
        window: int | None = None,
        scaler_name: str | None = None,
        scaler_params: dict[str, Any] | None = None,
        ar_var: str | None = None,
        self_operations: list[str] | None = None,
        cross_operations: list[str] | None = None,
        created_lags: dict[str, int] | None = None,
        steps_order: list[str] | None = None,
    ):
        """
        Data Preprocessor for Time Series Analysis

        Args:
            date_column (str): The column name of the date in the dataset
            target_variable (str): The target variable name
            input_columns (list): The input columns names in a list
            nan_treatment (str): The treatment for missing values \\
                *Options: 'drop', 'fill linear', 'linear interpolation'*
            lag_train (dict): The lags for each variable to be applyed during training \\
                *Format: {'variable_name': lag}*
            lag_transform (dict): The lags for each variable to be applyed during transformation \\
                *Format: {'variable_name': lag}*
            start_date (str): The start date for filtering data
            end_date (str): The end date for filtering data
            date_format (str | None): Frontend date format for start/end (e.g. dd/MM/yyyy HH:mm:ss or MM/dd/yyyy HH:mm:ss). When set, parsing matches the CSV date column.
            removed_intervals (list): List of tuples with intervals to remove from data \\
                *Format: [('start_date', 'end_date'), ...]*
            static_threshold (int): The number of repeated values to be considered as static
            low_lim (dict): The lower limits for each variable \\
                *Format: {'variable_name': limit}*
            upp_lim (dict): The upper limits for each variable \\
                *Format: {'variable_name': limit}*
            window (int): The window size for rolling window. **Not implemented yet**
            scaler_name (str): The scaler name. If no scaler is used, it is 'None' \\
                *Options: 'None', 'Standard Scaler'*
            scaler_params (dict): The parameters for the scaler object, if it is used \\
                *Format for Standard Scaler: {'variable_name': {'mean': mean, 'variance': variance}}*
            ar_var (str): The autoregressive variable name. If None, it is not created
            self_operations (list): The operations for feature creation using the same variable \\
                *Format: ['{variable_name}\\_{operation}\\_{scalar}']* \\
                *Operations: 'exp', 'pow', 'log', 'root'*
            cross_operations (list): The operations for feature creation using two variables \\
                *Format: ['{variable_name1}\\_{operation}\\_{variable_name2}']* \\
                *Operations: '\\*', '/'*
            created_lags (dict): Variables created by lagging existing ones \\
                *Format: {'original_variable_name': lag}*
            steps_order (list): The order of the steps to be executed in the pipeline \\
                *Options for list:  'Discontinuity Treatment',
                                    'Lag Selection',
                                    'Range Selection & Data Removal',
                                    'Static Window Removal',
                                    'Define Variables Limits',
                                    'Normalization',
                                    'Feature Creation',
                                    'Lag Creation'*

        Returns:
            DataPreprocessor: The data preprocessor object
        """
        self.date_column = date_column
        self.target_variable = target_variable
        self.input_columns = input_columns
        self.nan_treatment = nan_treatment
        self.lag_train = lag_train if lag_train else {}
        self.lag_transform = lag_transform if lag_transform else {}
        self.start_date = start_date
        self.end_date = end_date
        self.date_format = date_format
        self.removed_intervals = removed_intervals if removed_intervals else []
        self.ar_var = ar_var
        self.self_operations = self_operations
        self.cross_operations = cross_operations
        self.created_lags = created_lags
        self.static_threshold = static_threshold
        self.low_lim = low_lim
        self.upp_lim = upp_lim
        # self.window = window  # Not implemented yet
        self.scaler_name = scaler_name
        self.scaler_params = scaler_params
        self.feature_names_order: list[str] = []  # Initialize to avoid AttributeError
        self._fitted_feature_order: list[str] | None = None  # Track feature order after fit

        if self.scaler_name == 'Standard Scaler':
            self.scaler = StandardScaler()
        elif self.scaler_name == 'None':
            self.scaler = None
        else:
            self.scaler = None

        # Filter steps for preprocessor class
        possible_steps = [
            DISCONTINUITY_TREATMENT,
            LAG_SELECTION,
            RANGE_SELECTION,
            STATIC_WINDOW_REMOVAL,
            DEFINE_VARIABLES_LIMITS,
            NORMALIZATION,
            FEATURE_CREATION,
            LAG_CREATION,
        ]
        self.steps_order = steps_order or possible_steps
        for step in possible_steps:
            if step not in self.steps_order:
                self.steps_order.append(step)

    def get_required_columns(self, existing_columns: list) -> list:
        """
        Get the required columns to generate the input columns

        Args:
            existing_columns (list): The existing columns in the data

        Returns:
            list: The required columns
        """
        required_columns: list[str] = []

        # Columns for feature creation
        if self.self_operations is not None:
            for name in self.self_operations:
                var, operation, scalar = name.split('}_{')
                var = var.split('{')[1]
                operation = operation.split('}')[0]
                scalar = scalar.split('}')[0]
                required_columns.append(var)
        if self.cross_operations is not None:
            for name in self.cross_operations:
                var1, operation, var2 = name.split('}_{')
                var1 = var1.split('{')[1]
                operation = operation.split('}')[0]
                var2 = var2.split('}')[0]
                required_columns.append(var1)
                required_columns.append(var2)

        # Columns for lag creation
        if self.created_lags is not None:
            for var in self.created_lags.keys():
                required_columns.append(var)

        # Check if any column in required_columns is not in existing_columns
        required_columns = list(set(required_columns))
        _to_remove: list[str] = []
        for column in required_columns:
            # If column was already in self_operations list, remove it
            if (
                self.self_operations is not None
                and column not in existing_columns
                and column in self.self_operations
            ):
                _to_remove.append(column)
            # If column was already in cross_operations list, remove it
            if (
                self.cross_operations is not None
                and column not in existing_columns
                and column in self.cross_operations
            ):
                _to_remove.append(column)
            # If column was already in created_lags list, remove it
            if (
                self.created_lags is not None
                and column not in existing_columns
                and column in self.created_lags
            ):
                _to_remove.append(column)
        for column in set(_to_remove):
            required_columns.remove(column)

        return required_columns

    def get_scaler(self) -> Any:
        """
        Get the scaler object

        Returns:
            Scaler: The scaler object
        """
        return self.scaler

    def treat_discontinuities(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Treat the discontinuities in the data

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data
        """
        if self.nan_treatment:
            # Map 'linear interpolation' to 'fill linear' for compatibility
            treatment = self.nan_treatment
            if treatment == 'linear interpolation':
                treatment = 'fill linear'
            input_data = treat_nan(input_data, treatment)
        return input_data

    def _parse_datetime(self, date_str: str | None) -> pd.Timestamp | None:
        """Parse a date string to Timestamp using date_format when set.

        When date_format is set (e.g. dd/MM/yyyy HH:mm:ss or MM/dd/yyyy HH:mm:ss),
        parsing matches the CSV date column so start_date/end_date filter correctly.
        """
        if not date_str:
            return None
        try:
            python_fmt = _frontend_date_format_to_strftime(self.date_format) if self.date_format else None
            if python_fmt:
                return pd.to_datetime(date_str, format=python_fmt)
            return pd.to_datetime(date_str)
        except (ValueError, TypeError):
            return None

    def _filter_by_date_range(
        self, input_data: pd.DataFrame, start: pd.Timestamp | None, end: pd.Timestamp | None
    ) -> pd.DataFrame:
        """Filter DataFrame by start and end dates."""
        if start is not None:
            input_data = input_data[input_data.index >= start]
        if end is not None:
            input_data = input_data[input_data.index <= end]
        return input_data

    def _remove_interval(self, input_data: pd.DataFrame, interval: tuple | list) -> pd.DataFrame:
        """Remove a single interval from the DataFrame."""
        if len(interval) < 2:
            return input_data
        interval_start = self._parse_datetime(interval[0])
        interval_end = self._parse_datetime(interval[1])
        if interval_start is None or interval_end is None:
            return input_data
        mask = ~((input_data.index >= interval_start) & (input_data.index <= interval_end))
        return input_data[mask]

    def range_selection(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Filter data by date range and remove specified intervals.

        Args:
            input_data (pandas.DataFrame): The input data with datetime index

        Returns:
            pandas.DataFrame: The filtered data
        """
        start = self._parse_datetime(self.start_date)
        end = self._parse_datetime(self.end_date)
        input_data = self._filter_by_date_range(input_data, start, end)

        if self.removed_intervals:
            for interval in self.removed_intervals:
                input_data = self._remove_interval(input_data, interval)

        return input_data

    def lag_selection(self, input_data: pd.DataFrame, lag_dict: dict) -> pd.DataFrame:
        """
        Select the lags for the variables

        Args:
            input_data (pandas.DataFrame): The input data
            lag_dict (dict): The lags for each variable \\
                *Format: {'variable_name': lag}*

        Returns:
            pandas.DataFrame: The treated data

        Note:
            This method modifies input_data in-place. Ensure the caller passes
            a copy if the original DataFrame needs to be preserved.
        """
        if lag_dict:
            for var, lag in lag_dict.items():
                if lag > 0:
                    input_data[var] = input_data[var].shift(lag)
            # WARNING: Modifies DataFrame in-place
            input_data.dropna(inplace=True)
        return input_data

    def treat_static_windows(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Treat the static windows in the data

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data
        """
        if self.static_threshold:
            ts_analyzer = TimeSeriesDiscontinuityAnalyzer(input_data)
            ts_analyzer.infer_frequency()
            for col in input_data.columns:
                ts_analyzer.identify_static_windows(column=col, threshold=self.static_threshold)
                ts_analyzer.treat_static_windows(
                    column=col, remove_window=True, threshold=self.static_threshold
                )
                ts_analyzer.update_total_discontinuities(col)
                input_data = ts_analyzer.get_treated_data()
        return input_data

    def adjust_limits(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Adjust the limits for the variables

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data
        """
        input_data, self.low_lim, self.upp_lim = limit_dataset(
            input_data, self.low_lim, self.upp_lim
        )
        return input_data

    def create_features(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Create features in the data

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data
        """
        input_data = create_features(input_data, self.self_operations, self.cross_operations)
        return input_data

    def create_ar(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Create the autoregressive variable in the data

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data

        Note:
            This method modifies input_data in-place.
        """
        if self.ar_var:
            input_data[self.ar_var] = input_data[self.target_variable].shift(1)
            # WARNING: Modifies DataFrame in-place
            input_data.dropna(inplace=True)
        return input_data

    def create_lags(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Create additional lags in the data

        Args:
            input_data (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data

        Note:
            This method modifies input_data in-place.
        """
        if self.created_lags:
            for var, lag in self.created_lags.items():
                if lag > 0 and var in input_data.columns:
                    new_col = f'{var}_lag{lag}'
                    input_data[new_col] = input_data[var].shift(lag)
            # WARNING: Modifies DataFrame in-place
            input_data.dropna(inplace=True)
        return input_data

    def fit(self, x: pd.DataFrame, y: None | pd.Series = None) -> 'DataPreprocessor':
        """
        Function to preprocess the data and split it into training and testing sets

        Args:
            x (pandas.DataFrame): The input data
            y (pandas.Series): The target variable

        Returns:
            DataPreprocessor: The data preprocessor object
        """
        if x is not None and y is not None:
            data_treat = pd.concat([x.copy(), y.copy()], axis=1)
        elif x is not None:
            data_treat = x.copy()
        else:
            raise ValueError('No data was provided')
        assert self.input_columns is not None, 'input_columns must be set'
        existing_columns = [col for col in data_treat.columns if col in self.input_columns]
        data_treat = data_treat[existing_columns + [self.target_variable]]

        for step in self.steps_order:
            # Discontinuity Treatment
            if step == DISCONTINUITY_TREATMENT:
                data_treat = self.treat_discontinuities(data_treat)

            # Lag for Model Training
            if step == LAG_SELECTION:
                data_treat = self.lag_selection(data_treat, self.lag_train)

            # Range Selection & Data Removal
            if step == RANGE_SELECTION:
                data_treat = self.range_selection(data_treat)

            # Static Window Treatment
            if step == STATIC_WINDOW_REMOVAL:
                data_treat = self.treat_static_windows(data_treat)

            # Adjust limits
            if step == DEFINE_VARIABLES_LIMITS:
                data_treat = self.adjust_limits(data_treat)

            # Normalization
            if step == NORMALIZATION and self.scaler:
                self.scaler = self.scaler.fit(data_treat[existing_columns])
                self.feature_names_order = list(data_treat[existing_columns].columns)
                data_treat[existing_columns] = self.scaler.transform(data_treat[existing_columns])

                # Save scaler parameters
                assert self.scaler_params is not None, 'scaler_params must be initialized'
                for index, column in enumerate(list(existing_columns)):
                    mean = self.scaler.mean_[index]
                    variance = self.scaler.var_[index]
                    self.scaler_params[column] = {
                        'mean': round(mean, 3),
                        'variance': round(variance, 3),
                    }

        # Store fitted feature order for predict method
        self._fitted_feature_order = list(existing_columns)

        return self

    def transform(self, x: pd.DataFrame) -> pd.DataFrame:
        """
        Function to preprocess the data

        Args:
            x (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The treated data
        """
        if 'timestamp' in x.columns:
            data_treat = x.drop(columns='timestamp')
        else:
            data_treat = x.copy()
        assert self.input_columns is not None, 'input_columns must be set'
        existing_columns = [col for col in data_treat.columns if col in self.input_columns]
        required_columns = self.get_required_columns(existing_columns)
        all_cols = required_columns + existing_columns + [self.target_variable]
        all_cols = list(set(all_cols))
        data_treat = data_treat[all_cols]

        for step in self.steps_order:
            # Discontinuity Treatment
            if step == DISCONTINUITY_TREATMENT:
                data_treat = self.treat_discontinuities(data_treat)

            # Lag for Model Training
            if step == LAG_SELECTION:
                data_treat = self.lag_selection(data_treat, self.lag_transform)

            # Range Selection & Data Removal (typically skipped in transform)
            if step == RANGE_SELECTION:
                data_treat = self.range_selection(data_treat)

            # Static Window Treatment
            if step == STATIC_WINDOW_REMOVAL:
                data_treat = self.treat_static_windows(data_treat)

            # Adjust limits
            if step == DEFINE_VARIABLES_LIMITS:
                data_treat = self.adjust_limits(data_treat)

            # Normalization
            if step == NORMALIZATION and self.scaler:
                # Only transform feature columns, preserve target and any other required columns
                feature_cols = self.feature_names_order
                data_treat[feature_cols] = self.scaler.transform(data_treat[feature_cols])

            # Feature Creation
            if step == FEATURE_CREATION:
                data_treat = self.create_features(data_treat)

            # Lag Creation
            if step == LAG_CREATION:
                # Autoregressive Variable
                if self.input_columns is not None and self.ar_var in self.input_columns:
                    data_treat = self.create_ar(data_treat)

                # Additonal Lags
                data_treat = self.create_lags(data_treat)

        return data_treat

    def predict(self, x: pd.DataFrame) -> pd.DataFrame:
        """
        Transform data for prediction (removes target variable).

        This method is a wrapper around transform() that:
        1. Transforms the input data
        2. Removes the target variable column
        3. Ensures features are in the same order as during fit

        Args:
            x (pandas.DataFrame): The input data

        Returns:
            pandas.DataFrame: The transformed data without target variable,
                            with features in the same order as during fit
        """
        data_treat = self.transform(x)

        # Remove target variable if present
        if self.target_variable in data_treat.columns:  # pragma: no branch
            data_treat = data_treat.drop(columns=self.target_variable)

        # Ensure features are in the same order as during fit
        if self._fitted_feature_order is not None:
            # Filter to only include columns that exist in both
            available_cols = [c for c in self._fitted_feature_order if c in data_treat.columns]
            data_treat = data_treat[available_cols]

        return data_treat