Module futureexpert.associator

Contains the models with the configuration for the associator and the result format.

Functions

def export_associator_results_to_pandas(results: AssociatorResult) ‑> pandas.core.frame.DataFrame

Expand source code

def export_associator_results_to_pandas(results: AssociatorResult) -> pd.DataFrame:
    """Export associator results to a pandas DataFrame.

    Parameters
    ----------
    results: futureexpert.associator.AssociatorResult
        Associator results.

    Returns
    -------
    A pandas DataFrame with columns 'ts_name', 'group_id', 'trend_label', 'slope', 'num_obs'.
    return: pandas.core.frame.DataFrame

    """

    cluster_df = pd.DataFrame.from_records([x.model_dump() for x in results.clustering.group_assignments])
    trend_df = pd.DataFrame.from_records([x.model_dump() for x in results.trend.trend_labels])
    return cluster_df.merge(trend_df, how='outer', on='ts_name', validate='1:1')

Export associator results to a pandas DataFrame.

Parameters

results : AssociatorResult

Associator results.

Returns

A pandas DataFrame with columns 'ts_name', 'group_id', 'trend_label', 'slope', 'num_obs'.

return : pandas.core.frame.DataFrame

 

Classes

class AssociatorConfig (**data: Any)

Expand source code

class AssociatorConfig(BaseConfig):
    """Service configuration.

    Parameters
    ----------
    data_selection: futureexpert.associator.DataSelection
        Configuration on the selection of time series used for carrying out the service.
    trend_detection: futureexpert.associator.TrendDetectionConfiguration
        Configuration for trend detection.
    clustering: futureexpert.associator.ClusteringConfiguration
        Configuration for clustering.
    report_note: builtins.str
        User-defined string to be included in the report.
    db_name: typing.Optional[builtins.str]
        Only accessible for internal use. Name of the database to use for storing the results.
    """

    data_selection: DataSelection = Field(default_factory=DataSelection)
    trend_detection: TrendDetectionConfiguration = Field(default_factory=TrendDetectionConfiguration)
    clustering: ClusteringConfiguration = Field(default_factory=ClusteringConfiguration)
    report_note: str
    db_name: Optional[str] = None

Service configuration.

Parameters

data_selection : DataSelection

Configuration on the selection of time series used for carrying out the service.

trend_detection : TrendDetectionConfiguration

Configuration for trend detection.

clustering : ClusteringConfiguration

Configuration for clustering.

report_note : builtins.str

User-defined string to be included in the report.

db_name : typing.Optional[builtins.str]

Only accessible for internal use. Name of the database to use for storing the results.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var clustering : ClusteringConfiguration

var data_selection : DataSelection

var db_name : str | None

var model_config

var report_note : str

var trend_detection : TrendDetectionConfiguration

class AssociatorResult (**data: Any)

Expand source code

class AssociatorResult(BaseConfig):
    """Result of the associator service.

    Parameters
    ----------
    input: builtins.list[futureexpert.shared_models.TimeSeries]
        List of time series used as input.
    trend: futureexpert.associator.TrendResult
        Result of trend detection.
    clustering: futureexpert.associator.ClusteringResult
        Result of clustering.
    """
    input: list[TimeSeries]
    trend: TrendResult
    clustering: ClusteringResult

Result of the associator service.

Parameters

input : builtins.list[TimeSeries]

List of time series used as input.

trend : TrendResult

Result of trend detection.

clustering : ClusteringResult

Result of clustering.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var clustering : ClusteringResult

var input : list[TimeSeries]

var model_config

var trend : TrendResult

class ClusteringConfiguration (**data: Any)

Expand source code

