Merge pull request #941 from mindsdb/staging

Release 22.7.3.0

Author: paxcema

docssrc/source/tutorials/tutorial_time_series/tutorial_time_series.ipynb

@@ -1,6 +1,6 @@
 __title__ = 'lightwood'
 __package_name__ = 'lightwood'
-__version__ = '22.7.2.2'
+__version__ = '22.7.3.0'
 __description__ = "Lightwood is a toolkit for automatic machine learning model building"
 __email__ = "community@mindsdb.com"
 __author__ = 'MindsDB Inc'

lightwood/__about__.py

@@ -54,12 +54,9 @@ def explain(data: pd.DataFrame,
             for col in timeseries_settings.group_by:
                 row_insights[f'group_{col}'] = data[col]
 
-        for col in timeseries_settings.order_by:
-            row_insights[f'order_{col}'] = data[col]
-
-        for col in timeseries_settings.order_by:
-            row_insights[f'order_{col}'] = get_inferred_timestamps(
-                row_insights, col, ts_analysis['deltas'], timeseries_settings)
+        row_insights[f'order_{timeseries_settings.order_by}'] = data[timeseries_settings.order_by]
+        row_insights[f'order_{timeseries_settings.order_by}'] = get_inferred_timestamps(
+            row_insights, timeseries_settings.order_by, ts_analysis['deltas'], timeseries_settings)
 
     kwargs = {
         'data': data,

lightwood/analysis/explain.py

@@ -38,7 +38,7 @@ def analyze(self, info: Dict[str, object], **kwargs) -> Dict[str, object]:
         else:
             empty_input_accuracy = {}
             ignorable_input_cols = [x for x in ns.input_cols if (not ns.tss.is_timeseries or
-                                                                 (x not in ns.tss.order_by and
+                                                                 (x != ns.tss.order_by and
                                                                   x not in ns.tss.historical_columns))]
             for col in ignorable_input_cols:
                 partial_data = deepcopy(ns.encoded_val_data)

lightwood/analysis/helpers/feature_importance.py

@@ -9,7 +9,7 @@
 
 from lightwood.api.dtype import dtype
 from lightwood.api.types import PredictionArguments
-from lightwood.helpers.ts import add_tn_conf_bounds
+from lightwood.helpers.ts import add_tn_num_conf_bounds, add_tn_cat_conf_bounds
 
 from lightwood.data import EncodedDs
 from lightwood.analysis.base import BaseAnalysisBlock
@@ -120,7 +120,7 @@ def analyze(self, info: Dict[str, object], **kwargs) -> Dict[str, object]:
             # fit additional ICPs in time series tasks with grouped columns
             if ns.tss.is_timeseries and ns.tss.group_by:
                 # generate a multiindex
-                midx = pd.MultiIndex.from_frame(icp_df[[*ns.tss.group_by, f'__mdb_original_{ns.tss.order_by[0]}']])
+                midx = pd.MultiIndex.from_frame(icp_df[[*ns.tss.group_by, f'__mdb_original_{ns.tss.order_by}']])
                 icp_df.index = midx
 
                 # create an ICP for each possible group
@@ -157,7 +157,7 @@ def analyze(self, info: Dict[str, object], **kwargs) -> Dict[str, object]:
 
                 # add all predictions to DF
                 icps_df = deepcopy(ns.data)
-                midx = pd.MultiIndex.from_frame(icps_df[[*ns.tss.group_by, f'__mdb_original_{ns.tss.order_by[0]}']])
+                midx = pd.MultiIndex.from_frame(icps_df[[*ns.tss.group_by, f'__mdb_original_{ns.tss.order_by}']])
                 icps_df.index = midx
                 if ns.is_multi_ts or pred_is_list:
                     icps_df[f'__predicted_{ns.target}'] = np.array([p[0] for p in ns.normal_predictions['prediction']])
@@ -380,8 +380,11 @@ def explain(self, row_insights: pd.DataFrame, global_insights: Dict[str, object]
                                               cooldown=ns.pred_args.anomaly_cooldown)
                     row_insights['anomaly'] = anomalies
 
-            if ns.tss.is_timeseries and ns.tss.horizon > 1 and is_numerical:
-                row_insights = add_tn_conf_bounds(row_insights, ns.tss)
+            if ns.tss.is_timeseries and ns.tss.horizon > 1:
+                if is_numerical:
+                    row_insights = add_tn_num_conf_bounds(row_insights, ns.tss)
+                else:
+                    row_insights = add_tn_cat_conf_bounds(row_insights, ns.tss)
 
             # clip bounds if necessary
             if is_numerical:

lightwood/analysis/nc/calibrate.py

@@ -135,7 +135,7 @@ def lookup_encoder(
     # Time-series representations require more advanced flags
     if tss.is_timeseries:
         gby = tss.group_by if tss.group_by is not None else []
-        if col_name in tss.order_by:
+        if col_name == tss.order_by:
             encoder_dict["module"] = "ArrayEncoder"
             encoder_dict["args"]["original_type"] = f'"{tss.target_type}"'
             encoder_dict["args"]["window"] = f"{tss.window}"
@@ -575,6 +575,7 @@ def _add_implicit_values(json_ai: JsonAI) -> JsonAI:
             mixers[i]["args"]["tss"] = mixers[i]["args"].get("tss", "$problem_definition.timeseries_settings")
             mixers[i]["args"]["ts_analysis"] = mixers[i]["args"].get("ts_analysis", "$ts_analysis")
             mixers[i]["args"]["fit_on_dev"] = mixers[i]["args"].get("fit_on_dev", "True")
+            mixers[i]["args"]["use_stl"] = mixers[i]["args"].get("use_stl", "False")
 
         elif mixers[i]["module"] == "NHitsMixer":
             mixers[i]["args"]["target"] = mixers[i]["args"].get("target", "$target")
@@ -752,27 +753,29 @@ def code_from_json_ai(json_ai: JsonAI) -> str:
         encoder_dict[col_name] = call(encoder)
 
     # Populate time-series specific details
+    # TODO: consider moving this to a `JsonAI override` phase
     tss = json_ai.problem_definition.timeseries_settings
-    if tss.is_timeseries and tss.use_previous_target:
-        col_name = f"__mdb_ts_previous_{json_ai.problem_definition.target}"
-        target_type = json_ai.dtype_dict[json_ai.problem_definition.target]
-        json_ai.problem_definition.timeseries_settings.target_type = target_type
-        encoder_dict[col_name] = call(
-            lookup_encoder(
-                target_type,
-                col_name,
-                False,
-                json_ai.problem_definition,
-                False,
-                None,
+    if tss.is_timeseries:
+        if tss.use_previous_target:
+            col_name = f"__mdb_ts_previous_{json_ai.problem_definition.target}"
+            target_type = json_ai.dtype_dict[json_ai.problem_definition.target]
+            json_ai.problem_definition.timeseries_settings.target_type = target_type
+            encoder_dict[col_name] = call(
+                lookup_encoder(
+                    target_type,
+                    col_name,
+                    False,
+                    json_ai.problem_definition,
+                    False,
+                    None,
+                )
             )
-        )
 
-        dtype_dict[col_name] = target_type
-        # @TODO: Is populating the json_ai at this stage even necessary?
-        json_ai.encoders[col_name] = encoder_dict[col_name]
-        json_ai.dtype_dict[col_name] = target_type
-        json_ai.dependency_dict[col_name] = []
+            dtype_dict[col_name] = target_type
+            # @TODO: Is populating the json_ai at this stage even necessary?
+            json_ai.encoders[col_name] = encoder_dict[col_name]
+            json_ai.dtype_dict[col_name] = target_type
+            json_ai.dependency_dict[col_name] = []
 
     # ----------------- #
 

lightwood/api/json_ai.py

@@ -110,11 +110,11 @@ class TimeseriesSettings:
 
     :param is_timeseries: Whether the input data should be treated as time series; if true, this flag is checked in \
         subsequent internal steps to ensure processing is appropriate for time-series data.
-    :param order_by: A list of columns by which the data should be ordered.
+    :param order_by: Column by which the data should be ordered.
     :param group_by: Optional list of columns by which the data should be grouped. Each different combination of values\
          for these columns will yield a different series.
     :param window: The temporal horizon (number of rows) that a model intakes to "look back" into when making a\
-         prediction, after the rows are ordered by order_by columns and split into groups if applicable.
+         prediction, after the rows are ordered by the order_by column and split into groups if applicable.
     :param horizon: The number of points in the future that predictions should be made for, defaults to 1. Once \
         trained, the model will be able to predict up to this many points into the future.
     :param historical_columns: The temporal dynamics of these columns will be used as additional context to train the \
@@ -128,7 +128,7 @@ class TimeseriesSettings:
     """  # noqa
 
     is_timeseries: bool
-    order_by: List[str] = None
+    order_by: str = None
     window: int = None
     group_by: List[str] = None
     use_previous_target: bool = True
@@ -152,11 +152,15 @@ def from_dict(obj: Dict):
         :returns: A populated ``TimeseriesSettings`` object.
         """ # noqa
         if len(obj) > 0:
-            for mandatory_setting in ["order_by", "window"]:
+            for mandatory_setting, etype in zip(["order_by", "window"], [str, int]):
                 if mandatory_setting not in obj:
                     err = f"Missing mandatory timeseries setting: {mandatory_setting}"
                     log.error(err)
                     raise Exception(err)
+                if obj[mandatory_setting] and not isinstance(obj[mandatory_setting], etype):
+                    err = f"Wrong type for mandatory timeseries setting '{mandatory_setting}': found '{type(obj[mandatory_setting])}', expected '{etype}'"  # noqa
+                    log.error(err)
+                    raise Exception(err)
 
             timeseries_settings = TimeseriesSettings(
                 is_timeseries=True,

lightwood/api/types.py

@@ -394,7 +394,7 @@ def clean_timeseries(df: pd.DataFrame, tss: TimeseriesSettings) -> pd.DataFrame:
     invalid_rows = []
 
     for idx, row in df.iterrows():
-        if pd.isna(row[tss.order_by[0]]):
+        if pd.isna(row[tss.order_by]):
             invalid_rows.append(idx)
 
     df = df.drop(invalid_rows)

lightwood/data/cleaner.py

@@ -93,7 +93,8 @@ def stratify(data: pd.DataFrame,
              pct_test: float,
              stratify_on: List[str],
              seed: int,
-             reshuffle: bool) -> List[pd.DataFrame]:
+             reshuffle: bool,
+             atol: float = 0.05) -> List[pd.DataFrame]:
     """
     Stratified data splitter.
 
@@ -109,6 +110,7 @@ def stratify(data: pd.DataFrame,
     :param stratify_on: Columns to consider when stratifying
     :param seed: Random state for pandas data-frame shuffling
     :param reshuffle: specify if reshuffling should be done post-split
+    :param atol: absolute tolerance for difference in stratification percentages. If violated, reverts to a non-stratified split.
 
     :returns Stratified train, dev, test dataframes
     """  # noqa
@@ -136,10 +138,13 @@ def stratify(data: pd.DataFrame,
                                      for df in [train_st, dev_st, test_st]]
 
     # check that stratified lengths conform to expected percentages
-    if not np.isclose(len(train_st) / len(data), pct_train, atol=0.01) or \
-            not np.isclose(len(dev_st) / len(data), pct_dev, atol=0.01) or \
-            not np.isclose(len(test_st) / len(data), pct_test, atol=0.01):
-        log.info("Could not stratify; reverting to simple split")
+    emp_tr = len(train_st) / len(data)
+    emp_dev = len(dev_st) / len(data)
+    emp_te = len(test_st) / len(data)
+    if not np.isclose(emp_tr, pct_train, atol=atol) or \
+            not np.isclose(emp_dev, pct_dev, atol=atol) or \
+            not np.isclose(emp_te, pct_test, atol=atol):
+        log.warning(f"Stratification is outside of imposed tolerance ({atol}) ({emp_tr} train - {emp_dev} dev - {emp_te} test), reverting to a simple split.")  # noqa
         train_st, dev_st, test_st = simple_split(data, pct_train, pct_dev, pct_test)
 
     return [train_st, dev_st, test_st]

lightwood/data/splitter.py

@@ -10,7 +10,7 @@
 
 from lightwood.api.types import TimeseriesSettings
 from lightwood.api.dtype import dtype
-from lightwood.helpers.ts import get_ts_groups, get_delta, get_group_matches, Differencer
+from lightwood.helpers.ts import get_ts_groups, get_delta, get_group_matches, Differencer, max_pacf
 from lightwood.helpers.log import log
 from lightwood.encoder.time_series.helpers.common import generate_target_group_normalizers
 
@@ -41,6 +41,7 @@ def timeseries_analyzer(data: Dict[str, pd.DataFrame], dtype_dict: Dict[str, str
     normalizers = generate_target_group_normalizers(data['train'], target, dtype_dict, groups, tss)
 
     if dtype_dict[target] in (dtype.integer, dtype.float, dtype.num_tsarray):
+        periods = max_pacf(data['train'], groups, target, tss)  # override with PACF output
         naive_forecast_residuals, scale_factor = get_grouped_naive_residuals(data['dev'],
                                                                              target,
                                                                              tss,
@@ -96,9 +97,10 @@ def get_grouped_naive_residuals(
     group_scale_factors = {}
     for group in group_combinations:
         idxs, subset = get_group_matches(info, group, tss.group_by)
-        residuals, scale_factor = get_naive_residuals(subset[target])  # @TODO: pass m once we handle seasonality
-        group_residuals[group] = residuals
-        group_scale_factors[group] = scale_factor
+        if subset.shape[0] > 1:
+            residuals, scale_factor = get_naive_residuals(subset[target])  # @TODO: pass m once we handle seasonality
+            group_residuals[group] = residuals
+            group_scale_factors[group] = scale_factor
     return group_residuals, group_scale_factors
 
 
@@ -119,34 +121,38 @@ def get_stls(train_df: pd.DataFrame,
              groups: list,
              tss: TimeseriesSettings
              ) -> Dict[str, object]:
-    stls = {}
+    stls = {'__default': None}
     for group in groups:
-        _, tr_subset = get_group_matches(train_df, group, tss.group_by)
-        _, dev_subset = get_group_matches(dev_df, group, tss.group_by)
-        group_freq = tr_subset['__mdb_inferred_freq'].iloc[0]
-        tr_subset = deepcopy(tr_subset)[target]
-        dev_subset = deepcopy(dev_subset)[target]
-        tr_subset.index = pd.date_range(start=tr_subset.iloc[0], freq=group_freq, periods=len(tr_subset)).to_period()
-        dev_subset.index = pd.date_range(start=dev_subset.iloc[0], freq=group_freq, periods=len(dev_subset)).to_period()
-        stl = _pick_ST(tr_subset, dev_subset, sps[group])
-        log.info(f'Best STL decomposition params for group {group} are: {stl["best_params"]}')
-        stls[group] = stl
+        if group != '__default':
+            _, tr_subset = get_group_matches(train_df, group, tss.group_by)
+            _, dev_subset = get_group_matches(dev_df, group, tss.group_by)
+            if tr_subset.shape[0] > 0 and dev_subset.shape[0] > 0 and sps.get(group, False):
+                group_freq = tr_subset['__mdb_inferred_freq'].iloc[0]
+                tr_subset = deepcopy(tr_subset)[target]
+                dev_subset = deepcopy(dev_subset)[target]
+                tr_subset.index = pd.date_range(start=tr_subset.iloc[0], freq=group_freq,
+                                                periods=len(tr_subset)).to_period()
+                dev_subset.index = pd.date_range(start=dev_subset.iloc[0], freq=group_freq,
+                                                 periods=len(dev_subset)).to_period()
+                stl = _pick_ST(tr_subset, dev_subset, sps[group])
+                log.info(f'Best STL decomposition params for group {group} are: {stl["best_params"]}')
+                stls[group] = stl
     return stls
 
 
-def _pick_ST(tr_subset: pd.Series, dev_subset: pd.Series, sp: int):
+def _pick_ST(tr_subset: pd.Series, dev_subset: pd.Series, sp: list):
     """
     Perform hyperparam search with optuna to find best combination of ST transforms for a time series.
 
     :param tr_subset: training series used for fitting blocks. Index should be datetime, and values are the actual time series.
     :param dev_subset: dev series used for computing loss. Index should be datetime, and values are the actual time series.
-    :param sp: seasonal period
+    :param sp: list of candidate seasonal periods
     :return: best deseasonalizer and detrender combination based on dev_loss
     """  # noqa
 
     def _ST_objective(trial: optuna.Trial):
         trend_degree = trial.suggest_categorical("trend_degree", [1, 2])
-        ds_sp = trial.suggest_categorical("ds_sp", [sp])  # seasonality period to use in deseasonalizer
+        ds_sp = trial.suggest_categorical("ds_sp", sp)  # seasonality period to use in deseasonalizer
         if min(min(tr_subset), min(dev_subset)) <= 0:
             decomp_type = trial.suggest_categorical("decomp_type", ['additive'])
         else:
@@ -161,7 +167,8 @@ def _ST_objective(trial: optuna.Trial):
         trial.set_user_attr("transformer", transformer)
         return np.power(residuals, 2).sum()
 
-    study = optuna.create_study()
+    space = {"trend_degree": [1, 2], "ds_sp": sp, "decomp_type": ['additive', 'multiplicative']}
+    study = optuna.create_study(sampler=optuna.samplers.GridSampler(space))
     study.optimize(_ST_objective, n_trials=8)
 
     return {