Update model_predictor.py

model_predictor.py

@@ -190,4 +190,236 @@ class XGBoostModel:
             print("=== 漲幅預測結果 ===")
             for key, value in result.items():
                 days = key.split('_')[2][1:]  # 提取天數
-                direction = "↗️ 上漲" if value > 0
+                direction = "上漲" if value > 0 else "下跌"
+                print(f"  {days}日後預測: {value:+.2f}% ({direction})")
+            
+            return result
+            
+        except Exception as e:
+            print(f"預測過程中發生錯誤：{e}")
+            import traceback
+            traceback.print_exc()
+            return None
+
+    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_v1_3_percentage_output', 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("預測失敗！")