연금술사

fixture 란?

단위 테스트를 위해서, fixture을 이용하면 사전에 미리 정의된 신뢰할만하고 일관성있는 테스트를 이용할 수  있다. fixture은 환경이나 데이터셋에 대해서도 테스트해볼 수 있다. 이 fixture은 "step"과 "data"을 정의하고 이 테스트를 

fixture을 이용한 서비스, 상태, 조작환경등이 인자(argument)을 통해 모두 접근가능하고, 각 fixture는 보통 파라미터를 지정해준다. 그리고 pytest을 이용할 때, 어떤 함수를 테스트할것인지 @pytest.fixture 와 같이 "@"데코레이터를 이용해서 설정할 수 있다.

@pytest.mark.parametrize(ids=?) 내 ids란?? 

parameterize로 전달해주는 파라미터에 대해서 string으로 각 아이디를 줄 수 있고, 이는 "-k"을 옵션을 이용해서 특별한 파라미터(케이스)에 대해서만 실행해볼 수도 있다.

@pytest.fixture(scope=??):

scope범위에 따라 데코레이터를 달아준 함수의 실행범위가 달라진다. 예를 들어, scope="function"인 경우는 데코레이터를 달아준 함수를 매번 call해서 사용하고, 소멸된다. 한편, scope="module"로 지정하면 ".py"파일 내에서는 무조건 1번만 생성된다. 한편 scope="session"인 경우는 최초 테스트 실행시 단 한번만 객체가 실행되며, 각 테스트에서 이 하나의 객체가 호출된다. 

Reference: https://velog.io/@gyuseok-dev/pytest-pytest.fixture

https://velog.io/@dahunyoo/Test-fixture-on-Pytest

yeild 사용

yield은 보통 제너레이터에서 lazy하게 데이터를 불러올떄 사용한다. pytest에서는 테스트 실해 후에 환경정리, 테스트 데이터들을 정리할 때 yield 을 이용할 수 있다. yield는 호출한 테스트케이스가 실행된 후 종료되어도 마지막까지 실행되게끔 해주는 키워드이다.

역전파알고리즘은 1986년 데이비드 등이 개발한 알고리즘이다. 핵심은 "은닉 유닛(hidden unit)이 무엇을 해야하는지는 모르지만, 은닉층의 활성도를 바꿀때 얼마나 빨리 오차가 변하는지는 계산할 수 있다".

반환되는 Outcome $\hat{y}$가 변화할 때, 오차율(E)가 얼마나 변화할 것인지를 다음과 같이 미분식을 이용해보자. 아랫첨자 j은 데이터포인트이다. 

$\frac{\delta E}{\delta W_{1}} = \frac{\delta E}{\delta o_{1}} \frac{\delta o_{1}}{\delta z_{1}} \frac{\delta z_{1}}{\delta_{W_{1}}}$

첫번째 곱인 $\frac{\delta E}{\delta o_{1}}$ 은 MSE와 구하는 공식이 같아서 구할 수 있고, 

두번쨰 곱인 시그모이드(sigmoid)함수의 미분은 $\sigma'(\cdot) = \sigma(\cdot)(1-\sigma(\cdot))$ 으로 구할 수 있다.

세번째 곱은 W와 H1의 곱이 Z1이기 때문에,  $\frac{\delta z_{1}}{\delta_{W_{1}}} = h_{1}$

이를 모두 곱해주면된다. 곱한 모든 식의 아래와 같이 업데이트하는 과정을 이어나가면 끝이다.

$W_{1}^{new} = W_{1} -\alpha \frac{\delta E}{\delta W_{1}}$

위와 같은 과정을 $\hat{y}$ 을 이전레이어의 출력값으로 변경해서 층마다 업데이트하면된다.

예제를 위해서, 아래의 간단한 MLP(Multi-layer perceptron)을 생성해보았다.

import tensorflow as tf
model = tf.keras.models.Sequential()
model.add(tf.keras.layers.InputLayer(input_shape=(100)))
model.add(tf.keras.layers.Dense(10))
model.add(tf.keras.layers.Dense(1, activation='sigmoid'))

model.compile(loss='binary_crossentropy')

model.summary()

그리고 랜덤으로 xs, ys을 만들어주고, 학습시켰다.

xs = tf.constant(tf.random.uniform(shape=(100, 10)))
ys = tf.round(tf.random.uniform(shape=(100, 1)))

model.fit(xs, ys)

다행히 tensorflow2.x 은 역전파알고리즘을 tf.Gradient 객체로 한번에 구할 수 있게 만들어주어, 이를 사용하면된다.

with tf.GradientTape(persistent=True) as tape:
    tape.watch(xs)
    w = model.trainable_weights
    ys = model(xs)
    
    
dE_dW = tape.gradient(ys, w)

아래의 gradient descent방법과 같이 update을 해주면된다.

lr = 0.005
Ws = model.trainable_weights

