Skip to content

FrameSet


FrameSet

Group of Ticks.

A FrameSet group Ticks into Frames.

Attributes:

Name Type Description
unit estrade.enums.Unit

unit used to group ticks.

unit_quantity int

quantity of unit

max_frames_in_memory int

number of frames to keep in memory

epic estrade.epic.Epic

epic updating this instance

frames List[estrade.graph.frame_set.Frame]

List of Frame

indicators Dict[str, estrade.graph.base_indicator.BaseIndicator]

Dict of indicators calculated on every Frame with their ref as index.

ref str

reference of this instance (see estrade.mixins.ref.RefMixin)

current: Optional[estrade.graph.frame_set.Frame] property readonly

Return the current Frame.

ref: str inherited property writable

Return ref of current instance.

Returns:

Type Description
str

reference of current instance.

__init__(self, unit, unit_quantity, ref=None, max_frames_in_memory=100) special

Instanciate a new FrameSet.

Parameters:

Name Type Description Default
unit Unit

unit used to group ticks.

required
unit_quantity int

quantity of unit

required
max_frames_in_memory int

number of frames to keep in memory

100
ref Optional[str]

reference of this instance

None
Source code in estrade/graph/frame_set.py
def __init__(
    self,
    unit: Unit,
    unit_quantity: int,
    ref: Optional[str] = None,
    max_frames_in_memory: int = 100,
) -> None:
    """
    Instanciate a new [`FrameSet`][estrade.graph.frame_set.FrameSet].

    Arguments:
        unit: unit used to group ticks.
        unit_quantity: quantity of unit
        max_frames_in_memory: number of frames to keep in memory
        ref: reference of this instance
    """
    TimeframeMixin.__init__(self, unit, unit_quantity)
    RefMixin.__init__(self, ref)

    self._is_timeframe_over_func = self.is_frame_over_timed
    self._create_new_frame_func = self._create_new_frame_timed
    if self.unit == Unit.TICK:
        self._is_timeframe_over_func = self.is_frame_over_tick
        self._create_new_frame_func = self._create_new_frame_tick

    self.epic: Optional["Epic"] = None
    self.frames: List[Frame] = []
    self.indicators: Dict[str, "BaseIndicator"] = {}
    self.max_frames_in_memory = max_frames_in_memory

get_frame_end(self, frame_start) inherited

Get the timeframe end from a input datetime representing a timeframe start.

Parameters:

Name Type Description Default
frame_start Arrow

datetime to find the timeframe end from.

required

Returns:

Type Description
Arrow

End of the timeframe.

Source code in estrade/graph/frame_set.py
def get_frame_end(self, frame_start: "Arrow") -> "Arrow":
    """
    Get the timeframe end from a input datetime representing a timeframe start.

    Arguments:
        frame_start: datetime to find the timeframe end from.

    Returns:
        End of the timeframe.
    """
    if not self._timedelta:
        return frame_start
    frame_end = frame_start.shift(**self._timedelta)
    return frame_end

get_frame_start(self, dt) inherited

Get the timeframe start from a input datetime.

Parameters:

Name Type Description Default
dt Arrow

datetime to find the timeframe start from.

required

Returns:

Type Description
Arrow

Start of the timeframe.

Source code in estrade/graph/frame_set.py
def get_frame_start(self, dt: "Arrow") -> "Arrow":
    """
    Get the timeframe start from a input datetime.

    Arguments:
        dt: datetime to find the timeframe start from.

    Returns:
        Start of the timeframe.
    """
    timeframe_start = self._get_frame_start_func(dt)
    return timeframe_start

on_new_tick(self, tick)

Update the current instance with a new Tick.

This method will either:

  • add the received tick to the current instance.
  • create a new Frame if the current frame is over.

Parameters:

Name Type Description Default
tick Tick

new tick to update the current instance with.

