Module data

Module to manage complex data pipelines.

Sub-modules

backtest.data.datapipe

backtest.data.json_extension

backtest.data.pipes

backtest.data.utils

Functions

def CausalImpute(frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Dict[str, pandas.core.frame.DataFrame]], **kwargs) ‑> Dict[str, pandas.core.frame.DataFrame]

This pipe will impute the missing values in the time series data using the causal imputation method. (It will copy the last time step value to the missing values)

Example

a = [1, 2, 3, nan, 5, 6, nan, 8, 9] causal_impute(a) -> [1, 2, 3, 3, 5, 6, 6, 8, 9]

IN: dict[str, pd.Dataframe] where the values are the charts as Dataframe and the keys are the tickers

OUT: dict[str, TSData] where the values are the TSData and the keys are the tickers

Args:

frm: From datetime
to: to datetime
args: Args passed to the pipe
po: Previous Pipe output
kwargs: The keyword arguments passed to the pipe

Returns:

pd.DataFrame

def FilterNoneCharts(frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Dict[str, Optional[pandas.core.frame.DataFrame]]], **kwargs) ‑> Dict[str, pandas.core.frame.DataFrame]

This pipe will filter out the tickers that doesn't have a chart. (Chart is None)

IN: dict[str, Optional[pd.DataFrame]] where the values are the charts and the keys are the tickers

OUT: dict[str, pd.DataFrame] where the values are the charts and the keys are the tickers

Args:

frm: From datetime
to: to datetime
args: Args passed to the pipe
po: Previous Pipe output
kwargs: The keyword arguments passed to the pipe

Returns:

:

def PadNan(frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Dict[str, Optional[pandas.core.frame.DataFrame]]], **kwargs) ‑> Dict[str, Optional[pandas.core.frame.DataFrame]]

This pipe will pad the charts (dataframes) with NaNs. It will reindex them with the longest index found in the data. The index will be the same for all the charts.

IN: dict[str, Optional[pd.DataFrame]] where the values are the charts and the keys are the tickers

OUT: dict[str, pd.DataFrame] where the values are the charts and the keys are the tickers, but the charts are all the same length, and padded with Nans.

Args:

frm: From datetime
to: to datetime
args: Args passed to the pipe
po: Previous Pipe output. Must be a dictionary of charts (ticker, pd.DataFrame)
kwargs: The keyword arguments passed to the pipe

Returns:

pd.DataFrame

def ToTSData(frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Dict[str, pandas.core.frame.DataFrame]], **kwargs) ‑> List[Dict[str, TSData]]

This pipe will convert the charts into a time series object (TSData).

IN: dict[str, pd.DataFrame] where the values are the charts and the keys are the tickers

OUT: list[dict[str, TSData]] where the values are the TSData and the keys are the tickers (len of list = 1)

Args:

frm: From datetime
to: to datetime
args: Args passed to the pipe
po: Previous Pipe output
kwargs: The keyword arguments passed to the pipe

Returns:

pd.DataFrame

Classes

class CacheObject (value: Any, pipe_id: int, pipe_hash: str, next_revalidate: Optional[datetime.datetime] = None, max_request: Optional[int] = None, current_n_requests: int = 0)

This class holds a the output of a pipe and some metadata to be able to revalidate the cache.

Args:

value: The output of the pipe
pipe_id: The pipe_id of the pipe that generated the cache
pipe_hash: Used to check if the pipe structure has changed
next_revalidate: The next time the cache should be revalidated
max_request: The maximum number of requests that can be made before revalidating the cache
current_n_requests: The current number of requests made to the cache

Static methods

def load(pipe_id: int) ‑> CacheObject

Call this method to load a cache from the .cache folder.

Args:

pipe_id: The pipe_id that stored the cache

Returns:

