Candle Set

A CandleSet create candles, add ticks and close a list of estrade.Candles based on its timeframe.

Parameters

Name Type Description Default
timeframe str Define either a number of time units either a number of ticks. required
indicators List[estrade.mixins.candle_set_indicator_mixin.CandleSetIndicatorMixin] List of indicators attached to CandleSet None
max_candles_in_memory int Number of candles kept in memory 100

timeframe

  • unit must always be plural ('1hours' and not '1hour')
  • inside a CandleSet timeframe is splitted between:
    • ut (time unit, eg. ticks, minutes, hours etc.)
    • nb (nb of time unit)
  • eg: '3ticks', '56ticks', '30seconds', '3minutes', '4hours'.

max_candles_in_memory

  • note: the bigger the number of candles, the slower your program will be
  • warning: keep this number high enough for your indicators using candle to work properly (eg. if you use an indicator for moving average 200, the max number of candles in memory should at least be 200)

Example

Example of usage of candle set

from estrade import Epic, CandleSet

# create candle sets with your timeframes and assign them to your epics
cs_21ticks = CandleSet(timeframe='21ticks')
cs_1mn1 = CandleSet(timeframe='1minutes')
epic1 = Epic(ref='EPIC1', candle_sets=[cs_21ticks, cs_1mn])

cs_1mn2 = CandleSet(timeframe='1minutes')
cs_4hours = CandleSet(timeframe='4hours')
epic2 = EPIC(ref='EPIC2', candle_sets=[cs_1mn2, cs4hours])

current_candle: Union[estrade.candle.Candle, NoneType] (property, readonly)

Returns

Type Description
Union[estrade.candle.Candle, NoneType] return the candle that is currently fed by ticks received.

epic: Epic (property, writable)

Returns

Type Description
Epic CandleSet Epic

indicators: List[estrade.mixins.candle_set_indicator_mixin.CandleSetIndicatorMixin] (property, writable)

Returns

Type Description
List[estrade.mixins.candle_set_indicator_mixin.CandleSetIndicatorMixin] Return list of indicators attached to candle set

is_closing_candle: bool (property, readonly)

This method check if the last candle will be closed on next tick.

Returns

Type Description
bool True if the current candle will be closed on next tick, else False

Exceptions

Type Description
CandleSetException if attempt to use this method on a timed CandleSet.

Warning

This method can only be used when this instance UT is ticks. It cannot be used on time based candlesets.

last_closed_candle: Union[estrade.candle.Candle, NoneType] (property, readonly)

Returns

Type Description
Union[estrade.candle.Candle, NoneType] return the last candle that was closed.

Info

At beginning of Market run, the ticks feed a the first candle, while this first candle is not closed (=its timeframe over) the last_closed_candle is empty. As a consequence, in your strategies, it is strongly advised that your check that the last_closed_candle is not None before using it.

nb: int (property, writable)

Returns

Type Description
int Return the number of time units (see nb and timeframe setters)

Example

if the CandleSet was initated with a timeframe of 4hours, then the nb is 4.

nb_closed_candles: int (property, readonly)

Returns

Type Description
int The number of closed candles

Warning

when the number of closed candle reach the max_candles_in_memory value set on init, then this value will always returns the max_candles_in_memory

new_candle_opened: bool (property, readonly)

Check if current candle is newly opened (only one tick is in the last candle)

Returns

Type Description
bool True if the current candle has only one tick.

ref: str (property, readonly)

Returns

Type Description
str the reference (~name) of a Candle Set.

Info

Ref of a candle set cannot be set manually, it is a fixed name with format <epic_ref>-<nb><ut>

example: NASDAQ-4hours

timeframe: str (property, writable)

Returns

Type Description
str the timeframe value as provided in CandleSet instanciation.

ut: str (property, writable)

Returns

Type Description
str return CandleSet UT (time unit).

Example

if the CandleSet was initated with a timeframe of 4hours, then the ut is hours

check_timeframe_coherence(ut, nb) (staticmethod)

Show source code in estrade/candle_set.py
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
    @staticmethod
    def check_timeframe_coherence(ut: str, nb: int) -> bool:
        """
        Check that ut and nb (result of timeframe split) are coherent.

         - if ut is in minutes/seconds, nb must be inferior to 60 and nb/60 must be
         an integer.
         - if ut is in hours, nb must be inferior to 12 and nb/12 must be an integer.
         - as only one day period is supported: if ut is 'days' nb must be 1.

        !!! note
            nb and ut values individual consistencies are checked in ut and nb setters.

        """
        if ut in ['seconds', 'minutes'] and int(60 / nb) != 60 / nb:
            raise CandleSetException(
                'Inconsistent UT: munutes/seconds timeframe must be '
                'inferior to 60 and {}/60 must be an integer.'.format(nb)
            )
        elif ut in ['hours'] and int(12 / nb) != 12 / nb:
            raise CandleSetException(
                'Inconsistent UT: hour timeframe must be inferor or equal to 12. '
                'And {}/12 must be an int'.format(nb)
            )
        elif ut == 'days' and nb != 1:
            raise CandleSetException(
                'Inconsistent UT, only 1days timeframe is allowed.'
            )

        return True