required
Source code in estrade/graph/frame_set.py
def on_new_tick(self, tick: "Tick") -> None:
    """
    Update the current instance with a new [`Tick`][estrade.tick.Tick].

    This method will either:

     - add the received tick to the current instance.
     - create a new [`Frame`][estrade.graph.frame_set.Frame] if the current
            frame is over.

    Arguments:
        tick: new tick to update the current instance with.

    """
    if self._is_timeframe_over_func(tick):
        self._create_new_frame_func(tick)
    else:
        self.add_tick_to_last(tick)

Frame

A Frame is a group Ticks.

Attributes:

Name Type Description
parent_frameset estrade.graph.frame_set.FrameSet

parent FrameSet

first_tick estrade.tick.Tick

first tick registered on this instance.

high_tick estrade.tick.Tick

highest tick registered on this instance.

low_tick estrade.tick.Tick

lowest tick registered on this instance.

last_tick estrade.tick.Tick

last tick registered on this instance.

period_start arrow.Arrow

open datetime of this instance.

period_end arrow.Arrow

scheduled end datetime of this instance.

previous_frame Optional[estrade.graph.frame_set.Frame]

previous frame.

next_frame Optional[estrade.graph.frame_set.Frame]

next frame.

nb_ticks int

Number of ticks received by this instance.

indicators (Dict[str, Optional[estrade.graph.base_indicator.BaseIndicatorValue]]

Indicators values indexed by their reference.

closed: bool property readonly

Check if the current instance is closed.

An instance is considered as closed when its next_frame value is set.

__init__(self, parent_frameset, first_tick, period_start, period_end=None, previous_frame=None, empty=False) special

Create a new Frame.

Parameters:

Name Type Description Default
parent_frameset FrameSet

parent FrameSet

required
first_tick Tick

first tick registered on this instance.

required
period_start Arrow

open datetime of this instance.

required
period_end Optional[Arrow]

scheduled end datetime of this instance.

None
previous_frame Optional[Frame]

previous frame.

None
Source code in estrade/graph/frame_set.py
def __init__(
    self,
    parent_frameset: "FrameSet",
    first_tick: "Tick",
    period_start: "Arrow",
    period_end: Optional["Arrow"] = None,
    previous_frame: Optional["Frame"] = None,
    empty: bool = False,
) -> None:
    """
    Create a new Frame.

    Arguments:
        parent_frameset: parent FrameSet
        first_tick: first tick registered on this instance.
        period_start: open datetime of this instance.
        period_end: scheduled end datetime of this instance.
        previous_frame: previous frame.
    """
    self.parent_frameset = parent_frameset
    self.first_tick = first_tick
    self.period_start = period_start
    self.period_end = period_end
    self.previous_frame = previous_frame
    self.next_frame: Optional["Frame"] = None
    self.empty = empty

    self.nb_ticks = 1
    self.last_tick = first_tick
    self.high_tick = first_tick
    self.low_tick = first_tick
    self.indicators: Dict[str, Optional["BaseIndicatorValue"]] = {}

    self._upsert_indicators(first_tick)

on_new_tick(self, tick)

Update the current instance with a new Tick.

  • Update nb_ticks
  • Update last_tick, high_tick, low_tick
  • Call each indicator on_new_tick method.

Parameters:

Name Type Description Default
tick Tick

new received Tick.

required
Source code in estrade/graph/frame_set.py
def on_new_tick(self, tick: "Tick"):
    """
    Update the current instance with a new [`Tick`][estrade.tick.Tick].

     - Update `nb_ticks`
     - Update `last_tick`, `high_tick`, `low_tick`
     - Call each indicator `on_new_tick` method.

    Arguments:
        tick: new received Tick.
    """
    logger.debug("Update frame with new tick")
    self.nb_ticks += 1
    self.last_tick = tick

    if tick.value > self.high_tick.value:
        self.high_tick = tick
    elif tick.value < self.low_tick.value:
        self.low_tick = tick

    self._upsert_indicators(tick)