model_predictor.py

# model_predictor.py - 支援漲幅百分比輸出的XGBoost模型預測器
# 修改版本：輸出改為漲幅百分比而非絕對價格

# model_predictor.py - 修正版本，對應訓練腳本的確切配置

import os
import numpy as np
import pandas as pd
import xgboost as xgb
from sklearn.preprocessing import MinMaxScaler
import joblib
import warnings
warnings.filterwarnings('ignore')

class XGBoostModel:
    def __init__(self):
        """
        初始化 XGBoost 模型類別
        根據訓練腳本 xgboost_for_stock_trend_&_prices_prediction_gpu_v_2_1_3.py 的配置
        """
        # 根據訓練腳本的 new_feature_columns，確保順序完全一致
        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'   # 成交量加權報酬率
        ]
        
        # 預測目標對應（根據訓練腳本的 train_y）
        self.prediction_mapping = {
            'Change_pct_t1_pred': 1,   # 1天後漲幅%
            'Change_pct_t5_pred': 5,   # 5天後漲幅%
            'Change_pct_t10_pred': 10, # 10天後漲幅%
            'Change_pct_t20_pred': 20  # 20天後漲幅%
        }
        
        self.model = None
        self.scaler = None
        self.is_model_loaded = False
        
        # 模型檔案路徑
        self.model_path = 'xgboost_model.json'
        self.scaler_path = 'feature_scaler.pkl'

    def create_features_from_stock_data(self, stock_data):
        """
        從股票資料創建所需的特徵
        完全對應訓練腳本中的 create_new_features 函數
        
        Args:
            stock_data: yfinance 格式的股票資料 DataFrame
            
        Returns:
            processed_df: 包含所有特徵的 DataFrame
        """
        df = stock_data.copy()
        
        # 確保必要的基礎欄位存在
        required_base_columns = ['Close', 'Volume', 'High', 'Low']
        for col in required_base_columns:
            if col not in df.columns:
                raise ValueError(f"缺少必要的基礎欄位: {col}")
        
        # 統一欄位名稱（yfinance 使用大寫）
        df['close'] = df['Close']
        df['volume'] = df['Volume']
        
        # 1. return_t-1 — 前一日報酬率
        df['return_t-1'] = df['close'].pct_change()
        
        # 2. return_t-5 — 過去 5 日累積報酬率
        df['return_t-5'] = (df['close'] / df['close'].shift(5) - 1)
        
        # 3. MA5_close — 5 日移動平均價
        df['MA5_close'] = df['close'].rolling(window=5).mean()
        
        # 4. volatility_5d — 5 日報酬標準差
        df['volatility_5d'] = df['return_t-1'].rolling(window=5).std()
        
        # 5. volume_ratio_5d — 今日成交量 ÷ 5 日均量
        df['volume_5d_avg'] = df['volume'].rolling(window=5).mean()
        df['volume_ratio_5d'] = df['volume'] / df['volume_5d_avg']
        
        # 6. MACD_diff — MACD - signal
        exp1 = df['close'].ewm(span=12).mean()
        exp2 = df['close'].ewm(span=26).mean()
        macd_line = exp1 - exp2
        signal_line = macd_line.ewm(span=9).mean()
        df['MACD_diff'] = macd_line - signal_line
        
        # 7-8. 美股指數報酬率（需要外部資料，暫設為0）
        df['dji_return_t-1'] = 0.0  # 這需要從外部獲取道瓊指數資料
        df['sox_return_t-1'] = 0.0  # 這需要從外部獲取費半指數資料
        
        # 9. NEWS — 新聞情緒分數（需要外部資料，暫設為0）
        df['NEWS'] = 0.0
        
        # 10. MACDvol — MACD柱狀圖
        df['MACDvol'] = macd_line - signal_line
        
        # 11. RSI_14 — 14日RSI
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['RSI_14'] = 100 - (100 / (1 + rs))
        
        # 12. ADX — 平均趨向指標
        df['up_move'] = df['High'] - df['High'].shift(1)
        df['down_move'] = df['Low'].shift(1) - df['Low']
        df['+DM'] = np.where((df['up_move'] > df['down_move']) & (df['up_move'] > 0), df['up_move'], 0)
        df['-DM'] = np.where((df['down_move'] > df['up_move']) & (df['down_move'] > 0), df['down_move'], 0)
        
        high_low = df['High'] - df['Low']
        high_close_prev = np.abs(df['High'] - df['close'].shift(1))
        low_close_prev = np.abs(df['Low'] - df['close'].shift(1))
        df['TR'] = np.maximum.reduce([high_low, high_close_prev, low_close_prev])
        
        df['+DI'] = (df['+DM'].ewm(com=13, adjust=False).mean() / df['TR'].ewm(com=13, adjust=False).mean()) * 100
        df['-DI'] = (df['-DM'].ewm(com=13, adjust=False).mean() / df['TR'].ewm(com=13, adjust=False).mean()) * 100
        df['DX'] = np.abs(df['+DI'] - df['-DI']) / (df['+DI'] + df['-DI']) * 100
        df['ADX'] = df['DX'].ewm(com=13, adjust=False).mean()
        
        # 13. volume_weighted_return — 成交量加權報酬率
        df['volume_weighted_return'] = np.abs(df['return_t-1']) * df['volume']
        
        # 清理輔助欄位
        cleanup_columns = ['volume_5d_avg', 'up_move', 'down_move', '+DM', '-DM', 'TR', '+DI', '-DI', 'DX']
        df.drop(columns=[col for col in cleanup_columns if col in df.columns], inplace=True)
        
        # 填補 NaN 值
        df.fillna(method='ffill', inplace=True)
        df.fillna(0, inplace=True)  # 剩餘的 NaN 用 0 填補
        
        return df

    def load_model(self, model_name='xgboost_model'):
        """
        載入訓練好的模型和標準化器
        
        Args:
            model_name: 模型名稱
            
        Returns:
            bool: 載入是否成功
        """
        try:
            # 載入 XGBoost 模型
            if os.path.exists(self.model_path):
                self.model = xgb.XGBRegressor()
                self.model.load_model(self.model_path)
                print(f"成功載入模型: {self.model_path}")
            else:
                print(f"警告：模型檔案 {self.model_path} 不存在")
                return False
                
            # 嘗試載入標準化器（如果存在）
            if os.path.exists(self.scaler_path):
                self.scaler = joblib.load(self.scaler_path)
                print(f"成功載入標準化器: {self.scaler_path}")
            else:
                print(f"警告：未找到標準化器檔案 {self.scaler_path}，將使用原始數據進行預測")
                # 根據訓練腳本，模型沒有使用標準化，所以這是正常的
                self.scaler = None
                
            self.is_model_loaded = True
            return True
            
        except Exception as e:
            print(f"載入模型時發生錯誤: {e}")
            return False

    def predict(self, model_name, input_data):
        """
        使用載入的模型進行預測
        
        Args:
            model_name: 模型名稱（保持接口一致性）
            input_data: 輸入特徵 DataFrame 或 numpy array
            
        Returns:
            dict: 預測結果字典，包含各時間框架的漲幅百分比
        """
        if not self.is_model_loaded:
            if not self.load_model(model_name):
                raise RuntimeError("模型載入失敗，無法進行預測")
        
        try:
            # 確保輸入是 DataFrame 格式
            if isinstance(input_data, np.ndarray):
                if input_data.shape[1] != len(self.feature_columns):
                    raise ValueError(f"輸入特徵數量不匹配。期望: {len(self.feature_columns)}, 實際: {input_data.shape[1]}")
                input_df = pd.DataFrame(input_data, columns=self.feature_columns)
            elif isinstance(input_data, pd.DataFrame):
                input_df = input_data.copy()
            else:
                raise ValueError("輸入數據必須是 DataFrame 或 numpy array")
            
            # 確保所有必需的特徵都存在
            missing_features = [col for col in self.feature_columns if col not in input_df.columns]
            if missing_features:
                raise ValueError(f"缺少必要的特徵欄位: {missing_features}")
            
            # 選擇並排序特徵
            input_features = input_df[self.feature_columns]
            
            # 檢查 NaN 值
            if input_features.isnull().any().any():
                print("警告：輸入數據包含 NaN 值，將用 0 填補")
                input_features = input_features.fillna(0)
            
            # 應用標準化（如果有的話）
            if self.scaler is not None:
                input_features_scaled = self.scaler.transform(input_features)
            else:
                input_features_scaled = input_features.values
            
            # 進行預測
            predictions = self.model.predict(input_features_scaled)
            
            # 處理預測結果的維度
            if predictions.ndim == 1:
                # 如果是單一樣本的預測，reshape 成 (1, 4)
                if len(predictions) == 4:
                    predictions = predictions.reshape(1, -1)
                else:
                    raise ValueError(f"預測結果維度不正確: {predictions.shape}")
            
            # 確保結果是 (n_samples, 4) 的形狀
            if predictions.shape[1] != 4:
                raise ValueError(f"模型預測輸出維度錯誤，期望 4 個輸出，實際: {predictions.shape[1]}")
            
            # 構建預測結果字典（取第一個樣本的預測）
            result = {}
            prediction_keys = ['Change_pct_t1_pred', 'Change_pct_t5_pred', 'Change_pct_t10_pred', 'Change_pct_t20_pred']
            
            for i, key in enumerate(prediction_keys):
                result[key] = float(predictions[0, i])  # 取第一個樣本的第 i 個預測
            
            return result
            
        except Exception as e:
            print(f"預測過程中發生錯誤: {e}")
            raise

    def predict_single_timeframe(self, stock_data, days, news_score=0.0, us_market_data=None):
        """
        預測單一時間框架的漲幅
        
        Args:
            stock_data: 股票歷史數據 (yfinance格式)
            days: 預測天數 (1, 5, 10, 20)
            news_score: 新聞情緒分數
            us_market_data: 美股市場數據 (可選)
            
        Returns:
            float: 預測的漲幅百分比
        """
        try:
            # 創建特徵
            processed_df = self.create_features_from_stock_data(stock_data)
            
            # 使用最新的數據點
            latest_data = processed_df.iloc[-1:].copy()
            
            # 更新新聞分數
            latest_data.loc[latest_data.index[0], 'NEWS'] = news_score
            
            # 更新美股數據（如果提供）
            if us_market_data:
                if 'DJI' in us_market_data and len(us_market_data) > 1:
                    dji_return = (us_market_data['DJI'][-1] - us_market_data['DJI'][-2]) / us_market_data['DJI'][-2]
                    latest_data.loc[latest_data.index[0], 'dji_return_t-1'] = dji_return
                    
                if 'SOX' in us_market_data and len(us_market_data) > 1:
                    sox_return = (us_market_data['SOX'][-1] - us_market_data['SOX'][-2]) / us_market_data['SOX'][-2]
                    latest_data.loc[latest_data.index[0], 'sox_return_t-1'] = sox_return
            
            # 進行預測
            predictions = self.predict('xgboost_model', latest_data)
            
            # 根據天數返回對應的預測值
            if days == 1:
                return predictions['Change_pct_t1_pred']
            elif days == 5:
                return predictions['Change_pct_t5_pred']
            elif days == 10:
                return predictions['Change_pct_t10_pred']
            elif days == 20:
                return predictions['Change_pct_t20_pred']
            else:
                # 對於其他天數，使用最接近的預測值
                if days <= 3:
                    return predictions['Change_pct_t1_pred']
                elif days <= 7:
                    return predictions['Change_pct_t5_pred']
                elif days <= 15:
                    return predictions['Change_pct_t10_pred']
                else:
                    return predictions['Change_pct_t20_pred']
                    
        except Exception as e:
            print(f"單一時間框架預測失敗: {e}")
            return 0.0

    def validate_input_features(self, input_data):
        """
        驗證輸入特徵的完整性和有效性
        
        Args:
            input_data: 輸入的特徵數據
            
        Returns:
            dict: 驗證結果
        """
        validation_result = {
            'is_valid': True,
            'missing_features': [],
            'invalid_values': [],
            'warnings': []
        }
        
        try:
            if isinstance(input_data, np.ndarray):
                if input_data.shape[1] != len(self.feature_columns):
                    validation_result['is_valid'] = False
                    validation_result['warnings'].append(f"特徵數量不匹配: 期望{len(self.feature_columns)}, 實際{input_data.shape[1]}")
                return validation_result
                
            # 檢查缺失特徵
            if isinstance(input_data, pd.DataFrame):
                missing_features = [col for col in self.feature_columns if col not in input_data.columns]
                if missing_features:
                    validation_result['missing_features'] = missing_features
                    validation_result['is_valid'] = False
                
                # 檢查數值有效性
                for feature in self.feature_columns:
                    if feature in input_data.columns:
                        if input_data[feature].isnull().any():
                            validation_result['invalid_values'].append(f"{feature}: 包含NaN值")
                        
                        if np.isinf(input_data[feature]).any():
                            validation_result['invalid_values'].append(f"{feature}: 包含無限值")
            
            return validation_result
            
        except Exception as e:
            validation_result['is_valid'] = False
            validation_result['warnings'].append(f"驗證過程出錯: {e}")
            return validation_result

    def get_feature_importance(self):
        """
        獲取模型的特徵重要性
        
        Returns:
            dict: 特徵重要性字典
        """
        if not self.is_model_loaded:
            return {}
            
        try:
            importance_scores = self.model.feature_importances_
            importance_dict = {}
            
            for i, feature in enumerate(self.feature_columns):
                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 {}

    def get_prediction_confidence(self, input_data):
        """
        估算預測信心度
        
        Args:
            input_data: 輸入特徵數據
            
        Returns:
            float: 信心度分數 (0-1)
        """
        try:
            # 基礎信心度檢查
            validation_result = self.validate_input_features(input_data)
            
            if not validation_result['is_valid']:
                return 0.3  # 數據有問題時給予較低信心度
            
            # 根據特徵完整性調整信心度
            base_confidence = 0.7
            
            if validation_result['missing_features']:
                base_confidence -= len(validation_result['missing_features']) * 0.05
            
            if validation_result['invalid_values']:
                base_confidence -= len(validation_result['invalid_values']) * 0.05
            
            return max(0.3, min(0.9, 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("預測失敗！")