功能描述

这个智能金融数据分析系统提供以下功能：

1. 多源金融数据采集（股票、外汇、加密货币、宏观经济指标）
2. 高级时间序列分析与特征工程
3. 机器学习与深度学习模型集成
4. 投资组合优化与风险管理
5. 实时市场监控与异常检测
6. 自动化交易信号生成
7. 回测框架与绩效评估
8. 可视化分析与报告生成
9. 新闻情绪分析
10. 预测模型解释与可解释AI

代码实现

import numpy as np
import pandas as pd
import yfinance as yf
import ccxt
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple, Callable, Any, Union
from enum import Enum, auto
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.preprocessing import MinMaxScaler, StandardScaler
from sklearn.model_selection import TimeSeriesSplit
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.metrics import mean_squared_error, mean_absolute_error
import tensorflow as tf
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.layers import LSTM, Dense, Dropout, Input, Conv1D, MaxPooling1D, Flatten
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
import statsmodels.api as sm
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller
from pymongo import MongoClient
import pytz
from newsapi import NewsApiClient
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from pypfopt import EfficientFrontier, risk_models, expected_returns
from pypfopt.discrete_allocation import DiscreteAllocation
import warnings
import logging
import json
import os
import hashlib
import pickle
import concurrent.futures
import schedule
import time
from bs4 import BeautifulSoup
import re
from alpha_vantage.timeseries import TimeSeries
from fredapi import Fred

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('FinancialAnalysisSystem')

warnings.filterwarnings('ignore')
plt.style.use('seaborn')
pd.set_option('display.max_columns', None)

class AssetType(Enum):
    STOCK = auto()
    CRYPTO = auto()
    FOREX = auto()
    COMMODITY = auto()
    INDEX = auto()

class DataSource(Enum):
    YAHOO = auto()
    CCXT = auto()
    ALPHAVANTAGE = auto()
    FRED = auto()
    CUSTOM = auto()

class TimeFrame(Enum):
    DAILY = auto()
    HOURLY = auto()
    MINUTE_30 = auto()
    MINUTE_15 = auto()
    MINUTE_5 = auto()
    MINUTE_1 = auto()

class ModelType(Enum):
    LSTM = auto()
    CNN_LSTM = auto()
    RANDOM_FOREST = auto()
    GRADIENT_BOOSTING = auto()
    ARIMA = auto()
    PROPHET = auto()
    ENSEMBLE = auto()

class TradeSignal(Enum):
    STRONG_BUY = auto()
    BUY = auto()
    NEUTRAL = auto()
    SELL = auto()
    STRONG_SELL = auto()

@dataclass
class Asset:
    symbol: str
    name: str
    type: AssetType
    data_source: DataSource
    source_symbol: Optional[str] = None
    is_active: bool = True

@dataclass
class HistoricalData:
    asset: Asset
    df: pd.DataFrame
    timeframe: TimeFrame
    last_updated: datetime

@dataclass
class FeatureConfig:
    rolling_windows: List[int] = field(default_factory=lambda: [3, 5, 7, 10, 14, 21, 30, 50, 100, 200])
    technical_indicators: List[str] = field(default_factory=lambda: [
        'sma', 'ema', 'rsi', 'macd', 'bollinger', 'stochastic', 
        'atr', 'adx', 'obv', 'cci', 'williams_r'
    ])
    lag_features: List[int] = field(default_factory=lambda: [1, 2, 3, 5, 7])
    include_volume: bool = True
    include_news_sentiment: bool = True
    include_macro: bool = True

@dataclass
class ModelConfig:
    model_type: ModelType
    lookback_window: int = 60
    forecast_horizon: int = 5
    train_test_split: float = 0.8
    epochs: int = 100
    batch_size: int = 32
    early_stopping_patience: int = 10
    learning_rate: float = 0.001
    lstm_units: List[int] = field(default_factory=lambda: [64, 32])
    dropout_rate: float = 0.2
    random_forest_estimators: int = 200
    gb_estimators: int = 150

