Delete model_predictor.py

model_predictor.py

@@ -1,308 +0,0 @@
-# model_predictor.py - 支援漲幅百分比輸出的XGBoost模型預測器
-# 修改版本：輸出改為漲幅百分比而非絕對價格
-
-import os
-import pandas as pd
-import numpy as np
-import xgboost as xgb
-from sklearn.preprocessing import StandardScaler
-import pickle
-import joblib
-
-class XGBoostModel:
-    """
-    用於載入和使用預先訓練好的 XGBoost 模型的類別。
-    """
-    # 使用類別變數儲存所有可用的模型名稱及其對應的檔案名稱
-    MODELS = {
-        'xgboost_model': 'xgboost_model.json'
-    }
-
-    def __init__(self, default_model='xgboost_model'):
-        """
-        初始化時自動載入預設模型。
-        """
-        self.current_model_name = default_model
-        self.model = self._load_model(self.current_model_name)
-
-    def _load_model(self, model_name):
-        """
-        從檔案載入 XGBoost 模型。
-        """
-        if model_name not in self.MODELS:
-            raise ValueError(f"找不到模型 '{model_name}'。可用的模型名稱：{list(self.MODELS.keys())}")
-                
-        filename = self.MODELS[model_name]
-        try:
-            # 建立一個新的 XGBoost 模型實例
-            model = xgb.XGBRegressor()
-            # 使用 XGBoost 內建的 load_model 方法載入檔案
-            model.load_model(filename)
-            print(f"成功載入模型檔案: {filename}")
-            return model
-        except Exception as e:
-            raise FileNotFoundError(f"無法在本地找到或載入模型檔案 '{filename}'：{e}")
-
-    def predict(self, model_name, input_df):
-        """
-        使用載入的模型進行預測。
-
-        Args:
-            model_name (str): 要使用的模型名稱。
-            input_df (pd.DataFrame): 包含特徵數據的 DataFrame，應只有一筆紀錄。
-
-        Returns:
-            dict: 包含四個預測目標的預測結果字典。
-                  {'Change_pct_t1_pred': float, 'Change_pct_t5_pred': float, ...}
-        """
-        # 如果請求的模型名稱與目前載入的不同，則動態載入
-        if model_name != self.current_model_name:
-            self.model = self._load_model(model_name)
-            self.current_model_name = model_name
-        
-        # 進行預測
-        # model.predict 會回傳一個 numpy 陣列，形狀為 (n_samples, n_targets)
-        # 在我們的案例中，n_samples=1, n_targets=4
-        predictions = self.model.predict(input_df)
-
-        # 【【核心修正】】
-        # 您的模型是多輸出模型，預測結果是一個包含4個值的陣列。
-        # 我們需要將這個陣列轉換為一個包含各預測目標的字典，以便 app.py 使用。
-        # predictions[0] 會取得第一筆樣本的所有預測值 (一個有4個元素的陣列)
-        if predictions.ndim == 2 and predictions.shape[0] > 0:
-            pred_values = predictions[0]
-        elif predictions.ndim == 1:
-            pred_values = predictions
-        else:
-             raise ValueError("預測結果的格式不符合預期。")
-        
-        result = {
-            'Change_pct_t1_pred': pred_values[0],
-            'Change_pct_t5_pred': pred_values[1],
-            'Change_pct_t10_pred': pred_values[2],
-            'Change_pct_t20_pred': pred_values[3]
-        }
-        return result
-
-    def predict_single_timeframe(self, model_name, input_df, days):
-        """
-        預測特定時間框架的漲幅
-        
-        Args:
-            model_name (str): 模型名稱
-            input_df (pd.DataFrame): 輸入特徵
-            days (int): 預測天數 (1, 5, 10, 20)
-        
-        Returns:
-            float: 預測的漲幅百分比
-        """
-        try:
-            predictions = self.predict(model_name, input_df)
-            if predictions is None:
-                return None
-            
-            # 根據天數選擇對應的預測結果
-            target_key = f'Change_pct_t{days}_pred'
-            
-            if target_key in predictions:
-                return predictions[target_key]
-            else:
-                print(f"警告：找不到 {days} 日預測結果")
-                return None
-                
-        except Exception as e:
-            print(f"單一時間框架預測時發生錯誤：{e}")
-            return None
-
-    def get_prediction_confidence(self, input_df):
-        """
-        評估預測的信心度
-        
-        Args:
-            input_df (pd.DataFrame): 輸入特徵
-            
-        Returns:
-            float: 信心度 (0-1)
-        """
-        try:
-            # 基於特徵完整性和質量評估信心度
-            feature_completeness = 0
-            total_features = len(self.feature_columns)
-            
-            for feature in self.feature_columns:
-                if feature in input_df.columns:
-                    value = input_df[feature].iloc[0]
-                    if not pd.isna(value) and value != 0:
-                        feature_completeness += 1
-            
-            completeness_ratio = feature_completeness / total_features
-            
-            # 基於數據質量調整信心度
-            base_confidence = max(0.5, completeness_ratio)
-            
-            # 如果重要特徵缺失，降低信心度
-            important_features = ['close', 'return_t-1', 'MA5_close']
-            missing_important = 0
-            for feature in important_features:
-                if feature not in input_df.columns or pd.isna(input_df[feature].iloc[0]):
-                    missing_important += 1
-            
-            if missing_important > 0:
-                base_confidence *= (1 - missing_important * 0.1)
-            
-            return min(0.9, max(0.3, base_confidence))
-            
-        except Exception as e:
-            print(f"計算信心度時發生錯誤：{e}")
-            return 0.5
-
-    def validate_input(self, input_df):
-        """
-        驗證輸入數據的有效性
-        
-        Args:
-            input_df (pd.DataFrame): 輸入特徵
-            
-        Returns:
-            tuple: (是否有效, 錯誤訊息列表)
-        """
-        errors = []
-        
-        try:
-            # 檢查是否為空
-            if input_df.empty:
-                errors.append("輸入數據為空")
-            
-            # 檢查必要特徵
-            required_features = ['close', 'return_t-1']
-            for feature in required_features:
-                if feature not in input_df.columns:
-                    errors.append(f"缺少必要特徵：{feature}")
-                elif pd.isna(input_df[feature].iloc[0]):
-                    errors.append(f"必要特徵包含空值：{feature}")
-            
-            # 檢查數據合理性
-            if 'close' in input_df.columns:
-                close_price = input_df['close'].iloc[0]
-                if close_price <= 0:
-                    errors.append(f"收盤價不合理：{close_price}")
-            
-            if 'return_t-1' in input_df.columns:
-                return_val = input_df['return_t-1'].iloc[0]
-                if abs(return_val) > 0.5:  # 單日漲跌幅超過50%可能有問題
-                    errors.append(f"報酬率異常：{return_val:.3f}")
-            
-            return len(errors) == 0, errors
-            
-        except Exception as e:
-            errors.append(f"驗證過程發生錯誤：{e}")
-            return False, errors
-
-    def get_feature_importance(self):
-        """
-        獲取特徵重要性
-        
-        Returns:
-            dict: 特徵重要性字典
-        """
-        try:
-            if self.model is None:
-                return None
-            
-            # 獲取特徵重要性
-            importance_scores = self.model.feature_importances_
-            
-            # 創建特徵重要性字典
-            importance_dict = {}
-            for i, feature in enumerate(self.feature_columns):
-                if i < len(importance_scores):
-                    importance_dict[feature] = float(importance_scores[i])
-            
-            # 按重要性排序
-            sorted_importance = dict(sorted(importance_dict.items(), 
-                                          key=lambda x: x[1], 
-                                          reverse=True))
-            
-            return sorted_importance
-            
-        except Exception as e:
-            print(f"獲取特徵重要性時發生錯誤：{e}")
-            return None
-
-    def explain_prediction(self, input_df, predictions):
-        """
-        解釋預測結果
-        
-        Args:
-            input_df (pd.DataFrame): 輸入特徵
-            predictions (dict): 預測結果
-            
-        Returns:
-            str: 解釋文本
-        """
-        try:
-            explanation = []
-            explanation.append("=== 預測解釋 ===")
-            
-            # 分析主要驅動因素
-            feature_importance = self.get_feature_importance()
-            if feature_importance:
-                explanation.append("主要影響因素:")
-                top_features = list(feature_importance.keys())[:3]
-                for feature in top_features:
-                    if feature in input_df.columns:
-                        value = input_df[feature].iloc[0]
-                        importance = feature_importance[feature]
-                        explanation.append(f"  - {feature}: {value:.4f} (重要性: {importance:.3f})")
-            
-            # 分析預測趨勢
-            explanation.append("\n預測趨勢分析:")
-            for key, value in predictions.items():
-                days = key.split('_')[2][1:]
-                trend = "看漲" if value > 1 else "看跌" if value < -1 else "持平"
-                explanation.append(f"  - {days}日: {value:+.2f}% ({trend})")
-            
-            return "\n".join(explanation)
-            
-        except Exception as e:
-            return f"解釋生成失敗: {e}"
-
-# 範例��用方式
-if __name__ == "__main__":
-    # 初始化模型
-    model = XGBoostModel()
-    
-    # 準備測試數據
-    test_data = pd.DataFrame({
-        'close': [150.0],
-        'return_t-1': [0.02],
-        'return_t-5': [0.05],
-        'MA5_close': [148.0],
-        'volatility_5d': [0.025],
-        'volume_ratio_5d': [1.2],
-        'MACD_diff': [0.5],
-        'dji_return_t-1': [0.01],
-        'sox_return_t-1': [0.015],
-        'NEWS': [0.1]
-    })
-    
-    print("測試模型預測器...")
-    print("輸入特徵:")
-    print(test_data)
-    
-    # 進行預測
-    predictions = model.predict('xgboost_model', test_data)
-    
-    if predictions:
-        print("\n預測成功！")
-        print("結果說明：輸出為相對於當前價格的漲幅百分比")
-        
-        # 解釋預測
-        explanation = model.explain_prediction(test_data, predictions)
-        print(f"\n{explanation}")
-        
-        # 計算信心度
-        confidence = model.get_prediction_confidence(test_data)
-        print(f"\n預測信心度: {confidence:.2%}")
-    else:
-        print("預測失敗！")