import asyncio import operator from collections import OrderedDict import numpy as np import pandas as pd import toolz from ..collection import Streaming, _stream_types, OperatorMixin from ..sources import Source from ..utils import M from . import aggregations from .utils import is_dataframe_like, is_series_like, is_index_like, \ get_base_frame_type, get_dataframe_package class BaseFrame(Streaming): def round(self, decimals=0): """ Round elements in frame """ return self.map_partitions(M.round, self, decimals=decimals) def reset_index(self): """ Reset Index """ return self.map_partitions(M.reset_index, self) def set_index(self, index, **kwargs): """ Set Index """ return self.map_partitions(M.set_index, self, index, **kwargs) def tail(self, n=5): """ Round elements in frame """ return self.map_partitions(M.tail, self, n=n) def astype(self, dt): return self.map_partitions(M.astype, self, dt) @property def index(self): return self.map_partitions(lambda x: x.index, self) def map(self, func, na_action=None): return self.map_partitions(self._subtype.map, self, func, na_action=na_action) class Frame(BaseFrame): _stream_type = 'streaming' def groupby(self, other): """ Groupby aggregations """ return GroupBy(self, other) def aggregate(self, aggregation, start=None): return self.accumulate_partitions(aggregations.accumulator, agg=aggregation, start=start, stream_type='updating', returns_state=True) def sum(self, start=None): """ Sum frame. Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self.aggregate(aggregations.Sum(), start) def count(self, start=None): """ Count of frame Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self.aggregate(aggregations.Count(), start) @property def size(self): """ size of frame """ return self.aggregate(aggregations.Size()) def mean(self, start=None): """ Average frame Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self.aggregate(aggregations.Mean(), start) def rolling(self, window, min_periods=1, with_state=False, start=()): """ Compute rolling aggregations When followed by an aggregation method like ``sum``, ``mean``, or ``std`` this produces a new Streaming dataframe whose values are aggregated over that window. The window parameter can be either a number of rows or a timedelta like ``"2 minutes"` in which case the index should be a datetime index. This operates by keeping enough of a backlog of records to maintain an accurate stream. It performs a copy at every added dataframe. Because of this it may be slow if the rolling window is much larger than the average stream element. Parameters ---------- window: int or timedelta Window over which to roll with_state: bool (False) Whether to return the state along with the result as a tuple (state, result). State may be needed downstream for a number of reasons like checkpointing. start: () or resulting Python object type from the operation Accepts a valid start state. Returns ------- Rolling object See Also -------- DataFrame.window: more generic window operations """ return Rolling(self, window, min_periods, with_state, start) def window(self, n=None, value=None, with_state=False, start=None): """ Sliding window operations Windowed operations are defined over a sliding window of data, either with a fixed number of elements:: >>> df.window(n=10).sum() # sum of the last ten elements or over an index value range (index must be monotonic):: >>> df.window(value='2h').mean() # average over the last two hours Windowed dataframes support all normal arithmetic, aggregations, and groupby-aggregations. Parameters ---------- n: int Window of number of elements over which to roll value: str Window of time over which to roll with_state: bool (False) Whether to return the state along with the result as a tuple (state, result). State may be needed downstream for a number of reasons like checkpointing. start: None or resulting Python object type from the operation Accepts a valid start state. Examples -------- >>> df.window(n=10).std() >>> df.window(value='2h').count() >>> w = df.window(n=100) >>> w.groupby(w.name).amount.sum() >>> w.groupby(w.x % 10).y.var() See Also -------- DataFrame.rolling: mimic's Pandas rolling aggregations """ return Window(self, n=n, value=value, with_state=with_state, start=start) def expanding(self, with_state=False, start=None): return Expanding(self, n=1, with_state=with_state, start=start) def ewm(self, com=None, span=None, halflife=None, alpha=None, with_state=False, start=None): return EWM(self, n=1, com=com, span=span, halflife=halflife, alpha=alpha, with_state=with_state, start=start) def _cumulative_aggregation(self, op): return self.accumulate_partitions(_cumulative_accumulator, returns_state=True, start=(), op=op) def cumsum(self): """ Cumulative sum """ return self._cumulative_aggregation(op='cumsum') def cumprod(self): """ Cumulative product """ return self._cumulative_aggregation(op='cumprod') def cummin(self): """ Cumulative minimum """ return self._cumulative_aggregation(op='cummin') def cummax(self): """ Cumulative maximum """ return self._cumulative_aggregation(op='cummax') class Frames(BaseFrame): _stream_type = 'updating' def sum(self, **kwargs): return self.map_partitions(M.sum, self, **kwargs) def mean(self, **kwargs): return self.map_partitions(M.mean, self, **kwargs) def std(self, **kwargs): return self.map_partitions(M.std, self, **kwargs) def var(self, **kwargs): return self.map_partitions(M.var, self, **kwargs) @property def size(self, **kwargs): return self.map_partitions(M.size, self, **kwargs) def count(self, **kwargs): return self.map_partitions(M.count, self, **kwargs) def nlargest(self, n, *args, **kwargs): return self.map_partitions(M.nlargest, self, n, *args, **kwargs) def tail(self, n=5): """ Round elements in frame """ return self.map_partitions(M.tail, self, n=n) class _DataFrameMixin(object): @property def columns(self): return self.example.columns @property def dtypes(self): return self.example.dtypes def __getitem__(self, index): return self.map_partitions(operator.getitem, self, index) def __getattr__(self, key): if key in self.columns or not len(self.columns): return self.map_partitions(getattr, self, key) else: raise AttributeError("DataFrame has no attribute %r" % key) def __dir__(self): o = set(dir(type(self))) o.update(self.__dict__) o.update(c for c in self.columns if (isinstance(c, str) and c.isidentifier())) return list(o) def assign(self, **kwargs): """ Assign new columns to this dataframe Alternatively use setitem syntax Examples -------- >>> sdf = sdf.assign(z=sdf.x + sdf.y) # doctest: +SKIP >>> sdf['z'] = sdf.x + sdf.y # doctest: +SKIP """ kvs = list(toolz.concat(kwargs.items())) def _assign(df, *kvs): keys = kvs[::2] values = kvs[1::2] kwargs = OrderedDict(zip(keys, values)) return df.assign(**kwargs) return self.map_partitions(_assign, self, *kvs) def to_frame(self): """ Convert to a streaming dataframe """ return self def __setitem__(self, key, value): if isinstance(value, Series): result = self.assign(**{key: value}) elif isinstance(value, DataFrame): result = self.assign(**{k: value[c] for k, c in zip(key, value.columns)}) else: example = self.example.copy() example[key] = value df_type = type(self.example) result = self.map_partitions(df_type.assign, self, **{key: value}) self.stream = result.stream self.example = result.example return self def query(self, expr, **kwargs): df_type = type(self.example) return self.map_partitions(df_type.query, self, expr, **kwargs) class DataFrame(Frame, _DataFrameMixin): """ A Streaming Dataframe This is a logical collection over a stream of Pandas dataframes. Operations on this object will translate to the appropriate operations on the underlying Pandas dataframes. See Also -------- Series """ def __init__(self, *args, **kwargs): # {'x': sdf.x + 1, 'y': sdf.y - 1} if len(args) == 1 and not kwargs and isinstance(args[0], dict): def concat(tup, module=None, columns=None): result = module.concat(tup, axis=1) result.columns = columns return result columns, values = zip(*args[0].items()) base_frame_type = values[0]._subtype df_package = get_dataframe_package(base_frame_type) stream = type(values[0].stream).zip(*[v.stream for v in values]) stream = stream.map(concat, module=df_package, columns=list(columns)) example = df_package.DataFrame({k: getattr(v, 'example', v) for k, v in args[0].items()}) DataFrame.__init__(self, stream, example) else: example = None if "example" in kwargs: example = kwargs.get('example') elif len(args) > 1: example = args[1] if callable(example): example = example() kwargs["example"] = example self._subtype = get_base_frame_type(self.__class__.__name__, is_dataframe_like, example) super(DataFrame, self).__init__(*args, **kwargs) def verify(self, x): """ Verify consistency of elements that pass through this stream """ super(DataFrame, self).verify(x) if list(x.columns) != list(self.example.columns): raise IndexError("Input expected to have columns %s, got %s" % (self.example.columns, x.columns)) @property def plot(self): try: # import has side-effect of attaching .hvplot attribute import hvplot.streamz # # noqa: F401 except ImportError as err: # pragma: no cover raise ImportError("Streamz dataframe plotting requires hvplot") from err return self.hvplot class _SeriesMixin(object): @property def dtype(self): return self.example.dtype def to_frame(self): """ Convert to a streaming dataframe """ return self.map_partitions(M.to_frame, self) class Series(Frame, _SeriesMixin): """ A Streaming Series This is a logical collection over a stream of Pandas series objects. Operations on this object will translate to the appropriate operations on the underlying Pandas series. See Also -------- DataFrame """ def __init__(self, *args, **kwargs): example = None if "example" in kwargs: example = kwargs.get('example') elif len(args) > 1: example = args[1] if isinstance(self, Index): self._subtype = get_base_frame_type(self.__class__.__name__, is_index_like, example) else: self._subtype = get_base_frame_type(self.__class__.__name__, is_series_like, example) super(Series, self).__init__(*args, **kwargs) def value_counts(self): return self.accumulate_partitions(aggregations.accumulator, agg=aggregations.ValueCounts(), start=None, stream_type='updating', returns_state=True) class Index(Series): pass class DataFrames(Frames, _DataFrameMixin): pass class Seriess(Frames, _SeriesMixin): pass def _cumulative_accumulator(state, new, op=None): if not len(new): return state, new if not len(state): df = new else: df_package = get_dataframe_package(new) df = df_package.concat([state, new]) # ouch, full copy result = getattr(df, op)() new_state = result.iloc[-1:] if len(state): result = result[1:] return new_state, result class Rolling(object): """ Rolling aggregations This intermediate class enables rolling aggregations across either a fixed number of rows or a time window. Examples -------- >>> sdf.rolling(10).x.mean() # doctest: +SKIP >>> sdf.rolling('100ms').x.mean() # doctest: +SKIP """ def __init__(self, sdf, window, min_periods, with_state, start): self.root = sdf if not isinstance(window, int): window = pd.Timedelta(window) min_periods = 1 self.window = window self.min_periods = min_periods self.with_state = with_state self.start = start def __getitem__(self, key): sdf = self.root[key] return Rolling(sdf, self.window, self.min_periods, self.with_state, self.start) def __getattr__(self, key): if key in self.root.columns or not len(self.root.columns): return self[key] else: raise AttributeError("Rolling has no attribute %r" % key) def _known_aggregation(self, op, *args, **kwargs): return self.root.accumulate_partitions(rolling_accumulator, window=self.window, op=op, args=args, kwargs=kwargs, start=self.start, returns_state=True, with_state=self.with_state) def sum(self): """ Rolling sum """ return self._known_aggregation('sum') def mean(self): """ Rolling mean """ return self._known_aggregation('mean') def min(self): """ Rolling minimum """ return self._known_aggregation('min') def max(self): """ Rolling maximum """ return self._known_aggregation('max') def median(self): """ Rolling median """ return self._known_aggregation('median') def std(self, *args, **kwargs): """ Rolling standard deviation """ return self._known_aggregation('std', *args, **kwargs) def var(self, *args, **kwargs): """ Rolling variance """ return self._known_aggregation('var', *args, **kwargs) def count(self, *args, **kwargs): """ Rolling count """ return self._known_aggregation('count', *args, **kwargs) def aggregate(self, *args, **kwargs): """ Rolling aggregation """ return self._known_aggregation('aggregate', *args, **kwargs) def quantile(self, *args, **kwargs): """ Rolling quantile """ return self._known_aggregation('quantile', *args, **kwargs) class Window(OperatorMixin): """ Windowed aggregations This provides a set of aggregations that can be applied over a sliding window of data. See Also -------- DataFrame.window: contains full docstring """ def __init__(self, sdf, n=None, value=None, with_state=False, start=None): if value is None and isinstance(n, (str, pd.Timedelta)): value = n n = None self.n = n self.root = sdf if isinstance(value, str) and isinstance(self.root.example.index, pd.DatetimeIndex): value = pd.Timedelta(value) self.value = value self.with_state = with_state self.start = start def __getitem__(self, key): sdf = self.root[key] return type(self)( sdf, n=self.n, value=self.value, with_state=self.with_state, start=self.start ) def __getattr__(self, key): if key in self.root.columns or not len(self.root.columns): return self[key] else: raise AttributeError(f"{type(self)} has no attribute {key}") def map_partitions(self, func, *args, **kwargs): args2 = [a.root if isinstance(a, type(self)) else a for a in args] root = self.root.map_partitions(func, *args2, **kwargs) return type(self)( root, n=self.n, value=self.value, with_state=self.with_state, start=self.start ) @property def index(self): return self.map_partitions(lambda x: x.index, self) @property def columns(self): return self.root.columns @property def dtypes(self): return self.root.dtypes @property def example(self): return self.root.example def reset_index(self): return type(self)(self.root.reset_index(), n=self.n, value=self.value) def aggregate(self, agg): if self.n is not None: diff = aggregations.diff_iloc window = self.n elif self.value is not None: diff = aggregations.diff_loc window = self.value return self.root.accumulate_partitions(aggregations.window_accumulator, diff=diff, window=window, agg=agg, start=self.start, returns_state=True, stream_type='updating', with_state=self.with_state) def full(self): return self.aggregate(aggregations.Full()) def apply(self, func): """ Apply an arbitrary function over each window of data """ result = self.aggregate(aggregations.Full()) return result.map_partitions(func, result) def sum(self): """ Sum elements within window """ return self.aggregate(aggregations.Sum()) def count(self): """ Count elements within window """ return self.aggregate(aggregations.Count()) def mean(self): """ Average elements within window """ return self.aggregate(aggregations.Mean()) def var(self, ddof=1): """ Compute variance of elements within window """ return self.aggregate(aggregations.Var(ddof=ddof)) def std(self, ddof=1): """ Compute standard deviation of elements within window """ return self.var(ddof=ddof) ** 0.5 @property def size(self): """ Number of elements within window """ return self.aggregate(aggregations.Size()) def value_counts(self): """ Count groups of elements within window """ return self.aggregate(aggregations.ValueCounts()) def groupby(self, other): """ Groupby-aggregations within window """ return WindowedGroupBy(self.root, other, None, self.n, self.value, self.with_state, self.start) class Expanding(Window): def aggregate(self, agg): window = self.n diff = aggregations.diff_expanding return self.root.accumulate_partitions(aggregations.window_accumulator, diff=diff, window=window, agg=agg, start=self.start, returns_state=True, stream_type='updating', with_state=self.with_state) def groupby(self, other): raise NotImplementedError class EWM(Expanding): def __init__( self, sdf, n=1, value=None, with_state=False, start=None, com=None, span=None, halflife=None, alpha=None ): super().__init__(sdf, n=n, value=value, with_state=with_state, start=start) self._com = self._get_com(com, span, halflife, alpha) self.com = com self.span = span self.alpha = alpha self.halflife = halflife def __getitem__(self, key): sdf = self.root[key] return type(self)( sdf, n=self.n, value=self.value, with_state=self.with_state, start=self.start, com=self.com, span=self.span, halflife=self.halflife, alpha=self.alpha ) @staticmethod def _get_com(com, span, halflife, alpha): if sum(var is not None for var in (com, span, halflife, alpha)) > 1: raise ValueError("Can only provide one of `com`, `span`, `halflife`, `alpha`.") # Convert to center of mass; domain checks ensure 0 < alpha <= 1 if com is not None: if com < 0: raise ValueError("com must satisfy: comass >= 0") elif span is not None: if span < 1: raise ValueError("span must satisfy: span >= 1") com = (span - 1) / 2 elif halflife is not None: if halflife <= 0: raise ValueError("halflife must satisfy: halflife > 0") decay = 1 - np.exp(np.log(0.5) / halflife) com = 1 / decay - 1 elif alpha is not None: if alpha <= 0 or alpha > 1: raise ValueError("alpha must satisfy: 0 < alpha <= 1") com = (1 - alpha) / alpha else: raise ValueError("Must pass one of com, span, halflife, or alpha") return float(com) def full(self): raise NotImplementedError def apply(self, func): """ Apply an arbitrary function over each window of data """ raise NotImplementedError def sum(self): """ Sum elements within window """ raise NotImplementedError def count(self): """ Count elements within window """ raise NotImplementedError def mean(self): """ Average elements within window """ return self.aggregate(aggregations.EWMean(self._com)) def var(self, ddof=1): """ Compute variance of elements within window """ raise NotImplementedError def std(self, ddof=1): """ Compute standard deviation of elements within window """ raise NotImplementedError @property def size(self): """ Number of elements within window """ raise NotImplementedError def value_counts(self): """ Count groups of elements within window """ raise NotImplementedError def rolling_accumulator(acc, new, window=None, op=None, with_state=False, args=(), kwargs={}): if len(acc): df_package = get_dataframe_package(new) df = df_package.concat([acc, new]) else: df = new result = getattr(df.rolling(window), op)(*args, **kwargs) if isinstance(window, int): new_acc = df.iloc[-window:] else: new_acc = df.loc[result.index.max() - window:] result = result.iloc[len(acc):] return new_acc, result def _accumulate_mean(accumulator, new): accumulator = accumulator.copy() accumulator['sums'] += new.sum() accumulator['counts'] += new.count() result = accumulator['sums'] / accumulator['counts'] return accumulator, result def _accumulate_sum(accumulator, new): return accumulator + new.sum() def _accumulate_size(accumulator, new): return accumulator + new.size() class GroupBy(object): """ Groupby aggregations on streaming dataframes """ def __init__(self, root, grouper, index=None): self.root = root self.grouper = grouper self.index = index def __getitem__(self, index): return GroupBy(self.root, self.grouper, index) def __getattr__(self, key): if key in self.root.columns or not len(self.root.columns): return self[key] else: raise AttributeError("GroupBy has no attribute %r" % key) def _accumulate(self, Agg, with_state=False, start=None, **kwargs): stream_type = 'updating' if isinstance(self.grouper, Streaming): stream = self.root.stream.zip(self.grouper.stream) grouper_example = self.grouper.example agg = Agg(self.index, grouper=None, **kwargs) else: stream = self.root.stream grouper_example = self.grouper agg = Agg(self.index, grouper=self.grouper, **kwargs) # Compute example state = agg.initial(self.root.example, grouper=grouper_example) if hasattr(grouper_example, 'iloc'): grouper_example = grouper_example.iloc[:0] elif isinstance(grouper_example, np.ndarray) or is_index_like(grouper_example): grouper_example = grouper_example[:0] _, example = agg.on_new(state, self.root.example.iloc[:0], grouper=grouper_example) outstream = stream.accumulate(aggregations.groupby_accumulator, agg=agg, start=start, returns_state=True, with_state=with_state) for fn, s_type in _stream_types[stream_type]: """Function checks if example is of a specific frame type""" if fn(example): return s_type(outstream, example) return Streaming(outstream, example, stream_type=stream_type) def count(self, start=None): """ Groupby-count Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self._accumulate(aggregations.GroupbyCount, start=start) def mean(self, with_state=False, start=None): """ Groupby-mean Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self._accumulate(aggregations.GroupbyMean, with_state=with_state, start=start) def size(self): """ Groupby-size """ return self._accumulate(aggregations.GroupbySize) def std(self, ddof=1): """ Groupby-std """ return self.var(ddof=ddof) ** 0.5 def sum(self, start=None): """ Groupby-sum Parameters ---------- start: None or resulting Python object type from the operation Accepts a valid start state. """ return self._accumulate(aggregations.GroupbySum, start=start) def var(self, ddof=1): """ Groupby-variance """ return self._accumulate(aggregations.GroupbyVar, ddof=ddof) class WindowedGroupBy(GroupBy): """ Groupby aggregations over a window of data """ def __init__(self, root, grouper, index=None, n=None, value=None, with_state=False, start=None): self.root = root self.grouper = grouper self.index = index self.n = n if isinstance(value, str) and isinstance(self.root.example.index, pd.DatetimeIndex): value = pd.Timedelta(value) self.value = value self.with_state = with_state self.start = start def __getitem__(self, index): return WindowedGroupBy(self.root, self.grouper, index, self.n, self.value, self.with_state, self.start) def _accumulate(self, Agg, **kwargs): stream_type = 'updating' if isinstance(self.grouper, Streaming): stream = self.root.stream.zip(self.grouper.stream) grouper_example = self.grouper.example agg = Agg(self.index, grouper=None, **kwargs) elif isinstance(self.grouper, Window): stream = self.root.stream.zip(self.grouper.root.stream) grouper_example = self.grouper.root.example agg = Agg(self.index, grouper=None, **kwargs) else: stream = self.root.stream grouper_example = self.grouper agg = Agg(self.index, grouper=self.grouper, **kwargs) # Compute example state = agg.initial(self.root.example, grouper=grouper_example) if hasattr(grouper_example, 'iloc'): grouper_example = grouper_example.iloc[:0] elif isinstance(grouper_example, np.ndarray) or is_index_like(grouper_example): grouper_example = grouper_example[:0] _, example = agg.on_new(state, self.root.example.iloc[:0], grouper=grouper_example) if self.n is not None: diff = aggregations.diff_iloc window = self.n elif self.value is not None: diff = aggregations.diff_loc window = self.value outstream = stream.accumulate(aggregations.windowed_groupby_accumulator, agg=agg, start=self.start, returns_state=True, diff=diff, window=window, with_state=self.with_state) for fn, s_type in _stream_types[stream_type]: """Function checks if example is of a specific frame type""" if fn(example): return s_type(outstream, example) return Streaming(outstream, example, stream_type=stream_type) def random_datapoint(now=None, **kwargs): """Example of querying a single current value""" if now is None: now = pd.Timestamp.now() return pd.DataFrame( {'a': np.random.random(1)}, index=[now]) def random_datablock(last, now, **kwargs): """ Example of querying over a time range since last update Parameters ---------- last: pd.Timestamp Time of previous call to this function. now: pd.Timestamp Current time. freq: pd.Timedelta, optional The time interval between individual records to be returned. For good throughput, should be much smaller than the interval at which this function is called. Returns a pd.DataFrame with random values where: The x column is uniformly distributed. The y column is Poisson distributed. The z column is normally distributed. """ freq = kwargs.get("freq", pd.Timedelta("100ms")) index = pd.date_range(start=last + freq, end=now, freq=freq) df = pd.DataFrame({'x': np.random.random(len(index)), 'y': np.random.poisson(size=len(index)), 'z': np.random.normal(0, 1, size=len(index))}, index=index) return df @DataFrame.register_api(staticmethod, "from_periodic") class PeriodicDataFrame(DataFrame): """A streaming dataframe using the asyncio ioloop to poll a callback fn Parameters ---------- datafn: callable Callback function accepting **kwargs and returning a pd.DataFrame. kwargs will include at least 'last' (pd.Timestamp.now() when datafn was last invoked), and 'now' (current pd.Timestamp.now()). interval: timedelta The time interval between new dataframes. dask: boolean If true, uses a DaskStream instead of a regular Source. **kwargs: Optional keyword arguments to be passed into the callback function. By default, returns a three-column random pd.DataFrame generated by the 'random_datablock' function. Example ------- >>> df = PeriodicDataFrame(interval='1s', datafn=random_datapoint) # doctest: +SKIP """ def __init__(self, datafn=random_datablock, interval='500ms', dask=False, start=True, **kwargs): if dask: from streamz.dask import DaskStream source = DaskStream() else: source = Source() self.loop = source.loop self.interval = pd.Timedelta(interval).total_seconds() self.source = source self.continue_ = [False] # like the oppose of self.stopped self.kwargs = kwargs stream = self.source.map(lambda x: datafn(**x, **kwargs)) example = datafn(last=pd.Timestamp.now(), now=pd.Timestamp.now(), **kwargs) super(PeriodicDataFrame, self).__init__(stream, example) if start: self.start() def start(self): if not self.continue_[0]: self.continue_[0] = True self.loop.add_callback(self._cb, self.interval, self.source, self.continue_) def __del__(self): self.stop() def stop(self): self.continue_[0] = False @staticmethod async def _cb(interval, source, continue_): last = pd.Timestamp.now() while continue_[0]: await asyncio.sleep(interval) now = pd.Timestamp.now() await asyncio.gather(*source._emit(dict(last=last, now=now))) last = now @DataFrame.register_api(staticmethod, "random") class Random(PeriodicDataFrame): """PeriodicDataFrame providing random values by default Accepts same parameters as PeriodicDataFrame, plus `freq`, a string that will be converted to a pd.Timedelta and passed to the 'datafn'. Useful mainly for examples and docs. Example ------- >>> source = Random(freq='100ms', interval='1s') # doctest: +SKIP """ def __init__(self, freq='100ms', interval='500ms', dask=False, start=True, datafn=random_datablock): super(Random, self).__init__(datafn, interval, dask, start, freq=pd.Timedelta(freq)) _stream_types['streaming'].append((is_dataframe_like, DataFrame)) _stream_types['streaming'].append((is_index_like, Index)) _stream_types['streaming'].append((is_series_like, Series)) _stream_types['updating'].append((is_dataframe_like, DataFrames)) _stream_types['updating'].append((is_series_like, Seriess))