""" Limits for the solver variables """
import wip.modules.ops as operations
[docs]class Limits:
[docs] @staticmethod
def limits_by_rolling_mean(df, rolling, quant_one, quant_two):
roll_mean = df.rolling(rolling).mean().quantile
min_limit = roll_mean(quant_one)
max_limit = roll_mean(quant_two)
return min_limit, max_limit
[docs] @staticmethod
def parse_limits(limits):
"""
Parse json limits:
# tag limits have tags with the same limits as others
"""
for tag, value in limits.items():
if isinstance(value, str):
limits[tag] = limits[value]
return limits
[docs] @staticmethod
def read_limits(limits, feature=None):
"""Read and return the limit's values"""
limits = Limits.parse_limits(limits)
# In this case, the operation applied over that tag is different
if feature and feature not in limits.keys():
return None
return limits
[docs] @staticmethod
def define_limit_by_quantile(feature, models_features, production_query, limits):
"""
Define upper and lower bounds by percentiles.
Parameters
----------
feature : str
Feature name and the array with the feature values
models_features : dict
Dictionary with each of the model's features
production_query : pd.Series
A `pandas.Series` that contains the query to filter values with.
limits : dict
Dictionary with the limits for each feature.
"""
if not limits:
return None, None
feature_quantile = models_features[feature][production_query].quantile
if "min" in limits[feature].keys() and "max" in limits[feature].keys():
min_value, max_value = limits[feature]["min"], limits[feature]["max"]
return feature_quantile(min_value), feature_quantile(max_value)
if "min" in limits[feature].keys():
return feature_quantile(limits[feature]["min"]), None
return None, feature_quantile(limits[feature]["max"])
[docs] @staticmethod
def define_limit_by_normalization(scalers, feature, limits):
"""
Define upper and lower bounds by variable normalization.
Returns both values when lmin and lmax are defined
Return a left value when only lmin is defined
Return a right value when only lmax is defined
"""
arguments = {"scalers": scalers, "feature": feature}
if isinstance(limits[feature], str):
limits[feature] = limits[limits[feature]]
if len(limits[feature]) == 2:
arg_min = {**arguments, "norm_value": limits[feature]["min"]}
arg_max = {**arguments, "norm_value": limits[feature]["max"]}
return operations.normalize_feature(
**arg_min
), operations.normalize_feature(**arg_max)
# in this case has only one of the values
# getting the min or max value in the limits
try:
if "min" in limits[feature].keys():
args = {**arguments, "norm_value": limits[feature]["min"]}
return operations.normalize_feature(**args), None
except Exception as exc:
raise Exception(feature) from exc
args = {**arguments, "norm_value": limits[feature]["max"]}
return None, operations.normalize_feature(**args)
[docs] @staticmethod
def define_work_dataset(feature, datasets, production_query):
dataset = list(
filter(lambda col: feature in datasets[col].columns, datasets.keys())
)[0]
production_query = production_query.rename("production")
df = datasets[dataset].join(production_query)
query = "production == True"
if "status" in df.columns:
query += " and status == 1"
return df.query(query)[feature]
[docs] @staticmethod
def define_limit_by_rolling_mean(
feature, production_query, datasets, rolling_limits
):
"""
Set lower- and upper-bound for a variable defined inside `wip.files.rolling_limits`.
Parameters
----------
feature : str
The tag name defined as dictionary key inside `wip.files.rolling_limits`.
production_query :
datasets : Dict[str, pd.DataFrame]
Dictionary of `pandas.DataFrame` with the input datasets.
rolling_limits : Dict[str, Dict[str, int | float]]
Dictionary containing the rolling limits to be used for defining the
optimization variable's lower- and upper-bounds.
Returns
-------
Tuple[float, float]
Lower- and upper-bounds calculated from the rolling limit.
"""
df = Limits.define_work_dataset(feature, datasets, production_query)
return Limits.limits_by_rolling_mean(
df,
rolling_limits[feature]["rolling"],
rolling_limits[feature]["quant_one"],
rolling_limits[feature]["quant_two"],
)
[docs] @staticmethod
def define_bentonita_limit(feature, datasets, production_query, scalers):
df = Limits.define_work_dataset(feature, datasets, production_query)
# lmin, _ = Limits.limits_by_rolling_mean(df, scalers, feature, 24, 0.25, 0.25)
lmin = 0
aux = (
df.rolling(24).mean().quantile(0.25)
+ df.rolling(24).mean().quantile(0.25) * 0.05
)
aux = 0.0053 if aux <= 0.0053 else aux
lmax = operations.normalize_feature(scalers, feature, aux)
return lmin, lmax
[docs] @staticmethod
def define_flotcor_limit(feature, scalers):
lmin = 0
aux = 4.99
lmax = operations.normalize_feature(scalers, feature, aux)
return lmin, lmax
[docs] @staticmethod
def define_constant_limits(feature, limits):
for key in limits.keys():
if feature.startswith(key):
feature = key
break
return limits[feature]['min'], limits[feature]['max']