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)
위 결과는 학습과 검증 세트 간의 큰 차이는 없습니다. 그러나 이 값들만으로 의미를 찾기는 어렵습니다. 대신에 결정계수(R2)로 비교해 봅니다. 이것은 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 |
댓글
댓글 쓰기