Day1

🟢feature importance 확인

import pandas as pd
df = pd.DataFrame(dt.feature_importances_).T

df.columns=dt.feature_names_in_

df.T.sort_values(by=[0],ascending=False)

	0
concave_points1	0.691420
concave_points3	0.065651
texture1	0.058478
radius3	0.052299
perimeter3	0.051494

feature importance

• 파레토 법칙에 따라 너무 값이 작은 건 쳐낼 수도 있음.
  • 모든 현상에서 이런 경향성을 보인다.
  • drop out이라는 기법으로 이런 현상을 해결하려고 한다.

🔵feature importance 시각화

from sklearn.tree import plot_tree
import matplotlib.pyplot as plt

plt.figure(figsize=(15,15))
plot_tree(dt)
plt.show()

• 트리는 nonlinear의 대표적인 structure

🔵그래프 활용

• 노드의 크기를 통해 비교 분석.
• Neo4j. → 그래프 DB

🔵지니계수

• 정보 이득은 크게. 불순도는 작게 설계해야 함.

🟢하이퍼파라미터 튜닝 & 교차검증을 동시에

from sklearn.model_selection import GridSearchCV

params = {'max_depth' : [3, 4, 5, 6], 'min_samples_split' : [2,3,4,5,6]}
gs = GridSearchCV(DecisionTreeClassifier(random_state=42), params, n_jobs=-1)

gs.fit(X_train,y_train)

• 원하는 파라미터 종류 작성하면 코드가 알아서 조합을 짜서 확인함

• 한 번에 n개의 작업을 돌릴 수 있는 것 - threads

• threads - dead lock

• 파이썬은 GIL 걸려있는 상황

• 그리드서치에서 n_jobs 파라미터가 여기서 병렬처리를 몇 개 거는지 결정

gs.best_estimator_

gs.best_params_

{'max_depth': 3, 'min_samples_split': 2}

gs.cv_results_['mean_test_score']

array([0.92967033, 0.92967033, 0.92967033, 0.92967033, 0.92967033,
0.92307692, 0.92307692, 0.92087912, 0.92087912, 0.92307692,
0.92527473, 0.92527473, 0.92747253, 0.92307692, 0.92527473,
0.91648352, 0.91648352, 0.91868132, 0.91868132, 0.91648352])

import numpy as np
from sklearn.model_selection import GridSearchCV

params = {'min_impurity_decrease': np.arange(0.0001,0.001,0.0001),
          'max_depth':range(3,20,1),
          'min_samples_split':range(2,100,10)
          }

gs = GridSearchCV(DecisionTreeClassifier(random_state=42), params, n_jobs=-1)

gs.fit(X_train,y_train)

• 좀 더 다양한 파라미터를 사용하여 하이퍼 파라미터 튜닝

• 갤럭시 환경에서는 쓰로틀링 있을 수 있음
  • cpu 능력치 발열 위험으로 인한 제한

gs.cv_results_['mean_test_score'].__len__()

1530

gs.best_params_

{'max_depth': 4,
'min_impurity_decrease': np.float64(0.0001),
'min_samples_split': 12}

np.max(gs.cv_results_['mean_test_score'])

🔵찾아낸 최적의 파라미터로 모델을 학습시킨다

from sklearn.metrics import accuracy_score

dt = DecisionTreeClassifier(max_depth=4,
 min_impurity_decrease= np.float64(0.0001),
 min_samples_split= 12)

dt.fit(X_train,y_train)

accuracy_score(y_test, dt.predict(X_test))

0.9473684210526315

🔵uniform distribution 기반 RandomizedSearch

from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import uniform, randint
params = {
    'min_impurity_decrease' : uniform(0.0001, 0.001),
    'max_depth' : randint(3,100), 
    'min_samples_split' : randint(2,30), 
    'min_samples_leaf' : randint(1,30)
}

rs = RandomizedSearchCV(DecisionTreeClassifier(random_state=42),  
                   params, n_iter=100, n_jobs=-1, random_state=42)

rs.fit(X_train,y_train)

rs.best_params_

{'max_depth': 54,
'min_impurity_decrease': np.float64(0.0006632755719763837),
'min_samples_leaf': 4,
'min_samples_split': 6}

rs.best_estimator_

🟢MLFlow

🔵오늘의 목적 : MLOps 도커에 빌드하기

https://pypi.org/project/mlflow/

• databricks에서 개발

🔵도커 환경 시작

1. docker ps -a
2. sk25 존재 확인
3. docker start sk25

🔵Docker에 라이브러리 설정

• requirements.txt 권한 - 644
• :wq
  • 쓰고 나가라는 명령어

rosie@ming9:~/encore$ history |grep build
  208  docker build -t sk25:0.1 .
  237  history |grep build
rosie@ming9:~/encore$

• 명령어 history : 과거에 쳤던 명령어 이력

rosie@ming9:~/encore$  docker build -t sk25:0.2 .

• requirements.txt 수정 후 다시 빌드

환경 구축 자동화 - 젠킨스. 못생김.

현재 docker images 저장 목록들

docker images

docker stop sk25 

docker run -itd --name sk25_mlflow -p 9000:8888 -p 5000:5000 -v ~/mydata:/home/ict/work sk25:0.2

• mlflow는 포트 5000번 사용
• 주피터 서버 열어서 vscode에 연결

• Existing Jupyter server에서 주피터 노트북 서버 연결

• 멀쩡히 연결됨을 확인

• 주피터 서버 연결되어있는 곳에서 서버 열기

# bash
root@a5ba42c017e5:/home/ict/work/wk6# mlflow server --port 5000 --host 0.0.0.0
Backend store URI not provided. Using sqlite:///mlflow.db

• -ops로 끝나는 키워드는 모두 컨테이너 기반이다.
• 현재 하고 있는 것 : 컨테이너 기반 머신러닝 모델링.
• 기술 스택 indeed → 그 기술스택 쓰는 채용공고 연결

rosie@ming9:~/encore$ docker exec -it sk25_mlflow /bin/bash

mlflow server --port 5000 --host 0.0.0.0

🔵실험 띄우기

import mlflow
mlflow.set_tracking_uri("http://localhost:5000")

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split 
import pandas as pd

mlflow.set_experiment('sk25_first')

• 바닐라 모델 (=baseline)

🔵mlflow 돌려보기 -수동모드

from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import accuracy_score, confusion_matrix
from sklearn.model_selection import train_test_split

iris = load_iris()
X = pd.DataFrame(iris.data, columns=iris.feature_names)
y = iris.target

# 데이터 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

from sklearn.metrics import accuracy_score, classification_report
with mlflow.start_run(run_name = 'baseline'):
    params = {
        'max_depth' : 3,
        'random_state' : 42
    }

    mlflow.log_params(params)
    dt = DecisionTreeClassifier(**params)
    dt.fit(X_train, y_train)

    y_pred = dt.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)

    mlflow.log_metric('accuracy', accuracy)

    mlflow.sklearn.log_model(dt, 'decision_model')

    report = classification_report(y_test, y_pred)
    mlflow.log_dict(report, 'classification_report.json')

    print(mlflow.active_run().info.run_id)
    

🔵mlflow 돌려보기 -자동 모드

mlflow.set_experiment('sk25_second')
mlflow.sklearn.autolog()

with mlflow.start_run(run_name = 'haha'):
    model = DecisionTreeClassifier(**params)
    model.fit(X_train, y_train)
    y_pred = dt.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)

🟢모델 서빙

🔵모델 등록 후 다시 호출

import mlflow

mlflow.set_tracking_uri("http://localhost:5000")

model = mlflow.sklearn.load_model("models:/sk25/1")

import numpy as np
model.predict(np.array([5.1,3.5,1.4,0.2]).reshape(-1,4))

🔵환경변수 선언하여 모델 서빙

rosie@ming9:~$ docker exec -it sk25_mlflow /bin/bash
root@a5ba42c017e5:/home/ict/work# export MLFLOW_TRACKING_URI=http://localhost:5000
root@a5ba42c017e5:/home/ict/work#

root@a5ba42c017e5:/home/ict/work# mlflow models serve -m "models:/sk25/1" -p 123 --env-manager=local

🔵서빙된 모델로 예측

import requests

url='http://127.0.0.1:123/invocations'

import pandas as pd
df = pd.DataFrame(
    [[5.1, 3.5, 1.4, 0.2]], 
    columns=["sepal length (cm)", "sepal width (cm)", "petal length (cm)", "petal width (cm)"]
)

{'dataframe_split': df.to_dict(orient='split')}

<output>
{'index': [0],
 'columns': ['sepal length (cm)',
  'sepal width (cm)',
  'petal length (cm)',
  'petal width (cm)'],
 'data': [[5.1, 3.5, 1.4, 0.2]]}

서버가 받는 데이텨 형태로 변형

requests.post(url,json={'dataframe_split': df.to_dict(orient='split')}).text

'{"predictions": [0]}'

🟢앙상블

🔵bagging	🔵voting	🔵boosting
다수의 decision tree	soft
• 레이블 값을 결정하는 확률값을 모두 더해서 사용	xgboost, lightgbm
다수 dt가 각자의 데이터를 독립적으로 사용(샘플링)	hard

• 사람이 하는 투표와 동일
• 다른 종류의 알고리즘 혼합 가능 | |
| ex) random forest | soft가 hard voting보다 성능이 앞선다고 알려짐 | |

클래스가 불균형할 때 사용 → smote

• 둘 중 적은 것에 개수 맞춤 → undersampling
• 둘 중 많은것에 맞춰 뻥튀기함→ oversampling , smote

🟢Optuna

mlflow랑 결합하여 추적할 때 사용 가능

Day2

🟢다중분류

from sklearn.metrics import classification_report

y_true = [0,0,0,0,0,1,1,1,2,2]
y_pred = [0,0,0,0,1,0,1,2,1,2]

print(classification_report(y_true, y_pred))

              precision    recall  f1-score   support

           0       0.80      0.80      0.80         5
           1       0.33      0.33      0.33         3
           2       0.50      0.50      0.50         2

    accuracy                           0.60        10
   macro avg       0.54      0.54      0.54        10