Check that ut and nb (result of timeframe split) are coherent.

  • if ut is in minutes/seconds, nb must be inferior to 60 and nb/60 must be an integer.
  • if ut is in hours, nb must be inferior to 12 and nb/12 must be an integer.
  • as only one day period is supported: if ut is 'days' nb must be 1.

Note

nb and ut values individual consistencies are checked in ut and nb setters.

closed_candles(self, nb=None)

Show source code in estrade/candle_set.py
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
    def closed_candles(self, nb: int = None) -> List[Candle]:
        """
        Get a list of the last closed candles

        Arguments:
            nb: nb of closed candles to retrieve

        """
        if nb and nb > self.nb_closed_candles:
            raise CandleSetException(
                'Impossible to get {} candles from {} bc only contains {}'.format(
                    nb, self.ref, self.nb_closed_candles
                )
            )
        elif nb is None:
            nb = self.nb_closed_candles

        return self.candles[((nb + 1) * -1) : -1]

Get a list of the last closed candles

Parameters

Name Type Description Default
nb int nb of closed candles to retrieve None

indicator(self, indicator_name)

Show source code in estrade/candle_set.py
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
    def indicator(self, indicator_name: str) -> Optional[CandleSetIndicatorMixin]:
        """
        Arguments:
            indicator_name: name of the indicator to fetch

        Raises:
            CandleSetException: if no indicator is found with this name

        Returns:
            the an indicator in this instance indicators by its name.
        """
        for indicator in self.indicators:
            if indicator.name == indicator_name:
                return indicator
        raise CandleSetException(
            'No indicator found with name {}'.format(indicator_name)
        )

Parameters

Name Type Description Default
indicator_name str name of the indicator to fetch required

Exceptions

Type Description
CandleSetException if no indicator is found with this name

Returns

Type Description
Union[estrade.mixins.candle_set_indicator_mixin.CandleSetIndicatorMixin, NoneType] the an indicator in this instance indicators by its name.

on_new_tick(self, tick)

Show source code in estrade/candle_set.py
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
    def on_new_tick(self, tick: Tick) -> None:
        """
        This method is the entry point called on every new tick.

         - update candles (close and open candles if required or add tick
         to current candle)
         - send events (to update indicators, call strategies candle events etc.)

        Arguments:
            tick: new tick received to be dispatched to candles and indicators

        """
        logger.debug('%s : add tick in candle set: %s' % (self.timeframe, tick))
        if not self.epic:
            raise CandleSetException(
                'Impossible to add tick to a candle set with no epic'
            )

        self.fire('candle_set_before_on_new_tick_{}'.format(tick.epic.ref), tick=tick)
        new_candle_created = False
        current_candle_closed = False
        if not self.candles:
            logger.debug('%s : init list of candles with a new candle' % self.timeframe)
            self._create_new_candle(tick)
            new_candle_created = True
        else:
            logger.debug(
                '%s: check if tick datetime %s require to close candle'
                % (self.timeframe, tick.datetime)
            )
            if self._is_current_candle_finished(tick.datetime):
                logger.debug('%s: close candle' % self.timeframe)
                self._close_last_candle()
                current_candle_closed = True

                logger.debug('%s: create a new candle' % self.timeframe)
                self._create_new_candle(tick)
                new_candle_created = True

            else:
                logger.debug('%s: add new tick to current candle' % self.timeframe)
                self.current_candle.on_new_tick(tick)

        # update indicators
        for indicator in self.indicators:

            if current_candle_closed:
                indicator.on_candle_close(closed_candle=self.candles[-2])
            if new_candle_created:
                indicator.on_new_candle(new_candle=self.candles[-1])

            indicator.on_new_tick(tick=tick)

        self.fire('candle_set_after_on_new_tick_{}'.format(tick.epic.ref), tick=tick)

This method is the entry point called on every new tick.

  • update candles (close and open candles if required or add tick to current candle)
  • send events (to update indicators, call strategies candle events etc.)

Parameters

Name Type Description Default
tick Tick new tick received to be dispatched to candles and indicators required