Skip to content

contango.stream.historical_stream

contango.stream.historical_stream

HistoricalStream 

HistoricalStream(source: Stream[R], window: int)

Bases: Stream[R], Generic[R]

Wraps any stream and records a window of its past values.

Attributes:

Name Type Description
R

The return type for the historical stream.

Initializes HistoricalStream.

Parameters:

Name Type Description Default
source Stream[R]

The stream source to wrap.

 required
window int

The maximum amount of lookback for the history of the stream.

 required
Source code in src\contango\stream\historical_stream.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, source: Stream[R], window: int) -> None:
    """
    Initializes `HistoricalStream`.

    Args:
        source: The stream source to wrap.
        window: The maximum amount of lookback for the history of the stream.
    """
    self._source = source
    self._history: Deque[R] = deque(maxlen=window)

value property 

value: R

Returns the most recent value of the historical stream.

previous 

previous(n: int = 1) -> R | None

Returns the previous n amount of values backwards. For example, n=1 will be the previous value relative to the current one.

Parameters:

Name Type Description Default
n int

The amount of lookback.

 1

Returns:

Name Type Description
R R | None

The stream value at n amount of lookback.
None R | None

If the history deque is shorter than the amount of lookback.
Source code in src\contango\stream\historical_stream.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def previous(self, n: int = 1) -> R | None:
    """
    Returns the previous n amount of values backwards.
    For example, n=1 will be the previous value relative to the current one.

    Args:
        n: The amount of lookback.

    Returns:
        R: The stream value at n amount of lookback.
        None: If the history deque is shorter than the amount of lookback.
    """
    idx = -1 - n
    if len(self._history) <= n:
        return None
    return self._history[idx]

update 

update(event: MarketDataEvent) -> R

Updates the state of the historical stream given a market data event.

Source code in src\contango\stream\historical_stream.py 
32
33
34
35
36
37
38
def update(self, event: MarketDataEvent) -> R:
    """
    Updates the state of the historical stream given a market data event.
    """
    self._source.update(event)
    self._history.append(self._source.value)
    return self.value