Module futureexpert.shared_models

Shared models used across multiple modules.

Classes

class BaseConfig (**data: Any)

Expand source code

class BaseConfig(BaseModel):
    """Base configuration that is used for most models."""
    model_config = ConfigDict(allow_inf_nan=False,
                              extra='forbid',
                              arbitrary_types_allowed=True)

Base configuration that is used for most models.

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

• pydantic.main.BaseModel

Subclasses

• AssociatorConfig
• AssociatorResult
• ClusteringConfiguration
• ClusteringResult
• DataSelection
• Group
• GroupAssignment
• TrendDetectionConfiguration
• TrendLabel
• TrendResult
• ForecastingConfig
• MethodSelectionConfig
• PreprocessingConfig
• ReportConfig
• WorkingDayAdaptionsConfig
• MakeForecastConsistentConfiguration
• MakeForecastConsistentDataSelection
• ReconciliationConfig
• MatcherConfig
• ResultScenario
• Scenario
• ScenarioValuesConfig
• ShaperConfig

Class variables

var model_config

class ChainedReportIdentifier (**data: Any)

Expand source code

class ChainedReportIdentifier(ReportIdentifier):
    """Extended report identifier with prerequisites.

    Parameters
    ----------
    report_id: builtins.int

    settings_id: typing.Optional[builtins.int]

    prerequisites: builtins.list[futureexpert.shared_models.ReportIdentifier]

    """
    prerequisites: list[ReportIdentifier]

    @classmethod
    def of(cls, final_report_identifier: ReportIdentifier, prerequisites: list[ReportIdentifier]) -> ChainedReportIdentifier:
        return cls(report_id=final_report_identifier.report_id,
                   settings_id=final_report_identifier.settings_id,
                   prerequisites=prerequisites)

Extended report identifier with prerequisites.

Parameters

report_id : builtins.int

 

settings_id : typing.Optional[builtins.int]

 

prerequisites : builtins.list[ReportIdentifier]

 

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

• ReportIdentifier
• pydantic.main.BaseModel

Class variables

var model_config

var prerequisites : list[ReportIdentifier]

Static methods

def of(final_report_identifier: ReportIdentifier,
prerequisites: list[ReportIdentifier]) ‑> ChainedReportIdentifier

class Covariate (**data: Any)

Expand source code

class Covariate(BaseModel):
    """Covariate.

    Parameters
    ----------
    ts: futureexpert.shared_models.TimeSeries
        Time series object of the covariate. Not lagged.
    lag: builtins.int
        Lag by which the covariate was used.
    """
    ts: TimeSeries
    lag: int

Covariate.

Parameters

ts : TimeSeries

Time series object of the covariate. Not lagged.

lag : builtins.int

Lag by which the covariate was used.

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

• pydantic.main.BaseModel

Class variables

var lag : int

var model_config

var ts : TimeSeries

class CovariateRef (**data: Any)

Expand source code

class CovariateRef(BaseModel):
    """Covariate reference.

    Parameters
    ----------
    name: builtins.str
        Name of the Covariate
    lag: builtins.int
        Lag by which the covariate was used.
    """
    name: str
    lag: int

Covariate reference.

Parameters

name : builtins.str

Name of the Covariate

lag : builtins.int

Lag by which the covariate was used.

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

• pydantic.main.BaseModel

Class variables

var lag : int

var model_config

var name : str

class ErrorReason (**data: Any)

Expand source code

class ErrorReason(BaseModel):
    """Details about a specific error in a report.

    Parameters
    ----------
    status: builtins.str
        The status of the run ('Error' or 'NoEvaluation').
    error_message: typing.Optional[builtins.str]
        The error message describing what went wrong.
    timeseries: builtins.list[builtins.str]
        List of time series names that encountered this error.
    """
    status: str
    error_message: Optional[str]
    timeseries: list[str]

    @staticmethod
    def parse_error_reasons(customer_specific: dict[str, Any]) -> list[ErrorReason]:
        """Creates error reasons from raw customer_specific object.

    Parameters
    ----------
    customer_specific: builtins.dict[builtins.str, typing.Any]

    return: builtins.list[futureexpert.shared_models.ErrorReason]

    """
        log_messages = customer_specific.get('log_messages', None)
        assert log_messages is not None, 'missing log_messages property in customer_specific'
        assert isinstance(log_messages, list), 'unexpected type of log_messages'
        return [ErrorReason.model_validate(msg) for msg in log_messages]

Details about a specific error in a report.

Parameters

status : builtins.str

The status of the run ('Error' or 'NoEvaluation').

