Theoretical Power¶

Overview¶

The FeatureCalcPowerTheoretical class is a subclass of FeatureCalculator that calculates the theoretical power of a wind turbine using the internal methodology. This is the most accurate way to calculate the theoretical power but may vary in comparison to simpler methods such as contractual ones.

This class uses a pre-trained neural network model that represents the wind turbine power curve to calculate the theoretical power. The model is implemented in Tensorflow and trained using multiple input features such as wind speed, turbulence, ambient temperature, etc. The model is trained per wind turbine and the script used to do this training process can be found in the performance server at manual_routines\postgres_fit_wtg_power_curves.

Calculation Logic¶

The calculation logic is described in the constructor of the class, shown below in the Class Definition section.

Database Requirements¶

Feature attribute server_calc_type must be set to theoretical_active_power.
Calculation model with name fitted_power_curve must be present in the database and must be associated with the wind turbine object. This can be trained using the script manual_routines\postgres_fit_wtg_power_curves.
The following object attributes for the object that is being calculated:
- Required:
  - neighbor_wind_turbines: List of wind turbines that are considered neighbors to the wind turbine object, in order of proximity.
  - reference_met_masts: List of met masts that are considered reference met masts to the wind turbine object, in order of proximity.
  - spe_name: Name of the SPE of the wind turbine object.
  - nominal_power: Nominal power of the wind turbine in kW. This is used to clip the results avoiding unrealistic power values.
- Optional:
  - neighbor_active_power_regressions: Dict that has the slope and offset to be considered when using active power data from neighbor wind turbines. It is in the following format:
```
{
    "neighbor1": {"slope": 1, "offset": 0},
    "neighbor2": {"slope": 1, "offset": 0},
    ...
}
```
    If this is not present a default value of 1 for slope and 0 for offset will be used.
  - neighbor_wind_speed_regressions: Dict that has the slope and offset to be considered when using wind speed data from neighbor wind turbines. It is in the following format:
```
{
    "neighbor1": {"slope": 1, "offset": 0},
    "neighbor2": {"slope": 1, "offset": 0},
    ...
}
```
    If this is not present a default value of 1 for slope and 0 for offset will be used.
  - met_mast_wind_speed_regressions: Dict that has the slope and offset to be considered when using wind speed data from reference met masts. It is in the following format:
```
{
    "met_mast1": {"slope": 1, "offset": 0},
    "met_mast2": {"slope": 1, "offset": 0},
    ...
}
```
    If this is not present a default value of 1 for slope and 0 for offset will be used.
The following features for the object that is being calculated:
- wind_speed: Wind speed in m/s.
- wind_speed_estimated: Estimated wind speed in m/s.
- active_power: Active power in kW.
- curtailment_state: Curtailment state of the wind turbine.
- iec_operation_state: IEC operation state of the wind turbine.
If additional features were used in the training of the model, they will also be required. The needed features for the model will be dynamically loaded from the trained object in the database.
In case neighbor wind turbines are used, the following features for the neighbor wind turbines:
- active_power: Active power in kW.
- wind_speed: Wind speed in m/s.
- curtailment_state: Curtailment state of the wind turbine.
- iec_operation_state: IEC operation state of the wind turbine.
If additional features were used in the training of the model, they will also be required for the neighbor wind turbines. These are the same features as the ones required for the object that is being calculated.
In case reference met masts are used, the features required for the power curve model will also be required for the reference met masts. Usually these are:
- wind_speed: Wind speed in m/s that is retrieved from wind_speed_1_avg.
- wind_speed_std: Wind speed standard deviation in m/s that is retrieved from wind_speed_1_std.
- ambiente_temperature: Ambient temperature in degrees Celsius that is retrieved from temperature_1_avg.

Class Definition¶

`FeatureCalcPowerTheoretical(object_name, feature)` ¶

Class used to calculate theoretical active power for a wind turbine.

It uses Echoenergia's internal calculation methodology for better accuracy.

This will follow the following steps trying to fill all timestamps. If one step is not enough to fill all timestamps, the next step will be used until there are no more timestamps with NaN values.

