feat(Models): added lightGBM, moved other models to separate files (#128)

* feat(Models): added lightGBM, moved other models to separate files

* feat(Models): added non-working statsmodel wrapper

* fix(Models): added work-in-progress comment to StatsModels
This commit is contained in:
Mark Aron Szulyovszky
2022-01-08 12:02:42 +01:00
committed by GitHub
parent 6982187872
commit fc5eba4e2d
12 changed files with 165 additions and 73 deletions
-67
View File
@@ -1,14 +1,9 @@
from __future__ import annotations
from typing import Literal, Optional, Union
from sklearn.base import clone
from abc import ABC, abstractmethod
import numpy as np
import copy
import pytorch_lightning as pl
import numpy as np
from data_loader.pytorch_dataset import get_dataloader
class Model(ABC):
@@ -39,68 +34,6 @@ class Model(ABC):
def initialize_network(self, input_dim:int, output_dim:int):
pass
class SKLearnModel(Model):
data_scaling = 'scaled'
only_column = None
feature_selection = 'on'
model_type = 'ml'
predict_window_size = 'single_timestamp'
def __init__(self, model):
self.model = model
def fit(self, X: np.ndarray, y: np.ndarray) -> None:
self.model.fit(X, y)
def predict(self, X) -> tuple[float, np.ndarray]:
pred = self.model.predict(X).item()
probability = self.model.predict_proba(X).squeeze()
return (pred, probability)
def clone(self) -> SKLearnModel:
return SKLearnModel(clone(self.model))
def get_name(self) -> str:
return self.model.__class__.__name__
def initialize_network(self, input_dim:int, output_dim:int):
pass
class LightningNeuralNetModel(Model):
data_scaling = 'scaled'
only_column = None
feature_selection = 'off'
model_type = 'ml'
''' Standard lightning methods '''
def __init__(self, model, max_epochs=5):
self.model = model
self.trainer = pl.Trainer(max_epochs=max_epochs)
def fit(self, X: np.ndarray, y: np.ndarray) -> None:
train_dataloader = self.__prepare_data(X.astype(float), y.astype(float))
self.trainer.fit(self.model, train_dataloader)
def predict(self, X: np.ndarray) -> tuple[float, np.ndarray]:
return self.model(X)
def clone(self):
model_copy = copy.deepcopy(self.model)
return LightningNeuralNetModel(model_copy)
''' Non-standard lightning methods '''
def __prepare_data(self, X:np.ndarray, y:np.ndarray):
dataloader = get_dataloader(X, y)
return dataloader
def initialize_network(self, input_dim:int, output_dim:int):
self.model.initialize_network(input_dim, output_dim)
def get_name(self) -> str:
return self.model.__class__.__name__