error_message : typing.Optional[builtins.str]

The error message describing what went wrong.

timeseries : builtins.list[builtins.str]

List of time series names that encountered this error.

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

• pydantic.main.BaseModel

Class variables

var error_message : str | None

var model_config

var status : str

var timeseries : list[str]

Static methods

def parse_error_reasons(customer_specific: dict[str, Any]) ‑> list[ErrorReason]

Expand source code

@staticmethod
def parse_error_reasons(customer_specific: dict[str, Any]) -> list[ErrorReason]:
    """Creates error reasons from raw customer_specific object.

Parameters
----------
customer_specific: builtins.dict[builtins.str, typing.Any]

return: builtins.list[futureexpert.shared_models.ErrorReason]

"""
    log_messages = customer_specific.get('log_messages', None)
    assert log_messages is not None, 'missing log_messages property in customer_specific'
    assert isinstance(log_messages, list), 'unexpected type of log_messages'
    return [ErrorReason.model_validate(msg) for msg in log_messages]

Creates error reasons from raw customer_specific object.

Parameters

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

 

return : builtins.list[ErrorReason]

 

class PositiveInt (value: int)

Expand source code

class PositiveInt(int):
    def __new__(cls, value: int) -> PositiveInt:
        if value < 1:
            raise ValueError('The value must be a positive integer.')
        return super().__new__(cls, value)

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

>>> int('0b100', base=0)
4

Ancestors

• builtins.int

class PydanticModelList (*args, **kwargs)

Expand source code

class PydanticModelList(Generic[TBaseModel], list[TBaseModel]):
    def to_df(self) -> pd.DataFrame:
        """Converts the list into a Pandas Data Frame.

    Parameters
    ----------
    return: pandas.DataFrame

    """
        return pd.DataFrame([item.model_dump() for item in self])

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Ancestors

• typing.Generic
• builtins.list

Methods

def to_df(self) ‑> pandas.DataFrame

Expand source code

def to_df(self) -> pd.DataFrame:
    """Converts the list into a Pandas Data Frame.

Parameters
----------
return: pandas.DataFrame

"""
    return pd.DataFrame([item.model_dump() for item in self])

Converts the list into a Pandas Data Frame.

Parameters

return : pandas.DataFrame

 

class ReportIdentifier (**data: Any)

Expand source code

class ReportIdentifier(BaseModel):
    """Report ID and Settings ID of a report. Required to identify the report, e.g. when retrieving the results.

    Parameters
    ----------
    report_id: builtins.int

    settings_id: typing.Optional[builtins.int]

    """
    report_id: int
    settings_id: Optional[int]

Report ID and Settings ID of a report. Required to identify the report, e.g. when retrieving the results.

Parameters

report_id : builtins.int

 

settings_id : typing.Optional[builtins.int]

 

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

• pydantic.main.BaseModel

Subclasses

• ChainedReportIdentifier

Class variables

var model_config

var report_id : int

var settings_id : int | None

class ReportStatus (**data: Any)

Expand source code

class ReportStatus(BaseModel):
    """Status of a forecast or matcher report.

    Parameters
    ----------
    id: futureexpert.shared_models.ReportIdentifier
        The identifier of the report.
    description: builtins.str
        The description of the report.
    result_type: builtins.str
        The result type of the report.
    progress: futureexpert.shared_models.ReportStatusProgress
        Progress summary of the report.
    results: futureexpert.shared_models.ReportStatusResults
        Success/error summary of the report.
    error_reasons: typing.Optional[builtins.list[futureexpert.shared_models.ErrorReason]]
        Details about the errors of the report. Each error reason contains the status,
        error message, and list of affected time series.
    prerequisites: builtins.list[futureexpert.shared_models.ReportStatus]
        If the status was requested for a report that depends on other reports (ChainedReportIdentifier)
        all other report statuses are contained in the prerequisites in order to get an easy overview.
    """
    id: ReportIdentifier
    description: str
    result_type: str
    progress: ReportStatusProgress
    results: ReportStatusResults
    error_reasons: Optional[list[ErrorReason]] = None
    prerequisites: list[ReportStatus] = Field(default_factory=list)

    @property
    def is_finished(self) -> bool:
        """Indicates whether a forecasting report is finished."""
        return self.progress.pending == 0

    def print(self, print_prerequisites: bool = True, print_error_reasons: bool = True) -> None:
        """Prints a summary of the status.

        Parameters
        ----------
        print_prerequisites: builtins.bool
            Enable or disable printing of prerequisite reports.
        print_error_reasons: builtins.bool
            Enable or disable printing of error reasons.
        return: builtins.NoneType

    """
        title = f'Status of report "{self.description}" of type "{self.result_type}":'
        run_description = 'time series' if self.result_type in ['forecast', 'matcher'] else 'runs'
        if print_prerequisites:
            for prerequisite in self.prerequisites:
                prerequisite.print(print_error_reasons=print_error_reasons)

        if self.progress.requested == 0:
            print(f'{title}\n  No {run_description} created')
            return

        pct_txt = f'{round(self.progress.finished/self.progress.requested*100)} % are finished'
        overall = f'{self.progress.requested} {run_description} requested for calculation'
        finished_txt = f'{self.progress.finished} {run_description} finished'
        noeval_txt = f'{self.results.no_evaluation} {run_description} without evaluation'
        error_txt = f'{self.results.error} {run_description} ran into an error'
        print(f'{title}\n {pct_txt} \n {overall} \n {finished_txt} \n {noeval_txt} \n {error_txt}')

        if print_error_reasons and self.error_reasons is not None and len(self.error_reasons) > 0:
            print('\nError reasons:')
            for error_reason in self.error_reasons:
                ts_count = len(error_reason.timeseries)
                ts_names = ', '.join(error_reason.timeseries[:3])  # Show first 3 time series
                if ts_count > 3:
                    ts_names += f' ... and {ts_count - 3} more'
                print(f'  [{error_reason.status}] {error_reason.error_message if error_reason.error_message else ""}')
                print(f'    Affected time series ({ts_count}): {ts_names}')