Step 1: Use the NN power curve model of this turbine with data from this turbine. The NN model is stored in the database with the name fitted_power_curve.
Step 2: Use active power from the neighbor turbines corrected by neighbor_active_power_regressions if available. Neighbor data will only be used if it meets IEC-OperationState_10min.REP and CurtailmentState_10min.REP requirements.
Step 3: Use the NN power curve model of this turbine with data from neighbor turbines. Neighbor wind speed will be corrected by neighbor_wind_speed_regressions object attribute if available.
Step 4: Use the NN power curve model of this turbine with data from reference met masts. Wind speed will be corrected by met_mast_wind_speed_regressions object attribute if available.
Step 5: Consider the average active power of all turbines of this turbine wind farm, as long as it meets IEC-OperationState_10min.REP and CurtailmentState_10min.REP requirements.

Parameters:

object_name ¶
(str) –

Name of the object for which the feature is calculated. It must exist in performance_db.
feature ¶
(str) –

Feature of the object that is calculated. It must exist in performance_db.

Source code in echo_energycalc/feature_calc_power_theoretical.py

def __init__(
    self,
    object_name: str,
    feature: str,
) -> None:
    """Class used to calculate theoretical active power for a wind turbine.

    It uses Echoenergia's internal calculation methodology for better accuracy.

    This will follow the following steps trying to fill all timestamps. If one step is not enough to fill all timestamps, the next step will be used until there are no more timestamps with NaN values.

    - **Step 1**: Use the NN power curve model of this turbine with data from this turbine. The NN model is stored in the database with the name `fitted_power_curve`.
    - **Step 2**: Use active power from the neighbor turbines corrected by `neighbor_active_power_regressions` if available. Neighbor data will only be used if it meets `IEC-OperationState_10min.REP` and `CurtailmentState_10min.REP` requirements.
    - **Step 3**: Use the NN power curve model of this turbine with data from neighbor turbines. Neighbor wind speed will be corrected by `neighbor_wind_speed_regressions` object attribute if available.
    - **Step 4**: Use the NN power curve model of this turbine with data from reference met masts. Wind speed will be corrected by `met_mast_wind_speed_regressions` object attribute if available.
    - **Step 5**: Consider the average active power of all turbines of this turbine wind farm, as long as it meets `IEC-OperationState_10min.REP` and `CurtailmentState_10min.REP` requirements.

    Parameters
    ----------
    object_name : str
        Name of the object for which the feature is calculated. It must exist in performance_db.
    feature : str
        Feature of the object that is calculated. It must exist in performance_db.
    """
    # initialize parent class
    super().__init__(object_name, feature)

    # base requirements for the feature calculator
    self._add_requirement(RequiredCalcModels(calc_models={self.object: [{"model_name": "^fitted_power_curve$", "model_type": None}]}))
    self._add_requirement(
        RequiredObjectAttributes(
            {
                self.object: [
                    "neighbor_wind_turbines",
                    "nominal_power",
                    "reference_met_masts",
                    "spe_name",
                ],
            },
        ),
    )
    self._add_requirement(
        RequiredObjectAttributes(
            {
                self.object: [
                    "neighbor_active_power_regressions",
                    "neighbor_wind_speed_regressions",
                    "met_mast_wind_speed_regressions",
                ],
            },
            optional=True,
        ),
    )
    self._get_required_data()

    # loading calculation model from file
    try:
        self._pc_model: PowerCurveNNPredictiveModel = self._get_requirement_data("RequiredCalcModels")[self.object][
            "fitted_power_curve"
        ]["model"]
        self._pc_model._deserialize_model()  # noqa: SLF001
    except Exception as e:
        raise RuntimeError(f"'{self.object}' failed to load power curve model.") from e

    # defining required features
    needed_features = [
        "WindSpeed_10min.AVG",
        "WindSpeedEstimated_10min.AVG",
        "ActivePower_10min.AVG",
        "CurtailmentState_10min.REP",
        "IEC-OperationState_10min.REP",
    ]
    needed_features = list(set(needed_features).union(set(self._pc_model.model_arguments.reference_features)))
    self._add_requirement(RequiredFeatures({self.object: needed_features}))

    # amount of timestamps that is acceptable to have NaN values to avoid long calculations trying to fill all NaNs
    self._max_nan = 5

`feature` `property` ¶

Feature that is calculated. This will be defined in the constructor and cannot be changed.

