Source code for babao.models.modelBase
"""Base class for any model"""
import os
from abc import ABC, abstractmethod
from typing import List, Type, Optional, TypeVar, Union
import numpy as np
import babao.config as conf
import babao.utils.log as log
import babao.inputs.ledger.ledgerManager as lm
import babao.inputs.inputBase as ib
MODELS = [] # type: List[ABCModel]
NODE = TypeVar("NODE", "ABCModel", ib.ABCInput)
os.environ['KERAS_BACKEND'] = 'theano'
[docs]def getVerbose() -> int:
"""Transform our verbose level to match keras one"""
return int(log.VERBOSE / 2) if log.VERBOSE != 1 else 1
[docs]def reshape(arr):
"""Reshape the features to be keras-proof"""
return np.reshape(arr, (arr.shape[0], 1, arr.shape[1]))
[docs]def addLookbacks(df, look_back):
"""Add lookback(s) (shifted columns) to each df columns"""
for i in range(1, look_back + 1):
for col in df.columns:
if "lookback" not in col:
df[col + "_lookback_" + str(i)] = df[col].shift(i)
return df.dropna()
[docs]def addLookbacks3d(arr, look_back):
"""
Add lookback(s) (shifted columns) to each df columns
Reshape the features to be keras-proof (3d)
"""
res = None
for i in range(len(arr) - look_back):
if res is None:
res = np.array([arr[i:i + look_back]])
else:
res = np.append(
res,
np.array([arr[i: i + look_back]]),
axis=0
)
return res[look_back:] # dropna
[docs]class ABCModel(ABC):
"""Base class for any model """
@property
@abstractmethod
def dependencies_class(
self
) -> List[Union[Type["ABCModel"], Type[ib.ABCInput]]]:
"""
List of models or inputs needed by the current model
These should be class, not instances!
"""
pass
@property
@abstractmethod
def need_training(self) -> bool:
"""Specify if the current model need to be trained"""
pass
@staticmethod
def _getNodeFromList(
node_class: Type[NODE],
node_list: List[NODE]
) -> NODE:
"""Find a node instance in the MODELS list, or throw an error"""
return node_list[
[n.__class__ for n in node_list].index(node_class)
]
@staticmethod
def _getNodeInstance(
node_class: Type[NODE],
node_list: Optional[List[NODE]] = None
) -> NODE:
"""Find a node instance in the MODELS list, or instantiate a new one"""
if node_list is None:
if issubclass(node_class, ib.ABCInput):
node_list = ib.INPUTS
# elif issubclass(node_class, ABCModel):
else: # we are all grown up here
node_list = MODELS
try:
return ABCModel._getNodeFromList(node_class, node_list)
except ValueError:
pass
node = node_class() # recursive horror
# the node_class wasn't in the list so we instantiate it...
# but its dependencies may have added and instance of node_class!
if node_class in [n.__class__ for n in node_list]:
return ABCModel._getNodeFromList(node_class, node_list)
node_list.append(node) # TODO: sort the MODELS by priority order
return node
def _initDeps(self):
"""Instantiate all the needed dependencies"""
if not MODELS:
MODELS.append(self)
ib.INPUTS.extend(lm.LEDGERS.values())
ib.INPUTS.extend(lm.TRADES.values())
for node_class in self.dependencies_class:
self.dependencies.append(self._getNodeInstance(node_class))
def __init__(self):
self.model = None
self.model_file = os.path.join(
conf.DATA_DIR, self.__class__.__name__ + ".model"
)
try:
self.load()
except OSError:
log.warning("Couldn't load", self.__class__.__name__)
self.dependencies = []
self._initDeps()
[docs] @abstractmethod
def predict(self, since):
"""Return a dataframe of prediction starting from ´since´ timestamp"""
raise NotImplementedError("Implement me!")
[docs] @abstractmethod
def train(self, since):
"""
Train the model with data starting from ´since´ timestamp
Return the score of model.
"""
raise NotImplementedError("Implement me!")
[docs] @abstractmethod
def plot(self, since):
"""Plot the model predictions from ´since´ timestamp"""
raise NotImplementedError("Implement me!")
[docs] @abstractmethod
def save(self):
"""Save the model to self.model_file"""
raise NotImplementedError("Implement me!")
[docs] @abstractmethod
def load(self):
"""Load the model from self.model_file"""
raise NotImplementedError("Implement me!")