Multiple quantile regression with Tensorflow - Part 1

Abstract

이전 포스팅에서는 Multiple quantile에 대해서 소개했다. 이번 포스팅에서는 Tensorflow를 활용하여 동시에 Multiple quantile을 추정하며, Crossing problem을 방지할 수 있는 방법에 대해 서술한다.

Main

Deep learning을 사용하여 Multiple quantile regression을 어떻게 적합할 수 있을까?
많은 방법이 있겠지만 이전 포스팅에서 소개한 Cannon의 접근법을 사용해봤다.

Transform data

먼저, 아래처럼 학습 데이터를 입력 quantile들의 개수만큼 복제하는 DataTransformer class를 만들었다.

# mcqrnn/transform.py
from typing import Union, Tuple
import numpy as np


def _mcqrnn_transform(
    x: np.ndarray,
    taus: np.ndarray,
    y: Union[np.ndarray, None] = None,
) -> Tuple[np.ndarray, ...]:
    """
    Transform x, y, taus into the trainable form
    Args:
        x (np.ndarray): input
        y (np.ndarray): output
        taus (np.ndarray): quantiles
    Return
        Tuple[np.ndarray]: transformed x, y
    """
    _len_taus = len(taus)
    _len_x = len(x)
    _len_data = _len_x * _len_taus

    x_trans = np.repeat(x, _len_taus, axis=0).astype("float32")
    taus_trans = np.tile(taus, _len_x).reshape((_len_data, 1)).astype("float32")
    if y is not None:
        y_trans = np.repeat(y, _len_taus, axis=0).astype("float32")
        y_trans = y_trans.reshape((_len_data, 1))
        return x_trans, y_trans, taus_trans
    else:
        return x_trans, taus_trans


