Source code for babao.inputs.trades.tradesInputBase

"""
Module containing base class for any trades input
"""

from abc import abstractmethod

import numpy as np

import babao.inputs.inputBase as ib


class ABCTradesInput(ib.ABCInput):
    """Base class for any trades input"""

    raw_columns = [
        "price", "volume"
    ]
    resampled_columns = [
        "open", "high", "low", "close", "vwap", "volume", "count"
    ]

    @property
    @abstractmethod
    def quote(self):
        """
        Overide this method with the desired QuoteEnum
        ex: self.quote = QuoteEnum.EUR
        """
        pass

    @property
    @abstractmethod
    def crypto(self):
        """
        Overide this method with the desired CryptoEnum
        ex: self.crypto = CryptoEnum.XBT
        """
        pass

    def _resample(self, raw_data):
        p = ib.resampleSerie(raw_data["price"])
        resampled_data = p.ohlc()

        # tmp var for ordering
        v = ib.resampleSerie(raw_data["volume"]).sum()
        resampled_data["vwap"] = ib.resampleSerie(
            raw_data["price"] * raw_data["volume"]
        ).sum() / v
        resampled_data["volume"] = v
        resampled_data["count"] = p.count()

        return resampled_data

    def fillMissing(self, resampled_data):
        resampled_data["count"].fillna(0, inplace=True)
        resampled_data["volume"].fillna(0, inplace=True)
        resampled_data["vwap"].replace(np.inf, np.nan, inplace=True)

        i = resampled_data.index[0]
        for col in ["vwap", "close"]:
            if np.isnan(resampled_data.loc[i, col]):
                if self.current_row is not None:
                    resampled_data.loc[i, col] = self.current_row.price
                else:
                    resampled_data.loc[i, col] = 0
            resampled_data[col].ffill(inplace=True)

        c = resampled_data["close"]
        resampled_data["open"].fillna(c, inplace=True)
        resampled_data["high"].fillna(c, inplace=True)
        resampled_data["low"].fillna(c, inplace=True)

        return resampled_data

    @abstractmethod
    def fetch(self):
        pass