model_predictor.py

# 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:
    def __init__(self):
        """
        初始化 XGBoost 模型預測器
        
        【重要更新】
        - 模型現在輸出漲幅百分比而非絕對價格
        - 支援 1日、5日、10日、20日的漲幅預測
        """
        self.model = None
        self.scaler = None
        # 【【修改點】】更新特徵欄位列表以包含新特徵
        self.feature_columns = [
            'close',             # 前一日收盤價  
            'return_t-1',        # 前一日報酬率
            'return_t-5',        # 過去 5 日累積報酬率
            'MA5_close',         # 5 日移動平均價
            'volatility_5d',     # 5 日報酬標準差
            'volume_ratio_5d',   # 今日成交量 ÷ 5 日均量
            'MACD_diff',         # MACD - signal
            'dji_return_t-1',    # 前一日道瓊指數報酬率
            'sox_return_t-1',    # 前一日費半指數報酬率
            'NEWS',              # 新聞情緒分數
            'MACDvol',           # 成交量MACD
            'RSI_14',            # 14日RSI
            'ADX',               # ADX趨勢指標
            'volume_weighted_return' # 成交量加權報酬率
        ]
        
        # 【新增】輸出目標對應表
        self.output_targets = {
            1: 'Change_pct_t1_pred',    # 1天後漲幅%
            5: 'Change_pct_t5_pred',    # 5天後漲幅%
            10: 'Change_pct_t10_pred',  # 10天後漲幅%
            20: 'Change_pct_t20_pred'   # 20天後漲幅%
        }
        
        print("XGBoost 模型預測器初始化完成")
        print(f"輸出格式：漲幅百分比 (1日, 5日, 10日, 20日)")
        print(f"預期特徵數量: {len(self.feature_columns)}")

    def load_model(self, model_path):
        """
        載入預訓練的 XGBoost 模型
        
        Args:
            model_path (str): 模型檔案路徑 (.json 格式)
        
        Returns:
            bool: 是否成功載入
        """
        try:
            # 檢查模型檔案是否存在
            if not os.path.exists(model_path):
                print(f"錯誤：找不到模型檔案 {model_path}")
                return False
            
            # 載入 XGBoost 模型
            self.model = xgb.XGBRegressor()
            self.model.load_model(model_path)
            
            print(f"成功載入模型：{model_path}")
            print(f"預期特徵數量：{len(self.feature_columns)}")
            
            return True
            
        except Exception as e:
            print(f"載入模型時發生錯誤：{e}")
            return False

    def load_scaler(self, scaler_path):
        """停用標準化流程"""
        print("⚠️ 已停用標準化：模型使用原始特徵進行預測。")
        self.scaler = None
        return False

    def preprocess_features(self, input_df):
        # 確保特徵齊全
        missing_features = [f for f in self.feature_columns if f not in input_df.columns]
        if missing_features:
            print(f"警告：缺少以下特徵：{missing_features}")
            for feature in missing_features:
                input_df[feature] = 0
    
        input_df = input_df[self.feature_columns].fillna(0)

        # ✅ 直接回傳原始特徵
        return input_df

    def predict(self, model_name, input_df):
        """
        進行股價漲幅預測
        
        Args:
            model_name (str): 模型名稱（用於載入對應模型）
            input_df (pd.DataFrame): 輸入特徵
        
        Returns:
            dict: 預測結果，包含各時間點的漲幅百分比
        """
        try:
            # 載入模型（如果尚未載入）
            if self.model is None:
                model_path = f"{model_name}.json"
                if not self.load_model(model_path):
                    return None
            
            # 載入標準化器（如果存在）
            if self.scaler is None:
                scaler_path = f"{model_name}_scaler.pkl"
                self.load_scaler(scaler_path)
            
            # 預處理特徵
            processed_df = self.preprocess_features(input_df.copy())
            
            # 進行預測
            predictions = self.model.predict(processed_df)
            
            # 【重要修改】將預測結果格式化為漲幅百分比
            if predictions.ndim == 1:
                # 如果只有一個輸出，假設是 1 日預測
                result = {
                    'Change_pct_t1_pred': float(predictions[0])
                }
            else:
                # 多輸出情況：1日, 5日, 10日, 20日
                result = {
                    'Change_pct_t1_pred': float(predictions[0][0]) if len(predictions[0]) > 0 else 0.0,
                    'Change_pct_t5_pred': float(predictions[0][1]) if len(predictions[0]) > 1 else 0.0,
                    'Change_pct_t10_pred': float(predictions[0][2]) if len(predictions[0]) > 2 else 0.0,
                    'Change_pct_t20_pred': float(predictions[0][3]) if len(predictions[0]) > 3 else 0.0
                }
            
            # 輸出預測結果摘要
            print("=== 漲幅預測結果 ===")
            for key, value in result.items():
                days = key.split('_')[2][1:]  # 提取天數
                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],
        'MACDvol': [0.015],
        'RSI_14': [0.015],
        'ADX': [0.015],
        'volume_weighted_return': [0.015]
    })
    
    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("預測失敗！")