# 機器學習期末報告 AI股票
## 介紹
此為訓練單一股票之預測模型
由於教室的網路速度有點慢(比烏龜慢)，因此本次的報告使用==google colab==進行
>[!Important]
>**請注意，以下所獲取之預測資料僅供參考，本組只進行技術教授與說明，不負任何投資損失責任**
## 參照程式碼
https://github.com/lex-hue/Stock-Predictor-V4
https://hackmd.io/@s02260441/Hki9NN5jL
## 預測Model的建置順序
### install.py
本次會用到的套件我們這次透過function來直接將其包含起來，也能知道我們在這過程中下載了哪些的套件與工具。
```python=
def install_dependencies():
packages = [
"pandas",
"ta",
"yfinance",
"numpy",
"scikit-learn",
"matplotlib",
"tensorflow",
"statsmodels",
"pdfplumber"
"mplfinance"
]
total_packages = len(packages)
progress = 0
for package in packages:
progress += 1
print(f"Installing {package}... ({progress}/{total_packages})")
!pip install {package}
install_dependencies()
```
### 1-1.取得台灣股市資訊
本報告在網路上取得一份PDF檔，裡面包含台灣大部分的股市的名稱與代號，本程式將會去爬該PDF的內容，將檔案內所有的股市資訊給整理成一份字典，並且製作成下拉選單，方便選取想要的股市資訊。
PDF 下載連結：
https://drive.google.com/file/d/1G54CplA-WuDdJttQ6W5UuU_JjwR2myA8/view?usp=sharing
```python=
from IPython.display import display
import ipywidgets as widgets
import pdfplumber
options_mapping = {}
pdf = pdfplumber.open('SE1026.pdf')
for i, page in enumerate(pdf.pages):
tables = page.extract_table()
if tables:
if tables[0][0] == "股票代號" and tables[0][1] == "股票名稱":
tables = tables[1:]
for table in tables:
for index in range(0, len(table), 2):
options_mapping[table[index+1]] = table[index]
pdf.close()
stock_name = None
stock_symbol = None
# 下拉選單
combobox = widgets.Combobox(
options=list(options_mapping.keys()), # 使用顯示值作為選項
description='請選擇:'
)
# 創建處理函數
def handle_combobox_change(change):
global stock_name, stock_symbol
stock_name = change.new
stock_symbol = options_mapping.get(stock_name)
# 將處理函數綁定到下拉選單的變化事件
combobox.observe(handle_combobox_change, names='value')
# 顯示下拉選單
display(combobox)
```
> stock_name 和 stock_symbol是全域變數
> 只要在下拉方塊選擇了新的值，那麼這兩個變數就會及時更新
> 後面的程式若需要輸入股票代號就會採用這些資訊。
>
### 1-2.檢查下拉選單的選擇值
```python=
if stock_name:
print("你選擇了:", stock_name)
print("股票代號為:", stock_symbol)
else:
print("尚未選擇.")
```
### 2.抓取所選股市的資訊
根據下拉選單所選的股票代號以及所輸入的日期範圍，以下程式碼會抓該股票代號的相關資訊
匯出成csv，並且整理、運算出其他常見的技術分析指標
```python=
import os
import pandas as pd
import numpy as np
import ta
import matplotlib.pyplot as plt
import yfinance as yf
from statsmodels.tsa.seasonal import seasonal_decompose
def download_and_prepare_data(ticker_symbol,start_date, end_date):
data_dir = 'data'
if not os.path.exists(data_dir):
os.makedirs(data_dir)
data = yf.download(ticker_symbol, period="max", interval="1d", start=start_date, end=end_date)
df = pd.DataFrame(data)
data_file = f"./data/{ticker_symbol}.csv"
df.to_csv(data_file)
print("資料已下載並儲存至", data_file)
def preprocess_data(ticker_symbol):
data_file = f"./data/{ticker_symbol}.csv"
df = pd.read_csv(data_file)
# 使用 ta 庫計算技術指標(均線、相對強弱指標、指數平滑異同移動平均線)
df["SMA"] = ta.trend.sma_indicator(df["Close"], window=14)
df["RSI"] = ta.momentum.rsi(df["Close"], window=14)
df["MACD"] = ta.trend.macd_diff(df["Close"], window_slow=26, window_fast=12, window_sign=9)
df_bollinger = ta.volatility.BollingerBands(df["Close"], window=20)
df["upper_band"] = df_bollinger.bollinger_hband()
df["middle_band"] = df_bollinger.bollinger_mavg()
df["lower_band"] = df_bollinger.bollinger_lband()
aroon_indicator = ta.trend.AroonIndicator(high=df["High"], low=df["Low"], window=25)
df["aroon_up"] = aroon_indicator.aroon_up()
df["aroon_down"] = aroon_indicator.aroon_down()
open_prices = df["Open"]
close_prices = df["Close"]
kicking_pattern = np.zeros_like(open_prices)
for i in range(1, len(open_prices)):
if open_prices[i] < open_prices[i-1] and \
open_prices[i] > close_prices[i-1] and \
close_prices[i] > open_prices[i-1] and \
close_prices[i] < close_prices[i-1] and \
open_prices[i] - close_prices[i] > open_prices[i-1] - close_prices[i-1]:
kicking_pattern[i] = 100 # 一些正值來表示這個模式
# 將 kicking_pattern 作為新的欄位添加到 DataFrame 中
df["kicking"] = kicking_pattern
# 計算 ATR 和 SuperTrend
def calculate_atr(high, low, close, window=14):
true_ranges = np.maximum.reduce([high - low, np.abs(high - close.shift()), np.abs(low - close.shift())])
atr = np.zeros_like(high)
atr[window - 1] = np.mean(true_ranges[:window])
for i in range(window, len(high)):
atr[i] = (atr[i - 1] * (window - 1) + true_ranges[i]) / window
return atr
df["ATR"] = calculate_atr(df["High"], df["Low"], df["Close"], window=14)
# 使用減少的敏感度和所有指標計算 Supertrend 信號
df["upper_band_supertrend"] = df["High"] - (df["ATR"])
df["lower_band_supertrend"] = df["Low"] + (df["ATR"])
# 根據對指標的敏感度定義上升趨勢和下降趨勢條件
uptrend_conditions = [
(df["Close"] > df["lower_band_supertrend"]),
(df["Close"] > df["SMA"]),
(df["Close"] > df["middle_band"]),
(df["Close"] > df["MACD"]),
(df["RSI"] > 50),
(df["aroon_up"] > df["aroon_down"]),
(df["kicking"] == 1), # 假設 "kicking" 是指標，其中 1 表示上升趨勢。
(df["Close"] > df["upper_band_supertrend"])
]
downtrend_conditions = [
(df["Close"] < df["upper_band_supertrend"]),
(df["Close"] < df["SMA"]),
(df["Close"] < df["middle_band"]),
(df["Close"] < df["MACD"]),
(df["RSI"] < 50),
(df["aroon_up"] < df["aroon_down"]),
(df["kicking"] == -1), # 假設 "kicking" 是指標，其中 -1 表示下降趨勢。
(df["Close"] < df["lower_band_supertrend"])
]
# 將初始信號值設置為 0
df["supertrend_signal"] = 0
# 根據對指標的敏感度更新信號
df.loc[np.any(uptrend_conditions, axis=0), "supertrend_signal"] = 1
df.loc[np.any(downtrend_conditions, axis=0), "supertrend_signal"] = -1
# 分解時間序列數據
result = seasonal_decompose(df["Close"], model="additive", period=365)
# 將分解的組件添加到 DataFrame 中
df["trend"] = result.trend
df["seasonal"] = result.seasonal
df["residual"] = result.resid
# 按您希望的順序串聯欄位
df2 = pd.concat(
[
df["Date"],
df["Close"],
df["Open"],
df["Adj Close"],
df["Volume"],
df["High"],
df["Low"],
df["SMA"],
df["MACD"],
df["upper_band"],
df["middle_band"],
df["lower_band"],
df["supertrend_signal"],
df["RSI"],
df["aroon_up"],
df["aroon_down"],
df["kicking"],
df["upper_band_supertrend"],
df["lower_band_supertrend"],
df["trend"],
df["seasonal"],
df["residual"]
],
axis=1,
)
# 用 0 填充缺失值
df2.fillna(0, inplace=True)
# 將 DataFrame 保存到新 CSV 文件，包含指標和分解的組件
df2.to_csv("data.csv", index=False)
# 去除相同方向的連續信號（減少敏感度）
signal_changes = df["supertrend_signal"].diff().fillna(0)
consecutive_mask = (signal_changes == 0) & (signal_changes.shift(-1) == 0)
df.loc[consecutive_mask, "supertrend_signal"] = 0
# 繪製數據
fig, (ax1, ax2, ax3, ax4, ax5) = plt.subplots(5, 1, figsize=(12, 8), sharex=True)
ax1.plot(df["Open"], label="Open")
ax1.plot(df["Close"], label="Close")
ax1.plot(df["trend"], label="Trend")
ax1.plot(df["SMA"], label="SMA")
ax1.fill_between(
df.index, df["upper_band"], df["lower_band"], alpha=0.2, color="gray"
)
ax1.plot(df["upper_band"], linestyle="dashed", color="gray")
ax1.plot(df["middle_band"], linestyle="dashed", color="gray")
ax1.plot(df["lower_band"], linestyle="dashed", color="gray")
ax1.scatter(
df.index[df["supertrend_signal"] == 1],
df["Close"][df["supertrend_signal"] == 1],
marker="^",
color="green",
s=100,
)
ax1.scatter(
df.index[df["supertrend_signal"] == -1],
df["Close"][df["supertrend_signal"] == -1],
marker="v",
color="red",
s=100,
)
ax1.legend()
ax2.plot(df["aroon_up"], label="Aroon Up")
ax2.plot(df["aroon_down"], label="Aroon Down")
ax2.legend()
ax3.plot(df["RSI"], label="RSI")
ax3.legend()
ax4.plot(df["seasonal"], label="Seasonal")
ax4.legend()
ax5.plot(df["residual"], label="Residual")
ax5.legend()
plt.xlim(df.index[0], df.index[-1])
plt.show()
if __name__ == "__main__":
ticker_symbol = stock_symbol + ".TW"
tic = yf.Ticker(ticker_symbol)
info = tic.get_info()
start_date = input("請輸入開始日期(yyyy-mm-dd)：")
end_date = input("請輸入結束日期(yyyy-mm-dd)：")
print(f"{stock_name}({ticker_symbol}) 的信息:\n")
print(info["shortName"],"\n")
indicator = "未知"
if "recommendationKey" in info:
if info["recommendationKey"] == "buy":
indicator = "買入"
elif info["recommendationKey"] == "sell":
indicator = "賣出"
elif info["recommendationKey"] == "strong buy":
indicator = "強烈買入"
elif info["recommendationKey"] == "hold":
indicator = "持有"
elif info["recommendationKey"] == "underperform":
indicator = "表現不佳"
print(f"推薦趨勢為 {indicator}")
download = input("您是否要下載此股票代碼(y/n)？：").lower()
if download == "y":
download_and_prepare_data(ticker_symbol, start_date, end_date)
preprocess_data(ticker_symbol)
else:
print("退出腳本..")
```
#### 函數 download_and_prepare_data
* 參數
ticker_symbol: 股票代碼
start_date: 開始日期
end_date: 結束日期
* 流程
檢查資料夾data是否存在，不存在則創建。
使用yfinance庫下載股票數據。
將數據保存到本地CSV文件中。
#### 函數 preprocess_data
這個函數從CSV文件讀取數據，計算各種技術指標，並將結果保存到新的CSV文件中。
* 流程：
先從CSV文件中讀取數據。
1. 使用ta庫計算技術指標，包括均線（SMA）、相對強弱指標（RSI）、指數平滑異同移動平均線（MACD）、布林帶（Bollinger Bands）和Aroon指標。
2. 定義和計算“踢腿形態”（Kicking Pattern）。
3. 計算平均真實波幅（ATR）和SuperTrend指標
4. 分解時間序列數據成趨勢、季節性和殘差。
5. 將所有計算結果和原始數據合併，保存到新的CSV文件中。
6. 繪製數據圖表，包括開盤價、收盤價、SMA、布林帶、SuperTrend信號、Aroon指標、RSI、季節性成分和殘差成分。
* 簡單移動平均線（Simple Moving Average, SMA）
SMA 是一種常見的技術指標，用於平滑價格數據，消除短期波動，從而更容易識別價格趨勢。計算方法是將一定時間範圍內的收盤價格平均化。
* 布林帶（Bollinger Bands）
布林帶由三條線組成：中軌線（SMA）、上軌線和下軌線。它們用於衡量市場的波動性和相對價格水平。
* SuperTrend 信號
SuperTrend 指標基於平均真實波幅（ATR）計算，用於識別趨勢方向。SuperTrend 信號分為上升趨勢和下降趨勢。
* Aroon 指標
Aroon 指標用於測量趨勢的強度和變化，包含 Aroon Up 和 Aroon Down。
* 相對強弱指標（Relative Strength Index, RSI）
RSI 是一種動量指標，用於測量價格變動的速度和變動幅度。RSI 的值介於0和100之間。
* 季節性成分（Seasonal Component）
季節性成分來自於時間序列分解，用於揭示數據中的週期性變動。
* 殘差成分（Residual Component）
殘差成分是時間序列分解後，扣除趨勢和季節性成分後剩餘的部分，代表數據中的隨機波動。
#### 主程序
主程序首先要求用戶輸入股票代碼、開始日期和結束日期。
流程：
1. 使用yfinance庫獲取股票信息。
2. 顯示股票簡稱和推薦趨勢（例如，買入、賣出、持有等）。
3. 詢問用戶是否下載股票數據。
4. 如果用戶選擇下載，則調用download_and_prepare_data函數下載數據，然後調用preprocess_data函數進行數據處理和可視化。
### 3.訓練model
根據(2)所獲取的csv檔，開始訓練模型
```python=
import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import BatchNormalization, PReLU, LSTM, Dense
from sklearn.metrics import r2_score, mean_absolute_percentage_error
from sklearn.preprocessing import MaxAbsScaler
print("TensorFlow 版本:", tf.__version__)
# 定義一個函數來加載數據 (將 'data.csv' 替換為你的數據文件)
def load_data(file_path: str) -> pd.DataFrame:
data = pd.read_csv(file_path)
return data[['Close']] # 只加載 'Close' 列
# 定義獎勵函數
def get_reward(y_true, y_pred):
mape = mean_absolute_percentage_error(y_true, y_pred)
r2 = r2_score(y_true, y_pred)
reward = (((1 - mape) * 0.1) + ((r2) * 1.9)) / 2
return reward
# 定義一個函數來創建 LSTM 模型
def create_LSTM_model() -> Sequential:
model = Sequential()
model.add(LSTM(units=150, return_sequences=True))
model.add(PReLU())
model.add(PReLU())
model.add(PReLU())
model.add(LSTM(units=150))
model.add(PReLU())
model.add(PReLU())
model.add(PReLU())
# 添加最終輸出層
model.add(Dense(units=1, activation='linear'))
return model
# 加載數據
data = load_data("data.csv")
# 將數據分為訓練集和測試集
train_data = data.iloc[:int(0.8*len(data))]
test_data = data.iloc[int(0.8*len(data)):]
# 標準化數據 (只使用 'Close' 列)
scaler = MaxAbsScaler()
train_data_norm = scaler.fit_transform(train_data)
test_data_norm = scaler.transform(test_data)
# 定義時間步長
timesteps = 100
# 創建時間步長的序列 (只使用 'Close' 值)
def create_sequences(data, timesteps):
X = []
y = []
for i in range(timesteps, len(data)):
X.append(data[i-timesteps:i])
y.append(data[i, 0])
return np.array(X), np.array(y)
X_train, y_train = create_sequences(train_data_norm, timesteps)
X_test, y_test = create_sequences(test_data_norm, timesteps)
# 構建並編譯 LSTM 模型
model = create_LSTM_model()
# 編譯模型
model.compile(optimizer='adam', loss="huber")
# 訓練
epochs = 25
batch_size = 32
for i in range(epochs):
print(f"第 {i+1} 個訓練周期 / 共 {epochs} 個訓練周期")
history = model.fit(X_train, y_train, batch_size=batch_size, validation_data=(X_test, y_test), epochs=1)
# 在測試集上評估模型
y_pred_test = model.predict(X_test)
test_reward = get_reward(y_test, y_pred_test)
print("測試集獎勵:", test_reward)
if i == 0:
best_reward = test_reward
if test_reward >= best_reward:
best_reward = test_reward
print("模型已保存!")
model.save("model.keras")
# 加載最佳模型並評估
model = tf.keras.models.load_model("model.keras")
y_pred_test = model.predict(X_test)
test_reward = get_reward(y_test, y_pred_test)
test_loss = model.evaluate(X_test, y_test)
print("最終測試集獎勵:", test_reward)
print("最終測試集損失:", test_loss)
```
這段程式碼通過LSTM神經網絡模型，對股票收盤價進行時間序列預測，並使用獎勵函數來評估模型的預測性能。
### 4.評估模型的可用性
利用剛剛整理出的指標歷史數據來評估我們模型的可用性
```python=
import os
import sys
import pandas as pd
import numpy as np
import tensorflow as tf
from sklearn.preprocessing import MaxAbsScaler
from tensorflow.keras.models import load_model
from sklearn.metrics import (
mean_squared_error,
r2_score,
mean_absolute_percentage_error,
)
import matplotlib.pyplot as plt
print("TensorFlow 版本:", tf.__version__)
# 定義獎勵函數
def get_reward(y_true, y_pred):
mape = mean_absolute_percentage_error(y_true, y_pred)
r2 = r2_score(y_true, y_pred)
reward = (((1 - mape) * 0.1) + ((r2) * 1.9)) / 2
return reward
# 加載數據
data = pd.read_csv("data.csv")
# 將數據分為訓練集和測試集
test_data = data.iloc[int(0.8 * len(data)) :]
# 標準化數據
scaler = MaxAbsScaler()
test_data_norm = scaler.fit_transform(
test_data[
[
"Close"
]
]
)
# 定義時間步長
timesteps = 100
# 創建時間步長的序列
def create_sequences(data, timesteps):
X = []
y = []
for i in range(timesteps, len(data)):
X.append(data[i - timesteps : i])
y.append(data[i, 0])
return np.array(X), np.array(y)
X_test, y_test = create_sequences(test_data_norm, timesteps)
# 加載預訓練模型
model = load_model("model.keras")
print("\n評估模型")
# 評估模型
y_pred = model.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
mape = mean_absolute_percentage_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)
# 計算獎勵
reward = get_reward(y_test, y_pred)
# 打印評估指標
# MAPE 衡量的是預測值與實際值之間的百分比誤差，這個指標的值越低，表示預測結果越精確。
print("MAPE:", mape)
# R² 衡量的是模型解釋數據變異的程度，R² 的值範圍在 0 到 1 之間，值越接近 1，表示模型的解釋能力越強。
print("MSE:", mse)
# 獎勵的計算是將 (1 - MAPE) 與 R² 進行加權平均這裡將 (1 - MAPE) 和 R² 各自乘以不同的權重，然後取平均。選擇這些權重是為了更好地平衡兩個指標在獎勵值中的影響。
print("R2:", r2)
print("Reward:", reward)
# 繪製預測值與實際值的比較圖
plt.plot(y_test, label="Actual")
plt.plot(y_pred, label="Predicted")
plt.legend()
plt.show()
```
通過標準化、創建序列、加載模型和評估模型的方式，對股票數據進行了有效的預測和評估。可視化部分有助於直觀地理解模型的預測效果。
Q:使用歷史數據來驗證模型的準確性？
* 避免過擬合：將數據集劃分為訓練集和測試集，確保模型在訓練過程中不會看到測試數據，這樣可以避免過擬合，即模型在訓練數據上表現良好但在未知數據上表現不佳。
* 模型評估：使用測試集來評估模型，可以幫助我們了解模型在未見過的數據上的表現，即模型的泛化能力
### 5.微調模型
設定一個你期望達到的模型品質的值(0~1) 1大概率是不可能達到，建議0.9即可
加載預訓練的LSTM模型，對其進行評估並根據用戶定義的獎勵閾值（reward threshold）進行微調。程式碼結合了模型評估、重新訓練和保存最佳模型的過程，並提供了處理中斷信號的功能。
```python=
import os
import signal
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
import pandas as pd
import numpy as np
import tensorflow as tf
from sklearn.preprocessing import MaxAbsScaler
from tensorflow.keras.models import load_model
from sklearn.metrics import (
mean_squared_error,
r2_score,
mean_absolute_percentage_error,
)
import matplotlib.pyplot as plt
from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping
# 定義獎勵函數
def get_reward(y_true, y_pred):
mape = mean_absolute_percentage_error(y_true, y_pred)
r2 = r2_score(y_true, y_pred)
reward = (((1 - mape) * 0.1) + ((r2) * 1.9)) / 2
return reward
# 加載數據
data = pd.read_csv("data.csv")
# 分割數據為訓練集和測試集
train_data = data.iloc[: int(0.8 * len(data))]
test_data = data.iloc[int(0.8 * len(data)) :]
# 正規化數據
scaler = MaxAbsScaler()
train_data_norm = scaler.fit_transform(
train_data[
[
"Close"
]
]
)
test_data_norm = scaler.fit_transform(
test_data[
[
"Close"
]
]
)
# 定義時間步長
timesteps = 100
# 創建時間步長的序列
def create_sequences(data, timesteps):
X = []
y = []
for i in range(timesteps, len(data)):
X.append(data[i - timesteps : i])
y.append(data[i, 0])
return np.array(X), np.array(y)
X_train, y_train = create_sequences(train_data_norm, timesteps)
X_test, y_test = create_sequences(test_data_norm, timesteps)
# 定義獎勵閾值
reward_threshold = float(
input("輸入獎勵閾值 (0 - 1, 建議 0.9): ")
)
# 初始化獎勵和其他指標
rewards = []
mses = []
mapes = []
r2s = []
count = 0
# 處理 SIGINT 信號 (CTRL + C)
def handle_interrupt(signal, frame):
print("\n收到中斷信號.")
# 詢問用戶是否確認結束程序
user_input = input(
f"你確定要結束程序嗎？ (yes/no): "
)
if user_input.lower() == "yes":
exit(0)
else:
print("繼續進行微調過程")
# 註冊信號處理器
signal.signal(signal.SIGINT, handle_interrupt)
while True:
# 加載模型
model = load_model("model.keras")
print("\n評估模型")
# 評估模型
y_pred = model.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
mape = mean_absolute_percentage_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)
# 添加獎勵
reward = get_reward(y_test, y_pred)
best_reward1 = reward
rewards.append(reward)
mses.append(mse)
mapes.append(mape)
r2s.append(r2)
# 打印當前獎勵
print("獎勵:", rewards)
print("MAPE:", mape)
print("MSE:", mse)
print("R2:", r2)
count += 1
print("迴圈次數:", count)
# 檢查是否達到獎勵閾值
if len(rewards) >= 1 and sum(rewards[-1:]) >= reward_threshold:
print("達到獎勵閾值!")
model.save("model.keras")
break
else:
epochs = 10
print(f"訓練模型 {epochs} 次")
batch_size = 32
for i in range(epochs):
print(f"第 {i+1} / {epochs} 輪")
history = model.fit(X_train, y_train, batch_size=batch_size, validation_data=(X_test, y_test), epochs=1)
# 在測試集上評估模型
y_pred_test = model.predict(X_test)
test_reward = get_reward(y_test, y_pred_test)
print("測試獎勵:", test_reward)
if test_reward >= best_reward1:
print("模型已保存!")
best_reward1 = test_reward
model.save("model.keras")
if test_reward >= reward_threshold:
print("模型已達到獎勵閾值", test_reward, "。保存並停止訓練!")
model.save("model.keras")
break
```
這個函式實際上是一種優化算法，被稱為 "強化學習"。這種方法的核心思想是通過不斷地試驗和調整，以找到最優解決方案。
這種方法的有效性在於以下幾點：
>[!Tip]搜索空間：
>儘管看起來是一種 "重複" 的行為，但實際上是在搜索一個巨大的參數空間，以找到最佳的模型結構和參數設置。
>[!Tip]反饋機制：
> 通過不斷評估模型的性能，系統可以根據反饋信息調整策略，從而逐步提高模型的性能。
### 6.主動預測
主動去預測未來指定天數內的數據，並儲存下來欄位(Date, Close)。
```python=
import numpy as np
import numpy as np
import pandas as pd
from sklearn.preprocessing import MaxAbsScaler
import tensorflow as tf
# 載入模型
model = tf.keras.models.load_model('model.keras')
# 載入和準備數據
data = pd.read_csv(f'data/{ticker_symbol}.csv')
close_prices = data['Close'].values.reshape(-1, 1)
# 歸一化
scaler = MaxAbsScaler()
scaled_data = scaler.fit_transform(close_prices)
# 選擇最後的時間視窗資料作為輸入
timesteps = 100
last_sequence = scaled_data[-timesteps:]
last_sequence = np.expand_dims(last_sequence, axis=0)
# 預測未來N天
future_days = int(input('請輸入您想要預測的天數(建議可以設置30天)：'))
predictions = []
for _ in range(future_days):
# 獲取最新預測
current_pred = model.predict(last_sequence)
# 調整 current_pred 形狀以匹配 last_sequence
current_pred_reshaped = np.expand_dims(current_pred, axis=-1) # 使其成為 (1, 1, 1)
# 更新輸入資料
last_sequence = np.append(last_sequence[:, 1:, :], current_pred_reshaped, axis=1)
# 保存預測結果以反歸一化
predictions.append(current_pred_reshaped.flatten()[0])
# 反歸一化預測結果
predictions = np.array(predictions).reshape(-1, 1)
predicted_prices = scaler.inverse_transform(predictions)
predicted_prices = predicted_prices.flatten() # 展平陣列
# 列印預測結果
print("預測的未來 {} 天的價格是:".format(future_days))
print(predicted_prices)
# 原始收盤價
original_prices = data['Close'].values
# 生成日期索引
dates = pd.date_range(start=data['Date'].iloc[-1], periods=future_days)
predictions_df = pd.DataFrame({
"Date": dates,
"Predicted Close": predicted_prices
})
predictions_df.to_csv('predictions.csv', index=False)
```
通過設置獎勵閾值和進行多次迴圈訓練，程式碼確保了模型在不斷改進的同時，及時保存最佳模型。信號處理部分則提供了安全退出和繼續訓練的選擇。
### 7.比較
等到未來的指定天數後，使用剛剛預測出來的做比對
```python=
import os
import pandas as pd
import matplotlib.pyplot as plt
# 獲取"data"文件夾中的CSV文件列表
data_folder = "data"
csv_files = [file for file in os.listdir(data_folder) if file.endswith(".csv")]
# 顯示可用的CSV文件
print("可用的CSV文件:")
for i, file in enumerate(csv_files):
print(f"{i + 1}. {file}")
# 用戶選擇一個CSV文件
selected_file = None
while selected_file is None:
try:
file_number = int(input("請輸入您要選擇的CSV文件相應的編號: "))
if file_number < 1 or file_number > len(csv_files):
raise ValueError()
selected_file = csv_files[file_number - 1]
except ValueError:
print("無效的輸入。請輸入有效的編號。")
# 加載預測數據和實際數據
predicted_data = pd.read_csv("predictions.csv")
actual_data = pd.read_csv(os.path.join(data_folder, selected_file))
# 確保日期列是日期格式
predicted_data["Date"] = pd.to_datetime(predicted_data["Date"])
actual_data["Date"] = pd.to_datetime(actual_data["Date"])
# 如果您的實際數據包含未來的日期，這個步驟是必需的
actual_data = actual_data[actual_data["Date"] >= predicted_data["Date"].min()]
# 重命名列以增加清晰度
predicted_data.rename(columns={"Predicted Close": "Predict Close"}, inplace=True)
# 在日期列上結合預測數據和實際數據
combined_data = pd.merge(predicted_data, actual_data[['Date', 'Close']], on="Date", how='left')
combined_data.rename(columns={"Close": "Actual Close"}, inplace=True)
# 計算預測值和實際值之間的絕對百分比誤差
combined_data["Absolute % Error"] = (
abs(combined_data["Predict Close"] - combined_data["Actual Close"]) / combined_data["Actual Close"] * 100
)
# 計算平均絕對百分比誤差並打印它
mape = combined_data["Absolute % Error"].mean()
print(f"平均絕對百分比誤差: {mape:.2f}%")
# 找到絕對百分比誤差最大和最小的行並打印它們
min_error_row = combined_data.loc[combined_data["Absolute % Error"].idxmin()]
max_error_row = combined_data.loc[combined_data["Absolute % Error"].idxmax()]
print(f"\n最準確的預測:\n{min_error_row}\n")
print(f"最不準確的預測:\n{max_error_row}\n")
# 繪製預測和實際收盤價
plt.figure(figsize=(10, 6))
plt.plot(combined_data["Date"], combined_data["Predict Close"], label="Stock price forecasting")
plt.plot(combined_data["Date"], combined_data["Actual Close"], label="closing price")
plt.xlabel("date")
plt.ylabel("price")
plt.title("Compare")
plt.legend()
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
```
通過用戶選擇CSV文件，然後計算預測值與實際值之間的誤差，並將這些結果可視化展示出來，從而幫助用戶理解模型的準確性和改進空間。
### 8.利用抓取之資料繪製熟悉的 K棒圖
這個主要是可以利用最常規的手法，利用爬蟲去獲取股票資訊，之後繪製出k棒圖。
可以選擇我們先前所抓取的csv檔案，並且可以設定時間區間，使我們更能自由查看股票的波動。
```python=
import pandas as pd
import os
import mplfinance as mpf
from datetime import datetime
# 獲取"data"文件夾中的CSV文件列表
data_folder = "data"
csv_files = [file for file in os.listdir(data_folder) if file.endswith(".csv")]
# 顯示可用的CSV文件
print("可用的CSV文件:")
for i, file in enumerate(csv_files):
print(f"{i + 1}. {file}")
# 用戶選擇一個CSV文件
selected_file = None
while selected_file is None:
try:
file_number = int(input("請輸入您要選擇的CSV文件相應的編號: "))
if file_number < 1 or file_number > len(csv_files):
raise ValueError()
selected_file = csv_files[file_number - 1]
except ValueError:
print("無效的輸入。請輸入有效的編號。")
# 讀取CSV文件並設置日期列為DatetimeIndex
df = pd.read_csv(os.path.join(data_folder, selected_file), parse_dates=['Date'], index_col='Date')
# 讓使用者選擇分析區間
start_date = input("請輸入開始日期 (YYYY-MM-DD): ")
end_date = input("請輸入結束日期 (YYYY-MM-DD): ")
try:
start_date = datetime.strptime(start_date, '%Y-%m-%d')
end_date = datetime.strptime(end_date, '%Y-%m-%d')
df = df.loc[start_date:end_date]
except ValueError:
print("日期格式錯誤,請使用 YYYY-MM-DD 格式。")
exit()
mc = mpf.make_marketcolors(up='r', down='g', inherit=True)
s = mpf.make_mpf_style(base_mpf_style='yahoo', marketcolors=mc)
kwargs = dict(type='candle', mav=(10,20,60), volume=True, figratio=(16,9), figscale=2, title=selected_file, style=s)
mpf.plot(df, **kwargs)
```
根據用戶指定的日期範圍篩選數據，然後使用mplfinance庫繪製K線圖。這種方法可以用於分析股票或其他金融數據的歷史價格走勢，並可視化其在選定區間內的表現。
Sign in with Wallet
Connect another wallet