def do_inference()

in decisionai_plugin/sample/lr/lr_plugin_service.py [0:0]

• 32 lines of code
• 9 McCabe index (conditional complexity)

    def do_inference(self, model_dir, parameters, series, context):
        results = []

        start_time, end_time, gran = self.get_inference_time_range(parameters)
        traceback_window = parameters['instance']['params']['tracebackWindow']

        for factor in series:
            timestamps = []
            values = []
            mses = []
            r2scores = []

            df = pd.DataFrame(factor.value, columns=factor.fields)
            df = df[['time', '__VAL__']]
            df.columns = ['timestamp', 'value']
            df['timestamp'] = df['timestamp'].apply(str_to_dt)
            for timestamp in get_time_list(start_time, end_time, gran):
                sub_df = df[df['timestamp'] < timestamp]
                sub_df = sub_df.iloc[-traceback_window:]
                
                x = sub_df['timestamp'].apply(lambda x: x.timestamp()).to_numpy().reshape(-1, 1)
                y = sub_df['value'].to_numpy().reshape(-1, 1)
                model = linear_model.LinearRegression().fit(x, y)
                y_new = model.predict(x)

                timestamps.append(dt_to_str(timestamp))
                values.append(model.predict(np.array([timestamp.timestamp()]).reshape(-1,1))[0][0])
                mses.append(mean_squared_error(y, y_new))
                r2scores.append(r2_score(y, y_new))

            dimension = dict(seriesId=factor.series_id)
            metric_value = next(m for m in parameters['instance']['target']['metrics'] if m['metricName'] == 'value')
            metric_mse = next(m for m in parameters['instance']['target']['metrics'] if m['metricName'] == 'mse')
            metric_r2score = next(m for m in parameters['instance']['target']['metrics'] if m['metricName'] == 'r2score')

            results.append(dict(metricId=metric_value['metricId'], dimension=dimension, timestamps=timestamps, values=values))
            results.append(dict(metricId=metric_mse['metricId'], dimension=dimension, timestamps=timestamps, values=mses))
            results.append(dict(metricId=metric_r2score['metricId'], dimension=dimension, timestamps=timestamps, values=r2scores))

        return STATUS_SUCCESS, results, ''