Module engine

Functions

def FetchContext(frm: datetime.datetime, to: datetime.datetime, *args, po: Optional[PipeOutput], context, **kwargs)

This Pipe will return as a pipe output what has been passed as the context.

Classes

class BasicExtender (period: Union[TimePeriod, str])

This class provide an easy top implement extender for basic time resolutions such as 30 minutes, 1 hour, 4 hours, 1 day, 1 week, and 1 month. It assumes to have has main timestep (In the backtest object) a time resolution smaller than the value chose. For example, if the main timestep is 1 day, it will not work to extend the data to 30 minutes or 4h. If you have multiple static timestep that you want to extend, you can override the extend method to handle this case, or directly derive a new class from TimeResExtender.

Example

>>> extender = BasicExtender(TimePeriod.ONE_DAY) + BasicExtender("1w")
>>> # Given that the main timeresolution of the static data is smaller than one day.  Ex: 30 minutes
>>> # The data will have a time resolution of 30 minutes, 1 day and 1 week.
>>> data: List[Dict[str, TSData]] = ...
>>> data += extender.extend(data)

Parent

class TimePeriod (*args, **kwds)

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access:

Color.RED

  • value lookup:

Color(1)

  • name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Parent

  • enum.Enum

Class variables

var FOUR_HOURS
var ONE_DAY
var ONE_HOUR
var ONE_MONTH
var ONE_WEEK
var THIRTY_MINUTES
class TimeResExtender

Helper class that provides a standard way to create an ABC using inheritance.

Parent

  • abc.ABC

Methods

def extend(self, data: List[Dict[str, TSData]], main_timestep: int = 0) ‑> List[Dict[str, TSData]]

Extend the data with new time resolutions. By default, it uses the single_extend method to extend each timeseries data independently. However, it is possible to override this method and bypass the single_extend method to extend the data in a more complex way or to optimize the process.

Args:

data: The current data statically acquired. (Fetch from sources)
main_timestep: The index of the timeseries data in data list to use as the main series.

Returns:

The new data with the new time resolutions. (Do not mutate the original data)
def single_extend(self, data: TSData) ‑> Tuple[pandas.core.frame.DataFrame]

Extend a single timeseries data with new time resolutions. Override this method

Args:

data: The current data statically acquired. (Fetch from sources)

Returns:

The new data with the new time resolutions. (Do not mutate the original data)