for W, grad in zip(Ws, dE_dW):
    W = W - (lr * grad)

추가적으로 matrix derivation을 직접해보는경우 아래와같이 matrix shape이 안맞는 경우가 발생한다. 필자도 이 떄문에 한 일주일을 낭비했다..

마찬가지로, 우리가 구하고싶은것은 dE/dW 이므로, 아래와 같다.

https://medium.com/jun-devpblog/dl-3-backpropagation-98206058d72e

https://souryadey.github.io/teaching/material/Matrix_Calculus.pdf

import math
import numpy as np
import cv2

def ssim(img1, img2):
    C1 = (0.01 * 255)**2
    C2 = (0.03 * 255)**2

    img1 = img1.astype(np.float64)
    img2 = img2.astype(np.float64)
    kernel = cv2.getGaussianKernel(11, 1.5)
    window = np.outer(kernel, kernel.transpose())

    mu1 = cv2.filter2D(img1, -1, window)[5:-5, 5:-5]  # valid
    mu2 = cv2.filter2D(img2, -1, window)[5:-5, 5:-5]
    mu1_sq = mu1**2
    mu2_sq = mu2**2
    mu1_mu2 = mu1 * mu2
    sigma1_sq = cv2.filter2D(img1**2, -1, window)[5:-5, 5:-5] - mu1_sq
    sigma2_sq = cv2.filter2D(img2**2, -1, window)[5:-5, 5:-5] - mu2_sq
    sigma12 = cv2.filter2D(img1 * img2, -1, window)[5:-5, 5:-5] - mu1_mu2

    ssim_map = ((2 * mu1_mu2 + C1) * (2 * sigma12 + C2)) / ((mu1_sq + mu2_sq + C1) *
                                                            (sigma1_sq + sigma2_sq + C2))
    return ssim_map.mean()


def calculate_ssim(img1, img2):
    '''calculate SSIM
    the same outputs as MATLAB's
    img1, img2: [0, 255]
    '''
    if not img1.shape == img2.shape:
        raise ValueError('Input images must have the same dimensions.')
    if img1.ndim == 2:
        return ssim(img1, img2)
    elif img1.ndim == 3:
        if img1.shape[2] == 3:
            ssims = []
            for i in range(3):
                ssims.append(ssim(img1, img2))
            return np.array(ssims).mean()
        elif img1.shape[2] == 1:
            return ssim(np.squeeze(img1), np.squeeze(img2))
    else:
        raise ValueError('Wrong input image dimensions.')

Cox proportional hazard model

각 i번째 환자(데이터포인트) X가 있다고 하자. $X_{i}=(X_{i1}, ...X_{ip})$ 와 같이 표기할 수 있으며 i번쨰 환자의 각 p라는 변수(공변량, covariates)를 의미한다. Cox비례위험모형은 아래와 같이 작성할 수 있다[식(2)]. 

$\lambda(t|X_{i}) = \lambda_{0}(t)exp(\beta_{1}X_{i1}+...+\beta_{p}X_{ip}$ [식2]

이를 Matrix 표기($X_{i}$)로 변경하면 흔히 보는 식이다[식3]. 보통 이를 hazard function이라고 한다.

$\lambda(t|X_{i}) = \lambda_{0}(t)exp(X_{i}\beta)$ [식3]

위의 식을 들여다보면, X가 커지든 작아지든 t보다는 $\beta$와 연관이 있다.

1) 즉, 시점이 고정이라면(환자들끼리 같은 조건이라면), X가 커질수록 $\beta$가 커져서, 위험이 올라간다고 할 수 있다. 즉 시간에 따라서는 위험에 대한 비율이 일정하게 유지가 된다(비례위험ㄱ자ㅓㅇ)

2) 또한, t에 영향을 받는것은 $\lambda_{0}$만 영향을 받기때문에, 시간이 증가할수록 커진다.

그리고, 흔히 위험비(harzard ratio, HR)라고 불리는 위험에 대한 비율(집단A에 대한 집단 B에 대한 이벤트 발생에 대한 비율)을 다음과 같이 구할 수 있다. 위의 hazard function을 A집단에 있는 환자a와 B집단에 있는 환자b에 대해서 적용하고자면 아래와 같이 각각 구할 수 있다.

$\lambda(t|X_{a}) = \lambda_{0}(t)exp(\beta_{1}X_{i1a}+...+\beta_{p}X_{ia})$

$\lambda(t|X_{b}) = \lambda_{0}(t)exp(\beta_{1}X_{i1a}+...+\beta_{p}X_{ib})$

위의 두 식을 나누면 아래와 같은 식을 구할 수 있는데, 보면 $\lambda(t)$은 없어지고, $X, \beta$로만 이뤄져있다. 즉, 시점 T에는 무관하게 위험이 동일하다고 할 수 있다. 

