import numpy as np
import pandas as pd
from pyFTS.common import FuzzySet, SortedCollection, tree, Util
[docs]class FTS(object):
"""
Fuzzy Time Series object model
"""
def __init__(self, **kwargs):
"""
Create a Fuzzy Time Series model
"""
self.sets = {}
"""The list of fuzzy sets used on this model"""
self.flrgs = {}
"""The list of Fuzzy Logical Relationship Groups - FLRG"""
self.order = kwargs.get('order',1)
"""A integer with the model order (number of past lags are used on forecasting)"""
self.shortname = kwargs.get('name',"")
"""A string with a short name or alias for the model"""
self.name = kwargs.get('name',"")
"""A string with the model name"""
self.detail = kwargs.get('name',"")
"""A string with the model detailed information"""
self.is_high_order = False
"""A boolean value indicating if the model support orders greater than 1, default: False"""
self.min_order = 1
"""In high order models, this integer value indicates the minimal order supported for the model, default: 1"""
self.has_seasonality = False
"""A boolean value indicating if the model supports seasonal indexers, default: False"""
self.has_point_forecasting = True
"""A boolean value indicating if the model supports point forecasting, default: True"""
self.has_interval_forecasting = False
"""A boolean value indicating if the model supports interval forecasting, default: False"""
self.has_probability_forecasting = False
"""A boolean value indicating if the model support probabilistic forecasting, default: False"""
self.is_multivariate = False
"""A boolean value indicating if the model support multivariate time series (Pandas DataFrame), default: False"""
self.dump = False
self.transformations = []
"""A list with the data transformations (common.Transformations) applied on model pre and post processing, default: []"""
self.transformations_param = []
"""A list with the specific parameters for each data transformation"""
self.original_max = 0
"""A float with the upper limit of the Universe of Discourse, the maximal value found on training data"""
self.original_min = 0
"""A float with the lower limit of the Universe of Discourse, the minimal value found on training data"""
self.partitioner = kwargs.get("partitioner", None)
"""A pyFTS.partitioners.Partitioner object with the Universe of Discourse partitioner used on the model. This is a mandatory dependecy. """
if self.partitioner != None:
self.sets = self.partitioner.sets
self.auto_update = False
"""A boolean value indicating that model is incremental"""
self.benchmark_only = False
"""A boolean value indicating a façade for external (non-FTS) model used on benchmarks or ensembles."""
self.indexer = kwargs.get("indexer", None)
"""An pyFTS.models.seasonal.Indexer object for indexing the time series data"""
self.uod_clip = kwargs.get("uod_clip", True)
"""Flag indicating if the test data will be clipped inside the training Universe of Discourse"""
self.alpha_cut = kwargs.get("alpha_cut", 0.0)
"""A float with the minimal membership to be considered on fuzzyfication process"""
self.max_lag = self.order
"""A integer indicating the largest lag used by the model. This value also indicates the minimum number of past lags
needed to forecast a single step ahead"""
[docs] def fuzzy(self, data):
"""
Fuzzify a data point
:param data: data point
:return: maximum membership fuzzy set
"""
best = {"fuzzyset": "", "membership": 0.0}
for f in self.sets:
fset = self.sets[f]
if best["membership"] <= fset.membership(data):
best["fuzzyset"] = fset.name
best["membership"] = fset.membership(data)
return best
[docs] def predict(self, data, **kwargs):
"""
Forecast using trained model
:param data: time series with minimal length to the order of the model
:keyword type: the forecasting type, one of these values: point(default), interval or distribution.
:keyword steps_ahead: The forecasting horizon, i. e., the number of steps ahead to forecast
:keyword start: in the multi step forecasting, the index of the data where to start forecasting
:keyword distributed: boolean, indicate if the forecasting procedure will be distributed in a dispy cluster
:keyword nodes: a list with the dispy cluster nodes addresses
:return: a numpy array with the forecasted data
"""
if self.is_multivariate:
ndata = data
else:
ndata = self.apply_transformations(data)
if self.uod_clip:
ndata = np.clip(ndata, self.original_min, self.original_max)
if 'distributed' in kwargs:
distributed = kwargs.pop('distributed')
else:
distributed = False
if distributed is None or distributed == False:
if 'type' in kwargs:
type = kwargs.pop("type")
else:
type = 'point'
steps_ahead = kwargs.get("steps_ahead", None)
if steps_ahead == None or steps_ahead == 1:
if type == 'point':
ret = self.forecast(ndata, **kwargs)
elif type == 'interval':
ret = self.forecast_interval(ndata, **kwargs)
elif type == 'distribution':
ret = self.forecast_distribution(ndata, **kwargs)
elif steps_ahead > 1:
if type == 'point':
ret = self.forecast_ahead(ndata, steps_ahead, **kwargs)
elif type == 'interval':
ret = self.forecast_ahead_interval(ndata, steps_ahead, **kwargs)
elif type == 'distribution':
ret = self.forecast_ahead_distribution(ndata, steps_ahead, **kwargs)
if not ['point', 'interval', 'distribution'].__contains__(type):
raise ValueError('The argument \'type\' has an unknown value.')
else:
nodes = kwargs.get("nodes", ['127.0.0.1'])
num_batches = kwargs.get('num_batches', 10)
ret = Util.distributed_predict(self, kwargs, nodes, ndata, num_batches)
if not self.is_multivariate:
kwargs['type'] = type
ret = self.apply_inverse_transformations(ret, params=[data[self.max_lag - 1:]], **kwargs)
return ret
[docs] def forecast(self, data, **kwargs):
"""
Point forecast one step ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param kwargs: model specific parameters
:return: a list with the forecasted values
"""
raise NotImplementedError('This model do not perform one step ahead point forecasts!')
[docs] def forecast_interval(self, data, **kwargs):
"""
Interval forecast one step ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param kwargs: model specific parameters
:return: a list with the forecasted intervals
"""
raise NotImplementedError('This model do not perform one step ahead interval forecasts!')
[docs] def forecast_distribution(self, data, **kwargs):
"""
Probabilistic forecast one step ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param kwargs: model specific parameters
:return: a list with the forecasted Probability Distributions
"""
raise NotImplementedError('This model do not perform one step ahead distribution forecasts!')
[docs] def forecast_ahead(self, data, steps, **kwargs):
"""
Point forecast n steps ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param steps: the number of steps ahead to forecast
:param kwargs: model specific parameters
:return: a list with the forecasted values
"""
ret = []
for k in np.arange(0,steps):
tmp = self.forecast(data[-self.order:], **kwargs)
if isinstance(tmp,(list, np.ndarray)):
tmp = tmp[0]
ret.append(tmp)
data.append_rhs(tmp)
return ret
[docs] def forecast_ahead_interval(self, data, steps, **kwargs):
"""
Interval forecast n steps ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param steps: the number of steps ahead to forecast
:param kwargs: model specific parameters
:return: a list with the forecasted intervals
"""
raise NotImplementedError('This model do not perform multi step ahead interval forecasts!')
[docs] def forecast_ahead_distribution(self, data, steps, **kwargs):
"""
Probabilistic forecast n steps ahead
:param data: time series data with the minimal length equal to the max_lag of the model
:param steps: the number of steps ahead to forecast
:param kwargs: model specific parameters
:return: a list with the forecasted Probability Distributions
"""
raise NotImplementedError('This model do not perform multi step ahead distribution forecasts!')
[docs] def train(self, data, **kwargs):
"""
Method specific parameter fitting
:param data: training time series data
:param kwargs: Method specific parameters
"""
pass
[docs] def fit(self, ndata, **kwargs):
"""
Fit the model's parameters based on the training data.
:param ndata: training time series data
:param kwargs:
:keyword num_batches: split the training data in num_batches to save memory during the training process
:keyword save_model: save final model on disk
:keyword batch_save: save the model between each batch
:keyword file_path: path to save the model
:keyword distributed: boolean, indicate if the training procedure will be distributed in a dispy cluster
:keyword nodes: a list with the dispy cluster nodes addresses
"""
import datetime
if self.is_multivariate:
data = ndata
else:
data = self.apply_transformations(ndata)
self.original_min = np.nanmin(data)
self.original_max = np.nanmax(data)
if 'sets' in kwargs:
self.sets = kwargs.pop('sets')
if 'partitioner' in kwargs:
self.partitioner = kwargs.pop('partitioner')
if (self.sets is None or len(self.sets) == 0) and not self.benchmark_only and not self.is_multivariate:
if self.partitioner is not None:
self.sets = self.partitioner.sets
else:
raise Exception("Fuzzy sets were not provided for the model. Use 'sets' parameter or 'partitioner'. ")
if 'order' in kwargs:
self.order = kwargs.pop('order')
dump = kwargs.get('dump', None)
num_batches = kwargs.get('num_batches', None)
save = kwargs.get('save_model', False) # save model on disk
batch_save = kwargs.get('batch_save', False) #save model between batches
file_path = kwargs.get('file_path', None)
distributed = kwargs.get('distributed', False)
batch_save_interval = kwargs.get('batch_save_interval', 10)
if distributed:
nodes = kwargs.get('nodes', False)
train_method = kwargs.get('train_method', Util.simple_model_train)
Util.distributed_train(self, train_method, nodes, type(self), data, num_batches, {},
batch_save=batch_save, file_path=file_path,
batch_save_interval=batch_save_interval)
else:
if dump == 'time':
print("[{0: %H:%M:%S}] Start training".format(datetime.datetime.now()))
if num_batches is not None:
n = len(data)
batch_size = int(n / num_batches)
bcount = 1
rng = range(self.order, n, batch_size)
if dump == 'tqdm':
from tqdm import tqdm
rng = tqdm(rng)
for ct in rng:
if dump == 'time':
print("[{0: %H:%M:%S}] Starting batch ".format(datetime.datetime.now()) + str(bcount))
if self.is_multivariate:
mdata = data.iloc[ct - self.order:ct + batch_size]
else:
mdata = data[ct - self.order : ct + batch_size]
self.train(mdata, **kwargs)
if batch_save:
Util.persist_obj(self,file_path)
if dump == 'time':
print("[{0: %H:%M:%S}] Finish batch ".format(datetime.datetime.now()) + str(bcount))
bcount += 1
else:
self.train(data, **kwargs)
if dump == 'time':
print("[{0: %H:%M:%S}] Finish training".format(datetime.datetime.now()))
if save:
Util.persist_obj(self, file_path)
[docs] def clone_parameters(self, model):
"""
Import the parameters values from other model
:param model:
"""
self.order = model.order
self.shortname = model.shortname
self.name = model.name
self.detail = model.detail
self.is_high_order = model.is_high_order
self.min_order = model.min_order
self.has_seasonality = model.has_seasonality
self.has_point_forecasting = model.has_point_forecasting
self.has_interval_forecasting = model.has_interval_forecasting
self.has_probability_forecasting = model.has_probability_forecasting
self.is_multivariate = model.is_multivariate
self.dump = model.dump
self.transformations = model.transformations
self.transformations_param = model.transformations_param
self.original_max = model.original_max
self.original_min = model.original_min
self.partitioner = model.partitioner
self.sets = model.sets
self.auto_update = model.auto_update
self.benchmark_only = model.benchmark_only
self.indexer = model.indexer
[docs] def merge(self, model):
"""
Merge the FLRG rules from other model
:param model: source model
:return:
"""
for key in model.flrgs.keys():
flrg = model.flrgs[key]
if flrg.get_key() not in self.flrgs:
self.flrgs[flrg.get_key()] = flrg
else:
if isinstance(flrg.RHS, (list, set)):
for k in flrg.RHS:
self.flrgs[flrg.get_key()].append_rhs(k)
elif isinstance(flrg.RHS, dict):
for k in flrg.RHS.keys():
self.flrgs[flrg.get_key()].append_rhs(flrg.RHS[k])
else:
self.flrgs[flrg.get_key()].append_rhs(flrg.RHS)
[docs] def get_UoD(self):
return [self.original_min, self.original_max]
def __str__(self):
"""String representation of the model"""
tmp = self.name + ":\n"
for r in sorted(self.flrgs, key=lambda key: self.flrgs[key].get_midpoint(self.sets)):
tmp = tmp + str(self.flrgs[r]) + "\n"
return tmp
def __len__(self):
"""
The length (number of rules) of the model
:return: number of rules
"""
return len(self.flrgs)
[docs] def len_total(self):
return sum([len(k) for k in self.flrgs])