weighted avg       0.60      0.60      0.60        10

🟢KNN

🔵 지도학습

• y값을 따로 선언해주어야 한다.
• 개념 이해를 위한 것으로, 굳이 train test를 나누지 않는다.

from sklearn.neighbors import KNeighborsClassifier
from sklearn.datasets import load_iris

knn = KNeighborsClassifier()
knn.fit(load_iris().data, load_iris().target)

• 기본 이웃값 : 5
• 유클라디안 기본 값

🔵OPS(container & k8s)

• MLops, AIops. → 공부 필요

🟢파산데이터로 공부해보자

import pandas as pd
df= pd.read_csv("./data/credit_card_default.csv")

df.shape

y= df.pop('default_payment_next_month')
X= df.copy()

import numpy as np
np.unique(y,return_counts=True)

`(array([0, 1]), array([23364,  6636]))`

🔵 EDA

히스토그램 - 남녀별 나이 분포

import matplotlib.pyplot as plt 
import seaborn as sns 

fig, ax = plt.subplots()
sns.histplot(df.loc[df.sex == 'Male', 'age'])
# sns.kdeplot(df.loc[df.sex == 'Male', 'age'],  ax=ax)
plt.show()

커널밀도추정

남녀 동시에 커널밀도값 보기

• 둘이 분포가 비슷하므로, 남녀 모두 비슷한 양상을 보인다 할 수 있다.

범주형, 양적 데이터 상관없이 상관관계 한 번에 보기

• sns.pairplot

pair_plot = sns.pairplot(df[['age', 'limit_bal', 'previous_payment_sep']])
plt.show()

상관관계 표시

df.select_dtypes(include='number').corr()

삼각행렬 활용해서 삼각형 절반만 출력하기

# 상관계수 히트맵 그리기 
import numpy as np 
mask = np.zeros_like(df.select_dtypes(include='number').corr(), dtype=bool)
mask[np.triu_indices_from(mask)] = True
#df.select_dtypes(include='number').corr()
fig, ax = plt.subplots()
cmap = sns.diverging_palette(240, 10, n=9, as_cmap=True)
sns.heatmap(df.select_dtypes(include='number').corr(), mask=mask, vmax=3, cmap=cmap, 
            center=0, square=0, linewidths=0.5, ax = ax)
ax.set_title("correlation Matrix", fontsize=20)
plt.show()

• 나이는 다른 변수와 큰 관계성이 없는 것으로 보임

# 상관계수 히트맵 그리기 
import numpy as np 
matrix_corr = pd.concat([df, y], axis=1).select_dtypes(include='number').corr()
mask = np.zeros_like(  matrix_corr    , dtype=bool)
mask[np.triu_indices_from(mask)] = True
#df.select_dtypes(include='number').corr()
fig, ax = plt.subplots()
cmap = sns.diverging_palette(240, 10, n=9, as_cmap=True)
sns.heatmap(matrix_corr, mask=mask, vmax=0.3, cmap=cmap, 
            center=0, square=0, linewidths=0.5, ax = ax)
ax.set_title("correlation Matrix", fontsize=20)
plt.show()

vis_df=pd.concat([df['education'] , y],axis=1)

data=vis_df.groupby('education')['default_payment_next_month'].value_counts(normalize=True).plot(kind='barh')

vis_df=pd.concat([df['education'] , y],axis=1)

data=vis_df.groupby('education')['default_payment_next_month'].value_counts(normalize=True).unstack().plot(kind='barh',stacked=True)

#pip install missingno

import missingno
missingno.matrix(X)
plt.show()

missingno.bar(X)

타겟 인코더

!pip install category-encoders==2.6.3

import category_encoders 

import category_encoders 

target_en = category_encoders.TargetEncoder(smoothing=0)
target_en.fit(X.sex,y)

pd.concat([X.sex, target_en.transform(X.sex)] ,axis=1).iloc[:, 1].value_counts()

sex
0.207578    18027
0.241648    11823
0.246667      150
Name: count, dtype: int64

• 모델 성능이 올라갈 수 있다
• 원핫은 클래스가 많으면 컬럼이 많아지지만, 특징을 더 잡게 잡을 수 있다.
• https://freeislet.tistory.com/20
• 범주형 변수의 각각의 범주를 타겟 변수의 특정 통계치, 예를 들어 타겟값 평균으로 변환하는 방식이다.
• 통계치
  • 이진 분류- 타겟값이 1인 확률
  • 회귀 모델- 평균
  • 결국은 타겟값의 평균값으로 볼 수 있음
• 평균 대신 분산, 왜도(skewness), 첨도(kurtosis) 등 사용 가능

🟢결측치 처리

🔵simple imputer 기반

num_feautures=['age']
cat_features=['sex','education','marriage']

from sklearn.impute import SimpleImputer

num_imputer = SimpleImputer(strategy='median')
num_imputer.fit(X[['age']])
no_missing_values = num_imputer.transform(X[['age']])

X.loc[:, 'age_nomv' ]  = no_missing_values

• 로지스틱 회귀로 분류 테스크 적용할 경우 , 양적 데이터를 스케일링 해야한다.
• 그냥 디시젼 트리로 분류 테스크 수행할 경우, 스케일링은 필요없지만 인코딩 필수

오토인코더

• input 데이터와 output 데이터의 형태가 똑같다.
• VAE → Stable diffusion, 미드저니

🟢도커에서 설정을 만져본다

🔵requirements.txt 수정

명령어 모드에서 대문자 G → 맨 아래로 이동

소문자 g 2번 → 맨 위로 이동

:set nu → vim에 숫자번호 추가

/xgboost → / 키워드는 검색

dd → 라인 제거

u → 원상복구

[숫자]dd → 숫자에서 삭제

vim /etc/vim/vimrc → 환경파일

빈 줄에 set nu 입력 시 vim에 기본적으로 번호 설정

🔵Dockerfile 설정

1 FROM python:3.10.8-slim-bullseye
  2
  3
  4 # 작업 디렉토리 설정
  5 WORKDIR /home/ict/work
  6
  7 ENV MLFLOW_TRACKING_URI = http://localhost:5000
  8
  9
 10 COPY . /home/ict/work
 11
 12 RUN apt-get update && apt-get install -y \
 13     python3-dev default-libmysqlclient-dev build-essential pkg-config graphviz
 14
 15
 16 RUN pip install --upgrade pip
 17 RUN pip install -r requirements.txt
 18
 19 RUN jupyter notebook --generate-config
 20 COPY jupyter_notebook_config.py /root/.jupyter/
 21
 22
 23 EXPOSE 9000
 24
 25
 26 CMD ["jupyter", "lab", "--ip=0.0.0.0", "--allow-root"]
~
~
~
                    

• 7 ENV MLFLOW_TRACKING_URI = http://localhost:5000
  • 환경변수 추가

rosie@ming9:~/encore$ docker rmi sk25:0.1

• 전에 빌드했던 도커 이미지 제거

docker build -t sk25:0.3 .

• dockerfile과 requirements.txt 수정했으므로 이미지따라 다시 빌드

docker run -itd --name sk25_mlflow -p 9000:8888 -p 5000:5000 -v ~/mydata:/home/ict/work sk25:0.3

• 다시 서버 열기

🔵주피터에서 환경 잘 구축되었는지 보기

• 멀쩡 ^~^

🟢XGboost

# eXtreme Gradient Boosting
import xgboost as xgb
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report

import pandas as pd
df=pd.DataFrame(load_breast_cancer()['data'],columns=load_breast_cancer().feature_names)

🔵EDA 자동화

from ydata_profiling import ProfileReport
profile = ProfileReport(df, title = 'eda')
profile.to_file('./report.html')

🔵모델 선언 및 훈련

data = load_breast_cancer()
X = pd.DataFrame(data.data, columns=data.feature_names)
y = data.target

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

model = xgb.XGBClassifier(
    n_estimators=100, # 트리의 개수 
    max_depth=4,  # 깊이 제한
    learning_rate=0.1, 
    random_state=42,
    objective='binary:logistic',
    use_label_encoder=False,
    eval_metric='logloss'  # 회귀 : rmse , mae  # 분류 : logloss, auc 
)

model.fit(X_train, y_train)

🔵성능지표

print(classification_report(y_test, model.predict(X_test)))

🔵변수 중요도

import matplotlib.pyplot as plt
from xgboost import plot_importance
plt.figure(figsize=(10,8))
plot_importance(model, max_num_features=10)
plt.show()

🔵shap

• 협력적 게임이론에서 나오는 이론적 개념을 바탕으로 머신러닝 모델이 가지고 있는 해석을 할 수 있게 하는 법.

pip install shap --upgrade

• 버전 맞추기(numpy 최신것과 맞게)

#설명 가능한 인공지능(xai)

import shap
# 협력적 게임 이론에서 나오는 이론적 개념을 바탕으로 머신러닝 모델 해석에 적용한 방법론 
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)
shap.summary_plot(shap_values, X_test, feature_names=data.feature_names)

🟢optuna - 하이퍼파라미터 튜닝

import optuna

def objective(trial):
    param = {
            'n_estimators': trial.suggest_int('n_estimators', 100, 1000),
            'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
            'max_depth': trial.suggest_int('max_depth', 3, 10),
            'subsample': trial.suggest_float('subsample', 0.5, 1.0),
            'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
            'reg_alpha': trial.suggest_float('reg_alpha', 1e-8, 1.0, log=True), # L1
            'reg_lambda': trial.suggest_float('reg_lambda', 1e-8, 1.0, log=True), # l2
            'n_jobs': -1,
            'random_state': 42,}
    model=xgb.XGBClassifier(**param)
    pruning_callback=optuna.integration.XGBoostPruningCallback(trial, 'validation_0-logloss')

• pruning_callback
  • 콜백함수 활용하여 성능이 더 나아지지 않으면 작동을 멈춤

study = optuna.create_study(direction='maximize')

• 최대화시키는 방향으로 설정

🔵전체 코드

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=50)

# 콜백  정의
pruning_callback = optuna.integration.XGBoostPruningCallback(trial, "validation_0-logloss")

#하이퍼파라미터 설정
param = {
    'n_estimators': trial.suggest_int('n_estimators', 100, 1000),
    'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
    'max_depth': trial.suggest_int('max_depth', 3, 10),
    'subsample': trial.suggest_float('subsample', 0.5, 1.0),
    'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
    'reg_alpha': trial.suggest_float('reg_alpha', 1e-8, 1.0, log=True),
    'reg_lambda': trial.suggest_float('reg_lambda', 1e-8, 1.0, log=True),
    'n_jobs': -1,
    'random_state': 42,
    'eval_metric': 'logloss',
    'callbacks': [pruning_callback]
}

# 모델 정의
model = xgb.XGBClassifier(**param)

# 모델 학습
model.fit(
    X_train,
    y_train,
    eval_set=[(X_test, y_test)],
    verbose=False
)

preds = model.predict(X_test)
accuracy = accuracy_score(y_test, preds)
return accuracy



def objective(trial):
data = load_breast_cancer()
X, y = data.data, data.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

🔵mlflow로 상황보기

import optuna
import xgboost as xgb
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import mlflow
import mlflow.xgboost

mlflow.set_tracking_uri("http://localhost:5000")

mlflow.set_experiment("XGBoost_SHAP_Analysis")

data = load_breast_cancer()
X, y = data.data, data.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

def objective(trial):
    

    with mlflow.start_run(run_name=f"trial_{trial.number}", nested=True):

        # callback 함수 
        pruning_callback = optuna.integration.XGBoostPruningCallback(trial, "validation_0-logloss")

        # 하이퍼파라미터 설정
        param = {
            'n_estimators': trial.suggest_int('n_estimators', 100, 1000),
            'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3, log=True),
            'max_depth': trial.suggest_int('max_depth', 3, 10),
            'subsample': trial.suggest_float('subsample', 0.5, 1.0),
            'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
            'reg_alpha': trial.suggest_float('reg_alpha', 1e-8, 1.0, log=True),
            'reg_lambda': trial.suggest_float('reg_lambda', 1e-8, 1.0, log=True),
            'n_jobs': -1,
            'random_state': 42,
            'eval_metric': 'logloss',
            'callbacks': [pruning_callback] 
        }
        
        mlflow.xgboost.autolog(log_input_examples=True, log_model_signatures=True, silent=True)
        
    
        model = xgb.XGBClassifier(**param)
    
        # 모델 학습
        model.fit(
            X_train, 
            y_train, 
            eval_set=[(X_test, y_test)], 
            verbose=False
        )
    
        preds = model.predict(X_test)
        accuracy = accuracy_score(y_test, preds)
        
        mlflow.log_metric("val_accuracy", accuracy)
        
        return accuracy

with mlflow.start_run(run_name="HPO_Session_Main") as parent_run:
    print(f"Parent Run ID: {parent_run.info.run_id}")
    study = optuna.create_study(direction='maximize')
    print("XGBoost 최적화 시작...")
    study.optimize(objective, n_trials=50)

    
    print("\n================ 결과 ================")
    print(f"Best Trial Score: {study.best_value:.4f}")

🔵베이지안 옵티마이저

• optuna
• surrogate model - 추측하는 모델
• acquisition model - 판단하는 모델
  • 판단 및 활용

Day3

🟢전형적인 불균형 데이터를 모델링한다

import pandas as pd

df= pd.read_csv("./creditcard.csv")

df['Class'].value_counts(normalize=True)

Class
0    0.998273
1    0.001727
Name: proportion, dtype: float64

• 금융데이터는 비식별처리가 필수적이다

from sklearn.model_selection import train_test_split
y = df.pop('Class')
X_train, X_test, y_train,y_test= train_test_split(df,y,random_state=42,stratify=y)
print(y_train.value_counts(normalize=True))
print(y_test.value_counts(normalize=True))

Class
0    0.998273
1    0.001727
Name: proportion, dtype: float64
Class
0    0.998273
1    0.001727
Name: proportion, dtype: float64

🔵boxplot으로 x 값을 확인하다

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(10,6))
df.boxplot()
plt.show()

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(10,6))
df[['Amount']].boxplot()
plt.show()

X_train.drop(columns='Time',inplace=True)
X_test.drop(columns='Time',inplace=True)

• 무의미한 변수 제거

X_train.Amount.describe()

count    213605.000000
mean         88.216558
std         250.522258
min           0.000000
25%           5.640000
50%          22.000000
75%          77.500000
max       25691.160000
Name: Amount, dtype: float64

• 최댓값이 너무 큼

🔵두 종류로 스케일링 해본다.

from sklearn.preprocessing import StandardScaler

ss=StandardScaler()
amount_ss=ss.fit_transform(X_train[['Amount']])
amount_ss=pd.DataFrame(amount_ss)

from sklearn.preprocessing import MinMaxScaler

mm=MinMaxScaler()
amount_mm=mm.fit_transform(X_train[['Amount']])
amount_mm=pd.DataFrame(amount_mm)

• 이상치를 쉽게 해결하지 못한다.

🔵표준화

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(10,6))
amount_ss.boxplot()
plt.show()

• 최댓값 100.

🔵정규화

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(10,6))
amount_mm.boxplot()
plt.show()

• 최댓값 1
• 그래도 이상치 해결에 근본적 도움이 안됨

🔵로그변환 시도

• 데이터의 분포도가 심각하게 왜곡되어 있을 때 시도해볼 수 있다.

import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

log_df=pd.DataFrame(np.log1p(X_train[['Amount']]))
plt.figure(figsize=(10,6))
log_df.boxplot()
plt.show()

🟢딥러닝

• 비타민 기준 세션 14주차

https://brilliant.org/wiki/truth-tables/

https://brilliant.org/wiki/logic-gates/

• 퍼셉트론
  • and, or는 선형적 분리가 가능하나 xor는 하지 못했던 것에서 오는 문제.
  • xor - 두 값이 다르면 1을 출력함.

• 값과 비슷하지 않을 경우 가중치와 편향을 변형해야함
• 식의 경사를 알기 위해 사용한다.

모든 파라미터를 편미분을 활용하여 값을 조절함

• 그러나 시간이 아주 오래 걸림
• mnist 데이터

미분의 연쇄법칙 활용하여 품, 오차역전파

• 자료구조 중에서 그래프 구조에 해당
• 계산량이 확 줄어들 수 있게 됨

순방향, 역방향에 따른 계산법 상이

https://compmath.korea.ac.kr/deeplearning/BackPropagation.html

• 활성화 함수 - 시그모이드 함수
  • 선형을 비선형으로 바꿔줌.
  • sigmoid 미분값 : f(x)(1-x)

DNN은 유명한 완전연결층임

• fully connected neural network

각 동그라미를 node, neural이라고 부른다.

역전파

• 계산그래프 기준, 아메리카노 가격에 대한 지불 금액의 미분이 3.3
• 커피 값이 오르면 최종 금액은 그 금액의 3.3만큼 오른다.
• 만약 커피 최종 값이 8000원이 나오면, 그 3.3 을 통해 input이 2424원이었음을 예측할 수 있다.
• 모델 예측 값이 6600 이면 두 차는 1400원 . 1400/3.3 하면 424. input에 대해 424를 추가해야한다는 인사이트 도출 가능
• 역전파에서 덧셈을 마주할 경우 그냥 패스
  • 이런 개념에서 국소적 계산이 더욱 가능해짐
• 오차역전파를 통해서 미분을 복잡하게 안해도 계산할 수 있다.

🟢딥러닝에서의 성능지표

lose function과 cost function을 줄이는 방향

1. 회귀 - MSE
2. 분류 - entropy

왼쪽에서 오른쪽으로 간다 → 순전파

• 각 노드가 행하는 연산은 국소적이다(local)
  • 그래프 계산의 특징.

🔵확률적 경사 하강법 - SGD

• 값이 적어지는 방향으로 이동

• 데이터가 적을 경우 과적합 될 수 있음
• learning rate 작으면 ? ⇒ local minimum에 빠질 수 있음

🔵optimizer

• 가속도 붙어서 내려감 → 모멘텀
• 이미지 참고

🟢딥러닝 프레임워크

🔵tensorflow

• tensorflow 1 버전과 2버전은 너무 다름
• 1버전 - 직접 그래프 구조를 코드로 작성하여 사용해야함
• 2버전 - keras와 병합하여 사이킷런처럼 사용할 수 있도록 함

🔵pytorch

• 메타에서 제작

🔵편미분

🔵ollama 플랫폼

• 공개 오픈소스

🔵배치 학습법

• 밥 퍼먹을 때 숫가락 개수
• 데이터 한 번 소진 → 1 epoch

• gpu에서 중요한 것 → vram
• vram이 작으면 memory fault 생김

🔵딥러닝 3대장

• DNN
• CNN
• RNN - Recurrent?

🟢이상치 제거

• Boxplot ->        몸통 -> IQR    ( Q3 - Q1 )

q1  = np.percentile(X_train.Amount, 25)
q3  = np.percentile(X_train.Amount, 75)
iqr = q3 - q1

lowest = q1 - (iqr * 1.5)
highest = q3 + (iqr * 1.5)

X_train[(X_train.Amount < lowest) |  (X_train.Amount > highest)].index

remove_index = X_train[(X_train.Amount < lowest) |  (X_train.Amount > highest)].index
X_train_remove = X_train[~X_train.index.isin(remove_index)].copy()
y_train_remove = y_train[~y_train.index.isin(remove_index)].copy()

🟢smote

• 이상 데이터를 처리할 때 샘플링을 사용한다.
• over sampling

smote = SMOTE(random_state=42)
X_train_sm, y_train_sm = smote.fit_resample(X_train, y_train)

y_train_sm.value_counts()

Class
0    213236
1    213236
Name: count, dtype: int64

• 적던 클래스의 개수를 확장.