The cache object if it exists else None. (This function doesn't check if the cache valid)

Methods

def store(self)

Call this method to store the cache to the .cache folder. The cache file has the pipe_id as its name. This is why it is strongly recommended to delete the .cache folder when changing the pipeline structure, to avoid unexpected behavior.

class DataPipe (T: DataPipeType, name: Optional[str] = None)

The base element of the data module api. This object can represent the four types of pipes (Fetch, Process, Cache, and Collate). It is a recursive object that can be used to build complex pipelines. The pipeline is built using the pipe operator (|). Once the pipeline is built, it is run by calling the get method. To make custom pipes, you need to inherit from this class and implement one of the three methods (fetch, process, collate). If you want to implement a cache pipe, you need to implement the load, cache and revalidate methods. Do not forget to call the super().init method in the init method of your custom pipe and pass the appropriate DataPipeType. You can also pass a name to the pipe to make the pipeline more readable.

Args:

T: The pipe type (Fetch, Process, Cache, Collate)
name: The pipe name. If None, the name of the class is used

Parent

• abc.ABC

Static methods

def Collate(pipe1: DataPipe, pipe2: DataPipe) ‑> DataPipe

This method is used to build a collate pipe from two pipes. It is used top merge two branches of a pipeline.

Args:

pipe1: The first pipeline (or pipe if there is only one)
pipe2: The second pipeline (or pipe if there is only one)

Returns:

A new pipe that is the result of the concatenation of the two pipes. (Multiple pipes makes a pipeline)

Instance variables

var pipe_id : int

Get the pipe_id (Identifier unique to each pipe in the pipeline). It is given at built time and is deterministic. This means that each time the pipeline is built, the pipe_id will be the same for each pipe, given that the pipeline structure is the same.

Returns:

The pipe_id

Methods

def cache(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Any], **kwargs) ‑> None

This method is called to cache the output of the pipeline. It is called by the _run method when the pipe is a cache pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

None

def collate(self, frm: datetime.datetime, to: datetime.datetime, *args, po1: PipeOutput[typing.Any], po2: PipeOutput[typing.Any], **kwargs) ‑> PipeOutput

This method is called to merge the output of two pipes. It is called by the _run method when the pipe is a collate pipe.

Args:

frm: From datetime
to: To datetime
args: any args passed to the get method
po1: The output of the first pipeline (left)
po2: The output of the second pipeline (right)
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata

def fetch(self, frm: datetime.datetime, to: datetime.datetime, *args, po: Optional[PipeOutput[typing.Any]], **kwargs) ‑> PipeOutput

This method is called to fetch the data from an external source. It is called by the _run method when the pipe is a fetch pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata

def get(self, frm: datetime.datetime, to: datetime.datetime, *args, **kwargs) ‑> Any

Run the pipeline from the beginning to the end and return the output of the pipeline.

Args:

frm: From datetime, this is passed to all pipes and is used to fetch the data from this date to the 'to'date.
to: The end datetime, this is passed to all pipes and is used to fetch the data from the 'frm' date tothis date.
args: any args that needs to be passed to the subsequent pipes
kwargs: any keyword args that needs to be passed to the subsequent pipes

Returns:

The output of the pipeline

def hash(self) ‑> str

Make a hash that is consistent across runs and is deterministic given the object.

Returns:

The hash.

def load(self) ‑> Optional[CacheObject]

This method is called to load the cache from the disk. It is called by the _load_cache method when the pipe is a cache pipe. It will return the loaded cache object if found. Otherwise, it will return None.

Returns:

CacheObject if found else None

def process(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Any], **kwargs) ‑> PipeOutput

This method is called to process and transform the data. It is called by the _run method when the pipe is a process pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata

def render(self)

This method is used to render the pipeline as a string. It renders the pipe structure as a tree with each node being the name of the pipe. It is used by the str method to render the pipeline as a tree.

Returns:

The rendered pipeline as a string

def revalidate(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput, **kwargs) ‑> RevalidateAction

This method is called to revalidate the cache. It is called by the _run method when the pipe is a cache pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata

def set_id(self, pipe_id: int)

With this method, it is possible to manually set the pipe_id of the pipe. It is recommended to call this method only on the top-level pipes, and before running the pipeline with get. If the pipe is not the top-level pipe, its pipe_id may change, and you might run in unexpected behaviors. By running this method only before running the get method, this ensures that the pipe will have the same pipe_id if the pipeline structure is the same. This is due to the fact that pipe_ids are dynamically assigned at build time. However, when the pipe is run, the pipe becomes forged and no changes are allowed. This ensures that the pipe ids stays the same. When this method is called, it acts as if the pipe was run, and the pipe_ids are frozen. This means that the ids aren't verified. Make sure that the pipe_id given are unique to avoid conflicts. Note: The passed pipe_id will be assigned to the lowes-level pipe. The current pipe will have the highest pipe_id

Args:

pipe_id: The pipe_id to assign to the pipe.

Returns:

