Delete predictor_py.py

predictor_py.py

@@ -1,125 +0,0 @@
-# -*- 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