class DataTransformer:
    """
    A class to transform data into trainable form.
    Args:
        x (np.ndarray): input
        taus (np.ndarray): quantiles
        y (Union[np.ndarray, None]): output

    Methods:
        __call__:
            Return Tuple[np.ndarray, ...]:
        transform(input_taus: np.ndarray):
            Return transformed x with given input_taus
    """

    def __init__(
        self,
        x: np.ndarray,
        taus: np.ndarray,
        y: Union[np.ndarray, None] = None,
    ):
        self.x = x
        self.y = y
        self.taus = taus
        self.x_trans, self.y_trans, self.tau_trans = _mcqrnn_transform(
            x=self.x, y=self.y, taus=self.taus
        )

    def __call__(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        return self.x_trans, self.y_trans, self.tau_trans

    def transform(self, x: np.ndarray, input_taus: np.ndarray) -> np.ndarray:
        input_taus = input_taus.astype("float32")
        return _mcqrnn_transform(
            x=x,
            taus=input_taus,
        )

Layers

다음으로, 각각의 Layers에 대해서 정의한다.
Cannon의 연구에서는 Input에서 First hidden layer 에서는 quantile에 대해서만 exponential function 을 사용하여 양수로 고정하고, 뒤쪽의 모든 layer와 layer 사이의 weight는 exponential function을 사용하여 양수로 고정한다.
아래의 예시처럼 weight를 양수로 만들어 다음의McqrnnInputDense, McqrnnDense, McqrnnOutputDense 를 구현할 수 있다.

# mcqrnn/models.py
from typing import Callable
import numpy as np
import tensorflow as tf


class McqrnnInputDense(tf.keras.layers.Layer):
    """
    Mcqrnn Input dense network
    Args:
        out_features (int): the number of nodes in first hidden layer
        activation (Callable): activation function e.g. tf.nn.relu or tf.nn.sigmoid
    Methods:
        build:
            Set weight shape for first call
        call:
            Return dense layer with input activation and features
    """

    def __init__(
        self,
        out_features: int,
        activation: Callable,
        **kwargs,
    ):
        super(McqrnnInputDense, self).__init__(**kwargs)
        self.out_features = out_features
        self.activation = activation

    def build(
        self,
        input_shape,
    ):
        self.w_inputs = tf.Variable(
            tf.random.normal([input_shape[-1], self.out_features]), 
            name="w_inputs",
        )
        self.w_tau = tf.Variable(
        	tf.random.normal([1, self.out_features]),
	        name="w_tau",
        )
        self.b = tf.Variable(
        	tf.zeros([self.out_features]), 
            name="b",
        )

    def call(self, inputs, tau):
        outputs = (
            tf.matmul(inputs, self.w_inputs)
            + tf.matmul(tau, tf.exp(self.w_tau))
            + self.b
        )
        return self.activation(outputs)


class McqrnnDense(tf.keras.layers.Layer):
    """
    Mcqrnn dense network
    Args:
        dense_features (int): the number of nodes in hidden layer
        activation (Callable): activation function e.g. tf.nn.relu or tf.nn.sigmoid
    Methods:
        build:
            Set weight shape for first call
        call:
            Return dense layer with activation
    """

    def __init__(
        self,
        dense_features: int,
        activation: Callable,
        **kwargs,
    ):
        super(McqrnnDense, self).__init__(**kwargs)
        self.dense_features = dense_features
        self.activation = activation

    def build(
        self,
        input_shape,
    ):
        self.w = tf.Variable(
            tf.random.normal([input_shape[-1], self.dense_features]), 
            name="w",
        )
        self.b = tf.Variable(
        	tf.zeros([self.dense_features]), 
            name="b",
        )

    def call(
        self,
        inputs: np.ndarray,
    ):
        outputs = tf.matmul(inputs, tf.exp(self.w)) + self.b
        return self.activation(outputs)


class McqrnnOutputDense(tf.keras.layers.Layer):
    """
    Mcqrnn Output dense network
    Methods:
        build:
            Set weight shape for first call
        call:
            Return dense layer with activation
    """

    def __init__(
        self,
        **kwargs,
    ):
        super(McqrnnOutputDense, self).__init__(**kwargs)

    def build(
        self,
        input_shape,
    ):
        self.w = tf.Variable(
        	tf.random.normal([input_shape[-1], 1]),
            name="w",
        )
        self.b = tf.Variable(tf.zeros([1]), name="b")

    def call(
        self,
        inputs: np.ndarray,
    ):
        outputs = tf.matmul(inputs, tf.exp(self.w)) + self.b
        return outputs

Model

이렇게 구성된 Layer들을 가지고 모델을 구성할 수 있다.
예제로, Cannon의 연구에서 제안된 Hidden layer가 한 개인 Mcqrnn를 다음과 같이 구성했다.

# mcqrnn/models.py
from typing import Callable
import numpy as np
import tensorflow as tf
from mcqrnn.layers import (
    McqrnnInputDense,
    McqrnnDense,
    McqrnnOutputDense,
)


class Mcqrnn(tf.keras.Model):
    """
    Mcqrnn simple structure
    Note that the middle of dense network can be modified with McqrnnDense
    Args:
        out_features (int): the number of nodes in first hidden layer
        dense_features (int): the number of nodes in hidden layer
        activation (Callable): activation function e.g. tf.nn.relu or tf.nn.sigmoid
    Methods:
        call:
            Return dense layer with input activation and features
    """

    def __init__(
        self,
        out_features: int,
        dense_features: int,
        activation: Callable = tf.nn.relu,
        **kwargs,
    ):
        super(Mcqrnn, self).__init__(**kwargs)
        self.input_dense = McqrnnInputDense(
            out_features=out_features,
            activation=activation,
        )
        self.dense = McqrnnDense(
            dense_features=dense_features,
            activation=activation,
        )
        self.output_dense = McqrnnOutputDense()

    def call(
        self,
        inputs: np.ndarray,
        tau: np.ndarray,
    ):
        x = self.input_dense(inputs, tau)
        x = self.dense(x)
        outputs = self.output_dense(x)
        return outputs

Loss

마지막으로 최적화의 타겟이 될 목적함수를 TiltedAbsoluteLoss로 정의했다.

# mcqrnn/loss.py
from typing import Union
import tensorflow as tf
import numpy as np


class TiltedAbsoluteLoss(tf.keras.losses.Loss):
    """
    Tilted absolute loss function or check loss
    Args:
        y_true (Union[np.ndarray, tf.Tesnsor]): train target value
        y_pred (Union[np.ndarray, tf.Tesnsor]): pred value
        tau (Union[np.ndarray, tf.Tesnsor]): quantiles
    Return:
        tf.Tensor: tilted absolute loss
    """

    def __init__(self, tau: Union[np.ndarray, tf.Tensor], **kwargs):
        super(TiltedAbsoluteLoss, self).__init__(**kwargs)
        self._one = tf.cast(1, dtype=tau.dtype)
        self._tau = tf.cast(tau, dtype=tau.dtype)

    def call(
        self,
        y_true: Union[np.ndarray, tf.Tensor],
        y_pred: Union[np.ndarray, tf.Tensor],
    ) -> tf.Tensor:
        error = y_true - y_pred
        _loss = tf.math.maximum(self._tau * error, (self._tau - self._one) * error)
        return tf.reduce_mean(_loss)

이제는 모델의 모든 구성요소를 갖췄다. 다음으로 모델을 학습할 차례이다.
하지만, 학습과 시각화까지 소개하자니 코드가 너무 길어질 것 같다...
따라서, 해당 파트는 다음 포스팅에서 다루도록 하겠다.

Remarks

오늘은 Multiple quanitile을 위한 모델을 tensorflow를 이용하여 구성하는 방법을 소개했다.
위 코드는 모두 Github repo) 에서 확인할 수 있다.