None

class DataPipeType (*args, **kwds)

[Enum] - An enumeration representing the four types of pipes that can be used in the pipeline.

• FETCH: Fetch data from an external source (Web, DB, file, API, etc.).
• PROCESS: Process and transform the data (Filter, transform, etc.).
• CACHE: Cache the output of the pipeline to avoid repeating previous steps on each run.
• COLLATE: Collate the output of two pipes (This is a special case, used to merge two pipelines).

Parent

• enum.Enum

Class variables

var CACHE

var COLLATE

var FETCH

var PROCESS

class PipeOutput (value: ~T_, output_from: DataPipe, revalidate_action: Optional[ForwardRef('RevalidateAction')] = None, **kwargs)

This class holds the output of a pipe and some metadata that are passed through out the pipeline. This metadata is used for revalidation purpose.

Args

value

The value

output_from

The output from the pipe

revalidate_action

The revalidate action

**kwargs:

Parent

• typing.Generic

Instance variables

var revalidate

Get the revalidate action of the pipe

Returns:

The revalidate action of the pipe

var value

The output of the pipe

Returns:

The output value of the pipe

Methods

def set_output_from(self, output_from: DataPipe)

Set the output_from of the pipe. Can be useful when a pip makes a copy of the output of another pipe.

Args:

output_from: The pipe that generated the output

Returns:

None

def set_revalidate(self, action: RevalidateAction)

Set the revalidate action of the pipe.

Args:

action: The action to set

Returns:

None

class RevalidateAction (*args, **kwds)

The three possible revalidation action that a pipe can return.

• REVALIDATE: Revalidate the cache from the current position to the end of the pipeline
• FULL_REVALIDATE: Revalidate the cache from the beginning to the end of the pipeline
• NOOP: Do not revalidate the cache (The pipe will load from cache if it exists)

Parent

• enum.Enum

Class variables

var FULL_REVALIDATE

var NOOP

var REVALIDATE

class Cache (caching_cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], int, DataPipe, ...], None] = None, *, loading_cb: Callable[[int], CacheObject] = None, revalidate: Optional[datetime.datetime] = None, timeout: Optional[datetime.timedelta] = None, max_requests: Optional[int] = None, store: bool = True, revalidate_cb: Callable[[datetime.datetime, datetime.datetime, Tuple[Any, ...], PipeOutput[Any], dict[str, Any]], RevalidateAction] = None)

Function decorator designed to make a Cache pipe out of a function. The decorated function is expected to be the caching method. When using this, it is recommended to pass a load callback method to load the cache object from the storage. The caching method should be able to store the cache and load it from the storage. The cache object should be a CacheObject instance. It is also possible to pass a revalidate callback method to define a dynamic revalidate action. The revalidate action is a method that returns a RevalidateAction enum value. The revalidate action is used to determine if the cache should be revalidated or not. If no revalidate method is passed, the default revalidate method check if the cache is expired given an expired datetime, a timeout or a maximum number of requests.

Note

If timeout is None, max_requests is None and revalidate_cb is None, the cache will never be revalidated. If the structure of the underlying pipe changes or the parameters of the underlying pipe changes, the whole pipeline will be revalidated (full revalidation). This means that every cache pipe will revalidate their cache, not only the one that changed.

Examples:

>>> @Cache(loading_cb=JSON_load, store=True, timeout=timedelta(seconds=1))
... def MyCache(frm: datetime, to: datetime, *args, po: PipeOutput, pipe_id: int, revalidate: datetime,
...          timedelta, max_requests: int, n_requests: int, **kwargs):
...     value = {
...         "data": po.value,
...         "stored_dt": datetime.now().isoformat(),
...         "revalidate": revalidate.isoformat(),
...         "current_n_requests": n_requests,
...         "timeout": timeout.total_seconds(),
...         "max_requests": max_requests
...     }
...     if not os.path.exists(".cache"):
...         os.mkdir(".cache")
...     with open(f".cache/{pipe_id}.json", "w") as f:
...         json.dump(value, f)
>>> # The cache pipe can also be used as is:
>>> pipe = FetchN | Cache(timeout=timedelta(seconds=1))

Args:

