Module data

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 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

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

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
class JSONCacheObject (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

Parent