class ClusteringConfiguration(BaseConfig):
    """Configuration for clustering.

    If start_time or end_time is not provided, then the missing(s) of the two will be
    determined automatically; the final four parameters govern this process.

    Parameters
    ----------
    create_clusters: builtins.bool
        If True, then the service will attempt clustering.
    n_clusters: builtins.int
        Number of clusters of complete and non-constant time series.
    start_time: typing.Optional[datetime.datetime]
        Observations from start_time (inclusive) onwards will be considered during clustering.
    end_time: typing.Optional[datetime.datetime]
        Observations up to end_time (inclusive) will be considered during clustering.
    """
    create_clusters: bool = True
    n_clusters: int = Field(default=None, gt=0)
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None

    @model_validator(mode='after')
    def validate_times(self) -> 'ClusteringConfiguration':
        if self.start_time is not None and self.end_time is not None and self.start_time > self.end_time:
            raise ValueError('End time precedes start time.')
        return self

Configuration for clustering.

If start_time or end_time is not provided, then the missing(s) of the two will be determined automatically; the final four parameters govern this process.

Parameters

create_clusters : builtins.bool

If True, then the service will attempt clustering.

n_clusters : builtins.int

Number of clusters of complete and non-constant time series.

start_time : typing.Optional[datetime.datetime]

Observations from start_time (inclusive) onwards will be considered during clustering.

end_time : typing.Optional[datetime.datetime]

Observations up to end_time (inclusive) will be considered during clustering.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var create_clusters : bool

var end_time : datetime.datetime | None

var model_config

var n_clusters : int

var start_time : datetime.datetime | None

Methods

def validate_times(self) ‑> ClusteringConfiguration

Expand source code

@model_validator(mode='after')
def validate_times(self) -> 'ClusteringConfiguration':
    if self.start_time is not None and self.end_time is not None and self.start_time > self.end_time:
        raise ValueError('End time precedes start time.')
    return self

class ClusteringResult (**data: Any)

Expand source code

class ClusteringResult(BaseConfig):
    """Result of clustering.

    Parameters
    ----------
    start_time: datetime.datetime
        Start of the time span used for clustering.
    end_time: datetime.datetime
        End of the time span used for clustering.
    groups: builtins.list[futureexpert.associator.Group]
        List of groups.
    group_assignments: builtins.list[futureexpert.associator.GroupAssignment]
        List of group assignments.
    """
    start_time: datetime
    end_time: datetime
    groups: list[Group]
    group_assignments: list[GroupAssignment]

Result of clustering.

Parameters

start_time : datetime.datetime

Start of the time span used for clustering.

end_time : datetime.datetime

End of the time span used for clustering.

groups : builtins.list[Group]

List of groups.

group_assignments : builtins.list[GroupAssignment]

List of group assignments.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var end_time : datetime.datetime

var group_assignments : list[GroupAssignment]

var groups : list[Group]

var model_config

var start_time : datetime.datetime

class DataSelection (**data: Any)

Expand source code

class DataSelection(BaseConfig):
    """Time series selection.

    Parameters
    ----------
    version: typing.Optional[builtins.str]
        Time series version to be used. If None, then the latest version is used.
    filter: builtins.dict[builtins.str, typing.Any]
        Filter to select a subset of time series based on their metadata.
    """

    version: Optional[str] = None
    filter: dict[str, Any] = Field(default_factory=dict)

Time series selection.

Parameters

version : typing.Optional[builtins.str]

Time series version to be used. If None, then the latest version is used.

filter : builtins.dict[builtins.str, typing.Any]

Filter to select a subset of time series based on their metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var filter : dict[str, typing.Any]

var model_config

var version : str | None

class Group (**data: Any)

Expand source code

class Group(BaseConfig):
    """A group definition to which time series can be assigned.

    Parameters
    ----------
    group_id: builtins.int
        ID of the group.
    group_name: builtins.str
        Name of the group.
    group_size: builtins.int
        Number of time series in the group.
    """
    group_id: int
    group_name: str
    group_size: int

A group definition to which time series can be assigned.

Parameters

group_id : builtins.int

ID of the group.

group_name : builtins.str

