쿠버네티스,쿠버플로우
던전앤파이터 시세 예측[5]
세용용용용
2023. 9. 6. 17:22
아침에 지하철타고오다가 신박한 피처가 생각이나서 추가를 해볼 예정이다!!
내가 생각한 피처는 던파100만 골드당 현금 시세 피처를 추가하면 좋은거 같아서 추가해볼것이다 왜냐면... 인플레이션, 디플레이션 처럼 각나라의 현금 가치에 따라 물가변동률이 큰것으로 알고있기 떄문에 현재 골드 시세 컬럼을 추가해주면 모델 성능에 효과적이지 안을까??? 생각해보았다.
마리아db에 골드 시세 테이블을 추가하고 거기에 골드시세 정보를 넣어보자
CREATE TABLE goldprice (
date TEXT,
sell FLOAT,
buy FLOAT
);날짜와 판매, 구매 금액을 컬럼으로 두고
크롤링해서 테이블에 날짜별 골드 판매, 구매 시세 넣는 코드
from bs4 import BeautifulSoup
import requests
import pymysql
# MySQL 연결 설정
conn = pymysql.connect(
host='35.194.179.62', # 호스트명
user='root', # 데이터베이스 사용자명
password='1234', # 데이터베이스 암호
db='donpa_item', # 데이터베이스 이름
port=31381
)
# 커서 생성
cursor = conn.cursor()
date_list = []
sell_list = []
buy_list = []
# 검색 결과 페이지 URL
url = "http://dnfnow.xyz/invest"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
}
response = requests.get(url, headers=headers)
# 응답 데이터 파싱
soup = BeautifulSoup(response.text, "html.parser")
# 뉴스 타이틀, 링크 선택자
selector = "#week > script:nth-child(3)"
# 선택자에 해당하는 요소들을 모두 찾음
script_element = soup.select_one(selector)
# 원하는 JavaScript 코드 부분 추출
javascript_code = script_element.text
javascript_code = javascript_code.split('data')
date = javascript_code[1].split(' labels: [ ')[1].split(', ],\n ')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in date:
cleaned_date = i.strip().strip("'")
date_list.append(cleaned_date)
sell = javascript_code[3].split(': [ ')[1].split(', ]')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in sell:
cleaned_sell = i.strip().strip("'")
sell_list.append(float(cleaned_sell))
buy = javascript_code[4].split(': [ ')[1].split(', ]')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in buy:
cleaned_buy = i.strip().strip("'")
buy_list.append(float(cleaned_buy))
# 데이터베이스에 데이터 삽입 또는 업데이트
for i in range(len(date_list)):
date = date_list[i]
sell = sell_list[i]
buy = buy_list[i]
# 날짜가 이미 존재하는지 확인
cursor.execute("SELECT COUNT(*) FROM goldprice WHERE date = %s", (date,))
result = cursor.fetchone()
if result[0] == 0:
# 날짜가 존재하지 않으면 데이터 삽입
sql = "INSERT INTO goldprice (date, sell, buy) VALUES (%s, %s, %s)"
cursor.execute(sql, (date, sell, buy))
else:
# 날짜가 존재하면 데이터 업데이트 (오버라이팅)
sql = "UPDATE goldprice SET sell = %s, buy = %s WHERE date = %s"
cursor.execute(sql, (sell, buy, date))
# 변경사항 커밋
conn.commit()
# 연결 종료
conn.close()
이미지로 만들어서 쿠버네티스 크론잡으로 던져보자!!!
dockerfile
FROM python:latest
# 작업 디렉토리 설정
WORKDIR /app
# main, requirements 두 파일 작업디렉토리로 복사
COPY main.py /app/main.py
COPY requirements.txt /app
# 필요한 종속성 설치
RUN pip install --no-cache-dir -r requirements.txt
cMD [ "python" , "main.py" ]
main.py
from bs4 import BeautifulSoup
import requests
import pymysql
# MySQL 연결 설정
conn = pymysql.connect(
host='10.103.132.127', # 호스트명
user='root', # 데이터베이스 사용자명
password='1234', # 데이터베이스 암호
db='donpa_item', # 데이터베이스 이름
port=3306
)
# 커서 생성
cursor = conn.cursor()
date_list = []
sell_list = []
buy_list = []
# 검색 결과 페이지 URL
url = "http://dnfnow.xyz/invest"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
}
response = requests.get(url, headers=headers)
# 응답 데이터 파싱
soup = BeautifulSoup(response.text, "html.parser")
# 뉴스 타이틀, 링크 선택자
selector = "#week > script:nth-child(3)"
# 선택자에 해당하는 요소들을 모두 찾음
script_element = soup.select_one(selector)
# 원하는 JavaScript 코드 부분 추출
javascript_code = script_element.text
javascript_code = javascript_code.split('data')
date = javascript_code[1].split(' labels: [ ')[1].split(', ],\n ')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in date:
cleaned_date = i.strip().strip("'")
date_list.append(cleaned_date)
sell = javascript_code[3].split(': [ ')[1].split(', ]')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in sell:
cleaned_sell = i.strip().strip("'")
sell_list.append(float(cleaned_sell))
buy = javascript_code[4].split(': [ ')[1].split(', ]')[0].split(',')
# 공백 및 작은 따옴표 제거 후 출력
for i in buy:
cleaned_buy = i.strip().strip("'")
buy_list.append(float(cleaned_buy))
# 데이터베이스에 데이터 삽입 또는 업데이트
for i in range(len(date_list)):
date = date_list[i]
sell = sell_list[i]
buy = buy_list[i]
# 날짜가 이미 존재하는지 확인
cursor.execute("SELECT COUNT(*) FROM goldprice WHERE date = %s", (date,))
result = cursor.fetchone()
if result[0] == 0:
# 날짜가 존재하지 않으면 데이터 삽입
sql = "INSERT INTO goldprice (date, sell, buy) VALUES (%s, %s, %s)"
cursor.execute(sql, (date, sell, buy))
else:
# 날짜가 존재하면 데이터 업데이트 (오버라이팅)
sql = "UPDATE goldprice SET sell = %s, buy = %s WHERE date = %s"
cursor.execute(sql, (sell, buy, date))
# 변경사항 커밋
conn.commit()
# 연결 종료
conn.close()
requirements.txt
requests
beautifulsoup4
pymysql
이미지 저장하는 명령어
docker build -t sy02229/goldprice:latest .
이미지 푸쉬하는 명령어
docker push sy02229/goldprice
쿠버네티스 크론잡( 테스트로 2분마다 일단 실행시켜보자!!)
apiVersion: batch/v1beta1
kind: CronJob
metadata:
name: goldprice
spec:
schedule: "*/2 * * * *"
jobTemplate:
spec:
template:
metadata:
labels:
app: goldprice
spec:
restartPolicy: OnFailure
containers:
- name: goldprice
image: sy02229/goldprice:latest
일단 3개 완료된거 확인!!!!
테이블에 저장되었는지 확인해보자
# MySQL 클라이언트 컨테이너에 접속
kubectl exec -it mysql-client-75c74cbc5c-6xmb9 -- bash
# MySQL 클라이언트 실행 (예: root 계정으로 접속)
mysql -h 10.103.132.127 -u root -p
이제 해당 테이블을 사용하여 피처를 추가해보자!!!!!
우선 추가하기전 기존코드 성능을 확인해보자
1. 머신러닝
(1) 랜포
(2) xgboost
(3) lgbm
딥러닝(lstm)
추가후 성능을 확인해보자
1. 머신러닝
(1) 랜포
(2) xgboost
(3) lgbm
큰 변화는 없지만 어느정도 좋아진것을 확인 하지만 아직 과적합 문제가 해결되지 않음... 데이터가 아직은 많이 적다 row수가 200개... 데이터 수집의 한계가 있다.. ㅠㅠ
2. 딥러닝
훈련데이터 성능은 확실히 좋아졌지만 검증데이터 성능이 과적합이 아직 존재. 데이터 부족 문제일거 같다고 생각함.
이제 던파 아바타 시세를 예측해보자
목표는 하루마다 최적화된 모델을 저장시켜 장고를 통해 오늘의 예상 가격을 출력하는 것을 목표로 할것이다
일달 스케일링 객체, 인코딩 객체, 머신러닝 모델을 저장할 pvc를 생성해주자
vim scaler-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: scaler-pvc
spec:
accessModes:
- ReadWriteMany
resources:
requests:
storage: 10Gi
vim encoder-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: encoder-pvc
spec:
accessModes:
- ReadWriteMany
resources:
requests:
storage: 10Gi
vim model-pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: model-pvc
spec:
accessModes:
- ReadWriteMany
resources:
requests:
storage: 10Gi
kubectl create -f scaler-pvc.yaml
kubectl create -f encoder-pvc.yaml
kubectl create -f model-pvc.yaml
이제 아바타 데이터 스케일 객체, 인코딩 객체, 모델 객체를 저장하는 코드를 작성해보자
1. 추가해줄 골드 시세 데이터 가져오기
import pymysql
import pandas as pd
# 연결 정보 설정
host = "10.103.132.127"
port = 3306
database = "donpa_item"
user = "root"
password = "1234"
# 연결
connection = pymysql.connect(
host=host,
port=port,
user=user,
password=password,
database=database
)
# SQL 쿼리
query = "SELECT * FROM goldprice"
# 데이터프레임으로 변환
date_df = pd.read_sql(query, connection)
# 연결 종료
connection.close
date_df.head()
2. 수집중인 레어 아바타 데이터 불러오기
import pandas as pd
from pymongo import MongoClient
# MongoDB 연결 설정
client = MongoClient("mongodb://3.38.178.84:27017/")
db = client["donpa"] # 데이터베이스명
collection = db["donpa_aabata_rare"] # 컬렉션명
# 컬렉션 데이터 조회
data = collection.find()
# 데이터프레임으로 변환
df = pd.DataFrame(data)
# _id 컬럼은 제거해주자
df = df.drop(columns="_id")
df['price'] = df['price'].astype('float')
df.info()
df.head()
3. 피처 엔지니어링
df['soldDate'] = df['soldDate'].str.replace('-','')
# sell, buy컬럼 붙여주자!!
df = df.merge(date_df, left_on='soldDate', right_on='date', how='inner')
# date컬럼 삭제해주자
df.drop(columns = 'date', inplace=True)
from datetime import datetime
df['soldDate'] = pd.to_datetime(df['soldDate'])
# 년 컬럼 추가
df['year'] = df['soldDate'].dt.year
# 월 컬럼 추가
df['month'] = df['soldDate'].dt.month
# 일 컬럼 추가
df['day'] = df['soldDate'].dt.day
# 요일 컬럼 추가해주자
df['day_name'] = df['soldDate'].dt.day_name()
# 필요 없는 컬럼 삭제
df.drop(columns = 'soldDate', inplace=True)
# ava_rit도 어차피다 레어이니까 삭제
df.drop(columns = 'ava_rit', inplace=True)
# 타겟 컬러 맨뒤로 보내기
price_column = df.pop('price')
df['price'] = price_column
df.head()
4. 타겟값 분리해주자
X_train = df.drop(columns = 'price')
y_train = df['price']
display(X_train.head())
y_train.head()
5. 왜도 있는지 확인( 보통 절대값이 1이 넘어가면 왜도가 있다고 판단 로그변환 ㄱㄱ)
from scipy.stats import skew
# 왜도 있는 컬럼 모은 리스트
log_list = []
# 수치형 컬럼들만 돌려야됨
for i in X_train.select_dtypes(exclude='object').columns:
skewness = skew(X_train[i])
# 절대값이 1이 넘어가면 왜도가 있다고 판단
if abs(skewness) >= 1:
log_list.append(i)
print(log_list)
# 왜도 있는 컬럼들만 로그변환을 해주자
import numpy as np
for i in log_list:
if X_train[i].min() < 0:
X_train[i] = X_train.map(lambda x : x+abs(X_train[i].min()))
X_train[i] = np.log1p(X_train[i])
X_train.head()
6. 데이터를 분할하자 학습용8, 테스트2
from sklearn.model_selection import train_test_split
X_train, X_validation, Y_train, Y_validation = train_test_split(X_train,y_train, test_size=0.2, random_state=1)
print(X_train.shape, X_validation.shape, Y_train.shape, Y_validation.shape)
X_train.reset_index(drop=True, inplace=True)
X_validation.reset_index(drop=True, inplace=True)
7. 수치형 데이터 standardscaler해주자
obj_col = X_train.select_dtypes(include='object').columns
from sklearn.preprocessing import StandardScaler
sds = StandardScaler()
sds.fit(X_train.drop(columns = obj_col))
X_train_sc = sds.transform(X_train.drop(columns = obj_col))
X_train_sc = pd.DataFrame(X_train_sc, columns = X_train.drop(columns = obj_col).columns)
X_validation_sc = sds.transform(X_validation.drop(columns = obj_col))
X_validation_sc = pd.DataFrame(X_validation_sc, columns = X_validation.drop(columns = obj_col).columns)
# object 타입 컬럼 붙여주기
for i in obj_col:
X_train_sc[i] = X_train[i]
X_validation_sc[i] = X_validation[i]
# 스케일러 객체 저장
#import joblib
#joblib.dump(sds, '/home/cloudcccr3/scaler/scaler.pkl')
X_train_sc.head()
8. 원핫 인코딩 돌려주자
from sklearn.preprocessing import OneHotEncoder
import joblib
# OneHotEncoder 객체 생성
encoder = OneHotEncoder()
X_full = pd.concat([X_train_sc, X_validation_sc])
# 범주형 열만 선택
obj_df = X_full.select_dtypes(include='object')
# 숫자형 열만 선택
no_obj_df = X_full.select_dtypes(exclude='object')
# 범주형 열을 원핫 인코딩
encoded_features = encoder.fit_transform(obj_df)
# 인코딩된 결과를 데이터프레임으로 변환
encoded_df = pd.DataFrame(encoded_features.toarray(), columns=encoder.get_feature_names(obj_df.columns))
# 인코딩된 범주형 열과 숫자형 열을 합침
X_train_sc_ec = pd.concat([no_obj_df[:len(X_train_sc)] , encoded_df[:len(X_train_sc)]], axis = 1)
X_validation_sc_ec = pd.concat([no_obj_df[len(X_train_sc):] , encoded_df[len(X_train_sc):].reset_index(drop=True)], axis = 1)
# 인코딩 객체 저장
#joblib.dump(encoder, '/home/cloudcccr3/encoder/encoder.pkl')
X_train_sc_ec.head()
9. VIF로 다중공선성을 평가 보통 10이상이면 있다고 판단
!pip install statsmodels
import pandas as pd
from statsmodels.stats.outliers_influence import variance_inflation_factor
def calculate_vif(data_frame):
vif_data = pd.DataFrame()
vif_data["Variable"] = data_frame.columns
vif_data["VIF"] = [variance_inflation_factor(data_frame.values, i) for i in range(data_frame.shape[1])]
return vif_data
vif_values = calculate_vif(X_train_sc_ec.iloc[:,:5])
vif_values
10. 랜포
from sklearn.ensemble import RandomForestRegressor
RFC = RandomForestRegressor(random_state=10)
RFC.fit(X_train_sc_ec, Y_train)
# 모델 저장
import joblib
# 모델을 파일로 저장
# joblib.dump(RFC, '/home/cloudcccr3/modelsave/rf_model.pkl')
joblib.dump(RFC, 'rf_model.pkl')
pred_train = RFC.predict(X_train_sc_ec)
pred_validation = RFC.predict(X_validation_sc_ec)
# 성능평가
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
mse_train = mean_squared_error(Y_train, pred_train)
mse_validation = mean_squared_error(Y_validation, pred_validation)
mae_train = mean_absolute_error(Y_train, pred_train)
mae_validation = mean_absolute_error(Y_validation, pred_validation)
r2_train = r2_score(Y_train, pred_train)
r2_validation = r2_score(Y_validation, pred_validation)
print('mse_train',mse_train)
print('mae_train',mae_train)
print('r2_train',r2_train)
print('\n')
print('mse_validation',mse_validation)
print('mae_validation',mae_validation)
print('r2_validation',r2_validation)
위의 코드들을 정리해서 이미지로 만들어주자
일단 베이스이미지를 수정해주자 스케일, 인코딩, 모델 객체를 저장해줄 디렉토리를 컨테이너에 생성해줘야됨
FROM brightfly/kubeflow-jupyter-lab:tf2.0-cpu
RUN pip install --upgrade pip
# Install required Python packages
RUN pip install pymongo xgboost pymysql pandas scikit-learn
RUN mkdir scaler
RUN mkdir encoder
RUN mkdir model
이미지로 만들어주자
import os
import pymysql
import pandas as pd
from pymongo import MongoClient
from datetime import datetime
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import joblib
from sklearn.preprocessing import OneHotEncoder
from sklearn.ensemble import RandomForestRegressor
import os
class MyFashionMnist(object):
def train(self):
# 연결 정보 설정
host = "10.103.132.127"
port = 3306
database = "donpa_item"
user = "root"
password = "1234"
# 연결
connection = pymysql.connect(
host=host,
port=port,
user=user,
password=password,
database=database
)
# SQL 쿼리
query = "SELECT * FROM goldprice"
# 데이터프레임으로 변환
date_df = pd.read_sql(query, connection)
# 연결 종료
connection.close
# MongoDB 연결 설정
client = MongoClient("mongodb://3.38.178.84:27017/")
db = client["donpa"] # 데이터베이스명
collection = db["donpa_aabata_rare"] # 컬렉션명
# 컬렉션 데이터 조회
data = collection.find()
# 데이터프레임으로 변환
df = pd.DataFrame(data)
# _id 컬럼은 제거해주자
df = df.drop(columns="_id")
df['price'] = df['price'].astype('float')
df['soldDate'] = df['soldDate'].str.replace('-','')
# sell, buy컬럼 붙여주자!!
df = df.merge(date_df, left_on='soldDate', right_on='date', how='inner')
# date컬럼 삭제해주자
df.drop(columns = 'date', inplace=True)
df['soldDate'] = pd.to_datetime(df['soldDate'])
# 년 컬럼 추가
df['year'] = df['soldDate'].dt.year
# 월 컬럼 추가
df['month'] = df['soldDate'].dt.month
# 일 컬럼 추가
df['day'] = df['soldDate'].dt.day
# 요일 컬럼 추가해주자
df['day_name'] = df['soldDate'].dt.day_name()
# 필요 없는 컬럼 삭제
df.drop(columns = 'soldDate', inplace=True)
# ava_rit도 어차피다 레어이니까 삭제
df.drop(columns = 'ava_rit', inplace=True)
# 타겟 컬러 맨뒤로 보내기
price_column = df.pop('price')
df['price'] = price_column
X_train = df.drop(columns = 'price')
y_train = df['price']
X_train, X_validation, Y_train, Y_validation = train_test_split(X_train,y_train, test_size=0.2, random_state=1)
X_train.reset_index(drop=True, inplace=True)
X_validation.reset_index(drop=True, inplace=True)
obj_col = X_train.select_dtypes(include='object').columns
sds = StandardScaler()
sds.fit(X_train.drop(columns = obj_col))
X_train_sc = sds.transform(X_train.drop(columns = obj_col))
X_train_sc = pd.DataFrame(X_train_sc, columns = X_train.drop(columns = obj_col).columns)
X_validation_sc = sds.transform(X_validation.drop(columns = obj_col))
X_validation_sc = pd.DataFrame(X_validation_sc, columns = X_validation.drop(columns = obj_col).columns)
# object 타입 컬럼 붙여주기
for i in obj_col:
X_train_sc[i] = X_train[i]
X_validation_sc[i] = X_validation[i]
# 스케일러 객체 저장
joblib.dump(sds, '/home/jovyan/scaler/scaler.pkl')
# OneHotEncoder 객체 생성
encoder = OneHotEncoder()
X_full = pd.concat([X_train_sc, X_validation_sc])
# 범주형 열만 선택
obj_df = X_full.select_dtypes(include='object')
# 숫자형 열만 선택
no_obj_df = X_full.select_dtypes(exclude='object')
# 범주형 열을 원핫 인코딩
encoded_features = encoder.fit_transform(obj_df)
# 인코딩된 결과를 데이터프레임으로 변환
encoded_df = pd.DataFrame(encoded_features.toarray(), columns=encoder.get_feature_names(obj_df.columns))
# 인코딩된 범주형 열과 숫자형 열을 합침
X_train_sc_ec = pd.concat([no_obj_df[:len(X_train_sc)] , encoded_df[:len(X_train_sc)]], axis = 1)
X_validation_sc_ec = pd.concat([no_obj_df[len(X_train_sc):] , encoded_df[len(X_train_sc):].reset_index(drop=True)], axis = 1)
# 인코딩 객체 저장
joblib.dump(encoder, '/home/jovyan/encoder/encoder.pkl')
RFC = RandomForestRegressor(random_state=10)
RFC.fit(X_train_sc_ec, Y_train)
# 모델을 파일로 저장
joblib.dump(RFC, '/home/jovyan/model/model.pkl')
if __name__ == '__main__':
import os
if os.getenv('FAIRING_RUNTIME', None) is None:
# Kubeflow Fairing의 필요한 모듈을 가져옵니다.
from kubeflow.fairing.builders.append.append import AppendBuilder
from kubeflow.fairing.preprocessors.converted_notebook import ConvertNotebookPreprocessor
# Docker 이미지 관련 설정
DOCKER_REGISTRY = 'kubeflow-registry.default.svc.cluster.local:30000'
base_image = 'sy02229/new_image:latest'
image_name = 'fairing-job100'
# Fairing 빌더 설정
builder = AppendBuilder(
registry=DOCKER_REGISTRY,
image_name=image_name,
base_image=base_image,
push=True,
preprocessor=ConvertNotebookPreprocessor(
notebook_file="aabata_rare.ipynb" # 노트북 파일 지정
)
)
builder.build() # Docker 이미지 빌드 및 푸시
else:
# Kubeflow Fairing 런타임 환경에서 직접 실행되는 경우
myModel = MyFashionMnist()
myModel.train() # 모델 훈련 실행
해당 이미지를 테스트용으로 일단 돌려볼 크론잡
apiVersion: batch/v1beta1
kind: CronJob
metadata:
name: aabata-rare-cj
spec:
schedule: "*/5 * * * *"
jobTemplate:
spec:
template:
metadata:
labels:
app: aabata-rare-cj
spec:
restartPolicy: OnFailure
containers:
- name: aabata-rare-cj
image: kubeflow-registry.default.svc.cluster.local:30000/fairing-job100:57510100
volumeMounts:
- name: scaler-volume
mountPath: /home/jovyan/scaler
- name: encoder-volume
mountPath: /home/jovyan/encoder
- name: model-volume
mountPath: /home/jovyan/model
volumes:
- name: scaler-volume
persistentVolumeClaim:
claimName: scaler-pvc
- name: encoder-volume
persistentVolumeClaim:
claimName: encoder-pvc
- name: model-volume
persistentVolumeClaim:
claimName: model-pvc볼륨 마운트는 하나로 통합시켜줘야됨 아니면 마지막꺼만 실행됨
크론잡 실행후 pv에 저장되었는지 확인해보자
sudo ls -l /mnt/sharedfolder/default-scaler-pvc-pvc-c3833521-03b7-4a20-a6e3-53b96300b015
sudo ls -l /mnt/sharedfolder/default-encoder-pvc-pvc-9180bafa-27db-4e0a-8f71-a14da3e1b65a
sudo ls -l /mnt/sharedfolder/default-model-pvc-pvc-456da6be-dd49-4e3a-a6dd-e26243396ab0
이제 카팁으로 모델 최적의 매개변수를 찾아보자
카팁에 사용할 이미지를 만들어보자
import os
import pymysql
import pandas as pd
from pymongo import MongoClient
from datetime import datetime
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import joblib
from sklearn.preprocessing import OneHotEncoder
from sklearn.ensemble import RandomForestRegressor
import os
import argparse
class MyFashionMnist(object):
def train(self):
# 연결 정보 설정
host = "10.103.132.127"
port = 3306
database = "donpa_item"
user = "root"
password = "1234"
# 연결
connection = pymysql.connect(
host=host,
port=port,
user=user,
password=password,
database=database
)
# SQL 쿼리
query = "SELECT * FROM goldprice"
# 데이터프레임으로 변환
date_df = pd.read_sql(query, connection)
# 연결 종료
connection.close
# MongoDB 연결 설정
client = MongoClient("mongodb://3.38.178.84:27017/")
db = client["donpa"] # 데이터베이스명
collection = db["donpa_aabata_rare"] # 컬렉션명
# 컬렉션 데이터 조회
data = collection.find()
# 데이터프레임으로 변환
df = pd.DataFrame(data)
# _id 컬럼은 제거해주자
df = df.drop(columns="_id")
df['price'] = df['price'].astype('float')
df['soldDate'] = df['soldDate'].str.replace('-','')
# sell, buy컬럼 붙여주자!!
df = df.merge(date_df, left_on='soldDate', right_on='date', how='inner')
# date컬럼 삭제해주자
df.drop(columns = 'date', inplace=True)
df['soldDate'] = pd.to_datetime(df['soldDate'])
# 년 컬럼 추가
df['year'] = df['soldDate'].dt.year
# 월 컬럼 추가
df['month'] = df['soldDate'].dt.month
# 일 컬럼 추가
df['day'] = df['soldDate'].dt.day
# 요일 컬럼 추가해주자
df['day_name'] = df['soldDate'].dt.day_name()
# 필요 없는 컬럼 삭제
df.drop(columns = 'soldDate', inplace=True)
# ava_rit도 어차피다 레어이니까 삭제
df.drop(columns = 'ava_rit', inplace=True)
# 타겟 컬러 맨뒤로 보내기
price_column = df.pop('price')
df['price'] = price_column
X_train = df.drop(columns = 'price')
y_train = df['price']
X_train, X_validation, Y_train, Y_validation = train_test_split(X_train,y_train, test_size=0.2, random_state=1)
X_train.reset_index(drop=True, inplace=True)
X_validation.reset_index(drop=True, inplace=True)
obj_col = X_train.select_dtypes(include='object').columns
sds = StandardScaler()
sds.fit(X_train.drop(columns = obj_col))
X_train_sc = sds.transform(X_train.drop(columns = obj_col))
X_train_sc = pd.DataFrame(X_train_sc, columns = X_train.drop(columns = obj_col).columns)
X_validation_sc = sds.transform(X_validation.drop(columns = obj_col))
X_validation_sc = pd.DataFrame(X_validation_sc, columns = X_validation.drop(columns = obj_col).columns)
# object 타입 컬럼 붙여주기
for i in obj_col:
X_train_sc[i] = X_train[i]
X_validation_sc[i] = X_validation[i]
# 스케일러 객체 저장
joblib.dump(sds, '/home/jovyan/scaler/scaler.pkl')
# OneHotEncoder 객체 생성
encoder = OneHotEncoder()
X_full = pd.concat([X_train_sc, X_validation_sc])
# 범주형 열만 선택
obj_df = X_full.select_dtypes(include='object')
# 숫자형 열만 선택
no_obj_df = X_full.select_dtypes(exclude='object')
# 범주형 열을 원핫 인코딩
encoded_features = encoder.fit_transform(obj_df)
# 인코딩된 결과를 데이터프레임으로 변환
encoded_df = pd.DataFrame(encoded_features.toarray(), columns=encoder.get_feature_names(obj_df.columns))
# 인코딩된 범주형 열과 숫자형 열을 합침
X_train_sc_ec = pd.concat([no_obj_df[:len(X_train_sc)] , encoded_df[:len(X_train_sc)]], axis = 1)
X_validation_sc_ec = pd.concat([no_obj_df[len(X_train_sc):] , encoded_df[len(X_train_sc):].reset_index(drop=True)], axis = 1)
# 인코딩 객체 저장
joblib.dump(encoder, '/home/jovyan/encoder/encoder.pkl')
parser = argparse.ArgumentParser()
parser.add_argument('--n_estimators', required=False, type=int, default=100)
parser.add_argument('--max_depth', required=False, type=int, default=5)
parser.add_argument('--min_samples_split', required=False, type=int, default=2)
args = parser.parse_args()
RFC = RandomForestRegressor(random_state=10,
n_estimators=args.n_estimators,
max_depth=args.max_depth,
min_samples_split=args.min_samples_split
)
RFC.fit(X_train_sc_ec, Y_train)
# 모델을 파일로 저장
# joblib.dump(RFC, '/home/jovyan/model/model.pkl')
# 검증 데이터 예측
pred_validation = RFC.predict(X_validation_sc_ec)
# 성능평가
from sklearn.metrics import mean_absolute_error
mae_validation = mean_absolute_error(Y_validation, pred_validation)
print("mae_validation="+str(mae_validation))
if __name__ == '__main__':
import os
if os.getenv('FAIRING_RUNTIME', None) is None:
# Kubeflow Fairing의 필요한 모듈을 가져옵니다.
from kubeflow.fairing.builders.append.append import AppendBuilder
from kubeflow.fairing.preprocessors.converted_notebook import ConvertNotebookPreprocessor
# Docker 이미지 관련 설정
DOCKER_REGISTRY = 'kubeflow-registry.default.svc.cluster.local:30000'
base_image = 'sy02229/new_image:latest'
image_name = 'fairing-job100'
# Fairing 빌더 설정
builder = AppendBuilder(
registry=DOCKER_REGISTRY,
image_name=image_name,
base_image=base_image,
push=True,
preprocessor=ConvertNotebookPreprocessor(
notebook_file="aabata_rare.ipynb" # 노트북 파일 지정
)
)
builder.build() # Docker 이미지 빌드 및 푸시
else:
# Kubeflow Fairing 런타임 환경에서 직접 실행되는 경우
myModel = MyFashionMnist()
myModel.train() # 모델 훈련 실행
카팁 yaml
apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
namespace: gsoh
labels:
controller-tools.k8s.io: "1.0"
name: fashion-mnist-cpu-experiment-v2-1
spec:
parallelTrialCount: 2
maxTrialCount: 40
maxFailedTrialCount: 3
objective:
type: minimize # 작게해야되니까 크게하려면 maximize
goal: 3800000 # mae 목표를 설정
objectiveMetricName: mae_validation #변수 이름 넣어줘야됨
algorithm:
algorithmName: random
trialTemplate:
goTemplate:
rawTemplate: |-
apiVersion: batch/v1
kind: Job
metadata:
name: {{.Trial}}
namespace: {{.NameSpace}}
spec:
template:
spec:
containers:
- name: {{.Trial}}
image: kubeflow-registry.default.svc.cluster.local:30000/fairing-job100:86AE588D # 내가 생성한 이미지로 바꿔야됨
command:
- "python"
- "/app/aabata_rare.py" # 내 파이썬 파일 이름으로 설정
{{- with .HyperParameters}}
{{- range .}}
- "{{.Name}}={{.Value}}"
{{- end}}
{{- end}}
restartPolicy: Never
parameters:
- name: --n_estimators
parameterType: int
feasibleSpace:
min: "50"
max: "100"
- name: --max_depth
parameterType: int
feasibleSpace:
min: "1"
max: "30"
- name: --min_samples_split
parameterType: int
feasibleSpace:
min: "1"
max: "5"
카팁 돌아가는지 확인
모델을 xgboost로 수정해주자
import os
import pymysql
import pandas as pd
from pymongo import MongoClient
from datetime import datetime
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import joblib
from sklearn.preprocessing import OneHotEncoder
from sklearn.ensemble import RandomForestRegressor
import os
import argparse
import xgboost as xgb
class MyFashionMnist(object):
def train(self):
# 연결 정보 설정
host = "10.103.132.127"
port = 3306
database = "donpa_item"
user = "root"
password = "1234"
# 연결
connection = pymysql.connect(
host=host,
port=port,
user=user,
password=password,
database=database
)
# SQL 쿼리
query = "SELECT * FROM goldprice"
# 데이터프레임으로 변환
date_df = pd.read_sql(query, connection)
# 연결 종료
connection.close
# MongoDB 연결 설정
client = MongoClient("mongodb://3.38.178.84:27017/")
db = client["donpa"] # 데이터베이스명
collection = db["donpa_aabata_rare"] # 컬렉션명
# 컬렉션 데이터 조회
data = collection.find()
# 데이터프레임으로 변환
df = pd.DataFrame(data)
# _id 컬럼은 제거해주자
df = df.drop(columns="_id")
df['price'] = df['price'].astype('float')
df['soldDate'] = df['soldDate'].str.replace('-','')
# sell, buy컬럼 붙여주자!!
df = df.merge(date_df, left_on='soldDate', right_on='date', how='left')
# sell, buy값이 널값이면 가장 최근의 sell, buy값으로 대체해주자
null_check = df['sell'].isnull().sum()
if null_check != 0:
# date_df의 맨 마지막 값을 가져와서 NaN 값을 채우기
last_date_df_row = date_df.iloc[-1]
df['sell'].fillna(last_date_df_row['sell'], inplace=True)
df['buy'].fillna(last_date_df_row['buy'], inplace=True)
# date컬럼 삭제해주자
df.drop(columns = 'date', inplace=True)
df['soldDate'] = pd.to_datetime(df['soldDate'])
# 년 컬럼 추가
df['year'] = df['soldDate'].dt.year
# 월 컬럼 추가
df['month'] = df['soldDate'].dt.month
# 일 컬럼 추가
df['day'] = df['soldDate'].dt.day
# 요일 컬럼 추가해주자
df['day_name'] = df['soldDate'].dt.day_name()
# 필요 없는 컬럼 삭제
df.drop(columns = 'soldDate', inplace=True)
# ava_rit도 어차피다 레어이니까 삭제
df.drop(columns = 'ava_rit', inplace=True)
# 타겟 컬러 맨뒤로 보내기
price_column = df.pop('price')
df['price'] = price_column
X_train = df.drop(columns = 'price')
y_train = df['price']
X_train, X_validation, Y_train, Y_validation = train_test_split(X_train,y_train, test_size=0.2, random_state=1)
X_train.reset_index(drop=True, inplace=True)
X_validation.reset_index(drop=True, inplace=True)
obj_col = X_train.select_dtypes(include='object').columns
sds = StandardScaler()
sds.fit(X_train.drop(columns = obj_col))
X_train_sc = sds.transform(X_train.drop(columns = obj_col))
X_train_sc = pd.DataFrame(X_train_sc, columns = X_train.drop(columns = obj_col).columns)
X_validation_sc = sds.transform(X_validation.drop(columns = obj_col))
X_validation_sc = pd.DataFrame(X_validation_sc, columns = X_validation.drop(columns = obj_col).columns)
# object 타입 컬럼 붙여주기
for i in obj_col:
X_train_sc[i] = X_train[i]
X_validation_sc[i] = X_validation[i]
# 스케일러 객체 저장
joblib.dump(sds, '/home/jovyan/scaler/scaler.pkl')
# OneHotEncoder 객체 생성
encoder = OneHotEncoder()
X_full = pd.concat([X_train_sc, X_validation_sc])
# 범주형 열만 선택
obj_df = X_full.select_dtypes(include='object')
# 숫자형 열만 선택
no_obj_df = X_full.select_dtypes(exclude='object')
# 범주형 열을 원핫 인코딩
encoded_features = encoder.fit_transform(obj_df)
# 인코딩된 결과를 데이터프레임으로 변환
encoded_df = pd.DataFrame(encoded_features.toarray(), columns=encoder.get_feature_names(obj_df.columns))
# 인코딩된 범주형 열과 숫자형 열을 합침
X_train_sc_ec = pd.concat([no_obj_df[:len(X_train_sc)] , encoded_df[:len(X_train_sc)]], axis = 1)
X_validation_sc_ec = pd.concat([no_obj_df[len(X_train_sc):] , encoded_df[len(X_train_sc):].reset_index(drop=True)], axis = 1)
# 인코딩 객체 저장
joblib.dump(encoder, '/home/jovyan/encoder/encoder.pkl')
import re
# 데이터프레임의 컬럼 이름에서 특수 문자를 제거하고 변경할 새로운 컬럼 이름 리스트 생성
new_columns = []
for old_column in X_train_sc_ec.columns:
new_column = re.sub(r'[^\w\s]', '', old_column) # 특수 문자 제거
new_columns.append(new_column)
# 컬럼 이름을 새로운 이름으로 설정
X_train_sc_ec.columns = new_columns
X_validation_sc_ec.columns = new_columns
parser = argparse.ArgumentParser()
parser.add_argument('--learning_rate', required=False, type=float, default=0.1)
parser.add_argument('--n_estimators', required=False, type=int, default=100)
parser.add_argument('--max_depth', required=False, type=int, default=5)
args = parser.parse_args()
xgb_model = xgb.XGBRegressor(random_state=10,
learning_rate=args.learning_rate,
n_estimators=args.n_estimators,
max_depth=args.max_depth
)
xgb_model.fit(X_train_sc_ec, Y_train)
# 모델을 파일로 저장
# joblib.dump(RFC, '/home/jovyan/model/model.pkl')
# 검증 데이터 예측
pred_validation = xgb_model.predict(X_validation_sc_ec)
# 성능평가
from sklearn.metrics import mean_absolute_error
mae_validation = mean_absolute_error(Y_validation, pred_validation)
print("mae_validation="+str(mae_validation))
if __name__ == '__main__':
import os
if os.getenv('FAIRING_RUNTIME', None) is None:
# Kubeflow Fairing의 필요한 모듈을 가져옵니다.
from kubeflow.fairing.builders.append.append import AppendBuilder
from kubeflow.fairing.preprocessors.converted_notebook import ConvertNotebookPreprocessor
# Docker 이미지 관련 설정
DOCKER_REGISTRY = 'kubeflow-registry.default.svc.cluster.local:30000'
base_image = 'sy02229/new_image:latest'
image_name = 'fairing-job100'
# Fairing 빌더 설정
builder = AppendBuilder(
registry=DOCKER_REGISTRY,
image_name=image_name,
base_image=base_image,
push=True,
preprocessor=ConvertNotebookPreprocessor(
notebook_file="aabata_rare.ipynb" # 노트북 파일 지정
)
)
builder.build() # Docker 이미지 빌드 및 푸시
else:
# Kubeflow Fairing 런타임 환경에서 직접 실행되는 경우
myModel = MyFashionMnist()
myModel.train() # 모델 훈련 실행
카팁 야물
apiVersion: "kubeflow.org/v1alpha3"
kind: Experiment
metadata:
namespace: gsoh
labels:
controller-tools.k8s.io: "1.0"
name: fashion-mnist-cpu-experiment-v2-1
spec:
parallelTrialCount: 2
maxTrialCount: 40
maxFailedTrialCount: 3
objective:
type: minimize # 작게해야되니까 크게하려면 maximize
goal: 1400000 # mae 목표를 설정
objectiveMetricName: mae_validation #변수 이름 넣어줘야됨
algorithm:
algorithmName: random
trialTemplate:
goTemplate:
rawTemplate: |-
apiVersion: batch/v1
kind: Job
metadata:
name: {{.Trial}}
namespace: {{.NameSpace}}
spec:
template:
spec:
containers:
- name: {{.Trial}}
image: kubeflow-registry.default.svc.cluster.local:30000/fairing-job100:5D33A17C # 내가 생성한 이미지로 바꿔야됨
command:
- "python"
- "/app/aabata_rare.py" # 내 파이썬 파일 이름으로 설정
{{- with .HyperParameters}}
{{- range .}}
- "{{.Name}}={{.Value}}"
{{- end}}
{{- end}}
restartPolicy: Never
parameters:
- name: --learning_rate
parameterType: double
feasibleSpace:
min: "0.1"
max: "0.3"
- name: --n_estimators
parameterType: int
feasibleSpace:
min: "30"
max: "100"
- name: --max_depth
parameterType: int
feasibleSpace:
min: "1"
max: "30"
최적의 파라미터
최적의값