import matplotlib.pyplot as plt
import matplotlib.ticker as ticker
from sklearn.metrics import precision_recall_curve
%matplotlib inline

def precision_recall_curve_plot(y_test , pred_proba_c1):
    # threshold ndarray와 이 threshold에 따른 정밀도, 재현율 ndarray 추출. 
    precisions, recalls, thresholds = precision_recall_curve( y_test, pred_proba_c1)
    
    # X축을 threshold값으로, Y축은 정밀도, 재현율 값으로 각각 Plot 수행. 정밀도는 점선으로 표시
    plt.figure(figsize=(8,6))
    threshold_boundary = thresholds.shape[0]
    plt.plot(thresholds, precisions[0:threshold_boundary], linestyle='--', label='precision')
    plt.plot(thresholds, recalls[0:threshold_boundary],label='recall')
    
    # threshold 값 X 축의 Scale을 0.1 단위로 변경
    start, end = plt.xlim()
    plt.xticks(np.round(np.arange(start, end, 0.1),2))
    
    # x축, y축 label과 legend, 그리고 grid 설정
    plt.xlabel('Threshold value'); plt.ylabel('Precision and Recall value')
    plt.legend(); plt.grid()
    plt.show()
    

precision_recall_curve_plot( y_test, lr.predict_proba(X_test)[:, 1] )

• 로지스틱 회귀로 찍어본 성능지표

• xgboost로 찍어본 성능지표

jax → 최적화를 위한 라이브러리.

linux arm 체제 존재. 휴대폰 안에 들어있음

• arm - 영국 회사, soft bank

🔵코사인 유사도

• 벡터 간의 거리로 식 도출 가능
• 우리의 말이 숫자로 바뀌면 임베딩
• pca , 아이겐 벨류, 아이겐 벡터
• 코사인 90은 0

Day4

🟢딥러닝을 시작한다

🔵캐글 환경 구축

• 번호 인증 시 accelator → gpu 가능
• 잘 있는지 확인

🔵GPU 켜져있는지 확인

import torch
torch.cuda.is_available()

`True`

🔵CPU&GPU 버전 확인

🔵메모리 할당 여부 확인

🔵뉴럴 네트워크를 시작한다.(단순 선형회귀)

데이터 생성

import torch.nn as nn
import torch.optim as optim 

torch.manual_seed(42)
data= torch.randn(100,1)*10

시각화

import matplotlib.pyplot as plt

x = data 
y  = 2 * data + 1 +  torch.randn(100, 1) * 2
plt.scatter(x, y)
plt.show()

모델을 선언해본다

• torch는 모델을 클래스로 선언해야한다.
• torch는 nn.Module를 상속받도록 가이드하고 있다.

class LRModel(nn.Module):
    def __init__(self):
        pass
    def forword(self,y):
        pass

• 모델 기본 구조

초기함수에 사용 원하는 함수 설정 가능

• 부모를 호출해야함 - super

    def __init__(self):
        super(LRModel, self).__init__()
        self.linear = nn.Linear(1,1)

• 아직 선형회귀이므로 활성화 함수 넣지 않음(RELU, leaky RELU)

class LRModel(nn.Module):
    def __init__(self):
        super(LRModel, self).__init__()
        self.linear = nn.Linear(1,1)
        
    def forword(self,x):
        return self.linear(x)

🔵불러와보기

model=LRModel()
model.state_dict().items()

odict_items([('linear.weight', tensor([[-0.0098]])), ('linear.bias', tensor([-0.7605]))])

• 초기값 선언 때 랜덤으로 선언됨
• 확률분포를 알면 너가 대충 어디 출신인지 파악할 수 있다→ 모수적 통계기법

🔵성능지표

• loss function, cost function → 회귀 mse, 분류 entropy
• 현재로서는 회귀 문제이므로 mse 사용
• 옵티마이저는 파라미터 넘겨서 SGD 사용

criterion = nn.MSELoss()
optimizer = optim.SGD(model.parameters(), lr = 0.001)

🔵모델 학습시키기

epochs = 100
for epoch in range(epochs):
    pred_y = model(x)

    loss = criterion(pred_y, y)
    # print(optimizer.state_dict())
    optimizer.zero_grad()
    loss.backward()

    optimizer.step()
    if (epoch + 1) % 10 == 0:
        print(model.state_dict().items())
        print(f"{epoch+1} Loss: {loss.item():.3f}")

odict_items([('linear.weight', tensor([[1.8260]])), ('linear.bias', tensor([0.8543]))])
10 Loss: 7.860
odict_items([('linear.weight', tensor([[1.9830]])), ('linear.bias', tensor([0.8595]))])
20 Loss: 3.230
odict_items([('linear.weight', tensor([[2.0012]])), ('linear.bias', tensor([0.8638]))])
30 Loss: 3.165
odict_items([('linear.weight', tensor([[2.0034]])), ('linear.bias', tensor([0.8679]))])
40 Loss: 3.163
odict_items([('linear.weight', tensor([[2.0036]])), ('linear.bias', tensor([0.8719]))])
50 Loss: 3.161
odict_items([('linear.weight', tensor([[2.0036]])), ('linear.bias', tensor([0.8759]))])
60 Loss: 3.160
odict_items([('linear.weight', tensor([[2.0036]])), ('linear.bias', tensor([0.8797]))])
70 Loss: 3.158
odict_items([('linear.weight', tensor([[2.0035]])), ('linear.bias', tensor([0.8835]))])
80 Loss: 3.157
odict_items([('linear.weight', tensor([[2.0035]])), ('linear.bias', tensor([0.8872]))])
90 Loss: 3.155
odict_items([('linear.weight', tensor([[2.0035]])), ('linear.bias', tensor([0.8909]))])
100 Loss: 3.154

• loss.backward() : 예측값과 실제값 기준으로 값을 보정해줌.

model.to(’cuda’) ⇒ vram에 파라미터 올리기. 속도가 개선될 수 있음

predicted=model(x).detach().numpy()

명령어 설명

x (입력)
 ↓
model(x)        → 모델이 계산한 예측값 (Tensor)
 ↓
.detach()       → 학습 그래프에서 분리
 ↓
.numpy()        → NumPy 배열로 변환
 ↓
predicted       → "예측값"이 담김

🔵실제값과 예측값 시각화

predicted=model(x).detach().numpy()
plt.scatter(x.numpy(),y.numpy())
plt.scatter(x.numpy(),predicted, color='red')

🔵파이토치 버전 체크

if torch.cuda.is_available():
    DEVICE = torch.device('cuda')
else:
    DEVICE = torch.device('cpu')

print('Using PyTorch version:', torch.__version__, ' Device:', DEVICE)

🟢MNIST 데이터로 공부해본다.

DNN 기반 모델링을 시작한다………………

🔵데이터 자동 로드

from torchvision import transforms, datasets

BATCH_SIZE = 32
EPOCHS = 10
train_dataset = datasets.MNIST(root = "../data/MNIST",
                               train = True,
                               download = True,
                               transform = transforms.ToTensor())

test_dataset = datasets.MNIST(root = "../data/MNIST",
                              train = False,
                              transform = transforms.ToTensor())

• transform = numpy 데이터를 torch에 맞게 전환환

train_loader = torch.utils.data.DataLoader(dataset = train_dataset,
                                           batch_size = BATCH_SIZE,
                                           shuffle = True)

test_loader = torch.utils.data.DataLoader(dataset = test_dataset,
                                          batch_size = BATCH_SIZE,
                                          shuffle = False)

• loader를 통해 배치 사이즈 선언
• 즉 데이터를 몇 개씩 던져줄 것인지 선언하게 됨
• train_loader는 제너레이터. 이터레이터로 사용 가능

유명한 앤드류 응.

• 다운로드 받으면 컨테이너에 담기게 됨

• 서버 조절

🔵수동 모드로 데이터 로드하기

• 언젠가 필요하다

import gzip
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader

def load_ubyte_images(filepath):
    with gzip.open(filepath, 'rb') as f:
        # 헤더 읽기 (Magic number, Number of images, Rows, Cols)
        magic, num, rows, cols = np.frombuffer(f.read(16), dtype=np.uint32, count=4).byteswap()
        # 데이터 읽기 및 정규화 (0~255 -> 0~1)
        data = np.frombuffer(f.read(), dtype=np.uint8).reshape(num, 1, rows, cols)
        result = data.astype(np.float32)
        return result

def load_ubyte_labels(filepath):
    with gzip.open(filepath, 'rb') as f:
        # 헤더 읽기 (Magic number, Number of items)
        magic, num = np.frombuffer(f.read(8), dtype=np.uint32, count=2).byteswap()
        # 데이터 읽기
        return np.frombuffer(f.read(), dtype=np.uint8).astype(np.int64)

class MNISTManualDataset(Dataset):
    def __init__(self, image_path, label_path, transform=None):
        self.images = torch.from_numpy(load_ubyte_images(image_path))
        self.labels = torch.from_numpy(load_ubyte_labels(label_path))
        self.transform = transform

    def __len__(self):
        return len(self.images)

    def __getitem__(self, idx):
        image = self.images[idx]
        label = self.labels[idx]
        
        if self.transform:
            image = self.transform(image)
            
        return image, label

img_path = './data/MNIST/MNIST/raw/t10k-images-idx3-ubyte.gz'
lbl_path = './data/MNIST/MNIST/raw/t10k-labels-idx1-ubyte.gz'

# 데이터셋 인스턴스 생성
manual_dataset = MNISTManualDataset(img_path, lbl_path)

# 데이터로더 생성
manual_loader = DataLoader(
    manual_dataset, 
    batch_size=64, 
    shuffle=True
)

⇒ 데이터셋과 로더는 세트이다.

# 데이터 확인
images, labels = next(iter(manual_loader))
print(f"Batch Image Shape: {images.shape}") # [64, 1, 28, 28]
print(f"Batch Label Shape: {labels.shape}") # [64]

• manual_loader로 이터레이터 돌림
• 컬러면 이미지 shape이 64 3 28 28됨
• 64 → batch size, 1 → 색상 , 28 28 → 이미지 가로세로 픽셀 수