Status of a forecast or matcher report.

Parameters

id : ReportIdentifier

The identifier of the report.

description : builtins.str

The description of the report.

result_type : builtins.str

The result type of the report.

progress : ReportStatusProgress

Progress summary of the report.

results : ReportStatusResults

Success/error summary of the report.

error_reasons : typing.Optional[builtins.list[ErrorReason]]

Details about the errors of the report. Each error reason contains the status, error message, and list of affected time series.

prerequisites : builtins.list[ReportStatus]

If the status was requested for a report that depends on other reports (ChainedReportIdentifier) all other report statuses are contained in the prerequisites in order to get an easy overview.

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

• pydantic.main.BaseModel

Class variables

var description : str

var error_reasons : list[ErrorReason] | None

var id : ReportIdentifier

var model_config

var prerequisites : list[ReportStatus]

var progress : ReportStatusProgress

var result_type : str

var results : ReportStatusResults

Instance variables

prop is_finished : bool

Expand source code

@property
def is_finished(self) -> bool:
    """Indicates whether a forecasting report is finished."""
    return self.progress.pending == 0

Indicates whether a forecasting report is finished.

Methods

def print(self, print_prerequisites: bool = True, print_error_reasons: bool = True) ‑> None

Expand source code

def print(self, print_prerequisites: bool = True, print_error_reasons: bool = True) -> None:
    """Prints a summary of the status.

    Parameters
    ----------
    print_prerequisites: builtins.bool
        Enable or disable printing of prerequisite reports.
    print_error_reasons: builtins.bool
        Enable or disable printing of error reasons.
    return: builtins.NoneType

"""
    title = f'Status of report "{self.description}" of type "{self.result_type}":'
    run_description = 'time series' if self.result_type in ['forecast', 'matcher'] else 'runs'
    if print_prerequisites:
        for prerequisite in self.prerequisites:
            prerequisite.print(print_error_reasons=print_error_reasons)

    if self.progress.requested == 0:
        print(f'{title}\n  No {run_description} created')
        return

    pct_txt = f'{round(self.progress.finished/self.progress.requested*100)} % are finished'
    overall = f'{self.progress.requested} {run_description} requested for calculation'
    finished_txt = f'{self.progress.finished} {run_description} finished'
    noeval_txt = f'{self.results.no_evaluation} {run_description} without evaluation'
    error_txt = f'{self.results.error} {run_description} ran into an error'
    print(f'{title}\n {pct_txt} \n {overall} \n {finished_txt} \n {noeval_txt} \n {error_txt}')

    if print_error_reasons and self.error_reasons is not None and len(self.error_reasons) > 0:
        print('\nError reasons:')
        for error_reason in self.error_reasons:
            ts_count = len(error_reason.timeseries)
            ts_names = ', '.join(error_reason.timeseries[:3])  # Show first 3 time series
            if ts_count > 3:
                ts_names += f' ... and {ts_count - 3} more'
            print(f'  [{error_reason.status}] {error_reason.error_message if error_reason.error_message else ""}')
            print(f'    Affected time series ({ts_count}): {ts_names}')

