Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def moving_average(self, sampling_period,
- window_size=None,
- start=None, end=None,
- placement='center',
- pandas=False):
- """Averaging over regular intervals
- """
- start, end, mask = self._check_boundaries(start, end)
- # default to sampling_period if not given
- if window_size is None:
- window_size = sampling_period
- sampling_period = \
- self._check_regularization(start, end, sampling_period)
- # convert to datetime if the times are datetimes
- half_window = float(window_size) / 2
- full_window = float(window_size)
- if isinstance(start, datetime.datetime):
- half_window = datetime.timedelta(seconds=half_window)
- full_window = datetime.timedelta(seconds=full_window)
- result = []
- current_time = start
- while current_time <= end:
- if placement == 'center':
- window_start = current_time - half_window
- window_end = current_time + half_window
- elif placement == 'left':
- window_start = current_time
- window_end = current_time + full_window
- elif placement == 'right':
- window_start = current_time - full_window
- window_end = current_time
- else:
- msg = 'unknown placement "{}"'.format(placement)
- raise ValueError(msg)
- # calculate mean over window and add (t, v) tuple to list
- mean = self.mean(window_start, window_end)
- result.append((current_time, mean))
- current_time += sampling_period
- # convert to pandas Series if pandas=True
- if pandas:
- try:
- import pandas as pd
- except ImportError:
- msg = "can't have pandas=True if pandas is not installed"
- raise ImportError(msg)
- result = pd.Series(
- [v for t, v in result],
- index=[t for t, v in result],
- )
- return result
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement