[stock]SGDRegressor를 사용한 주가예측

SGDRegressor를 사용한 주가예측

Stochastic Gradient Descent(SGD) 알고리즘을 사용하여 선형모델을 학습시키는 클래스입니다.

sklearn.linear_mode.SGDRegressor() 클래스를 사용하며 선형모델을 생성합니다.

• loss: 손실함수 지정, 'squaredd_error'(최소제곱오차, 기본값), 'huber', 'epsilon_insensitive', 'squared_epsilon_insensitive'를 지정할 수 있음.
• penalty : 과적합 방지를 위한 적용할 규제 지정 . ('l1', 'l2', 'elasticnet'). 기본값은 None (규제 없음)입니다.
• alpha: 규제 강도를 조절하는 상수이며 클수록 규제가 강해집니다. 기본값은 0.0001입니다.
• l1_ratio: Elastic Net 규제 사용 시 L1 규제와 L2 규제의 혼합 비율을 결정합니다 (0 ≤ l1_ratio ≤ 1). penalty='elasticnet'일 때만 사용됩니다.
• fit_intercept: 모델에 절편 (bias) 항을 추가할지 여부를 결정. 기본값은 True.
• max_iter: 학습 데이터에 대한 최대 반복 횟수 (epoch)를 지정. 기본값은 1000.
• tol: 학습 종료 조건으로, 손실 함수의 감소량이 이 값보다 작으면 학습을 중단. 기본값은 1e-3.
• learning_rate: 학습률 스케줄링 방법을 지정('constant', 'optimal', 'invscaling', 'adaptive'). 기본값은 'invscaling'.
• eta0: 초기 학습률을 지정 (learning_rate='constant' 또는 learning_rate='invscaling'일 때 사용). 기본값은 0.01.
• random_state
• shuffle: 각 epoch마다 데이터를 섞을지 여부를 결정. 기본값은 True.

import numpy as np
import pandas as pd
import pandas_ta as ta
import matplotlib.pyplot as plt
import FinanceDataReader as fdr
import matplotlib.pyplot as plt

from sklearn.model_selection import GridSearchCV
from sklearn.linear_model import SGDRegressor
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, mean_squared_error, r2_score

st=pd.Timestamp(2023, 1,1)
et=pd.Timestamp(2025, 5,20)
trgnme="000660"
trg=fdr.DataReader(trgnme,  st, et)
df=trg[["Open", "High", "Low", "Close", "Volume"]]
#위 데이터를 기준으로 기술적 지표를 첨가한 새로운 자료 생성 
data=df.copy()
data["ema5"]=df.ta.ema(5)
data["ema20"]=df.ta.ema(20)
data[['bbl', 'bbm','bbu']]=df.ta.bbands().iloc[:,:3]
data.dropna(inplace=True)
data.head(1).round(0)

	Open	High	Low	Close	Volume	ema5	ema20	bbl	bbm	bbu
Date
2023-01-31	90800	90800	86800	88500	5185088	89697.0	85675.0	88293.0	90880.0	93467.0

자료 Close 예측을 위한 sgd 모델을 구축합니다. 모델 구축을 위해 데이터를 정규화하고 학습과 검증 데이터로 분리합니다. 또한 SGDRegressor()클래스의 다양한 하이퍼파라메터의 적정값으로 구성된 모델을 생성하기 위해 GridSearch() 클래스를 적용합니다.

d=df.copy()
ind=d.values
de=d["Close"].values
scalerX=StandardScaler().fit(ind[:-1, :])
X=scalerX.transform(ind[:-1,:])
Xfinal=scalerX.transform(ind[-1,:].reshape(-1, ind.shape[1]))
scalery=StandardScaler()
y=scalery.fit_transform(de[1:].reshape(-1,1)).flatten()
Xtr, Xte, ytr, yte=train_test_split(X, y, test_size=0.3, random_state=7)

param={"alpha":[0.0001, 0.001, 0.01, 0.1, 1], "l1_ratio":[0.0, 0.25, 0.5, 0.75, 1.0]}
sgd_reg=SGDRegressor(penalty="elasticnet", random_state=7)
sgd_search=GridSearchCV(sgd_reg, param, cv=5, scoring='neg_mean_squared_error',  n_jobs=-1)
sgd_search.fit(Xtr, ytr)
sgd_search.best_estimator_

SGDRegressor(alpha=0.001, l1_ratio=1.0, penalty='elasticnet', random_state=7)

sgd_search의 메소드 .score()는 음의 MSE값을 반환합니다.

sgd_search.score(Xtr, ytr), sgd_search.score(Xte, yte)

(-0.013411371990480763, -0.012495508936140981)

위 결과는 학습과 검증 세트 간의 큰 차이는 없습니다. 그러나 이 값들만으로 의미를 찾기는 어렵습니다. 대신에 결정계수(R²)로 비교해 봅니다. 이것은 sklearn.metrics.2_score(y, pre) 함수로 계산할 수 있습니다.

preTr=sgd_search.predict(Xtr)
preTe=sgd_search.predict(Xte)
r2tr=r2_score(ytr, preTr)
r2te=r2_score(yte, preTe)
print("r2_tr: {%.3f}, r2_te: {%.3f}" %(r2tr, r2te))

r2_tr: {0.987}, r2_te: {0.986}

최종예측을 위한 예측변수인 Xfinal에 대한 예측은 다음과 같습니다.

pre=sgd_search.predict(Xfinal)
scalery.inverse_transform(pre.reshape(-1,1))

array([[203426.69373404]])

위 과정을 하나의 함수로 작성하여 Open, High, Low, Close을 예측합니다.

#sgdregressor()에 의한 예측 함수 
def sgdPredict(data, deCol="Close", testSize=0.3, randomState=7):
    ind=data.values
    de=data[deCol].values
    scalerX=StandardScaler().fit(ind[:-1, :])
    X=scalerX.transform(ind[:-1,:])
    Xfinal=scalerX.transform(ind[-1,:].reshape(-1, ind.shape[1]))
    scalery=StandardScaler()
    y=scalery.fit_transform(de[1:].reshape(-1,1)).flatten()
    Xtr, Xte, ytr, yte=train_test_split(X, y, test_size=testSize, random_state=randomState)
    param={"alpha":[0.0001, 0.001, 0.01, 0.1, 1], "l1_ratio":[0.0, 0.25, 0.5, 0.75, 1.0]}
    sgd_reg=SGDRegressor(penalty="elasticnet", random_state=7)
    sgd_search=GridSearchCV(sgd_reg, param, cv=5, scoring='neg_mean_squared_error',  n_jobs=-1)
    sgd_search.fit(Xtr, ytr)
    pre=sgd_search.predict(Xfinal)
    return scalery.inverse_transform(pre.reshape(-1,1)).flatten()

#위 결과에 당일 Open을 고려한 결과를 반환 
#result: 위 홤수들의 결과, op: 당일  Open
def consider_Op(result, op):
    open_op=result["Open"]-op
    r=pd.DataFrame()
    for i, j in enumerate(result.index):
        r=pd.concat([r, result.iloc[i,:]-open_op.iloc[i]], axis=1)
    return r.T

result_sgd1={}
result_sgd2={}
for i in ["Open","High","Low", "Close"]:
    result_sgd1[i]=sgdPredict(df, deCol=i)
    result_sgd2[i]=sgdPredict(data, deCol=i)
result_sgd1=pd.DataFrame(result_sgd1)
result_sgd2=pd.DataFrame(result_sgd2)
result_sgd=pd.concat([result_sgd1, result_sgd2])
result_sgd.index=["OHLC", "addIdx"]
result_sgd.round(0)

	Open	High	Low	Close
OHLC	203432.0	206961.0	200091.0	203427.0
addIdx	202584.0	205584.0	199389.0	203113.0

위 결과에 알려진 시가를 고려합니다(consider_Op() 사용).

consider_Op(result_sgd, 204500).round(0)

	Open	High	Low	Close
OHLC	204500.0	208029.0	201159.0	204495.0
addIdx	204500.0	207500.0	201305.0	205028.0