Prints a summary of the status.

Parameters

print_prerequisites : builtins.bool

Enable or disable printing of prerequisite reports.

print_error_reasons : builtins.bool

Enable or disable printing of error reasons.

return : builtins.NoneType

 

class ReportStatusProgress (**data: Any)

Expand source code

class ReportStatusProgress(BaseModel):
    """Progress of a forecasting report.

    Parameters
    ----------
    requested: builtins.int

    pending: builtins.int

    finished: builtins.int

    """
    requested: int
    pending: int
    finished: int

Progress of a forecasting report.

Parameters

requested : builtins.int

 

pending : builtins.int

 

finished : builtins.int

 

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

• pydantic.main.BaseModel

Class variables

var finished : int

var model_config

var pending : int

var requested : int

class ReportStatusResults (**data: Any)

Expand source code

class ReportStatusResults(BaseModel):
    """Result status of a forecasting report.

    This only includes runs that are already finished.

    Parameters
    ----------
    successful: builtins.int

    no_evaluation: builtins.int

    error: builtins.int

    """
    successful: int
    no_evaluation: int
    error: int

Result status of a forecasting report.

This only includes runs that are already finished.

Parameters

successful : builtins.int

 

no_evaluation : builtins.int

 

error : builtins.int

 

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

• pydantic.main.BaseModel

Class variables

var error : int

var model_config

var no_evaluation : int

var successful : int

class ReportSummary (**data: Any)

Expand source code

class ReportSummary(BaseModel):
    """Report ID and description of a report.

    Parameters
    ----------
    report_id: builtins.int

    description: builtins.str

    result_type: builtins.str

    """
    report_id: int
    description: str
    result_type: str

Report ID and description of a report.

Parameters

report_id : builtins.int

 

description : builtins.str

 

result_type : builtins.str

 

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

• pydantic.main.BaseModel

Class variables

var description : str

var model_config

var report_id : int

var result_type : str

class TimeSeries (**data: Any)

Expand source code

class TimeSeries(BaseModel):
    """Time series data.

    Parameters
    ----------
    name: builtins.str
        Name of the time series.
    group: builtins.str
        Group of the time series.
    granularity: builtins.str
        Granularity of the time series.
    values: typing.Sequence[futureexpert.shared_models.TimeSeriesValue]
        The actual values of the time series.
    unit: typing.Optional[builtins.str]
        The unit of the time .
    unit_factors: typing.Optional[builtins.dict[builtins.str, builtins.float]]
        Factors to convert the time series in another unit.
    grouping: typing.Optional[builtins.dict[builtins.str, typing.Union[builtins.str, builtins.int]]]
        Hierarchy levels ot the time series.
    """
    name: Annotated[str, Field(min_length=1)]
    group: str
    granularity: Annotated[str, Field(min_length=1)]
    values: Annotated[Sequence[TimeSeriesValue], Field(min_length=1)]
    unit: Optional[str] = None
    unit_factors: Optional[dict[str, float]] = None
    grouping: Optional[dict[str, Union[str, int]]] = None

Time series data.

Parameters

name : builtins.str

Name of the time series.

group : builtins.str

Group of the time series.

granularity : builtins.str

Granularity of the time series.

values : typing.Sequence[TimeSeriesValue]

The actual values of the time series.

unit : typing.Optional[builtins.str]

The unit of the time .

unit_factors : typing.Optional[builtins.dict[builtins.str, builtins.float]]

Factors to convert the time series in another unit.

grouping : typing.Optional[builtins.dict[builtins.str, typing.Union[builtins.str, builtins.int]]]

Hierarchy levels ot 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

• pydantic.main.BaseModel

Class variables

var granularity : str

var group : str

var grouping : dict[str, str | int] | None

var model_config

var name : str

var unit : str | None

var unit_factors : dict[str, float] | None

var values : Sequence[TimeSeriesValue]

class TimeSeriesValue (**data: Any)

Expand source code

class TimeSeriesValue(BaseModel):
    """Value of a time series.

    Parameters
    ----------
    time_stamp_utc: datetime.datetime
        The time stamp of the value.
    value: builtins.float
        The value.
    """
    time_stamp_utc: datetime
    value: Annotated[float, Field(allow_inf_nan=False)]

Value of a time series.

Parameters

time_stamp_utc : datetime.datetime

The time stamp of the value.

value : builtins.float

The value.

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

• pydantic.main.BaseModel

Class variables

var model_config

var time_stamp_utc : datetime.datetime

var value : float