Returns:

str –

Name of the feature that is calculated.

`name` `property` ¶

Name of the feature calculator. Is defined in child classes of FeatureCalculator.

This must be equal to the "server_calc_type" attribute of the feature in performance_db.

Returns:

str –

Name of the feature calculator.

`object` `property` ¶

Object for which the feature is calculated. This will be defined in the constructor and cannot be changed.

Returns:

str –

Object name for which the feature is calculated.

`requirements` `property` ¶

List of requirements of the feature calculator. Is defined in child classes of FeatureCalculator.

Returns:

dict[str, list[CalculationRequirement]] –

Dict of requirements.

The keys are the names of the classes of the requirements and the values are lists of requirements of that class.

For example: {"RequiredFeatures": [RequiredFeatures(...), RequiredFeatures(...)], "RequiredObjects": [RequiredObjects(...)]}

`result` `property` ¶

Result of the calculation. This is None until the method "calculate" is called.

Returns:

Series | DataFrame | None: –

Result of the calculation if the method "calculate" was called. None otherwise.

`calculate(period, save_into=None, cached_data=None, **kwargs)` ¶

Method that will calculate the feature.

Parameters:

period ¶
(DateTimeRange) –

Period for which the feature will be calculated.
save_into ¶
(Literal['all', 'performance_db'] | None, default: None ) –

Argument that will be passed to the method "save". The options are: - "all": The feature will be saved in performance_db and bazefield. - "performance_db": the feature will be saved only in performance_db. - None: The feature will not be saved.

By default None.
cached_data ¶
(DataFrame | None, default: None ) –

DataFrame with features already queried/calculated. This is useful to avoid needing to query all the data again from performance_db, making chained calculations a lot more efficient. By default None
**kwargs ¶
(dict, default: {} ) –

Additional arguments that will be passed to the "save" method.

Returns:

Series –

Pandas Series with the calculated feature.

Source code in echo_energycalc/feature_calc_power_theoretical.py