@dataclass
class TradingConfig:
    initial_capital: float = 100000.0
    risk_per_trade: float = 0.02
    take_profit: float = 0.15
    stop_loss: float = 0.05
    max_position_size: float = 0.2
    commission: float = 0.001
    slippage: float = 0.0005

@dataclass
class BacktestResult:
    total_return: float
    annualized_return: float
    sharpe_ratio: float
    max_drawdown: float
    win_rate: float
    profit_factor: float
    trades: List[Dict[str, Any]]
    equity_curve: pd.DataFrame
    performance_metrics: Dict[str, float]

class FinancialAnalysisSystem:
    """智能金融数据分析与预测系统"""
    
    def __init__(self, 
                 db_connection_string: str = "mongodb://localhost:27017",
                 news_api_key: Optional[str] = None,
                 alpha_vantage_key: Optional[str] = None,
                 fred_api_key: Optional[str] = None,
                 data_cache_dir: str = "financial_data",
                 model_storage_dir: str = "saved_models"):
        
        self.db_connection_string = db_connection_string
        self.news_api_key = news_api_key
        self.alpha_vantage_key = alpha_vantage_key
        self.fred_api_key = fred_api_key
        
        # 初始化数据库连接
        self._init_db()
        
        # 创建目录
        os.makedirs(data_cache_dir, exist_ok=True)
        os.makedirs(model_storage_dir, exist_ok=True)
        
        self.data_cache_dir = data_cache_dir
        self.model_storage_dir = model_storage_dir
        
        # 初始化数据源客户端
        self._init_data_clients()
        
        # 缓存和状态
        self.assets: Dict[str, Asset] = {}
        self.historical_data: Dict[str, HistoricalData] = {}
        self.models: Dict[str, Any] = {}
        self.portfolio: Dict[str, float] = {}
        self.market_state: Dict[str, Any] = {}
        
        # 加载资产配置
        self._load_assets()
        
        logger.info("Financial analysis system initialized")
    
    def _init_db(self):
        """初始化数据库连接"""
        try:
            self.db_client = MongoClient(self.db_connection_string)
            self.db = self.db_client["financial_analysis"]
            self.assets_collection = self.db["assets"]
            self.historical_data_collection = self.db["historical_data"]
            self.models_collection = self.db["models"]
            self.portfolios_collection = self.db["portfolios"]
            self.trades_collection = self.db["trades"]
            self.news_collection = self.db["news"]
            
            logger.info("Connected to MongoDB at %s", self.db_connection_string)
        except Exception as e:
            logger.error("Failed to connect to MongoDB: %s", str(e))
            raise
    
    def _init_data_clients(self):
        """初始化数据源客户端"""
        try:
            # 新闻API客户端
            if self.news_api_key:
                self.news_client = NewsApiClient(api_key=self.news_api_key)
                self.sentiment_analyzer = SentimentIntensityAnalyzer()
            
            # Alpha Vantage客户端
            if self.alpha_vantage_key:
                self.alpha_vantage = TimeSeries(key=self.alpha_vantage_key, output_format='pandas')
            
            # FRED (Federal Reserve Economic Data) 客户端
            if self.fred_api_key:
                self.fred = Fred(api_key=self.fred_api_key)
            
            # CCXT 加密货币交易所客户端
            self.ccxt_exchanges = {
                'binance': ccxt.binance(),
                'coinbase': ccxt.coinbasepro(),
                'kraken': ccxt.kraken()
            }
            
            logger.info("Data clients initialized")
        except Exception as e:
            logger.error("Failed to initialize data clients: %s", str(e))
            raise
    
    def _load_assets(self):
        """从数据库加载资产配置"""
        try:
            for asset_data in self.assets_collection.find({"is_active": True}):
                asset = Asset(
                    symbol=asset_data["symbol"],
                    name=asset_data["name"],
                    type=AssetType[asset_data["type"]],
                    data_source=DataSource[asset_data["data_source"]],
                    source_symbol=asset_data.get("source_symbol"),
                    is_active=asset_data["is_active"]
                )
                self.assets[asset.symbol] = asset
            
            logger.info("Loaded %d assets from database", len(self.assets))
        except Exception as e:
            logger.error("Failed to load assets: %s", str(e))
            raise
    
    def add_asset(self, asset: Asset):
        """添加资产到系统"""
        try:
            self.assets[asset.symbol] = asset
            self.assets_collection.update_one(
                {"symbol": asset.symbol},
                {"$set": asdict(asset)},
                upsert=True
            )
            logger.info("Added asset: %s (%s)", asset.name, asset.symbol)
        except Exception as e:
            logger.error("Failed to add asset: %s", str(e))
            raise
    
    def fetch_historical_data(self, 
                            asset_symbol: str,
                            start_date: Union[str, datetime],
                            end_date: Union[str, datetime],
                            timeframe: TimeFrame = TimeFrame.DAILY,
                            force_update: bool = False) -> HistoricalData:
        """获取历史数据"""
        try:
            # 检查资产是否存在
            if asset_symbol not in self.assets:
                raise ValueError(f"Asset {asset_symbol} not found")
            
            asset = self.assets[asset_symbol]
            
            # 检查缓存中是否有数据
            cache_key = f"{asset_symbol}_{timeframe.name}"
            cached_data = self._get_cached_data(cache_key)
            
            if cached_data and not force_update:
                # 检查是否需要更新
                last_updated = cached_data.last_updated
                if timeframe == TimeFrame.DAILY and last_updated.date() >= datetime.now().date():
                    return cached_data
                elif timeframe != TimeFrame.DAILY and last_updated >= datetime.now() - timedelta(hours=1):
                    return cached_data
            
            # 根据数据源获取数据
            df = None
            if asset.data_source == DataSource.YAHOO:
                df = self._fetch_yahoo_data(asset, start_date, end_date, timeframe)
            elif asset.data_source == DataSource.CCXT:
                df = self._fetch_ccxt_data(asset, start_date, end_date, timeframe)
            elif asset.data_source == DataSource.ALPHAVANTAGE:
                df = self._fetch_alphavantage_data(asset, start_date, end_date, timeframe)
            elif asset.data_source == DataSource.FRED:
                df = self._fetch_fred_data(asset, start_date, end_date, timeframe)
            
            if df is None or df.empty:
                raise ValueError(f"Failed to fetch data for {asset_symbol}")
            
            # 处理数据
            df = self._process_raw_data(df, timeframe)
            
            # 创建历史数据对象
            historical_data = HistoricalData(
                asset=asset,
                df=df,
                timeframe=timeframe,
                last_updated=datetime.now()
            )
            
            # 缓存数据
            self._cache_data(cache_key, historical_data)
            
            # 保存到数据库
            self.historical_data_collection.update_one(
                {"asset.symbol": asset_symbol, "timeframe": timeframe.name},
                {"$set": asdict(historical_data)},
                upsert=True
            )
            
            # 更新内存缓存
            self.historical_data[cache_key] = historical_data
            
            logger.info("Fetched historical data for %s (%s to %s)", 
                       asset_symbol, start_date, end_date)
            
            return historical_data
        except Exception as e:
            logger.error("Failed to fetch historical data for %s: %s", asset_symbol, str(e))
            raise
    
    def _fetch_yahoo_data(self, 
                         asset: Asset,
                         start_date: Union[str, datetime],
                         end_date: Union[str, datetime],
                         timeframe: TimeFrame) -> pd.DataFrame:
        """从Yahoo Finance获取数据"""
        try:
            symbol = asset.source_symbol if asset.source_symbol else asset.symbol
            
            # 转换时间框架为Yahoo的间隔参数
            interval_map = {
                TimeFrame.DAILY: '1d',
                TimeFrame.HOURLY: '1h',
                TimeFrame.MINUTE_30: '30m',
                TimeFrame.MINUTE_15: '15m',
                TimeFrame.MINUTE_5: '5m',
                TimeFrame.MINUTE_1: '1m'
            }
            interval = interval_map.get(timeframe, '1d')
            
            # 获取数据
            df = yf.download(
                tickers=symbol,
                start=start_date,
                end=end_date,
                interval=interval,
                progress=False,
                auto_adjust=True
            )
            
            # 重命名列以保持一致性
            df = df.rename(columns={
                'Open': 'open',
                'High': 'high',
                'Low': 'low',
                'Close': 'close',
                'Adj Close': 'adj_close',
                'Volume': 'volume'
            })
            
            return df
        except Exception as e:
            logger.error("Yahoo Finance fetch error: %s", str(e))
            raise
    
    def _fetch_ccxt_data(self,
                        asset: Asset,
                        start_date: Union[str, datetime],
                        end_date: Union[str, datetime],
                        timeframe: TimeFrame) -> pd.DataFrame:
        """从CCXT交易所获取加密货币数据"""
        try:
            symbol = asset.source_symbol if asset.source_symbol else asset.symbol
            
            # 确定最佳交易所
            exchange = None
            for exch in self.ccxt_exchanges.values():
                if symbol in exch.load_markets():
                    exchange = exch
                    break
            
            if not exchange:
                raise ValueError(f"Symbol {symbol} not found in any exchange")
            
            # 转换时间框架为CCXT的间隔参数
            timeframe_map = {
                TimeFrame.DAILY: '1d',
                TimeFrame.HOURLY: '1h',
                TimeFrame.MINUTE_30: '30m',
                TimeFrame.MINUTE_15: '15m',
                TimeFrame.MINUTE_5: '5m',
                TimeFrame.MINUTE_1: '1m'
            }
            tf = timeframe_map.get(timeframe, '1d')
            
            # 转换日期为毫秒时间戳
            since = int(pd.to_datetime(start_date).timestamp() * 1000)
            now = int(pd.to_datetime(end_date).timestamp() * 1000)
            
            # 获取OHLCV数据
            all_ohlcv = []
            while since < now:
                ohlcv = exchange.fetch_ohlcv(symbol, tf, since)
                if not ohlcv:
                    break
                
                all_ohlcv += ohlcv
                since = ohlcv[-1][0] + 1
                
                # 避免API限制
                time.sleep(exchange.rateLimit / 1000)
            
            # 转换为DataFrame
            df = pd.DataFrame(all_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
            df = df.set_index('date')
            df = df.drop('timestamp', axis=1)
            
            return df
        except Exception as e:
            logger.error("CCXT fetch error: %s", str(e))
            raise
    
    def _fetch_alphavantage_data(self,
                               asset: Asset,
                               start_date: Union[str, datetime],
                               end_date: Union[str, datetime],
                               timeframe: TimeFrame) -> pd.DataFrame:
        """从Alpha Vantage获取数据"""
        try:
            symbol = asset.source_symbol if asset.source_symbol else asset.symbol
            
            # 确定适当的函数和参数
            if timeframe == TimeFrame.DAILY:
                data, _ = self.alpha_vantage.get_daily(symbol=symbol, outputsize='full')
            elif timeframe == TimeFrame.HOURLY:
                data, _ = self.alpha_vantage.get_intraday(symbol=symbol, interval='60min', outputsize='full')
            else:
                raise ValueError("Alpha Vantage only supports daily and hourly data")
            
            # 转换日期索引
            data.index = pd.to_datetime(data.index)
            
            # 筛选日期范围
            data = data.loc[start_date:end_date]
            
            # 重命名列
            data = data.rename(columns={
                '1. open': 'open',
                '2. high': 'high',
                '3. low': 'low',
                '4. close': 'close',
                '5. volume': 'volume'
            })
            
            return data
        except Exception as e:
            logger.error("Alpha Vantage fetch error: %s", str(e))
            raise
    
    def _fetch_fred_data(self,
                       asset: Asset,
                       start_date: Union[str, datetime],
                       end_date: Union[str, datetime],
                       timeframe: TimeFrame) -> pd.DataFrame:
        """从FRED获取宏观经济数据"""
        try:
            symbol = asset.source_symbol if asset.source_symbol else asset.symbol
            
            # 获取数据
            data = self.fred.get_series(symbol, start_date, end_date)
            
            # 转换为DataFrame
            df = pd.DataFrame(data, columns=['close'])
            df.index.name = 'date'
            
            # 对于宏观经济数据，我们通常只有收盘价
            df['open'] = df['close']
            df['high'] = df['close']
            df['low'] = df['close']
            df['volume'] = 0
            
            return df
        except Exception as e:
            logger.error("FRED fetch error: %s", str(e))
            raise
    
    def _process_raw_data(self, df: pd.DataFrame, timeframe: TimeFrame) -> pd.DataFrame:
        """处理原始数据"""
        try:
            # 确保索引是DatetimeIndex
            if not isinstance(df.index, pd.DatetimeIndex):
                df.index = pd.to_datetime(df.index)
            
            # 按日期排序
            df = df.sort_index()
            
            # 处理缺失值
            df = df.ffill()  # 前向填充
            df = df.bfill()  # 后向填充
            
            # 对于高频数据，可能需要重新采样
            if timeframe != TimeFrame.DAILY:
                # 确保数据均匀分布
                freq_map = {
                    TimeFrame.HOURLY: 'H',
                    TimeFrame.MINUTE_30: '30T',
                    TimeFrame.MINUTE_15: '15T',
                    TimeFrame.MINUTE_5: '5T',
                    TimeFrame.MINUTE_1: '1T'
                }
                freq = freq_map.get(timeframe, 'D')
                
                # 重新采样
                df = df.resample(freq).agg({
                    'open': 'first',
                    'high': 'max',
                    'low': 'min',
                    'close': 'last',
                    'volume': 'sum'
                })
            
            # 计算对数收益率
            df['log_return'] = np.log(df['close'] / df['close'].shift(1))
            
            # 计算简单移动平均
            df['sma_20'] = df['close'].rolling(window=20).mean()
            df['sma_50'] = df['close'].rolling(window=50).mean()
            df['sma_200'] = df['close'].rolling(window=200).mean()
            
            # 计算指数移动平均
            df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
            df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
            
            # 计算MACD
            ema_12 = df['close'].ewm(span=12, adjust=False).mean()
            ema_26 = df['close'].ewm(span=26, adjust=False).mean()
            df['macd'] = ema_12 - ema_26
            df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
            df['macd_hist'] = df['macd'] - df['macd_signal']
            
            # 计算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'] = 100 - (100 / (1 + rs))
            
            # 计算布林带
            sma = df['close'].rolling(window=20).mean()
            std = df['close'].rolling(window=20).std()
            df['bollinger_upper'] = sma + (std * 2)
            df['bollinger_lower'] = sma - (std * 2)
            
            # 计算ATR (平均真实波幅)
            high_low = df['high'] - df['low']
            high_close = np.abs(df['high'] - df['close'].shift())
            low_close = np.abs(df['low'] - df['close'].shift())
            ranges = pd.concat([high_low, high_close, low_close], axis=1)
            true_range = np.max(ranges, axis=1)
            df['atr'] = true_range.rolling(window=14).mean()
            
            # 添加日期特征
            df['day_of_week'] = df.index.dayofweek
            df['day_of_month'] = df.index.day
            df['month'] = df.index.month
            df['quarter'] = df.index.quarter
            df['year'] = df.index.year
            
            # 删除初始的NaN值
            df = df.dropna()
            
            return df
        except Exception as e:
            logger.error("Data processing error: %s", str(e))
            raise
    
    def _get_cached_data(self, cache_key: str) -> Optional[HistoricalData]:
        """从缓存获取数据"""
        try:
            # 检查内存缓存
            if cache_key in self.historical_data:
                return self.historical_data[cache_key]
            
            # 检查文件缓存
            cache_file = os.path.join(self.data_cache_dir, f"{cache_key}.pkl")
            if os.path.exists(cache_file):
                with open(cache_file, 'rb') as f:
                    return pickle.load(f)
            
            # 检查数据库
            cached_data = self.historical_data_collection.find_one(
                {"asset.symbol": cache_key.split('_')[0], "timeframe": cache_key.split('_')[1]},
                projection={"_id": 0}
            )
            
            if cached_data:
                # 转换DataFrame
                df = pd.DataFrame(cached_data['df'])
                df.index = pd.to_datetime(df.index)
                
                # 创建HistoricalData对象
                asset_data = cached_data['asset']
                asset = Asset(
                    symbol=asset_data['symbol'],
                    name=asset_data['name'],
                    type=AssetType[asset_data['type']],
                    data_source=DataSource[asset_data['data_source']],
                    source_symbol=asset_data.get('source_symbol'),
                    is_active=asset_data['is_active']
                )
                
                historical_data = HistoricalData(
                    asset=asset,
                    df=df,
                    timeframe=TimeFrame[cached_data['timeframe']],
                    last_updated=cached_data['last_updated']
                )
                
                # 保存到内存缓存
                self.historical_data[cache_key] = historical_data
                
                return historical_data
            
            return None
        except Exception as e:
            logger.error("Cache retrieval error: %s", str(e))
            return None
    
    def _cache_data(self, cache_key: str, data: HistoricalData):
        """缓存数据"""
        try:
            # 保存到内存缓存
            self.historical_data[cache_key] = data
            
            # 保存到文件缓存
            cache_file = os.path.join(self.data_cache_dir, f"{cache_key}.pkl")
            with open(cache_file, 'wb') as f:
                pickle.dump(data, f)
        except Exception as e:
            logger.error("Cache save error: %s", str(e))
    
    def engineer_features(self, 
                         historical_data: HistoricalData,
                         config: Optional[FeatureConfig] = None) -> pd.DataFrame:
        """特征工程"""
        try:
            if config is None:
                config = FeatureConfig()
            
            df = historical_data.df.copy()
            asset = historical_data.asset
            
            # 添加技术指标
            for window in config.rolling_windows:
                df[f'sma_{window}'] = df['close'].rolling(window=window).mean()
                df[f'ema_{window}'] = df['close'].ewm(span=window, adjust=False).mean()
                df[f'volatility_{window}'] = df['log_return'].rolling(window=window).std() * np.sqrt(window)
                df[f'momentum_{window}'] = df['close'] / df['close'].shift(window) - 1
            
            # 添加滞后特征
            for lag in config.lag_features:
                df[f'close_lag_{lag}'] = df['close'].shift(lag)
                df[f'volume_lag_{lag}'] = df['volume'].shift(lag)
                df[f'log_return_lag_{lag}'] = df['log_return'].shift(lag)
            
            # 添加交互特征
            df['sma_ratio'] = df['sma_20'] / df['sma_50']
            df['ema_ratio'] = df['ema_12'] / df['ema_26']
            df['volume_ma_ratio'] = df['volume'] / df['volume'].rolling(window=20).mean()
            
            # 添加新闻情绪数据
            if config.include_news_sentiment and self.news_api_key:
                news_sentiment = self._get_news_sentiment(asset)
                df = df.merge(news_sentiment, left_index=True, right_index=True, how='left')
                df[['news_sentiment', 'news_volume']] = df[['news_sentiment', 'news_volume']].fillna(0)
            
            # 添加宏观经济数据
            if config.include_macro and asset.type != AssetType.INDEX and self.fred_api_key:
                macro_data = self._get_macro_data(historical_data.df.index)
                df = df.merge(macro_data, left_index=True, right_index=True, how='left')
                df = df.fillna(method='ffill').fillna(method='bfill')
            
            # 添加目标变量 (未来收益率)
            df['target'] = df['close'].shift(-1) / df['close'] - 1
            
            # 删除NaN值
            df = df.dropna()
            
            return df
        except Exception as e:
            logger.error("Feature engineering error: %s", str(e))
            raise
    
    def _get_news_sentiment(self, asset: Asset) -> pd.DataFrame:
        """获取新闻情绪数据"""
        try:
            # 确定搜索词
            search_terms = [asset.name, asset.symbol]
            if asset.type == AssetType.STOCK:
                search_terms.append(f"{asset.name} stock")
            elif asset.type == AssetType.CRYPTO:
                search_terms.append(f"{asset.name} cryptocurrency")
            
            # 获取新闻文章
            all_articles = []
            for term in search_terms:
                articles = self.news_client.get_everything(
                    q=term,
                    language='en',
                    sort_by='relevancy',
                    page_size=100
                )
                all_articles.extend(articles['articles'])
            
            # 分析情绪
            sentiments = []
            for article in all_articles:
                title = article['title']
                description = article['description']
                content = title + " " + (description if description else "")
                
                # 使用VADER进行情绪分析
                sentiment = self.sentiment_analyzer.polarity_scores(content)
                date = pd.to_datetime(article['publishedAt']).date()
                
                sentiments.append({
                    'date': date,
                    'compound': sentiment['compound'],
                    'positive': sentiment['pos'],
                    'negative': sentiment['neg'],
                    'neutral': sentiment['neu'],
                    'title': title
                })
            
            # 创建DataFrame
            if not sentiments:
                return pd.DataFrame(columns=['date', 'news_sentiment', 'news_volume'])
            
            df = pd.DataFrame(sentiments)
            df = df.groupby('date').agg({
                'compound': 'mean',
                'positive': 'mean',
                'negative': 'mean',
                'neutral': 'mean',
                'title': 'count'
            }).rename(columns={
                'compound': 'news_sentiment',
                'title': 'news_volume'
            })
            
            # 保存到数据库
            records = df.reset_index().to_dict('records')
            if records:
                self.news_collection.insert_many(records)
            
            return df[['news_sentiment', 'news_volume']]
        except Exception as e:
            logger.error("News sentiment analysis error: %s", str(e))
            return pd.DataFrame(columns=['date', 'news_sentiment', 'news_volume'])
    
    def _get_macro_data(self, dates: pd.DatetimeIndex) -> pd.DataFrame:
        """获取宏观经济数据"""
        try:
            # 定义重要的宏观经济指标
            macro_indicators = {
                'DGS10': '10_year_treasury_yield',  # 10年期国债收益率
                'DCOILWTICO': 'wti_crude_price',    # WTI原油价格
                'CPIAUCSL': 'cpi',                  # 消费者价格指数
                'UNRATE': 'unemployment_rate',      # 失业率
                'FEDFUNDS': 'fed_funds_rate',       # 联邦基金利率
                'SP500': 'sp500_index'              # S&P 500指数
            }
            
            # 获取数据
            macro_data = pd.DataFrame(index=dates)
            
            for fred_code, col_name in macro_indicators.items():
                try:
                    series = self.fred.get_series(fred_code)
                    series = series.reindex(dates, method='ffill')
                    macro_data[col_name] = series
                except Exception as e:
                    logger.warning("Failed to fetch %s: %s", fred_code, str(e))
                    macro_data[col_name] = np.nan
            
            # 计算变化率
            for col in macro_data.columns:
                macro_data[f'{col}_change'] = macro_data[col].pct_change()
            
            # 填充缺失值
            macro_data = macro_data.ffill().bfill()
            
            return macro_data
        except Exception as e:
            logger.error("Macro data fetch error: %s", str(e))
            return pd.DataFrame(index=dates)
    
    def train_model(self, 
                   asset_symbol: str,
                   model_type: ModelType,
                   feature_config: Optional[FeatureConfig] = None,
                   model_config: Optional[ModelConfig] = None) -> Any:
        """训练预测模型"""
        try:
            # 检查资产是否存在
            if asset_symbol not in self.assets:
                raise ValueError(f"Asset {asset_symbol} not found")
            
            # 获取历史数据
            historical_data = self.historical_data.get(asset_symbol + "_DAILY")
            if not historical_data:
                historical_data = self.fetch_historical_data(
                    asset_symbol=asset_symbol,
                    start_date=datetime.now() - timedelta(days=365*5),  # 5年数据
                    end_date=datetime.now(),
                    timeframe=TimeFrame.DAILY
                )
            
            # 特征工程
            features_df = self.engineer_features(historical_data, feature_config)
            
            # 准备训练数据
            X = features_df.drop(['target'], axis=1)
            y = features_df['target']
            
            # 标准化特征
            scaler = StandardScaler()
            X_scaled = scaler.fit_transform(X)
            
            # 时间序列分割
            tscv = TimeSeriesSplit(n_splits=5)
            
            # 训练模型
            if model_type == ModelType.LSTM:
                model = self._train_lstm_model(X_scaled, y, model_config)
            elif model_type == ModelType.CNN_LSTM:
                model = self._train_cnn_lstm_model(X_scaled, y, model_config)
            elif model_type == ModelType.RANDOM_FOREST:
                model = self._train_random_forest_model(X_scaled, y, model_config)
            elif model_type == ModelType.GRADIENT_BOOSTING:
                model = self._train_gradient_boosting_model(X_scaled, y, model_config)
            elif model_type == ModelType.ARIMA:
                model = self._train_arima_model(X_scaled, y, model_config)
            else:
                raise ValueError(f"Unsupported model type: {model_type}")
            
            # 保存模型
            model_key = f"{asset_symbol}_{model_type.name}"
            self.models[model_key] = model
            
            # 保存到数据库
            model_bytes = pickle.dumps(model)
            self.models_collection.update_one(
                {"key": model_key},
                {"$set": {
                    "asset_symbol": asset_symbol,
                    "model_type": model_type.name,
                    "model_data": model_bytes,
                    "features": list(X.columns),
                    "trained_at": datetime.now()
                }},
                upsert=True
            )
            
            logger.info("Trained %s model for %s", model_type.name, asset_symbol)
            
            return model
        except Exception as e:
            logger.error("Model training error for %s: %s", asset_symbol, str(e))
            raise
    
    def _train_lstm_model(self, 
                        X: np.ndarray,
                        y: np.ndarray,
                        config: ModelConfig) -> Sequential:
        """训练LSTM模型"""
        try:
            # 准备LSTM输入数据
            X_reshaped, y_reshaped = self._prepare_sequences(X, y, config.lookback_window)
            
            # 创建模型
            model = Sequential()
            model.add(Input(shape=(X_reshaped.shape[1], X_reshaped.shape[2])))
            
            # 添加LSTM层
            for i, units in enumerate(config.lstm_units):
                return_sequences = i < len(config.lstm_units) - 1
                model.add(LSTM(units, return_sequences=return_sequences))
                model.add(Dropout(config.dropout_rate))
            
            # 输出层
            model.add(Dense(1))
            
            # 编译模型
            model.compile(
                optimizer=Adam(learning_rate=config.learning_rate),
                loss='mse',
                metrics=['mae']
            )
            
            # 定义回调
            callbacks = [
                EarlyStopping(monitor='val_loss', patience=config.early_stopping_patience, restore_best_weights=True),
                ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=5