"""Data preprocessing: normalization, feature engineering, and sequence creation.""" from typing import Tuple, Optional, Literal import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler import pandas_ta as ta class CryptoPreprocessor: """Preprocess cryptocurrency data with normalization and feature engineering.""" def __init__( self, lookback: int = 168, forecast_horizon: int = 24, price_change_threshold: float = 0.0 # 改为0,使用真实涨跌 ): """Initialize preprocessor. Args: lookback: Number of historical timesteps to use forecast_horizon: Number of timesteps to predict ahead price_change_threshold: Threshold for binary classification (e.g., 0.015 = 1.5%) """ self.lookback = lookback self.forecast_horizon = forecast_horizon self.price_change_threshold = price_change_threshold self.scaler = MinMaxScaler() self.feature_columns = [] self.is_fitted = False def add_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame: """添加技术指标作为特征 Args: df: 包含OHLCV数据的DataFrame Returns: 添加了技术指标的DataFrame """ df = df.copy() # 使用pandas_ta库 # RSI (14 periods) df['rsi'] = ta.rsi(df['close'], length=14) # MACD macd = ta.macd(df['close']) if macd is not None and not macd.empty: # pandas-ta 返回的列名可能不同,动态获取 macd_cols = macd.columns.tolist() if len(macd_cols) >= 3: df['macd'] = macd.iloc[:, 0] # MACD线 df['macd_hist'] = macd.iloc[:, 1] # MACD柱 df['macd_signal'] = macd.iloc[:, 2] # 信号线 else: # 如果格式不对,使用None df['macd'] = None df['macd_signal'] = None df['macd_hist'] = None else: df['macd'] = None df['macd_signal'] = None df['macd_hist'] = None # Bollinger Bands bbands = ta.bbands(df['close'], length=20) if bbands is not None and not bbands.empty: # pandas-ta 返回的列名可能不同,动态获取 bb_cols = bbands.columns.tolist() if len(bb_cols) >= 3: df['bb_lower'] = bbands.iloc[:, 0] # 第一列是下轨 df['bb_middle'] = bbands.iloc[:, 1] # 第二列是中轨 df['bb_upper'] = bbands.iloc[:, 2] # 第三列是上轨 df['bb_bandwidth'] = (df['bb_upper'] - df['bb_lower']) / (df['bb_middle'] + 1e-8) else: # 如果格式不对,手动计算 sma = df['close'].rolling(window=20).mean() std = df['close'].rolling(window=20).std() df['bb_middle'] = sma df['bb_upper'] = sma + (std * 2) df['bb_lower'] = sma - (std * 2) df['bb_bandwidth'] = (df['bb_upper'] - df['bb_lower']) / (df['bb_middle'] + 1e-8) else: # pandas-ta不可用,手动计算布林带 sma = df['close'].rolling(window=20).mean() std = df['close'].rolling(window=20).std() df['bb_middle'] = sma df['bb_upper'] = sma + (std * 2) df['bb_lower'] = sma - (std * 2) df['bb_bandwidth'] = (df['bb_upper'] - df['bb_lower']) / (df['bb_middle'] + 1e-8) # Volume indicators df['volume_sma'] = ta.sma(df['volume'], length=20) df['volume_ratio'] = df['volume'] / (df['volume_sma'] + 1e-8) # Price momentum df['returns'] = df['close'].pct_change() df['returns_5'] = df['close'].pct_change(5) df['returns_10'] = df['close'].pct_change(10) # Volatility df['volatility'] = df['returns'].rolling(window=20).std() return df def add_time_features(self, df: pd.DataFrame) -> pd.DataFrame: """Add time-based features. Args: df: DataFrame with timestamp column Returns: DataFrame with added time features """ df = df.copy() df['hour'] = df['timestamp'].dt.hour df['day_of_week'] = df['timestamp'].dt.dayofweek df['is_weekend'] = (df['day_of_week'] >= 5).astype(int) # Cyclical encoding for hour df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24) df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24) # Cyclical encoding for day of week df['dow_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7) df['dow_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7) return df def create_labels( self, df: pd.DataFrame, task: Literal['classification', 'regression'] ) -> pd.DataFrame: """Create target labels for classification or regression. Args: df: DataFrame with price data task: Task type ('classification' or 'regression') Returns: DataFrame with added label column """ df = df.copy() # Calculate future returns df['future_price'] = df['close'].shift(-self.forecast_horizon) df['future_return'] = (df['future_price'] - df['close']) / df['close'] if task == 'classification': # Binary classification: 1 if price increases above threshold, 0 otherwise df['label'] = (df['future_return'] > self.price_change_threshold).astype(int) else: # regression # Regression: predict percentage return df['label'] = df['future_return'] # Drop rows without labels (last forecast_horizon rows) df = df[:-self.forecast_horizon].copy() return df def fit(self, df: pd.DataFrame) -> 'CryptoPreprocessor': """Fit the scaler on training data. Args: df: Training DataFrame with features Returns: Self """ # Get feature columns (exclude timestamp and label) exclude_cols = ['timestamp', 'label', 'future_price', 'future_return'] self.feature_columns = [col for col in df.columns if col not in exclude_cols] # Fit scaler self.scaler.fit(df[self.feature_columns].fillna(0)) self.is_fitted = True return self def transform(self, df: pd.DataFrame) -> np.ndarray: """Transform data using fitted scaler. Args: df: DataFrame with features Returns: Scaled feature array """ if not self.is_fitted: raise ValueError("Preprocessor must be fitted before transform") # Fill NaN values df_filled = df[self.feature_columns].fillna(0) # Scale features scaled = self.scaler.transform(df_filled) return scaled def fit_transform(self, df: pd.DataFrame) -> np.ndarray: """Fit and transform in one step. Args: df: DataFrame with features Returns: Scaled feature array """ return self.fit(df).transform(df) def create_sequences( self, features: np.ndarray, labels: np.ndarray ) -> Tuple[np.ndarray, np.ndarray]: """Create sequences for time series prediction. Args: features: Feature array (n_samples, n_features) labels: Label array (n_samples,) Returns: Tuple of (X, y) where X is (n_sequences, lookback, n_features) and y is (n_sequences,) """ X, y = [], [] for i in range(len(features) - self.lookback): X.append(features[i:i + self.lookback]) y.append(labels[i + self.lookback]) return np.array(X), np.array(y) def preprocess_pipeline( self, df: pd.DataFrame, task: Literal['classification', 'regression'], fit: bool = True ) -> Tuple[np.ndarray, np.ndarray, pd.DataFrame]: """Full preprocessing pipeline. Args: df: Raw OHLCV DataFrame task: Task type fit: Whether to fit the scaler (True for training data) Returns: Tuple of (X, y, processed_df) where X is sequences, y is labels, and processed_df is the DataFrame after feature engineering """ # Add features df = self.add_technical_indicators(df) df = self.add_time_features(df) df = self.create_labels(df, task) # Drop initial NaN rows from indicators df = df.dropna().reset_index(drop=True) # Scale features if fit: features = self.fit_transform(df) else: features = self.transform(df) # Get labels labels = df['label'].values # Create sequences X, y = self.create_sequences(features, labels) print(f"Created {len(X)} sequences with shape {X.shape}") return X, y, df def split_data( X: np.ndarray, y: np.ndarray, train_split: float = 0.7, val_split: float = 0.15 ) -> Tuple[Tuple[np.ndarray, np.ndarray], ...]: """Split data into train/val/test sets using time-based split. Args: X: Feature sequences y: Labels train_split: Fraction for training val_split: Fraction for validation Returns: Tuple of ((X_train, y_train), (X_val, y_val), (X_test, y_test)) """ n = len(X) train_end = int(n * train_split) val_end = int(n * (train_split + val_split)) X_train, y_train = X[:train_end], y[:train_end] X_val, y_val = X[train_end:val_end], y[train_end:val_end] X_test, y_test = X[val_end:], y[val_end:] print(f"Train: {len(X_train)} samples") print(f"Val: {len(X_val)} samples") print(f"Test: {len(X_test)} samples") return (X_train, y_train), (X_val, y_val), (X_test, y_test) if __name__ == "__main__": # Test the preprocessor from fetcher import BinanceDataFetcher # Fetch data fetcher = BinanceDataFetcher() df = fetcher.fetch_ohlcv('BTC', '1h') # Preprocess for classification preprocessor = CryptoPreprocessor(lookback=168, forecast_horizon=24) X, y, processed_df = preprocessor.preprocess_pipeline(df, task='classification') print(f"\nFeatures shape: {X.shape}") print(f"Labels shape: {y.shape}") print(f"Label distribution: {np.bincount(y.astype(int))}") # Split data (X_train, y_train), (X_val, y_val), (X_test, y_test) = split_data(X, y)