def calculate(
    self,
    period: DateTimeRange,
    save_into: Literal["all", "performance_db"] | None = None,
    cached_data: DataFrame | None = None,
    **kwargs,
) -> Series:  # sourcery skip: extract-method
    """Method that will calculate the feature.

    Parameters
    ----------
    period : DateTimeRange
        Period for which the feature will be calculated.
    save_into : Literal["all", "performance_db"] | None, optional
        Argument that will be passed to the method "save". The options are:
        - "all": The feature will be saved in performance_db and bazefield.
        - "performance_db": the feature will be saved only in performance_db.
        - None: The feature will not be saved.

        By default None.
    cached_data : DataFrame | None, optional
        DataFrame with features already queried/calculated. This is useful to avoid needing to query all the data again from performance_db, making chained calculations a lot more efficient.
        By default None
    **kwargs : dict, optional
        Additional arguments that will be passed to the "save" method.

    Returns
    -------
    Series
        Pandas Series with the calculated feature.

    """
    t0 = perf_counter()

    # creating series for the result
    result = self._create_empty_result(period=period, result_type="Series")

    # * Step 1: using power curve model with this turbine wind speed

    # getting feature values
    self._get_required_data(period=period, reindex="10min", cached_data=cached_data)

    try:
        # adjusting features for this turbine
        df = self._adjust_features(self.object)

        # getting indexes where all reference data for this power curve model is present (no NaNs)
        idx = df[~df[self._pc_model.model_arguments.reference_features].isna().any(axis=1)].index

        if len(idx) > 0:
            # adjusting wind direction relative to be between -30 and 30
            # this is done because when the turbine is stopped it might to be pointing to the wind and the wind direction relative might be very high
            # this might be an issue for the power curve
            # TODO we need to think of a better way to handle this, currently we are just setting to 0 for values higher than 30 degrees
            temp_df = df.loc[idx, self._pc_model.model_arguments.reference_features].copy()
            if relative_cols := [col for col in temp_df.columns if "relative" in col.lower()]:
                relative_col = relative_cols[0]
                temp_df[relative_col] = temp_df[relative_col].apply(lambda x: 0 if abs(x) > 30 else x)

            # adjusting format for predictive model
            temp_df.index = MultiIndex.from_product([[self.object], temp_df.index], names=["object", "timestamp"])
            temp_df.index = temp_df.index.set_levels(temp_df.index.levels[0].astype("string"), level=0)
            temp_df = temp_df.astype("float32")

            # predicting active power theoretical using the power curve model
            result.loc[idx] = self._pc_model.predict(temp_df)[self._pc_model.model_arguments.target_features[0]].values

    except Exception:
        logger.exception(
            f"'{self.object}' - Error on step 1: calculating active power theoretical using neural network power curve model on this turbine wind speed",
        )

    t1 = perf_counter()

    # * Step 2: using active power from neighbor turbines

    # skipping if everything in result is filled already (no NaNs)
    if result.isna().sum() > self._max_nan:
        # getting neighbor turbines
        neighbor_turbines = self._get_requirement_data("RequiredObjectAttributes")[self.object]["neighbor_wind_turbines"]

        # iterating over neighbor turbines
        for neighbor in neighbor_turbines:
            # skipping if everything in result is filled already (no NaNs)
            if result.isna().sum() <= self._max_nan:
                continue

            try:
                # adjusting required period to be limited to the missing periods in result
                missing_period = DateTimeRange(result[result.isna()].index[0], result[result.isna()].index[-1])

                # add neighbor features to the requirements
                self._add_requirement(
                    RequiredFeatures(
                        {neighbor: ["ActivePower_10min.AVG", "CurtailmentState_10min.REP", "IEC-OperationState_10min.REP"]},
                    ),
                )
                # getting neighbor features
                self._get_required_data(period=missing_period, reindex="10min", cached_data=cached_data)

                # adjusting features for this neighbor turbine to remove periods where turbine is not producing power or is curtailed
                df = self._adjust_features(neighbor)

                # getting only indexes where result is NaN and neighbor ActivePower_10min.AVG is not NaN
                idx = result[result.isna()].index.intersection(df[df["ActivePower_10min.AVG"].notna()].index)

                if len(idx) == 0:
                    continue

                # filtering df to only include the indexes
                df = df.loc[idx, :].copy()

                # getting active_power_regressions for this neighbor turbine
                regression = {"slope": 1, "offset": 0}
                if "neighbor_active_power_regressions" in self._get_requirement_data("RequiredObjectAttributes")[self.object] and (
                    neighbor in self._get_requirement_data("RequiredObjectAttributes")[self.object]["neighbor_active_power_regressions"]
                ):
                    regression = self._get_requirement_data("RequiredObjectAttributes")[self.object][
                        "neighbor_active_power_regressions"
                    ][neighbor]

                # applying regression to neighbor active power
                df["ActivePower_10min.AVG"] = (df["ActivePower_10min.AVG"] * regression["slope"] + regression["offset"]).clip(
                    0,
                    None,
                )  # clip to be sure no negative active power appear after regression

                # adding neighbor active power to result
                result.loc[idx] = df["ActivePower_10min.AVG"]

            except Exception:
                logger.exception(f"'{self.object}' - Error on step 2: using active power from neighbor turbine '{neighbor}'")

    t2 = perf_counter()

    # * Step 3: using wind speed from neighbor turbines

    # skipping if everything in result is filled already (no NaNs)
    if result.isna().sum() > self._max_nan:
        # getting neighbor turbines
        neighbor_turbines = self._get_requirement_data("RequiredObjectAttributes")[self.object]["neighbor_wind_turbines"]

        # iterating over neighbor turbines
        for neighbor in neighbor_turbines:
            # skipping if everything in result is filled already (no NaNs)
            if result.isna().sum() <= self._max_nan:
                continue

            try:
                # adjusting required period to be limited to the missing periods in result
                missing_period = DateTimeRange(result[result.isna()].index[0], result[result.isna()].index[-1])

                # add neighbor features to the requirements
                self._add_requirement(RequiredFeatures({neighbor: self._pc_model.model_arguments.reference_features}))
                # getting neighbor features
                self._get_required_data(period=missing_period, reindex="10min", cached_data=cached_data)

                # adjusting features for this neighbor turbine to remove periods where turbine is not producing power or is curtailed
                df = self._adjust_features(neighbor)

                # getting indexes where all reference data for this power curve model is present (no NaNs)
                idx = df[~df[self._pc_model.model_arguments.reference_features].isna().any(axis=1)].index

                # getting only indexes where result is NaN
                idx = result[result.isna()].index.intersection(idx)

                if len(idx) == 0:
                    continue

                # filtering df to only include the indexes
                df = df.loc[idx, :].copy()

                # getting wind_speed_regressions for this neighbor turbine
                regression = {"slope": 1, "offset": 0}
                if "neighbor_wind_speed_regressions" in self._get_requirement_data("RequiredObjectAttributes")[self.object] and (
                    neighbor in self._get_requirement_data("RequiredObjectAttributes")[self.object]["neighbor_wind_speed_regressions"]
                ):
                    regression = self._get_requirement_data("RequiredObjectAttributes")[self.object]["neighbor_wind_speed_regressions"][
                        neighbor
                    ]

                # applying regression to neighbor wind speed
                df["WindSpeed_10min.AVG"] = (df["WindSpeed_10min.AVG"] * regression["slope"] + regression["offset"]).clip(
                    0,
                    None,
                )  # clip to be sure no negative wind speeds appear after regression

                # adjusting format for predictive model
                df = df[self._pc_model.model_arguments.reference_features]
                df.index = MultiIndex.from_product([[self.object], df.index], names=["object", "timestamp"])
                df.index = df.index.set_levels(df.index.levels[0].astype("string"), level=0)
                df = df.astype("float32")

                # predicting active power theoretical using the power curve model
                result.loc[idx] = self._pc_model.predict(df)[self._pc_model.model_arguments.target_features[0]].values

            except Exception:
                logger.exception(f"'{self.object}' - Error on step 3: using wind speed from neighbor turbine '{neighbor}'")

    t3 = perf_counter()

    # * Step 4: using reference met mast wind

    # skipping if everything in result is filled already (no NaNs)
    if result.isna().sum() > self._max_nan:
        # getting reference met mast
        reference_met_masts = self._get_requirement_data("RequiredObjectAttributes")[self.object]["reference_met_masts"]

        # iterating over reference met masts
        for ref_mast in reference_met_masts:
            # skipping if everything in result is filled already (no NaNs)
            if result.isna().sum() <= self._max_nan:
                continue

            try:
                # adjusting required period to be limited to the missing periods in result
                missing_period = DateTimeRange(result[result.isna()].index[0], result[result.isna()].index[-1])

                # selecting the features that are needed for the power curve model
                turbine_mast_conversion = {
                    "WindSpeed_10min.AVG": "WindSpeed1_10min.AVG",
                    "WindSpeed_10min.STD": "WindSpeed1_10min.STD",
                    "AmbTemp_10min.AVG": "AmbTemp1_10min.AVG",
                }
                met_features = [v for k, v in turbine_mast_conversion.items() if k in self._pc_model.model_arguments.reference_features]

                # add met mast features to the requirements
                self._add_requirement(RequiredFeatures({ref_mast: met_features}))
                # getting ref_mast features
                self._get_required_data(period=missing_period, reindex="10min", cached_data=cached_data)

                # adjusting features
                df = self._adjust_features(ref_mast, rename_dict={v: k for k, v in turbine_mast_conversion.items()})

                # shifting one timestamp as met mast data represents the start of period
                df = df.shift(periods=1)

                # adding WindDirectionRelative_10min.AVG in case needed (default is 0)
                if "WindDirectionRelative_10min.AVG" in self._pc_model.model_arguments.reference_features:
                    df["WindDirectionRelative_10min.AVG"] = 0

                # getting indexes where all reference data for this power curve model is present (no NaNs)
                idx = df[~df[self._pc_model.model_arguments.reference_features].isna().any(axis=1)].index

                # getting only indexes where result is NaN
                idx = result[result.isna()].index.intersection(idx)

                if len(idx) == 0:
                    continue

                # filtering df to only include the wanted indexes
                df = df.loc[idx, :].copy()

                # getting wind_speed_regressions for this met mast
                regression = {"slope": 1, "offset": 0}
                if "met_mast_wind_speed_regressions" in self._get_requirement_data("RequiredObjectAttributes")[self.object] and (
                    ref_mast in self._get_requirement_data("RequiredObjectAttributes")[self.object]["met_mast_wind_speed_regressions"]
                ):
                    regression = self._get_requirement_data("RequiredObjectAttributes")[self.object]["met_mast_wind_speed_regressions"][
                        ref_mast
                    ]

                # applying regression to ref_mast wind speed
                df["WindSpeed_10min.AVG"] = (df["WindSpeed_10min.AVG"] * regression["slope"] + regression["offset"]).clip(
                    0,
                    None,
                )  # clip to be sure no negative wind speeds appear after regression

                # adjusting format for predictive model
                df = df[self._pc_model.model_arguments.reference_features]
                df.index = MultiIndex.from_product([[self.object], df.index], names=["object", "timestamp"])
                df.index = df.index.set_levels(df.index.levels[0].astype("string"), level=0)
                df = df.astype("float32")

                # predicting active power theoretical using the power curve model
                result.loc[idx] = self._pc_model.predict(df)[self._pc_model.model_arguments.target_features[0]].values

            except Exception:
                logger.exception(f"'{self.object}' - Error on step 4: using reference met mast wind '{ref_mast}'")

    t4 = perf_counter()

    # * Step 5: using average of all turbines from the same wind farm

    # skipping if everything in result is filled already (no NaNs)
    if result.isna().sum() > self._max_nan:
        try:
            # adjusting required period to be limited to the missing periods in result
            missing_period = DateTimeRange(result[result.isna()].index[0], result[result.isna()].index[-1])

            # getting all turbines from the same wind farm
            wf_wtgs = self._perfdb.objects.instances.get(
                spe_names=[self._get_requirement_data("RequiredObjectAttributes")[self.object]["spe_name"]],
                object_types=["wind_turbine"],
            )
            wf_wtgs = list(wf_wtgs.keys())
            needed_wtgs = [w for w in wf_wtgs if w != self.object]

            # adding required features for all turbines from the same wind farm
            self._add_requirement(
                RequiredFeatures(
                    {w: ["ActivePower_10min.AVG", "CurtailmentState_10min.REP", "IEC-OperationState_10min.REP"] for w in needed_wtgs},
                ),
            )

            # getting all turbines from the same wind farm features
            self._get_required_data(period=missing_period, reindex="10min", cached_data=cached_data)

            # adjusting features
            df = self._adjust_features(needed_wtgs).loc[:, IndexSlice[:, "ActivePower_10min.AVG"]].droplevel(1, axis=1).copy()

            # calculating average of all turbines from the same wind farm
            df["average"] = df.mean(axis=1, skipna=True)

            # getting indexes where "average" is present (no NaNs)
            idx = df[~df["average"].isna()].index

            # getting only indexes where result is NaN
            idx = result[result.isna()].index.intersection(idx)

            if len(idx) > 0:
                # adding average to result
                result.loc[idx] = df.loc[idx, "average"]

        except Exception:
            logger.exception(f"'{self.object}' - Error on step 5: using average of all turbines from the same wind farm")

    t5 = perf_counter()

    # * Final check to see if all values are filled

    null_idx = result[result.isna()].index
    if len(null_idx) > 0:
        logger.error(
            f"'{self.object}' - Could not calculate active_power_theoretical for {len(null_idx) / len(result):.2%} ({len(null_idx)}) of timestamps: {null_idx.to_list()}",
        )

    # * clipping values from 0 to nominal_power

    result = result.clip(lower=0, upper=self._get_requirement_data("RequiredObjectAttributes")[self.object]["nominal_power"])

    # adding calculated feature to class result attribute
    self._result = result.copy()

    # saving results
    self.save(save_into=save_into, **kwargs)

    logger.debug(
        f"'{self.object}' - Time taken to calculate: {perf_counter() - t0:.2f}s. Step 1: {t1 - t0:.2f}s. Step 2: {t2 - t1:.2f}s. Step 3: {t3 - t2:.2f}s. Step 4: {t4 - t3:.2f}s. Step 5: {t5 - t4:.2f}s. Final check: {perf_counter() - t5:.2f}s.",
    )

    return result