🔵데이터 시각화

for (X_train, y_train) in train_loader:
    print('X_train:', X_train.size(), 'type:', X_train.type())
    print('y_train:', y_train.size(), 'type:', y_train.type())
    break
import matplotlib.pyplot as plt 
pltsize = 1
plt.figure(figsize=(10 * pltsize, pltsize))
for i in range(10):
    plt.subplot(1, 10, i + 1)
    plt.axis('off')
    plt.imshow(X_train[i, :, :, :].numpy().reshape(28, 28), cmap = "gray_r")
    plt.title('Class: ' + str(y_train[i].item()))

🔵모델 선언

초기함수

DNN 구조에서 노드 개수로 생각

• input layer - 28*28 ⇒ 512
• hidden layer - 512 ⇒256
• output layer → 256 ⇒ 10(숫자는 10개이므로.)

forward 함수

• 레이어 하나 지나고 활성화 함수 거치고를 반복.
• 최종 아웃풋의 클래스는 10개이다.
• 마지막 레이어의 노드는 확률값이 나온다
  • 마지막에 소프트맥스를 통과시키기 때문, 통상적으로 로그 소프트맥스가 성능이 낫다

import torch.nn.functional as F
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(28 * 28, 512)
        self.fc2 = nn.Linear(512, 256)
        self.fc3 = nn.Linear(256, 10)

    def forward(self, x):
        x = x.view(-1, 28 * 28)
        x = self.fc1(x)
        x = F.sigmoid(x)
        x = self.fc2(x)
        x = F.sigmoid(x)
        x = self.fc3(x)
        x = F.log_softmax(x, dim = 1)
        return x

model = Net().to(DEVICE)
optimizer = torch.optim.SGD(model.parameters(), lr = 0.01, momentum = 0.5)
criterion = nn.CrossEntropyLoss()

print(model)

Net(
  (fc1): Linear(in_features=784, out_features=512, bias=True)
  (fc2): Linear(in_features=512, out_features=256, bias=True)
  (fc3): Linear(in_features=256, out_features=10, bias=True)
)

from torchinfo import summary
summary(model, input_size=(32,784))

==========================================================================================
Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
Net                                      [32, 10]                  --
├─Linear: 1-1                            [32, 512]                 401,920
├─Linear: 1-2                            [32, 256]                 131,328
├─Linear: 1-3                            [32, 10]                  2,570
==========================================================================================
Total params: 535,818
Trainable params: 535,818

• linear 1-1 : 784*512 + 512
• linear 1-2 : 512*256 + 256
• linear 1-3 : 256*10 +10

각 간선은 wx+b 로 구성되어있으므로 양쪽 노드의 곱과 바이어스 개수의 합으로 파라미터의 수를 산출해낼 수 있다.

훈련 함수 정의

def train(model, train_loader, optimizer, log_interval):
    model.train()
    for batch_idx, (image, label) in enumerate(train_loader):
        image = image.to(DEVICE)
        label = label.to(DEVICE)
        optimizer.zero_grad()
        output = model(image)
        loss = criterion(output, label)
        loss.backward()
        optimizer.step()

        if batch_idx % log_interval == 0:
            print("Train Epoch: {} [{}/{} ({:.0f}%)]\tTrain Loss: {:.6f}".format(
                epoch, batch_idx * len(image), 
                len(train_loader.dataset), 100. * batch_idx / len(train_loader), 
                loss.item()))

평가 함수 정의

def evaluate(model, test_loader):
    model.eval()
    test_loss = 0
    correct = 0

    with torch.no_grad():
        for image, label in test_loader:
            image = image.to(DEVICE)
            label = label.to(DEVICE)
            output = model(image)
            test_loss += criterion(output, label).item()
            prediction = output.max(1, keepdim = True)[1]
            correct += prediction.eq(label.view_as(prediction)).sum().item()
    
    test_loss /= (len(test_loader.dataset) / BATCH_SIZE)
    test_accuracy = 100. * correct / len(test_loader.dataset)
    return test_loss, test_accuracy

진짜 훈련시켜보기

for epoch in range(1, EPOCHS + 1):
    train(model, train_loader, optimizer, log_interval = 200)
    test_loss, test_accuracy = evaluate(model, test_loader)
    print("\n[EPOCH: {}], \tTest Loss: {:.4f}, \tTest Accuracy: {:.2f} % \n".format(
        epoch, test_loss, test_accuracy))

농땡이 치는 간선 처리, 과적합 방지

• dropout

🟢오후가 됐습니다. 일어나서 코딩하세요.

• backward ⇒ 맨 처음 기준은 예측값이 엉망. 역전파 적용해서 값을 꾸준히 초기화한다.

🔵훈련 함수

def train(model, train_loader, optimizer, criterion, device, epoch, log_interval, batch_size):
    global loss, label
    model.train()
    correct = 0
    train_loss = 0
    total_samples = 0
    for batch_idx, (image, label) in enumerate(train_loader):
        image = image.to(device)
        label = label.to(device) 

        optimizer.zero_grad()

        output = model(image) 

        loss = criterion(output, label)

        train_loss += loss.item() * label.size(0)
        loss.backward()
        optimizer.step()

        preds = output.argmax(dim=1) 

        correct += (preds == label).sum().item()
        total_samples += label.size(0)

        if batch_idx % log_interval == 0:
            print("Train Epoch: {} [{}/{} ({:.0f}%)]\tTrain Loss: {:.6f}".format(
                epoch, total_samples,
                len(train_loader.dataset), 100. * batch_idx / len(train_loader),
                loss.item()))

    train_loss /= total_samples
    train_accuracy = 100. * correct / total_samples

    return train_loss, train_accuracy

• 드랍아웃 : 일꾼들의 n%는 비번을 주는 것. 추론 시에는 모두 사용해야한다.
• 훈련함수에서만 backward를 통한 step 함수가 적용된다. test와의 차이점

🔵평가함수

def evaluate(model, test_loader):
    model.eval() # 평가 모드 전환 (드롭아웃, 배치 정규화 등 비활성화)
    test_loss = 0
    correct = 0

    with torch.no_grad(): # 기울기 계산 비활성화 (메모리 절약 및 속도 향상)
        for images, labels in test_loader: # 복수형(images, labels)으로 통일 권장
            # 1. 데이터를 동일한 장치(GPU/CPU)로 이동
            images = images.to(DEVICE)
            labels = labels.to(DEVICE)

            # 2. 모델 예측
            outputs = model(images)

            # 3. 손실 계산 및 누적
            # criterion의 결과는 보통 배치당 평균이므로, 전체 평균을 구하기 위해 배치 크기를 곱해줍니다.
            loss = criterion(outputs, labels)
            test_loss += loss.item() * images.size(0) 

            # 4. 정확도 계산
            predictions = outputs.argmax(dim=1) # 가장 높은 확률을 가진 인덱스 추출
            correct += (predictions == labels).sum().item() # 현재 배치의 정답 수 누적

    # 전체 데이터셋에 대한 평균 손실과 정확도 계산
    test_loss /= len(test_loader.dataset)
    test_accuracy = 100. * correct / len(test_loader.dataset)

    return test_loss, test_accuracy

🔵early stopping 선언

class early_stopping:
  def __init__(self, patience, verbose, delta, path='checkpoint.pt'):
    self.patience = patience
    self.verbose = verbose
    self.delta = delta
    self.count = 0
    self.best_score = None
    self.early_stop = False
    self.val_loss_min = np.inf
    self.path = path

  def __call__(self, val_loss, model):
    score = -val_loss

    if self.best_score is None:
      self.best_score = score
      self.save_checkpoint(val_loss, model)
    elif score < self.best_score + self.delta:
      self.count += 1
      if self.verbose:
        print(f"Early Stopping counter: {self.count} out of {self.patience}")

      if self.count >= self.patience:
        self.early_stop = True  
    else:
      self.best_score = score
      self.save_checkpoint(val_loss, model)
      self.count =0  

  def save_checkpoint(self, val_loss, model):
    if self.verbose:
      print(f"Validation loss decreased ({self.val_loss_min:.6f}) --> {val_loss:.6f}. saving model..")
    torch.save(model.state_dict(), self.path)
    self.val_loss_min = val_loss

🔵early stopping 기준 코드 수행

patience = 2
EPOCHS = 50

import numpy as np
early_stopping = early_stopping(patience=2, verbose=True, path='best_model.pt', delta=0)
loss_hist_train     = [0] * EPOCHS
accuracy_hist_train = [0] * EPOCHS
loss_hist_valid     = [0] * EPOCHS
accuracy_hist_valid = [0] * EPOCHS
for epoch in range(1, EPOCHS + 1):
    loss_, acc_ = train(model, train_loader, optimizer, criterion, DEVICE, epoch, log_interval = 200, batch_size=BATCH_SIZE)
    loss_hist_train[epoch-1] = loss_
    accuracy_hist_train[epoch-1] = acc_
    test_loss, test_accuracy = evaluate(model, test_loader)
    loss_hist_valid[epoch-1] = test_loss
    accuracy_hist_valid[epoch-1] = test_accuracy
    print("\n[EPOCH: {}], \tTest Loss: {:.4f}, \tTest Accuracy: {:.2f} % \n".format(
        epoch, test_loss, test_accuracy))
    early_stopping(test_loss, model)
    if early_stopping.early_stop:
        print("Early stopping")
        break

• loss 값 기준으로 early stopping 진행

🔵성능 시각화

import matplotlib.pyplot as plt
epochs_range = np.arange(1,50)
epochs_range = np.arange(0, len(loss_hist_train))
plt.figure(figsize=(12, 5)) # 전체 그래프 크기 설정