caching_cb: The callback that is called after the pipeline wrapped by the Cache pipe is executed. It is used to store the cache.
loading_cb: The callback used to load the cache from the disk.
revalidate: The next datetime to revalidate the cache.
timeout: A timedelta object representing the max age of the cache before revalidating it. If it is oneday, the cache will be revalidated every day as opposed to the revalidate parameter, which revalidate only once.
max_requests: The maximum number of request that hit the cache before revalidating it.
store: Whether to store the cache on disk or keep it only in memory.
revalidate_cb: A callback method that returns a RevalidateAction enum value. It is used to determine ifthe cache should be revalidated or not. This callback can be provided for complex revalidation logic.

Parent

• DataPipe

class Collate (cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], PipeOutput, ...], Any])

Function decorator designed to make a Collate pipe out of a function. It is designed to collate data. The data can be combined, aggregated, concatenated, etc. The pipe is named after the function name.

Args:

cb: The collate callback method.

Parent

• DataPipe

class Fetch (cb: Callable[[datetime.datetime, datetime.datetime, ..., Optional[PipeOutput[Any]], ...], Any])

Function decorator designed to make a Fetch pipe out of a function. It is designed to fetch data from a source. The source can be a database, a web API, a file, etc. The pipe is named after the function name.

Args:

cb: The fetch callback method.

Parent

• DataPipe

class Process (cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], ...], Any])

Function decorator designed to make a Process pipe out of a function. It is designed to process data. The data can be transformed, cleaned, imputed, etc. The pipe is named after the function name.

Args:

cb: The process callback method.

Parent

• DataPipe

class FetchCharts (tickers=None, interval: str = '1d', progress: bool = False, throttle: float = 0.0, *args, **kwargs)

This pipe will fetch the charts from Yahoo Finance. It will fetch the charts for the specified tickers and the specified interval. The interval can be any interval supported by yfinance. The data will be returned as a dictionary of DataFrames where the keys are the tickers and the values are the DataFrames. The columns of the returned dataframes are the following: Open, High, Low, Close, Volume, Dividends, Stock Splits. The index is named "Date" for intervals of 1d and longer and "Datetime" for intervals shorter than 1d. The index is a datetime index. If the returned chart is empty, it will be returned as None. Consider removing the None charts in further preprocessing steps.

Warning

The returned index contains timezone information that might be inconsistent. Consider removing them.

IN: Optional[List[str]]: The list of tickers to fetch the charts for. If provided, it will override the ticker list passed during initialization.

OUT: dict[str, Optional[pd.DataFrame]] where the values are the charts and the keys are the tickers

Args:

T: The pipe type (Fetch, Process, Cache, Collate)
name: The pipe name. If None, the name of the class is used

Parent

• DataPipe

class JSONCache (*, revalidate: Optional[datetime.datetime] = None, timeout: Optional[datetime.timedelta] = None, max_requests: Optional[int] = None, store: bool = True, revalidate_cb: Callable[[datetime.datetime, datetime.datetime, Tuple[Any, ...], PipeOutput, dict[str, Any]], RevalidateAction] = None, custom_types: Optional[dict[str, type]] = None)

This pipe will cache the data as a JSON file instead of a pickle file. This class uses an extended JSON encoder and decoder to encode most datatypes. However, the encoded version might be suboptimal. To make objects JSON serializable, you can add a tojson method that returns a JSONable dictionary to your object. To make it loadable, you can add a fromjson class method that takes a JSONable dictionary and returns an instance of your object.

Args:

revalidate: The datetime to revalidate the cache
timeout: The timedelta at which the cache will expire and be revalidated periodically
max_requests: The maximum number of request before revaidating the cache
store: Whether to store on disk or only in memory
revalidate_cb: A revalidate callback function to implement a custom revalidation mechanism
custom_types: The custom types to add to the JSONEncoder and JSONDecoder to make them JSON serializableand loadable. Every type that implemented the tojson and fromjson methods should be registered here.

Parent

• Cache

Methods

def cache(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Any], **kwargs) ‑> None

This method is called to cache the data. This emthod will cache the data to the json format, and save it to the disk. It will also save as pickle the automatically detected types.

Args:

frm: Start Datetime
to: End Datetime
args: Any arguments passed to the pipeline
po: The previous PipeOutput
kwargs: Any keyword arguments passed to the pipeline

Returns:

None

def load(self) ‑> Optional[JSONCacheObject]

This method is called to load the cache. This method will load the cache from the disk, and return the CacheObject. If the cache is not found, it will return None.

Returns:

JSONCacheObject if cache is found and None otherwise