transforms

transforms

Time-series transformations for FPL data.

add_rolling_features

add_rolling_features(
    df: DataFrame,
    columns: list[str],
    windows: list[int] = [3, 5, 10],
    agg_funcs: list[str] = ["mean", "std"],
    min_periods: int = 1,
) -> DataFrame

Add rolling window features to dataframe.

PARAMETER	DESCRIPTION
df	Input dataframe with time-series data TYPE: DataFrame
columns	Columns to compute rolling features for TYPE: list[str]
windows	Window sizes for rolling computation TYPE: list[int] DEFAULT: [3, 5, 10]
agg_funcs	Aggregation functions ('mean', 'std', 'min', 'max', 'sum') TYPE: list[str] DEFAULT: ['mean', 'std']
min_periods	Minimum observations in window TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
DataFrame	DataFrame with added rolling features

Source code in fplx/timeseries/transforms.py

def add_rolling_features(
    df: pd.DataFrame,
    columns: list[str],
    windows: list[int] = [3, 5, 10],
    agg_funcs: list[str] = ["mean", "std"],
    min_periods: int = 1,
) -> pd.DataFrame:
    """
    Add rolling window features to dataframe.

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe with time-series data
    columns : list[str]
        Columns to compute rolling features for
    windows : list[int]
        Window sizes for rolling computation
    agg_funcs : list[str]
        Aggregation functions ('mean', 'std', 'min', 'max', 'sum')
    min_periods : int
        Minimum observations in window

    Returns
    -------
    pd.DataFrame
        DataFrame with added rolling features
    """
    df = df.copy()

    for col in columns:
        if col not in df.columns:
            continue

        for window in windows:
            for func in agg_funcs:
                feature_name = f"{col}_rolling_{window}_{func}"
                df[feature_name] = (
                    df[col].rolling(window=window, min_periods=min_periods).agg(func)
                )

    return df

add_lag_features

add_lag_features(
    df: DataFrame,
    columns: list[str],
    lags: list[int] = [1, 2, 3, 7],
) -> DataFrame

Add lagged features to dataframe.

PARAMETER	DESCRIPTION
df	Input dataframe TYPE: DataFrame
columns	Columns to create lags for TYPE: list[str]
lags	Lag periods TYPE: list[int] DEFAULT: [1, 2, 3, 7]

RETURNS	DESCRIPTION
DataFrame	DataFrame with lagged features

Source code in fplx/timeseries/transforms.py

def add_lag_features(
    df: pd.DataFrame, columns: list[str], lags: list[int] = [1, 2, 3, 7]
) -> pd.DataFrame:
    """
    Add lagged features to dataframe.

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe
    columns : list[str]
        Columns to create lags for
    lags : list[int]
        Lag periods

    Returns
    -------
    pd.DataFrame
        DataFrame with lagged features
    """
    df = df.copy()

    for col in columns:
        if col not in df.columns:
            continue

        for lag in lags:
            feature_name = f"{col}_lag_{lag}"
            df[feature_name] = df[col].shift(lag)

    return df

add_ewma_features

add_ewma_features(
    df: DataFrame,
    columns: list[str],
    alphas: list[float] = [0.3, 0.5, 0.7],
) -> DataFrame

Add exponentially weighted moving average features.

PARAMETER	DESCRIPTION
df	Input dataframe TYPE: DataFrame
columns	Columns to compute EWMA for TYPE: list[str]
alphas	Smoothing factors (0 < alpha < 1) TYPE: list[float] DEFAULT: [0.3, 0.5, 0.7]

RETURNS	DESCRIPTION
DataFrame	DataFrame with EWMA features

Source code in fplx/timeseries/transforms.py

def add_ewma_features(
    df: pd.DataFrame, columns: list[str], alphas: list[float] = [0.3, 0.5, 0.7]
) -> pd.DataFrame:
    """
    Add exponentially weighted moving average features.

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe
    columns : list[str]
        Columns to compute EWMA for
    alphas : list[float]
        Smoothing factors (0 < alpha < 1)

    Returns
    -------
    pd.DataFrame
        DataFrame with EWMA features
    """
    df = df.copy()

    for col in columns:
        if col not in df.columns:
            continue

        for alpha in alphas:
            feature_name = f"{col}_ewma_{int(alpha * 100)}"
            df[feature_name] = df[col].ewm(alpha=alpha, adjust=False).mean()

    return df