# --- 첫 번째 그래프: Loss (손실) ---
plt.subplot(1, 2, 1) # 1행 2열 중 첫 번째
plt.plot(epochs_range, loss_hist_train, label='Train Loss', color='blue', marker='o')
plt.plot(epochs_range, loss_hist_valid, label='Valid Loss', color='red', marker='s')
plt.title('Training and Validation Loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.grid(True)
plt.legend()

# --- 두 번째 그래프: Accuracy (정확도) ---
plt.subplot(1, 2, 2) # 1행 2열 중 두 번째
plt.plot(epochs_range, accuracy_hist_train, label='Train Accuracy', color='blue', marker='o')
plt.plot(epochs_range, accuracy_hist_valid, label='Valid Accuracy', color='red', marker='s')
plt.title('Training and Validation Accuracy')
plt.xlabel('Epochs')
plt.ylabel('Accuracy (%)')
plt.grid(True)
plt.legend()

plt.tight_layout() # 그래프 간 간격 자동 조절
plt.show()

🟢CIFAR 데이터

🔵데이터 호출

train_dataset = datasets.CIFAR10(root = "./data/CIFAR_10",
                                  train = True,
                                  download = True,
                                  transform = transforms.ToTensor())

test_dataset = datasets.CIFAR10(root = "./data/CIFAR_10",
                                train = False,
                                transform = transforms.ToTensor())

train_loader = torch.utils.data.DataLoader(dataset = train_dataset,
                                            batch_size = BATCH_SIZE,
                                            shuffle = True)

test_loader = torch.utils.data.DataLoader(dataset = test_dataset,
                                          batch_size = BATCH_SIZE,
                                          shuffle = False)

🔵크기확인

for (X_train, y_train) in train_loader:
    print('X_train:', X_train.size(), 'type:', X_train.type())
    print('y_train:', y_train.size(), 'type:', y_train.type())
    break

X_train.shape

🔵시각화

pltsize = 1
plt.figure(figsize=(10 * pltsize, pltsize))

for i in range(10):
    plt.subplot(1, 10, i + 1)
    plt.axis('off')
    plt.imshow(np.transpose(X_train[i], (1, 2, 0)))
    plt.title('Class: ' + str(y_train[i].item()))

• 인셉션, yolo는 유명한 이미지 모델
• cnn은 전이학습이 가능
• rag 미래에 진행할 예정

🔵모델 구성

import torch.nn.functional as F
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(32 * 32 * 3, 512)
        self.fc2 = nn.Linear(512, 256)
        self.fc3 = nn.Linear(256, 10)

    def forward(self, x):
        x = x.view(-1, 32 * 32 *3)
        x = self.fc1(x)
        x = F.sigmoid(x)
        x = self.fc2(x)
        x = F.sigmoid(x)
        x = self.fc3(x)
        x = F.log_softmax(x, dim = 1)
        return x

🔵파라미터 개수 확인

model = Net().to(DEVICE)
optimizer = torch.optim.SGD(model.parameters(), lr = 0.01, momentum = 0.5)
criterion = nn.CrossEntropyLoss()

from torchinfo import summary
summary(model, input_size=(64, 32*32*3))

Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
Net                                      [64, 10]                  --
├─Linear: 1-1                            [64, 512]                 1,573,376
├─Linear: 1-2                            [64, 256]                 131,328
├─Linear: 1-3                            [64, 10]                  2,570
==========================================================================================
Total params: 1,707,274
Trainable params: 1,707,274
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 109.27

🔵early stopping 선언 및 훈련 진행

import numpy as np

class early_stopping:
  def __init__(self, patience, verbose, delta, path='checkpoint.pt'):
    self.patience = patience
    self.verbose = verbose
    self.delta = delta
    self.count = 0
    self.best_score = None
    self.early_stop = False
    self.val_loss_min = np.inf
    self.path = path

  def __call__(self, val_loss, model):
    score = -val_loss

    if self.best_score is None:
      self.best_score = score
      self.save_checkpoint(val_loss, model)
    elif score < self.best_score + self.delta:
      self.count += 1
      if self.verbose:
        print(f"Early Stopping counter: {self.count} out of {self.patience}")

      if self.count >= self.patience:
        self.early_stop = True  
    else:
      self.best_score = score
      self.save_checkpoint(val_loss, model)
      self.count =0  

  def save_checkpoint(self, val_loss, model):
    if self.verbose:
      print(f"Validation loss decreased ({self.val_loss_min:.6f}) --> {val_loss:.6f}. saving model..")
    torch.save(model.state_dict(), self.path)
    self.val_loss_min = val_loss

patience = 2
EPOCHS = 50

early_stopping = early_stopping(patience=2, verbose=True, path='best_model.pt', delta=0)
loss_hist_train     = [0] * EPOCHS
accuracy_hist_train = [0] * EPOCHS
loss_hist_valid     = [0] * EPOCHS
accuracy_hist_valid = [0] * EPOCHS
for epoch in range(1, EPOCHS + 1):
    loss_, acc_ = train(model, train_loader, optimizer, criterion, DEVICE, epoch, log_interval = 200, batch_size=BATCH_SIZE)
    loss_hist_train[epoch-1] = loss_
    accuracy_hist_train[epoch-1] = acc_
    test_loss, test_accuracy = evaluate(model, test_loader)
    loss_hist_valid[epoch-1] = test_loss
    accuracy_hist_valid[epoch-1] = test_accuracy
    print("\n[EPOCH: {}], \tTest Loss: {:.4f}, \tTest Accuracy: {:.2f} % \n".format(
        epoch, test_loss, test_accuracy))
    early_stopping(test_loss, model)
    if early_stopping.early_stop:
        print("Early stopping")
        break

🔵성능 시각화

• early stopping 적용됨

🔵accuracy가 구리니 코드를 고치겠다.

• dropout 적용, 활성화 함수 변경

import torch.nn.functional as F
from torchinfo import summary
import torch.nn as nn

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(32 * 32 * 3, 512)
        self.fc2 = nn.Linear(512, 256)
        self.fc3 = nn.Linear(256, 10)
        self.dropout_prob = 0.5

    def forward(self, x):
        x = x.view(-1, 32 * 32 * 3)
        x = self.fc1(x)
        x = F.relu(x)
        x = F.dropout(x, training = self.training, p = self.dropout_prob)
        x = self.fc2(x)
        x = F.relu(x)
        x = F.dropout(x, training = self.training, p = self.dropout_prob)
        x = self.fc3(x)
        x = F.log_softmax(x, dim = 1)
        return x

🔵배치 학습, 온라인 학습

• 차이 인지 필요

🟢파이토치 정리

🔵모델 생성

• nn.module
• init
  • 사용할 레이어 구조 선언
  • 사용할 변수들 - dropout 변수
• forward

🔵데이터

• dataset - 전처리를 포함할 수 있음
• dataloader - batch size 모델 전달

🔵optimizer

🔵loss, cost function

🔵train

• predict → loss → backward → update
• 과적합 방지 → early stopping

🔵eval

• model.eval → 모델에게 추론 모드라고 지시(비활성)
• torch.no_grad()

🔵가중치 초기화

• xavier
• He

🔵배치 정규화, 레이어 정규화

🟢비지도학습

🔵고유벡터 확인하기

import numpy as np
A = np.array([[4, 2],
              [1, 3]])

np.linalg.eig(A)

EigResult(eigenvalues=array([5., 2.]), eigenvectors=array([[ 0.89442719, -0.70710678], [ 0.4472136 , 0.70710678]]))

eigenvalues, eigenvectors = np.linalg.eig(A)

🔵아이리스 데이터를 고유벡터 활용해서 pca해보자

• 편차들의 공분산을 참고하여 고유벡터와 고유값을 구하고 그걸로 추가.

import matplotlib.pyplot as plt
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import load_iris

iris = load_iris()
X = iris.data
y = iris.target
feature_names = iris.feature_names

X_means = X-np.mean(X,axis=0)
np.cov(X_means, rowvar=False)

• 직접 고유값 고유벡터 구해서 pca 진행

eigenvalues, eigenvectors = np.linalg.eig(cov_mat)
sorted_idex  = np.argsort(eigenvalues)[::-1]
sorted_eigenvectors = eigenvectors[:, sorted_idex]

eigenvector_subset = sorted_eigenvectors[:, :2]
X_reduced = np.dot(X_means,eigenvector_subset)

array([[-2.68412563, -0.31939725],
       [-2.71414169,  0.17700123],
       [-2.88899057,  0.14494943],
       [-2.74534286,  0.31829898],

🔵pca 불러와서 진행

pca = PCA(n_components=2)
X_pca=pca.fit_transform(X)

X_pca

array([[-2.68412563,  0.31939725],
       [-2.71414169, -0.17700123],
       [-2.88899057, -0.14494943],
       [-2.74534286, -0.31829898],

🔵각 벡터의 설명력

eigenvalues/eigenvalues.sum()
`array([0.92461872, 0.05306648, 0.01710261, 0.00521218])`

• 첫번째 특성과 두번째 특성의 합으로 약 97퍼를 설명할 수 있다.

🔵시각화

iris = load_iris()
X = iris.data
y = iris.target

# 2. 데이터 전처리: 표준화 (Standardization)
# PCA 수행 전 반드시 거쳐야 하는 과정입니다.
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 3. Scikit-learn을 이용한 PCA 수행
# 4차원의 데이터를 시각화가 가능한 2차원으로 축소합니다.
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)

# 4. 고유값(Explained Variance) 확인
# 각 주성분이 원본 데이터의 전체 분산 중 얼마만큼을 설명하는지 나타냅니다.
print("각 주성분의 분산 비율:", pca.explained_variance_ratio_)
print("총 보존된 정보량:", np.sum(pca.explained_variance_ratio_))

# 5. 시각화 결과 분석
plt.figure(figsize=(10, 7))
colors = ['navy', 'turquoise', 'darkorange']
for i, color, target_name in zip([0, 1, 2], colors, iris.target_names):
    plt.scatter(X_pca[y == i, 0], X_pca[y == i, 1], color=color, alpha=.8, lw=2,
                label=target_name)

plt.legend(loc='best', shadow=False, scatterpoints=1)
plt.title('PCA of IRIS dataset')
plt.xlabel('Principal Component 1')
plt.ylabel('Principal Component 2')
plt.grid(True)
plt.show()

각 주성분의 분산 비율: [0.72962445 0.22850762] 총 보존된 정보량: 0.9581320720000166

Day5

🟢CNN 개념 정리

🔵DNN + C 구조 추가

CNN → DNN

• CNN 이 특징을 잡은 후에 DNN에 보낸다.
• DNN은 어쨋든 펼쳐서 가야한다.
• 그 전에 CNN이 특징을 잘 잡도록 노력해야 한다.
• 얀르쿤이 발표함
• 합성곱 통해 진행

🔵alexnet

• cnn 구조 처음 활용

🔵convolution → 이미지 특징을 잡는다.

• 세션 17주차 1조 발표자료

• 단순히 convolution 연산 진행 시 외곽 정보 손실 가능

⇒ padding 적용하여 각 칸이 동일 연산 횟수를 가질 수 있게 함

맥스 풀링

• 평균 풀링보다 성능 좋다는 의견 존재.

🔵lenet 구조 활용

• cnn 개념 잘 정립

🔵출력 크기 연산

• 커널 사이즈가 짝수면 오히려 출력값이 커지는 경향 존재.

• CNN의 weight값은 각 커널 값 마다 담겨 있다.
• CNN의 bias는 각 커널마다 존재한다.

🟢캐글을 열걸아.

🔵데이터 다운로드

import torch
from torchvision import transforms, datasets
BATCH_SIZE = 64 

train_dataset = datasets.CIFAR10(root = "./data/CIFAR_10",
                                  train = True,
                                  download = True,
                                  transform = transforms.ToTensor())

test_dataset = datasets.CIFAR10(root = "./data/CIFAR_10",
                                train = False,
                                transform = transforms.ToTensor())

train_loader = torch.utils.data.DataLoader(dataset = train_dataset,
                                            batch_size = BATCH_SIZE,
                                            shuffle = True)

test_loader = torch.utils.data.DataLoader(dataset = test_dataset,
                                          batch_size = BATCH_SIZE,
                                          shuffle = False)

sample = iter(train_loader)
sample_data = next(sample)

🔵형상 확인

sample_data[0].shape

torch.Size([64, 3, 32, 32])

🔵시각화

from PIL import Image
import numpy as np
import pickle
with open("./data/CIFAR_10/cifar-10-batches-py/data_batch_1", 'rb') as f:
    dict_data = pickle.load(f, encoding='bytes')

dict_data[b'data'].shape

sample1 = dict_data[b'data'][0]

import matplotlib.pyplot as plt 

sample1 = sample1.reshape( 3, 32, 32   )
#torch, numpy와 matplotlib 축이 다르므로 전환시켜줌.
plt.imshow(sample1.transpose(1,2, 0))
plt.show()

torch와 matplotlib 축 순서 다름 유의

• CIFAR-10 원본 데이터 축 구조 - dict_data[b'data']

→ shape: (10000, 3072)→ (3072, ) = 3 × 32 × 32 👉 이건 채널 우선(Channel First) 구조

sample1.shape == (3, 32, 32)

PyTorch / NumPy에서 자주 쓰는 이미지 축 - (C, H, W)

• C: Channel (RGB)
• H: Height
• W: Width
• 딥러닝 모델 입력용 표준

---

matplotlib imshow()가 기대하는 축

• (H, W, C)
• 즉 채널 마지막(Channel Last) 구조

(32, 32, 3)

---

transpose(1, 2, 0) 의미

sample1.transpose(1, 2, 0) => (H, W, C)

• 축 재배치:
  • 1 → Height
  • 2 → Width
  • 0 → Channel

결과

• 🧠 모델 → (C, H, W)
• 🎨 시각화(imshow) → (H, W, C)

🔵모델 선언

import torch
import torch.nn as nn
import torch.nn.functional as F
class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(in_channels = 3, out_channels = 8, kernel_size = 3, padding = 1)
        self.conv2 = nn.Conv2d(in_channels = 8, out_channels = 16, kernel_size = 3, padding = 1)
        self.pool = nn.MaxPool2d(kernel_size = 2, stride = 2)
        self.fc1 = nn.Linear(8 * 8 * 16, 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, 10)
        
    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.pool(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = self.pool(x)
        
        x = x.view(-1, 8 * 8 * 16)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.fc2(x)
        x = F.relu(x)
        x = self.fc3(x)
        x = F.log_softmax(x)
        return x

• log_softmax 사용
  • softmax에 곱셈 연산이 있는데, log 전환 시 덧셈으로 변하여 안정적

🔵모델 등록

if torch.cuda.is_available():
    DEVICE = torch.device('cuda')
else:
    DEVICE = torch.device('cpu')
print('Using PyTorch version:', torch.__version__, ' Device:', DEVICE)

model = CNN().to(DEVICE)

🔵파라미터 수 확인

from torchinfo import summary
summary(model, input_size=(64, 3, 32, 32))

==========================================================================================
Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
CNN                                      [64, 10]                  --
├─Conv2d: 1-1                            [64, 8, 32, 32]           224
├─MaxPool2d: 1-2                         [64, 8, 16, 16]           --
├─Conv2d: 1-3                            [64, 16, 16, 16]          1,168
├─MaxPool2d: 1-4                         [64, 16, 8, 8]            --
├─Linear: 1-5                            [64, 64]                  65,600
├─Linear: 1-6                            [64, 32]                  2,080
├─Linear: 1-7                            [64, 10]                  330

🔵CNN 파라미터 & 출력 구조 정리

기본 개념

• 출력 채널 수 = 커널(필터) 수
• Convolution을 통과한 결과를 feature map이라고 한다
• input의 채널 수만큼 커널이 겹쳐서 계산되고, 그 결과가 하나의 출력 채널이 된다
• 학습되는 파라미터는 Conv, Linear 레이어에만 존재
• ReLU, Pooling, view, log_softmax에는 파라미터가 없다

---

Output Shape 규칙

• 출력 shape 기본 구조
  • (batch, channel, height, width)
• summary에서 앞의 64는
  • batch size = 한 번에 들어가는 데이터 개수

---

Conv2d 파라미터 계산 공식

• Conv2d 파라미터 수
  • (out_channels) × (in_channels × kernel_h × kernel_w + 1)
• +1은 bias
• 출력 채널마다 bias는 1개씩 존재

---

Linear 파라미터 계산 공식

• Linear 파라미터 수
  • (out_features) × (in_features + 1)
• 역시 +1은 bias

---

Conv1 파라미터 계산

• 입력 채널: 3
• 출력 채널: 8
• 커널 크기: 3 × 3
• 계산
  • (3 × 3 × 3 + 1) × 8
  • = (27 + 1) × 8
  • = 224

---

Conv2 파라미터 계산

• 입력 채널: 8
• 출력 채널: 16
• 커널 크기: 3 × 3
• 계산
  • (8 × 3 × 3 + 1) × 16
  • = (72 + 1) × 16
  • = 1,168

---

Pooling & Activation

• MaxPool2d
  • 가로, 세로 길이를 절반으로 줄인다
• ReLU
  • 비선형 함수
• 두 연산 모두
  • 파라미터 없음

---

Feature Map 크기 변화

• 입력: (3, 32, 32)
• Conv1 → (8, 32, 32)
• Pool → (8, 16, 16)
• Conv2 → (16, 16, 16)
• Pool → (16, 8, 8)

---

Flatten

• Conv 출력은 4차원
• Linear에 넣기 위해 펼침
• 펼친 크기
  • 16 × 8 × 8 = 1024

---

FC1 파라미터 계산

• 입력: 1024
• 출력: 64
• 계산
  • (1024 + 1) × 64
  • = 65,600

---

FC2 파라미터 계산

• 입력: 64
• 출력: 32
• 계산
  • (64 + 1) × 32
  • = 2,080

---

FC3 파라미터 계산

• 입력: 32
• 출력: 10
• 계산
  • (32 + 1) × 10
  • = 330

---

중요한 포인트

• FC 레이어에서 파라미터 수가 급증한다
  • Conv는 weight 공유
  • FC는 완전 연결 구조

---

Padding & Stride

• 커널 크기 3 × 3
• stride = 1
• padding = 1
• 입력과 출력의 가로·세로 크기가 동일해진다

🔵모델 돌리기

옵티마이저와 비용 함수 선언

optimizer = torch.optim.Adam(model.parameters(), lr = 0.001)
criterion = nn.CrossEntropyLoss()

훈련 함수 정의

def train(model, train_loader, optimizer, log_interval):
    model.train()
    for batch_idx, (image, label) in enumerate(train_loader):
        image = image.to(DEVICE)
        label = label.to(DEVICE)
        optimizer.zero_grad()
        output = model(image)
        loss = criterion(output, label)
        loss.backward()
        optimizer.step()

        if batch_idx % log_interval == 0:
            print("Train Epoch: {} [{}/{} ({:.0f}%)]\tTrain Loss: {:.6f}".format(
                epoch, batch_idx * len(image), 
                len(train_loader.dataset), 100. * batch_idx / len(train_loader), 
                loss.item()))

평가 함수 정의

def evaluate(model, test_loader):
    model.eval()
    test_loss = 0
    correct = 0

    with torch.no_grad():
        for image, label in test_loader:
            image = image.to(DEVICE)
            label = label.to(DEVICE)
            output = model(image)
            test_loss += criterion(output, label).item()
            prediction = output.max(1, keepdim = True)[1]
            correct += prediction.eq(label.view_as(prediction)).sum().item()
    
    test_loss /= (len(test_loader.dataset) / BATCH_SIZE)
    test_accuracy = 100. * correct / len(test_loader.dataset)
    return test_loss, test_accuracy

훈련 실행

EPOCHS = 30
optimizer = torch.optim.Adam(model.parameters(), lr = 0.001)
criterion = nn.CrossEntropyLoss()
for epoch in range(1, EPOCHS + 1):
    train(model, train_loader, optimizer, log_interval = 200)
    test_loss, test_accuracy = evaluate(model, test_loader)
    print("\n[EPOCH: {}], \tTest Loss: {:.4f}, \tTest Accuracy: {:.2f} % \n".format(
        epoch, test_loss, test_accuracy))

Train Epoch: 1 [0/50000 (0%)]	Train Loss: 2.328551
Train Epoch: 1 [12800/50000 (26%)]	Train Loss: 1.878364
Train Epoch: 1 [25600/50000 (51%)]	Train Loss: 1.585775
Train Epoch: 1 [38400/50000 (77%)]	Train Loss: 1.598529

[EPOCH: 1], 	Test Loss: 1.5576, 	Test Accuracy: 42.83 %

Train Epoch: 2 [0/50000 (0%)]	Train Loss: 1.432566
Train Epoch: 2 [12800/50000 (26%)]	Train Loss: 1.685357
Train Epoch: 2 [25600/50000 (51%)]	Train Loss: 1.448450
Train Epoch: 2 [38400/50000 (77%)]	Train Loss: 1.545207

🔵피쳐 맵 보기

img = next(iter(train_loader))

target_img = img[0][0]

target_tensor = target_img.unsqueeze(0)

with torch.no_grad():
    feature_map = model.conv1(target_img.to(DEVICE))

feature_map = feature_map.to('cpu')

img_numpy  = target_img.permute(1, 2,0).numpy()
plt.imshow(img_numpy)
plt.show()

fig, axes = plt.subplots(2, 4, figsize=(12, 6))
for i in range(8):
    ax = axes[i // 4, i % 4]
    f_map = feature_map[i].numpy() 

    ax.imshow(f_map, cmap='gray') # 특성 맵은 보통 그레이스케일로 봅니다
    ax.set_title(f"Filter {i+1}")
    ax.axis('off')
plt.tight_layout()
plt.show()

🟢카메라 연동

🔵환경 설정

pip install opencv-python

🔵코드

import cv2

capture = cv2.VideoCapture(0)
capture.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)

🔵카메라 실행

position = (30,50)
font = cv2.FONT_HERSHEY_SIMPLEX
scale = 1
color = (255,255,255)
thickness = 2

while True:
    ret, frame = capture.read()
    cv2.putText(frame, "Haha", position, font, scale, color, thickness)
    if not ret:
        print("카메라 오류")
        break
    print(type(frame))
    cv2.imshow("VideoFrame", frame)

    if cv2.waitKey(10) & 0xFF == ord('q'):
        capture.release()
        cv2.destroyAllWindows()
        break

🟢왜 웃는 데이터를 쓰라는걸까요

🔵환경설정

!pip3 install torch torchvision

🔵미소 탐지 모델 실험(데모, 로컬에서 돌림)

serving.py

import os
os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
os.environ["OMP_NUM_THREADS"] = "1"   # 선택: 충돌/과부하 줄이기

import torch
import torch.nn as nn
import torch.nn.functional as F
import cv2
from torchvision import transforms
trans = transforms.Compose(
    [
        transforms.Resize((64,64)),
    ]
)

class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(in_channels = 3, out_channels = 8, kernel_size = 3, padding = 1)
        self.conv2 = nn.Conv2d(in_channels = 8, out_channels = 16, kernel_size = 3, padding = 1)
        self.pool = nn.MaxPool2d(kernel_size = 2, stride = 2)
        self.fc1 = nn.Linear(16 * 16 * 16, 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, 1)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.pool(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = self.pool(x)

        x = x.view(-1, 16 * 16 * 16)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.fc2(x)
        x = F.relu(x)
        x = self.fc3(x)
        x = torch.sigmoid(x)
        return x

smile_model = torch.load("./smile.pt", weights_only=False, map_location=torch.device('cpu') )

capture = cv2.VideoCapture(0)
capture.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
position = (30,50)
font = cv2.FONT_HERSHEY_SIMPLEX
scale = 1
color = (255,255,255)
thickness = 2

while True:
    ret, frame = capture.read()
    target_data = trans(torch.from_numpy(frame.transpose(2, 0,1)))
    output =smile_model(target_data.reshape(-1, 3, 64, 64).float()).item()
    print(output)
    if (output >= 0.5):
        text = "Smile!!"
        cv2.putText(frame, text, position, font, scale, color, thickness)
    else:
        text = "Angry!!"
        cv2.putText(frame, text, position, font, scale, color, thickness)

    if not ret:
        print("카메라 오류")
        break
    print(type(frame))
    cv2.imshow("VideoFrame", frame)

    if cv2.waitKey(10) & 0xFF == ord('q'):
        capture.release()
        cv2.destroyAllWindows()
        break

🟢미소 탐지 실습

🔵캐글 데이터 호출

import kagglehub

# Download latest version
path = kagglehub.dataset_download("chazzer/smiling-or-not-face-data")

print("Path to dataset files:", path)

🔵데이터 쪼개기

import random
import os
random.seed(42)

smile = "/kaggle/input/smiling-or-not-face-data/smile/"
non_smile = "/kaggle/input/smiling-or-not-face-data/non_smile/"

smile_files = os.listdir(smile)
random.shuffle(smile_files)
smile_train = smile_files[:480]
smile_test = smile_files[480:]

non_smile_files = os.listdir(non_smile)
random.shuffle(non_smile_files)
non_smile_train = non_smile_files[:480]
non_smile_test = non_smile_files[480:]

• 배치 정규화, 레이어 정규화
  • 배치에 들어오는 데이터의 클래스를 균등하게 정규화시킨다
  • 왜?
  • 초반에 하나의 클래스만을 학습하면 특정 클래스에 대해서 먼저 학습하기 때문에 다른 클래스에 대해 대응이 어렵다.

🔵데이터 클래스 생성

• 이미지 증강 기능 추가.

from PIL import Image 
from torchvision import transforms
from torch.utils.data import Dataset, DataLoader

class facedata(Dataset):
    def __init__(self, smile_folder_path, smile_file_list, non_folder_path, non_file_list):
        self.smile_file_list = [os.path.join(smile_folder_path, x) for x in smile_file_list]
        self.non_file_list = [os.path.join(non_folder_path, x) for x in non_file_list]
        self.total_img = self.smile_file_list +  self.non_file_list
        random.shuffle(self.total_img)
        # 이미지 증강 (image augmentaion)
        self.transforms = transforms.Compose([
            transforms.ToTensor()
        ])

    def __len__(self):
        return len(self.total_img)

    def __getitem__(self, idx):
        img_name = self.total_img[idx]
        image = Image.open(img_name)
        img = self.transforms(image)
        label = 1 if 'non_smile' in img_name else 0
        return img, label

train_dataset = facedata(smile, smile_train, non_smile, non_smile_train )
X, y = next(iter(train_dataset))

X.shape
print(y)

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
iter_loader = iter(train_loader)

train_dataset = facedata(smile, smile_train, non_smile, non_smile_train )
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
train_test = facedata(smile, smile_test, non_smile, non_smile_test)
test_loader = DataLoader(train_test, batch_size=32, shuffle=True)

파이썬 문법 중 하나로 ,인지 필요

X, y = next(iter(train_dataset)) 

• iter() : dataset iterator로 만들기
• next() : iterator에서 다음 값 1개를 꺼내는 함수

🔵훈련 함수

import torch
import torch.nn as nn
import torch.nn.functional as F
class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(in_channels = 3, out_channels = 8, kernel_size = 3, padding = 1)
        self.conv2 = nn.Conv2d(in_channels = 8, out_channels = 16, kernel_size = 3, padding = 1)
        self.pool = nn.MaxPool2d(kernel_size = 2, stride = 2)
        self.fc1 = nn.Linear(16 * 16 * 16, 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, 1)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.pool(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = self.pool(x)

        x = x.view(-1, 16 * 16 * 16)
        x = self.fc1(x)
        x = F.relu(x)
        x = self.fc2(x)
        x = F.relu(x)
        x = self.fc3(x)
        x = torch.sigmoid(x)
        return x

model, optimizer, criterion 선언

model = CNN().to('cuda')
optimizer = torch.optim.RMSprop(model.parameters(), lr = 0.001)
criterion = nn.BCELoss()

훈련 함수 선언

def train(model, train_loader, optimizer, log_interval):
    global label, image, output
    model.train()
    for batch_idx, (image, label) in enumerate(train_loader):
        image = image.to(DEVICE)
        label = label.to(DEVICE)
        optimizer.zero_grad()
        output = model(image)
        loss = criterion(output, label.to(torch.float32).reshape(-1, 1))
        loss.backward()
        optimizer.step()

        if batch_idx % log_interval == 0:
            print("Train Epoch: {} [{}/{} ({:.0f}%)]\tTrain Loss: {:.6f}".format(
                epoch, batch_idx * len(image), 
                len(train_loader.dataset), 100. * batch_idx / len(train_loader), 
                loss.item()))

평가 함수 선언

def evaluate(model, test_loader):
    global label, image, output

    model.eval()
    test_loss = 0
    correct = 0

    with torch.no_grad():
        for image, label in test_loader:
            # print(image.shape)
            image = image.to(DEVICE)
            label = label.to(DEVICE)
            output = model(image)
    
            test_loss += criterion(output, label.to(torch.float32).reshape(-1, 1)).item()
            # print('-----')
            # print(output)
            # prediction = output.max(1, keepdim = True)[1]
            # print(prediction)
            # print("---")
            # print(label)
            correct += ((output > 0.5).long().reshape(-1) == label).reshape(-1).sum().item()
            # correct += prediction.eq(label.view_as(prediction)).sum().item()
    
    test_loss /= (len(test_loader.dataset) / BATCH_SIZE)
    test_accuracy = 100. * correct / len(test_loader.dataset)
    return test_loss, test_accuracy

훈련 실행

EPOCHS = 30
BATCH_SIZE = 32
for epoch in range(1, EPOCHS + 1):
    train(model, train_loader, optimizer, log_interval = 10)
    test_loss, test_accuracy = evaluate(model, test_loader)
    print("\n[EPOCH: {}], \tTest Loss: {:.4f}, \tTest Accuracy: {:.2f} % \n".format(
        epoch, test_loss, test_accuracy))