`save(save_into=None, **kwargs)` ¶

Method to save the calculated feature values in performance_db.

Parameters:

save_into ¶
(Literal['all', 'performance_db'] | None, default: None ) –

Argument that will be passed to the method "save". The options are: - "all": The feature will be saved in performance_db and bazefield. - "performance_db": the feature will be saved only in performance_db. - None: The feature will not be saved.

By default None.
**kwargs ¶
(dict, default: {} ) –

Not being used at the moment. Here only for compatibility.

Source code in echo_energycalc/feature_calc_core.py

def save(
    self,
    save_into: Literal["all", "performance_db"] | None = None,
    **kwargs,  # noqa: ARG002
) -> None:
    """
    Method to save the calculated feature values in performance_db.

    Parameters
    ----------
    save_into : Literal["all", "performance_db"] | None, optional
        Argument that will be passed to the method "save". The options are:
        - "all": The feature will be saved in performance_db and bazefield.
        - "performance_db": the feature will be saved only in performance_db.
        - None: The feature will not be saved.

        By default None.
    **kwargs : dict, optional
        Not being used at the moment. Here only for compatibility.
    """
    # checking arguments
    if not isinstance(save_into, str | type(None)):
        raise TypeError(f"save_into must be a string or None, not {type(save_into)}")
    if isinstance(save_into, str) and save_into not in ["all", "performance_db"]:
        raise ValueError(f"save_into must be 'all', 'performance_db' or None, not {save_into}")

    # checking if calculation was done
    if self.result is None:
        raise ValueError(
            "The calculation was not done. Cannot save the feature calculation results. Please make sure to do something like 'self._result = df[self.feature].copy()' in the method 'calculate' before calling 'self.save()'.",
        )

    if save_into is None:
        return

    if isinstance(save_into, str):
        if save_into not in ["performance_db", "all"]:
            raise ValueError(f"save_into must be 'performance_db' or 'all', not {save_into}.")
        upload_to_bazefield = save_into == "all"
    elif save_into is None:
        upload_to_bazefield = False
    else:
        raise TypeError(f"save_into must be a string or None, not {type(save_into)}.")

    # converting result series to DataFrame if needed
    if isinstance(self.result, Series):
        result_df = self.result.to_frame()
    elif isinstance(self.result, DataFrame):
        result_df = self.result.droplevel(0, axis=1)
    else:
        raise TypeError(f"result must be a pandas Series or DataFrame, not {type(self.result)}.")

    # adjusting DataFrame to be inserted in the database
    # making the columns a Multindex with levels object_name and feature_name
    result_df.columns = MultiIndex.from_product([[self.object], result_df.columns], names=["object_name", "feature_name"])

    self._perfdb.features.values.series.insert(
        df=result_df,
        on_conflict="update",
        bazefield_upload=upload_to_bazefield,
    )

