Upload predictor_py.py

predictor_py.py

@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+"""predictor.py
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1AHdt8sCCYTJXRC_ENNsgMFv7rK_R0Oa4
+"""
+
+# predictor.py
+
+import pandas as pd
+import numpy as np
+from tensorflow.keras.models import load_model
+import joblib
+import os
+
+# --- 全域變數，定義模型和相關檔案的路徑 ---
+# 這樣可以方便地在同一個地方管理檔案名稱
+MODEL_PATH = 'stock_lstm_model_v2.keras'
+SCALER_PATH = 'data_scaler_v2.gz'
+
+# 這個列表是關鍵：它定義了 v2 模型訓練時所使用的所有特徵
+# 順序也必須和訓練時完全一樣
+FEATURES_USED = [
+    'open', 'high', 'low', 'close', 'volume',
+    'MACD', 'KD_K', 'KD_D', 'RSI',
+    'DMI_ADX',
+    'Index'
+]
+TARGET_FEATURE = 'close' # 預測的目標
+
+# --- 內部輔助函式，用於載入元件 ---
+def _load_predictor_components():
+    """
+    載入模型和 Scaler。這是一個內部函式，外部不應該直接呼叫。
+    """
+    # 檢查檔案是否存在
+    if not os.path.exists(MODEL_PATH) or not os.path.exists(SCALER_PATH):
+        print(f"錯誤：找不到必要的檔案。請確保 '{MODEL_PATH}' 和 '{SCALER_PATH}' 存在於同一個目錄下。")
+        return None, None
+
+    try:
+        model = load_model(MODEL_PATH)
+        scaler = joblib.load(SCALER_PATH)
+        print("模型和 Scaler 載入成功！")
+        return model, scaler
+    except Exception as e:
+        print(f"載入預測器組件時發生錯誤: {e}")
+        return None, None
+
+# --- 這是您唯一需要呼叫的主函式 ---
+def predict_next_day_price(df_stock_original: pd.DataFrame, df_climate_original: pd.DataFrame) -> (float | None):
+    """
+    接收原始的台積電股價 DataFrame 和景氣燈號 DataFrame，
+    執行所有必要的預處理步驟，並回傳下一日的預測收盤價。
+
+    Args:
+        df_stock_original (pd.DataFrame): 包含 'Date' 和其他技術指標的原始股價資料。
+        df_climate_original (pd.DataFrame): 包含 'Date' 和 'Index' 的原始景氣燈號資料。
+
+    Returns:
+        float: 預測的收盤價。如果發生錯誤則回傳 None。
+    """
+    model, scaler = _load_predictor_components()
+    if model is None or scaler is None:
+        return None
+
+    # --- 1. 資料整合與預處理 (將模型訓練時的邏輯完全複製於此) ---
+    try:
+        # 複製一份資料以避免修改到原始傳入的 DataFrame
+        df_stock = df_stock_original.copy()
+        df_climate = df_climate_original.copy()
+
+        df_stock['Date'] = pd.to_datetime(df_stock['Date'])
+        df_stock.set_index('Date', inplace=True)
+        df_climate['Date'] = pd.to_datetime(df_climate['Date'], format='%Y-%m')
+        df_climate.set_index('Date', inplace=True)
+
+        df_merged = pd.merge(df_stock, df_climate, left_index=True, right_index=True, how='left')
+        df_merged['Index'].fillna(method='ffill', inplace=True)
+
+        # 確保所有需要的特徵都存在
+        if not all(feature in df_merged.columns for feature in FEATURES_USED):
+            missing = [f for f in FEATURES_USED if f not in df_merged.columns]
+            print(f"錯誤：輸入的資料中缺少模型所需特徵: {missing}")
+            return None
+
+        df_featured = df_merged[FEATURES_USED].copy()
+        df_featured.dropna(inplace=True)
+
+        # 檢查是否有足夠的資料 (至少需要 look_back 筆)
+        look_back = 60
+        if len(df_featured) < look_back:
+            print(f"錯誤：資料不足。模型需要至少 {look_back} 筆完整的歷史資料來進行預測，但只得到 {len(df_featured)} 筆。")
+            return None
+
+        # --- 2. 準備輸入模型的資料 ---
+        # 取得最新的 60 筆資料
+        last_60_days = df_featured[-look_back:].values
+
+        # 使用【已載入的】scaler 來轉換資料
+        scaled_last_60_days = scaler.transform(last_60_days)
+
+        # 重塑為模型期望的形狀: [1, 60, 特徵數量]
+        X_predict = np.reshape(scaled_last_60_days, (1, look_back, len(FEATURES_USED)))
+
+        # --- 3. 進行預測與反標準化 ---
+        predicted_price_scaled = model.predict(X_predict)
+
+        # 建立一個和 scaler 相同結構的 dummy 陣列來進行反標準化
+        dummy_array = np.zeros(shape=(1, len(FEATURES_USED)))
+
+        # 找到 'close' 在特徵列表中的位置
+        target_col_index = FEATURES_USED.index(TARGET_FEATURE)
+        dummy_array[0, target_col_index] = predicted_price_scaled[0, 0]
+
+        # 反標準化，取得真實的預測股價
+        predicted_price = scaler.inverse_transform(dummy_array)[0, target_col_index]
+
+        return float(predicted_price)
+
+    except Exception as e:
+        print(f"預測過程中發生未預期的錯誤: {e}")
+        return None