Name of the group.

group_size : builtins.int

Number of time series in the group.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var group_id : int

var group_name : str

var group_size : int

var model_config

class GroupAssignment (**data: Any)

Expand source code

class GroupAssignment(BaseConfig):
    """Assignment of a time series to a group.

    Parameters
    ----------
    ts_name: builtins.str
        Name of the time series.
    group_id: builtins.int
        ID of the group the time series is assigned to.
    """
    ts_name: str
    group_id: int

Assignment of a time series to a group.

Parameters

ts_name : builtins.str

Name of the time series.

group_id : builtins.int

ID of the group the time series is assigned to.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var group_id : int

var model_config

var ts_name : str

class TrendDetectionConfiguration (**data: Any)

Expand source code

class TrendDetectionConfiguration(BaseConfig):
    """Configuration for trend detection.

    Parameters
    ----------
    end_time: typing.Optional[datetime.datetime]
        End (inclusive) of the time span used for trend detection.
    max_number_of_obs: builtins.int
        Width of the time span used for trend detection; (leading and trailing) missing values
        are disregarded, that is, at most this number of observations are used for a given time series.
    number_of_nans_tolerated: builtins.int
        Leading and lagging missing values are dropped prior to running the trend detection; if this
        results in a loss of more than this number of observations lost, then the trend is considered
        undetermined.
    """
    end_time: Optional[datetime] = None
    max_number_of_obs: int = Field(default=6, gt=0)
    number_of_nans_tolerated: int = 2

Configuration for trend detection.

Parameters

end_time : typing.Optional[datetime.datetime]

End (inclusive) of the time span used for trend detection.

max_number_of_obs : builtins.int

Width of the time span used for trend detection; (leading and trailing) missing values are disregarded, that is, at most this number of observations are used for a given time series.

number_of_nans_tolerated : builtins.int

Leading and lagging missing values are dropped prior to running the trend detection; if this results in a loss of more than this number of observations lost, then the trend is considered undetermined.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var end_time : datetime.datetime | None

var max_number_of_obs : int

var model_config

var number_of_nans_tolerated : int

class TrendLabel (**data: Any)

Expand source code

class TrendLabel(BaseConfig):
    """Trend label for a time series.

    Parameters
    ----------
    ts_name: builtins.str
        Name of the time series.
    trend_label: builtins.str
        Trend label.
    slope: typing.Optional[builtins.float]
        Slope of the trend.
    num_obs: typing.Optional[builtins.int]
        Number of observations used for trend detection.
    """
    ts_name: str
    trend_label: str
    slope: Optional[float]
    num_obs: Optional[int]

Trend label for a time series.

Parameters

ts_name : builtins.str

Name of the time series.

trend_label : builtins.str

Trend label.

slope : typing.Optional[builtins.float]

Slope of the trend.

num_obs : typing.Optional[builtins.int]

Number of observations used for trend detection.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var model_config

var num_obs : int | None

var slope : float | None

var trend_label : str

var ts_name : str

class TrendResult (**data: Any)

Expand source code

class TrendResult(BaseConfig):
    """Result of trend detection.

    Parameters
    ----------
    start_time: datetime.datetime
        Start of the time span used for trend detection.
    end_time: datetime.datetime
        End of the time span used for trend detection.
    trend_labels: builtins.list[futureexpert.associator.TrendLabel]
        List of trend labels for the time series.
    """
    start_time: datetime
    end_time: datetime
    trend_labels: list[TrendLabel]

Result of trend detection.

Parameters

start_time : datetime.datetime

Start of the time span used for trend detection.

end_time : datetime.datetime

End of the time span used for trend detection.

trend_labels : builtins.list[TrendLabel]

List of trend labels for the time series.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• BaseConfig
• pydantic.main.BaseModel

Class variables

var end_time : datetime.datetime

var model_config

var start_time : datetime.datetime

var trend_labels : list[TrendLabel]