Theoretical Power¶

Overview¶

Calculation Logic¶

Database Requirements¶

Class Definition¶

`FeatureCalcPowerTheoretical(object_name, feature)` ¶

`object_name` ¶

`feature` ¶

`feature` `property` ¶

`name` `property` ¶

`object` `property` ¶

`requirements` `property` ¶

`result` `property` ¶

`calculate(period, save_into=None, cached_data=None, **kwargs)` ¶

`period` ¶

`save_into` ¶

`cached_data` ¶

`kwargs`** ¶

`save(save_into=None, **kwargs)` ¶

`save_into` ¶

`kwargs`** ¶

Theoretical Power¶

Overview¶

Calculation Logic¶

Database Requirements¶

Class Definition¶

FeatureCalcPowerTheoretical(object_name, feature) ¶

object_name ¶

feature ¶

feature property ¶

name property ¶

object property ¶

requirements property ¶

result property ¶

calculate(period, save_into=None, cached_data=None, **kwargs) ¶

period ¶

save_into ¶

cached_data ¶

**kwargs ¶

save(save_into=None, **kwargs) ¶

save_into ¶

**kwargs ¶

`FeatureCalcPowerTheoretical(object_name, feature)` ¶

`object_name` ¶

`feature` ¶

`feature` `property` ¶

`name` `property` ¶

`object` `property` ¶

`requirements` `property` ¶

`result` `property` ¶

`calculate(period, save_into=None, cached_data=None, **kwargs)` ¶

`period` ¶

`save_into` ¶

`cached_data` ¶

`kwargs`** ¶

`save(save_into=None, **kwargs)` ¶

`save_into` ¶

`kwargs`** ¶