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
|
indicators |
Dict[str, estrade.graph.base_indicator.BaseIndicator] |
Dict
of indicators calculated on every |
ref |
str |
reference of this instance
(see |
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
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)
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,
) -> 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.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)