$\frac{\lambda(t|X_{a})}{\lambda(t|X_{b})} 
= \frac{exp(\beta_{1}X_{i1a}+...+\beta_{p}X_{ia})}{exp(\beta_{1}X_{i1a}+...+\beta_{p}X_{ib})} 
= exp(\beta(X_{a}-X_{b})) $

$HR=exp(\beta(X_{a}-X_{b}) = exp(\beta_{1}(X_{a1}-X_{b1})+...+beta_{p}(X_{ap}-X_{bp}))$을 의미하고, 각 $\beta_{p}$은 $X_{p}$에 따른 위험을 의미한다. 만약 다른 환자 a와 환자b가 모든 변수가 동일하고 $x_{1}$만 다르면 어떻게될까?

$\frac{\lambda(t|X_{a})}{\lambda(t|X_{b})} 

= \frac{h(t|x_{1}=1)}{h(t|x_{1}=0)}
= exp(\beta_{1}(1-0))
= exp(\beta_{1})
$

위처럼 모든 $x_{1}$ 변수만 제외하고 모든 변수가 동일하였을때 계수 $\beta_{1}$이 $x_{1}$가 1단위 변화할때마다 $exp(\beta)$한다고 해석하고, 이를 harzard ratio 라고 한다.

Python3에서는 lifelines패키지가 생존분석을 지원하는데, 시각화가 R보다는 약하다. 아래의 예시는 Cox을 이용한 시각화 및 통계플롯인데, CI값이 안나와서 다소 아쉽지만. python3에서 지원하는게 어딘가 싶다. (https://buildmedia.readthedocs.org/media/pdf/lifelines/latest/lifelines.pdf)

from lifelines.datasets import load_rossi
from lifelines import CoxPHFitter

rossi = load_rossi()
cph = CoxPHFitter()
cph.fit(rossi, duration_col='week', event_col='arrest')
cph.print_summary()


cph.plot()

cph.plot_partial_effects_on_outcome(covariates='race', values=list(set(rossi.race)), cmap='coolwarm')

Machine learning 관련 페이퍼를 보다보면, supervise, semi-supervise, reinforcment, 등의 용어는 흔히 들어봐서 이해하지만, inductive, transductive, self-supervised 등은 약간 낯설기도하다. 비슷한 개념인것같은데 분류방식에 의한 방법으로 생각되어 다음과 같이 정리해보았다.

1. 문제의 종류 따른 분류

- Supervised,

- Unsupervised

- Reinforment.

2. Statistical inference: inference은 학습한 기계학습모델이 주어진 관찰하지 않은 데이터(X, unseen data)을 예측하는 작업을 의미한다. 아래의 3가지 작업에 "Learning", 과 "Inference"을 꼭 구분하여 읽어야한다. "inductive", "deductive"에만 꽂혀서는 의미를 알 수 없다.

- Inductive learning (귀납적 학습): 근거를 이용한 추론방법. 여태까지 주어진 데이터를 이용하여(근거,from specific historical examples), 학습하는 것을 의미한다.. 대부분의 머신러닝들이 위의 학습방법에 해당된다. 주어진데이터에 따라 머신러닝모델이 데이터를 어떻게 표현할지 파라미터를 알고있으면, 이 과정이 일반화되는 과정일 수 있다. 데이터를 기반으로 모델을학습하는 과정(데이터: 구체적인 사례, 머신러닝의 파라미터: 일반화)

- deductive inference (연역적 추론): inductive learning 의 반대다. 학습한 머신러닝의 모델을 각 개별 사례에 적용하는 것을 의미한다. 예측모델로 흔히 하고자하는 궁극적인 일이 이에 해당된다. 모델의 학습이 Induction(귀납)이라면, 모델의 사용은 연역이다.

- Transductive learning: 특정한 사례를 이용하여, 또다른 특정한 사례를 예측하는 것을 의미한다. (specific case to specific case)

[reference] https://machinelearningmastery.com/types-of-learning-in-machine-learning/

연속형변수의 구간화(Bucketized)

균등하게 자르는 방법이 있음. 아래와 같이 pandas.cut 메서드를 사용하면 8개의 구간으로 잘려 할당됨

import pandas as pd
from sklearn.datasets import load_iris

data = pd.DataFrame(load_iris()['data'], columns=load_iris()['feature_names'])
pd.cut(data['sepal width (cm)'], 8)

0      (3.2, 3.5]
1      (2.9, 3.2]
2      (2.9, 3.2]
3      (2.9, 3.2]
4      (3.5, 3.8]
          ...    
145    (2.9, 3.2]
146    (2.3, 2.6]
147    (2.9, 3.2]
148    (3.2, 3.5]
149    (2.9, 3.2]
Name: sepal width (cm), Length: 150, dtype: category
Categories (8, interval[float64]): [(1.998, 2.3] < (2.3, 2.6] < (2.6, 2.9] < (2.9, 3.2] < (3.2, 3.5] < (3.5, 3.8] < (3.8, 4.1] < (4.1, 4.4]]