Python으로 주식 데이터를 분석하고 미래 주가를 예측하는 것은 금융 데이터 분석에서 중요한 역할을 합니다.
이번 포스팅에서는 Darts라는 라이브러리를 활용해서 SCHD ETF의 주가를 예측하는 모델을 만들어 보도록 하겠습니다.
지난번 Prophet 라이브러리로 SCHD ETF의 주가를 분석한 것의 연장 버전입니다.
[머신러닝 with Python] Prophet 모델로 SCHD 주가 분석하기
1. Darts 라이브러리로 SCHD 주가 예측모델 만들기
1.1 Darts 라이브러리란?
- Darts는 시계열(Time Series) 데이터 분석과 예측을 위한 강력한 라이브러리입니다.
- 이 라이브러리는 딥러닝 모델과 전통적인 통계 모델을 모두 지원하여 다양한 시계열 분석이 가능합니다.
- Darts 라이브러리의 장점은 다음과 같은데요
a) ARIMA, Exponential Smoothing, Prophet 같은 전통적인 모델 지원
b) N-BEATS, Transformer 같은 딥러닝 기반 모델 제공
c) 한 번의 코드 실행으로 여러 모델을 쉽게 비교 가능
d)PyTorch 기반으로 GPU 가속 가능
1.2 SCHD란?
- SCHD는 Schwab U.S. Dividend Equity ETF 의 티커(Symbol)입니다.
- 이는, 배당 성장주에 투자하는 미국 주식 기반 **ETF(Exchange Traded Fund, 상장지수펀드)**입니다.
2. Darts 라이브러리를 활용해서 SCHD ETF 주가 예측모델 만들기
- 모델 구현은 구글 코랩 환경에서 진행하였습니다.
- 먼저, 도출된 결과를 Pyplot으로 시각화하며, 이때 한글폰트를 사용하기 위해 아래와 같은 nanum 폰트를 설치하고 설정해줍니다.
!apt-get install -y fonts-nanum
!fc-cache -fv
!rm -rf ~/.cache/matplotlib
import matplotlib.pyplot as plt
from matplotlib import font_manager
# 나눔 폰트 경로 확인
font_path = "/usr/share/fonts/truetype/nanum/NanumGothic.ttf"
# 폰트 설정
font_manager.fontManager.addfont(font_path)
plt.rc('font', family='NanumGothic') # 나눔고딕 설정
plt.rcParams['axes.unicode_minus'] = False # 마이너스 기호 깨짐 방지
- 이제 분석에 필요한 darts 라이브러리와 데이터를 쉽게 가져오기위한 yfinance 라이브러리를 설치해줍니다.
# 라이브러리 설치 (Colab 환경)
!pip install darts yfinance --quiet
- 이제 분석에 필요한 라이브러리들을 불러와줍니다.
# 필수 라이브러리 불러오기
import yfinance as yf
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from darts import TimeSeries
from darts.models import ARIMA, ExponentialSmoothing, Prophet, NBEATSModel, TransformerModel
from darts.metrics import mse, mae
import warnings
warnings.filterwarnings("ignore")
- 이제 데이터를 불러오고, darts 라이브러리에 활용하기 위해 전처리를 진행해줍니다.
- 사용할 데이터는 2015년 1월 1일 ~ 2024년 1월 1일 까지의 데이터입니다.
# ✅ 1️⃣ 데이터 다운로드 및 전처리
ticker = "SCHD"
df = yf.download(ticker, start="2015-01-01", end="2024-01-01", auto_adjust=True)
df = df[["Close"]]
df.reset_index(inplace=True)
df.columns = ["ds", "y"] # Prophet과 호환되도록 열 이름 변경
# 📌 결측값 처리 및 정규화
df.set_index("ds", inplace=True)
df = df.asfreq("B") # 영업일 기준 데이터로 보정
df.fillna(method="ffill", inplace=True) # 결측값 전방 채우기
# MinMax 정규화 적용 (N-BEATS 및 Transformer 안정성 향상)
scaler = MinMaxScaler()
df["y"] = scaler.fit_transform(df[["y"]])
# Darts의 TimeSeries 객체 변환
ts = TimeSeries.from_dataframe(df, fill_missing_dates=True, freq="B")
- 주어진 데이터학습 및 테스트 데이터 세트를최근 1년을 테스트로 활용하기 위해 분할해줍니다.
# 학습 및 테스트 데이터 분할 (최근 1년을 테스트 데이터로 사용)
train, test = ts[:-252], ts[-252:]
- 이제 학습할 모델들에 대해 하이퍼 파라미터들을 설정해줍니다.
* 활용할 모델은 ARIMA, Exponetial Smoothing, Prophet, N-Beats, Transformer 입니다.
# ✅ 2️⃣ 모델 정의 및 학습 설정 수정
models = {
"ARIMA": ARIMA(5, 1, 0),
"Exponential Smoothing": ExponentialSmoothing(),
"Prophet": Prophet(),
"N-BEATS": NBEATSModel(
input_chunk_length=30,
output_chunk_length=10,
n_epochs=50,
batch_size=16, # 🛠 배치 사이즈 조정
optimizer_kwargs={"lr": 1e-3}, # 🛠 학습률 조정
pl_trainer_kwargs={"accelerator": "gpu"}, # ✅ GPU 설정 수정
random_state=42,
),
"Transformer": TransformerModel(
input_chunk_length=30,
output_chunk_length=10,
n_epochs=50,
batch_size=16, # 🛠 배치 사이즈 조정
optimizer_kwargs={"lr": 1e-3}, # 🛠 학습률 조정
dropout=0.1, # 🛠 Dropout 추가
pl_trainer_kwargs={"accelerator": "gpu"}, # ✅ GPU 설정 수정
random_state=42,
),
}
# 결과 저장 딕셔너리
forecasts = {}
errors = {}
- 이제 모델을 학습시키고 예측성능을 비교해보겠습니다.
# ✅ 3️⃣ 모델 학습 및 예측
for name, model in models.items():
print(f"⏳ 모델 학습 중: {name}...")
try:
model.fit(train)
forecast = model.predict(len(test))
# 예측 결과 저장
forecasts[name] = forecast
# 성능 평가
mse_value = mse(test, forecast)
mae_value = mae(test, forecast)
errors[name] = {"MSE": mse_value, "MAE": mae_value}
print(f"✅ {name} 완료! MSE: {mse_value:.4f}, MAE: {mae_value:.4f}\n")
except Exception as e:
print(f"❌ {name} 모델 오류 발생: {e}\n")
# ✅ 4️⃣ 모델 성능 비교
errors_df = pd.DataFrame(errors).T
print("📊 모델별 MSE & MAE 비교:\n", errors_df)
- 각 모델이 예측한 1년 간의 주가를 시각화해보면 아래와 같습니다.
# ✅ 5️⃣ 시각화: 모든 모델 예측 비교
plt.figure(figsize=(12, 6))
plt.plot(test.time_index, test.values(), label="Actual", color="black", linewidth=2)
for name, forecast in forecasts.items():
plt.plot(test.time_index, forecast.values(), label=name, linestyle="--")
plt.title("SCHD Stock Prediction using various models")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.show()
- 가장 성능이 좋은 모델은 MSE와 MAE를 기준으로
# ✅ 6️⃣ 가장 성능이 좋은 모델 출력
best_model = errors_df["MSE"].idxmin()
print(f"🏆 가장 좋은 모델: {best_model} (MSE: {errors_df.loc[best_model, 'MSE']:.4f}, MAE: {errors_df.loc[best_model, 'MAE']:.4f})")
* Prophet 모델임을 알 수 있습니다.
* Prophet이 전반적인 트렌드를 잘 예측하고 있음을 알 수 있네요
이번에는 2015년 1월 1일부터 2025년 2월 28일까지의 데이터를 불러와서, 이 중 마지막 1년을 테스트 데이터로 설정해서 예측을 진행해보겠습니다.
(포스트 작성을 위해 실험을 하는 시간은 25년 3월 2일 입니다)
# ✅ 1️⃣ 데이터 다운로드 및 전처리
ticker = "SCHD"
df = yf.download(ticker, start="2015-01-01", end="2025-02-28", auto_adjust=True)
df = df[["Close"]]
df.reset_index(inplace=True)
df.columns = ["ds", "y"] # Prophet과 호환되도록 열 이름 변경
# 📌 결측값 처리 및 정규화
df.set_index("ds", inplace=True)
df = df.asfreq("B") # 영업일 기준 데이터로 보정
df.fillna(method="ffill", inplace=True) # 결측값 전방 채우기
# MinMax 정규화 적용 (N-BEATS 및 Transformer 안정성 향상)
scaler = MinMaxScaler()
df["y"] = scaler.fit_transform(df[["y"]])
# Darts의 TimeSeries 객체 변환
ts = TimeSeries.from_dataframe(df, fill_missing_dates=True, freq="B")
# 학습 및 테스트 데이터 분할 (최근 1년을 테스트 데이터로 사용)
train, test = ts[:-252], ts[-252:]
# ✅ 2️⃣ 모델 정의 및 학습 설정 수정
models = {
"ARIMA": ARIMA(5, 1, 0),
"Exponential Smoothing": ExponentialSmoothing(),
"Prophet": Prophet(),
"N-BEATS": NBEATSModel(
input_chunk_length=30,
output_chunk_length=10,
n_epochs=50,
batch_size=16, # 🛠 배치 사이즈 조정
optimizer_kwargs={"lr": 1e-3}, # 🛠 학습률 조정
pl_trainer_kwargs={"accelerator": "gpu"}, # ✅ GPU 설정 수정
random_state=42,
),
"Transformer": TransformerModel(
input_chunk_length=30,
output_chunk_length=10,
n_epochs=50,
batch_size=16, # 🛠 배치 사이즈 조정
optimizer_kwargs={"lr": 1e-3}, # 🛠 학습률 조정
dropout=0.1, # 🛠 Dropout 추가
pl_trainer_kwargs={"accelerator": "gpu"}, # ✅ GPU 설정 수정
random_state=42,
),
}
# 결과 저장 딕셔너리
forecasts = {}
errors = {}
# ✅ 3️⃣ 모델 학습 및 예측
for name, model in models.items():
print(f"⏳ 모델 학습 중: {name}...")
try:
model.fit(train)
forecast = model.predict(len(test))
# 예측 결과 저장
forecasts[name] = forecast
# 성능 평가
mse_value = mse(test, forecast)
mae_value = mae(test, forecast)
errors[name] = {"MSE": mse_value, "MAE": mae_value}
print(f"✅ {name} 완료! MSE: {mse_value:.4f}, MAE: {mae_value:.4f}\n")
except Exception as e:
print(f"❌ {name} 모델 오류 발생: {e}\n")
# ✅ 4️⃣ 모델 성능 비교
errors_df = pd.DataFrame(errors).T
print("📊 모델별 MSE & MAE 비교:\n", errors_df)
# ✅ 5️⃣ 시각화: 모든 모델 예측 비교
plt.figure(figsize=(12, 6))
plt.plot(test.time_index, test.values(), label="Actual", color="black", linewidth=2)
for name, forecast in forecasts.items():
plt.plot(test.time_index, forecast.values(), label=name, linestyle="--")
plt.title("SCHD Stock Prediction using various models")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.show()
# ✅ 6️⃣ 가장 성능이 좋은 모델 출력
best_model = errors_df["MSE"].idxmin()
print(f"🏆 가장 좋은 모델: {best_model} (MSE: {errors_df.loc[best_model, 'MSE']:.4f}, MAE: {errors_df.loc[best_model, 'MAE']:.4f})")
가장 좋은 모델은 Exponentail Smoothing인데요.
실제 결과를 확인해보면 결과가 딱히 맘에 들지는 않네요
역시 주가를 예측하는건 어려운것 같습니다.
댓글