add_trend_features

add_trend_features(
    df: DataFrame,
    columns: list[str],
    windows: list[int] = [5, 10],
) -> DataFrame

Add trend features (slope) using linear regression.

PARAMETER	DESCRIPTION
df	Input dataframe TYPE: DataFrame
columns	Columns to compute trends for TYPE: list[str]
windows	Window sizes for trend calculation TYPE: list[int] DEFAULT: [5, 10]

RETURNS	DESCRIPTION
DataFrame	DataFrame with trend features

Source code in fplx/timeseries/transforms.py

def add_trend_features(
    df: pd.DataFrame, columns: list[str], windows: list[int] = [5, 10]
) -> pd.DataFrame:
    """
    Add trend features (slope) using linear regression.

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe
    columns : list[str]
        Columns to compute trends for
    windows : list[int]
        Window sizes for trend calculation

    Returns
    -------
    pd.DataFrame
        DataFrame with trend features
    """
    df = df.copy()

    def calculate_slope(series):
        """Calculate slope of linear fit."""
        if len(series) < 2 or series.isna().all():
            return np.nan
        x = np.arange(len(series))
        y = series.values
        mask = ~np.isnan(y)
        if mask.sum() < 2:
            return np.nan
        slope = np.polyfit(x[mask], y[mask], 1)[0]
        return slope

    for col in columns:
        if col not in df.columns:
            continue

        for window in windows:
            feature_name = f"{col}_trend_{window}"
            df[feature_name] = (
                df[col]
                .rolling(window=window, min_periods=2)
                .apply(calculate_slope, raw=False)
            )

    return df

add_diff_features

add_diff_features(
    df: DataFrame,
    columns: list[str],
    periods: list[int] = [1, 2],
) -> DataFrame

Add difference features (current - previous).

PARAMETER	DESCRIPTION
df	Input dataframe TYPE: DataFrame
columns	Columns to compute differences for TYPE: list[str]
periods	Difference periods TYPE: list[int] DEFAULT: [1, 2]

RETURNS	DESCRIPTION
DataFrame	DataFrame with difference features

Source code in fplx/timeseries/transforms.py

def add_diff_features(
    df: pd.DataFrame, columns: list[str], periods: list[int] = [1, 2]
) -> pd.DataFrame:
    """
    Add difference features (current - previous).

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe
    columns : list[str]
        Columns to compute differences for
    periods : list[int]
        Difference periods

    Returns
    -------
    pd.DataFrame
        DataFrame with difference features
    """
    df = df.copy()

    for col in columns:
        if col not in df.columns:
            continue

        for period in periods:
            feature_name = f"{col}_diff_{period}"
            df[feature_name] = df[col].diff(periods=period)

    return df

add_consistency_features

add_consistency_features(
    df: DataFrame, columns: list[str], window: int = 5
) -> DataFrame

Add consistency measures (coefficient of variation).

PARAMETER	DESCRIPTION
df	Input dataframe TYPE: DataFrame
columns	Columns to measure consistency for TYPE: list[str]
window	Window size TYPE: int DEFAULT: 5

RETURNS	DESCRIPTION
DataFrame	DataFrame with consistency features

Source code in fplx/timeseries/transforms.py

def add_consistency_features(
    df: pd.DataFrame, columns: list[str], window: int = 5
) -> pd.DataFrame:
    """
    Add consistency measures (coefficient of variation).

    Parameters
    ----------
    df : pd.DataFrame
        Input dataframe
    columns : list[str]
        Columns to measure consistency for
    window : int
        Window size

    Returns
    -------
    pd.DataFrame
        DataFrame with consistency features
    """
    df = df.copy()

    for col in columns:
        if col not in df.columns:
            continue

        feature_name = f"{col}_consistency_{window}"
        rolling_mean = df[col].rolling(window=window, min_periods=1).mean()
        rolling_std = df[col].rolling(window=window, min_periods=1).std()

        # Coefficient of variation (lower = more consistent)
        df[feature_name] = rolling_std / (rolling_mean + 1e-6)

    return df