timeseries

timeseries

Time-series feature engineering and transformations.

FeatureEngineer

FeatureEngineer(config: Optional[dict] = None)

Feature engineering pipeline for player time-series data.

PARAMETER	DESCRIPTION
config	Feature configuration dictionary TYPE: Optional[Dict] DEFAULT: None

Source code in fplx/timeseries/features.py

def __init__(self, config: Optional[dict] = None):
    self.config = {**self.DEFAULT_CONFIG, **(config or {})}

fit_transform

fit_transform(df: DataFrame) -> DataFrame

Apply all feature engineering transformations.

PARAMETER	DESCRIPTION
df	Input player timeseries data TYPE: DataFrame

RETURNS	DESCRIPTION
DataFrame	Transformed data with engineered features

Source code in fplx/timeseries/features.py

def fit_transform(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Apply all feature engineering transformations.

    Parameters
    ----------
    df : pd.DataFrame
        Input player timeseries data

    Returns
    -------
    pd.DataFrame
        Transformed data with engineered features
    """
    df = df.copy()

    # Identify available columns and ensure they are numeric
    key_cols = [c for c in self.config["key_columns"] if c in df.columns]

    for col in key_cols:
        df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0)

    if not key_cols:
        logger.warning("No key columns found for feature engineering")
        return df

    # Apply transformations
    logger.info("Adding rolling features...")
    df = add_rolling_features(
        df,
        columns=key_cols,
        windows=self.config["rolling_windows"],
        agg_funcs=["mean", "std"],
    )

    logger.info("Adding lag features...")
    df = add_lag_features(df, columns=key_cols, lags=self.config["lag_periods"])

    logger.info("Adding EWMA features...")
    df = add_ewma_features(df, columns=key_cols, alphas=self.config["ewma_alphas"])

    logger.info("Adding trend features...")
    df = add_trend_features(
        df, columns=key_cols, windows=self.config["trend_windows"]
    )

    logger.info("Adding difference features...")
    df = add_diff_features(df, columns=key_cols, periods=[1])

    logger.info("Adding consistency features...")
    df = add_consistency_features(df, columns=["minutes", "points"], window=5)

    return df

get_feature_names

get_feature_names(base_columns: list[str]) -> list[str]

Get list of all generated feature names.

PARAMETER	DESCRIPTION
base_columns	Base column names TYPE: list[str]

RETURNS	DESCRIPTION
list[str]	Generated feature names

Source code in fplx/timeseries/features.py

def get_feature_names(self, base_columns: list[str]) -> list[str]:
    """
    Get list of all generated feature names.

    Parameters
    ----------
    base_columns : list[str]
        Base column names

    Returns
    -------
    list[str]
        Generated feature names
    """
    features = []

    for col in base_columns:
        # Rolling features
        for window in self.config["rolling_windows"]:
            features.extend([
                f"{col}_rolling_{window}_mean",
                f"{col}_rolling_{window}_std",
            ])

        # Lag features
        for lag in self.config["lag_periods"]:
            features.append(f"{col}_lag_{lag}")

        # EWMA features
        for alpha in self.config["ewma_alphas"]:
            features.append(f"{col}_ewma_{int(alpha * 100)}")

        # Trend features
        for window in self.config["trend_windows"]:
            features.append(f"{col}_trend_{window}")

        # Diff features
        features.append(f"{col}_diff_1")

    # Consistency features
    features.extend([
        "minutes_consistency_5",
        "points_consistency_5",
    ])

    return features

create_future_features

create_future_features(
    df: DataFrame, horizon: int
) -> DataFrame

Create features for future predictions.

This method extends the historical data by horizon periods, applies the full feature engineering pipeline, and returns the newly created future feature set.

PARAMETER	DESCRIPTION
df	Historical data TYPE: DataFrame
horizon	Number of future gameweeks to predict TYPE: int

RETURNS	DESCRIPTION
DataFrame	DataFrame with features for future gameweeks

Source code in fplx/timeseries/features.py

def create_future_features(self, df: pd.DataFrame, horizon: int) -> pd.DataFrame:
    """
    Create features for future predictions.

    This method extends the historical data by `horizon` periods,
    applies the full feature engineering pipeline, and returns
    the newly created future feature set.

    Parameters
    ----------
    df : pd.DataFrame
        Historical data
    horizon : int
        Number of future gameweeks to predict

    Returns
    -------
    pd.DataFrame
        DataFrame with features for future gameweeks
    """
    if df.empty:
        return pd.DataFrame()

    # Create future placeholders by repeating the last known data point
    last_row = df.iloc[-1:].copy()

    # Avoid duplicating index if it's a timestamp or gameweek
    is_numeric_index = pd.api.types.is_numeric_dtype(df.index)
    if isinstance(df.index, pd.DatetimeIndex) or is_numeric_index:
        last_index = df.index[-1]
        future_index = pd.RangeIndex(
            start=last_index + 1, stop=last_index + 1 + horizon
        )
        last_row.index = [future_index[0]]  # Temporarily align for concat
    else:
        future_index = pd.RangeIndex(start=len(df), stop=len(df) + horizon)

    future_rows = pd.concat([last_row] * horizon, ignore_index=True)
    if isinstance(df.index, pd.DatetimeIndex) or is_numeric_index:
        future_rows.index = future_index

    # Combine historical and future data
    combined_df = pd.concat([df, future_rows])

    # Run the full feature engineering pipeline on the combined data
    # This ensures that rolling/lag features are calculated correctly
    # based on the historical context.
    engineered_df = self.fit_transform(combined_df)

    # Return only the future part
    return engineered_df.tail(horizon)

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_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_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_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