Qiitaに記事を書いて得られたメリットについて整理する #Python – Qiita

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

# -----------------------------
# 1. データ取得 & 特徴量作成
# -----------------------------
ticker = "2897.T"
df = yf.download(ticker, start="1995-01-01", end="2025-04-04")
df = df[["Open", "High", "Low", "Close", "Volume"]].dropna()

# --- テクニカル指標（変更禁止） ---
df["MA_5"] = df["Close"].rolling(window=5).mean()
df["MA_25"] = df["Close"].rolling(window=25).mean()
df["MA_75"] = df["Close"].rolling(window=75).mean()
delta = df["Close"].diff()
delta = delta.values.reshape(-1)
gain = np.where(delta > 0, delta, 0)
loss = np.where(delta  0, -delta, 0)
avg_gain = pd.Series(gain, index=df.index).rolling(14).mean()
avg_loss = pd.Series(loss, index=df.index).rolling(14).mean()
rs = avg_gain / (avg_loss + 1e-10)
df["RSI"] = 100 - (100 / (1 + rs))
df["Volatility"] = (df["High"] - df["Low"]) / df["Close"]
df["MA5_diff"] = (df["Close"].values.reshape(-1) - df["MA_5"].values.reshape(-1)) / df["MA_5"].values.reshape(-1)
df["MA25_diff"] = (df["Close"].values.reshape(-1) - df["MA_25"].values.reshape(-1)) / df["MA_25"].values.reshape(-1)
df["MA75_diff"] = (df["Close"].values.reshape(-1) - df["MA_75"].values.reshape(-1)) / df["MA_75"].values.reshape(-1)
df["Return"] = df["Close"].pct_change()
df["LogReturn"] = np.log(df["Close"] / df["Close"].shift(1))

df = df.dropna()

# -----------------------------
# 2. sequence-to-sequence用データ作成
# -----------------------------
features = ["Close", "Volume", "RSI", "Volatility", "MA5_diff", "MA25_diff", "MA75_diff", "Return", "LogReturn"]
data = df[features].values.astype(np.float32)
target = df["LogReturn"].values.astype(np.float32)
close_series = df["Close"].values.astype(np.float32)
date_index = df.index

seq_len = 90
pred_len = 30  # 30日先の予測

# ★ 30点ahead対応 create_sequences ★
def create_seq2seq(data, target, close_series, date_index, seq_len, pred_len):
    xs, ys, cs, ds = [], [], [], []
    for i in range(len(data) - seq_len - pred_len):
        xs.append(data[i:i+seq_len])
        ys.append(target[i+seq_len:i+seq_len+pred_len])
        cs.append(close_series[i+seq_len:i+seq_len+pred_len])
        ds.append(date_index[i+seq_len:i+seq_len+pred_len])
    return np.array(xs), np.array(ys), np.array(cs), np.array(ds)

x_seq, y_seq, c_seq, d_seq = create_seq2seq(data, target, close_series, date_index, seq_len, pred_len)

train_size = int(len(x_seq) * 0.8)
x_train, y_train, c_train, d_train = x_seq[:train_size], y_seq[:train_size], c_seq[:train_size], d_seq[:train_size]
x_test, y_test, c_test, d_test = x_seq[train_size:], y_seq[train_size:], c_seq[train_size:], d_seq[train_size:]

train_dataset = TensorDataset(torch.tensor(x_train), torch.tensor(y_train))
test_dataset = TensorDataset(torch.tensor(x_test), torch.tensor(y_test))

# -----------------------------
# 3. seq2seq Transformer
# -----------------------------
class Seq2SeqTransformer(nn.Module):
    def __init__(self, input_dim, d_model=64, nhead=4, num_layers=2, pred_len=30):
        super().__init__()
        self.input_linear = nn.Linear(input_dim, d_model)
        encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead)
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.output_linear = nn.Linear(d_model, pred_len)

    def forward(self, x):
        x = self.input_linear(x)
        x = self.transformer(x.permute(1, 0, 2))
        x = x.mean(dim=0)
        return self.output_linear(x)

model = Seq2SeqTransformer(input_dim=len(features), pred_len=pred_len)

# -----------------------------
# 4. 学習
# -----------------------------
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

for epoch in range(30):
    model.train()
    losses = []
    for batch_x, batch_y in train_loader:
        optimizer.zero_grad()
        output = model(batch_x)
        loss = criterion(output, batch_y)
        loss.backward()
        optimizer.step()
        losses.append(loss.item())
    print(f"Epoch {epoch+1}, Loss: {np.mean(losses):.6f}")

# -----------------------------
# 5. 予測 & 復元 & グラフ（過去の株価と未来の予測をつなげる）
# -----------------------------
model.eval()
pred_logret_seq = model(torch.tensor(x_test)).detach().numpy()

# 予測結果の復元
c_test = np.array(c_test)
if c_test.ndim == 3:
    c_test = c_test.squeeze(-1)

predicted_close_seq = c_test * np.exp(np.cumsum(pred_logret_seq, axis=1))
actual_close_seq = c_test * np.exp(np.cumsum(y_test, axis=1))

# 1. 予測と実測の誤差を示す
sample = -1  # 例として最後のサンプルを使用
plt.figure(figsize=(12, 5))

# 実際の株価と予測された株価のプロット
plt.plot(d_test[sample], actual_close_seq[sample], label="Actual Close", color="blue")
plt.plot(d_test[sample], predicted_close_seq[sample], label="Predicted Close", color="orange")

# 予測と実測の誤差を計算して表示
error = actual_close_seq[sample] - predicted_close_seq[sample]
plt.plot(d_test[sample], error, label="Error (Actual - Predicted)", color="red", linestyle="--")

plt.legend()
plt.title("Actual vs Predicted Close Price (Error Plot)")
plt.xlabel("Date")
plt.ylabel("Price")
plt.gcf().autofmt_xdate()
plt.show()

# 2. 過去の株価に予測値をつなげてグラフ化
# 実際の株価（過去の）をプロット
plt.figure(figsize=(12, 5))
plt.plot(df.index, df['Close'], label="Actual Close", color="blue")

# 予測された株価（未来）をつなげてプロット
plt.plot(d_test[sample], predicted_close_seq[sample], label="Predicted Close (Future)", color="orange")

plt.legend()
plt.title("Past Stock Prices and Predicted Future Close Price")
plt.xlabel("Date")
plt.ylabel("Close Price")
plt.gcf().autofmt_xdate()
plt.show()
# -----------------------------
# 6. 30日後の予測結果（30日後を予測）
# -----------------------------
latest_data  = df[features].tail(seq_len).values.astype(np.float32)   # (90, 特徴量数)
latest_close = float(df["Close"].iloc[-1])                            # スカラー化しておく

# 30日後のログリターンを推論
future_pred_logret = model(torch.tensor(latest_data).unsqueeze(0)).detach().numpy()
predicted_future_logret = np.cumsum(future_pred_logret, axis=1)[0]   # shape = (30,)

# 株価に復元（30個のベクトル）
predicted_future_close = latest_close * np.exp(predicted_future_logret)  # shape = (30,)

# プロット
plt.figure(figsize=(12, 5))
plt.plot(range(1, 31), predicted_future_close,
         label="Predicted 30-day Future Close", color="purple")
plt.title("2897.T (Nissin Foods Holdings Co.,Ltd.)-Predicted Close Price for the Next 30 Days", fontname='Meiryo')
plt.xlabel("Days Ahead")
plt.ylabel("Predicted Close Price")
plt.legend()
plt.show()