Очистка OHLCV данных в Pandas: Удаление выбросов и NaN

import pandas as pd
import numpy as np

# 1. Создание примера OHLCV данных с NaN и выбросами
def create_sample_ohlcv_data():
    dates = pd.date_range(start='2023-01-01', periods=100, freq='D')
    data = {
        'Open': np.random.uniform(100, 110, 100),
        'High': np.random.uniform(110, 120, 100),
        'Low': np.random.uniform(90, 100, 100),
        'Close': np.random.uniform(105, 115, 100),
        'Volume': np.random.randint(10000, 50000, 100)
    }
    df = pd.DataFrame(data, index=dates)

    # Добавляем NaN
    df.loc[df.sample(frac=0.05).index, ['Close', 'Volume']] = np.nan
    df.loc[df.sample(frac=0.02).index, 'Open'] = np.nan

    # Добавляем выбросы
    df.loc[10, 'Close'] = 500  # Выброс вверх
    df.loc[25, 'Close'] = 50   # Выброс вниз
    df.loc[40, 'Volume'] = 500000 # Выброс объема
    df.loc[60, 'High'] = 1000 # Выброс High
    df.loc[75, 'Low'] = 10 # Выброс Low

    return df

df = create_sample_ohlcv_data()
print("Исходные данные с NaN и выбросами:")
print(df.head(15))
print("\nКоличество NaN до обработки:")
print(df.isnull().sum())

# --- Обработка пропущенных значений (NaN) ---

def handle_nan_values(df_input, method='ffill', columns=None):
    df = df_input.copy()
    if columns is None:
        columns = df.columns
    
    if method == 'ffill':
        df[columns] = df[columns].fillna(method='ffill')
    elif method == 'bfill':
        df[columns] = df[columns].fillna(method='bfill')
    elif method == 'mean':
        df[columns] = df[columns].fillna(df[columns].mean())
    elif method == 'median':
        df[columns] = df[columns].fillna(df[columns].median())
    elif method == 'drop':
        df = df.dropna(subset=columns)
    else:
        raise ValueError("Метод заполнения NaN должен быть 'ffill', 'bfill', 'mean', 'median' или 'drop'.")
    return df

# Пример 1: Заполнение NaN методом 'ffill' (forward fill)
df_cleaned_ffill = handle_nan_values(df.copy(), method='ffill')
print("\nДанные после заполнения NaN методом 'ffill':")
print(df_cleaned_ffill.isnull().sum())

# Пример 2: Заполнение NaN методом 'bfill' (backward fill)
# Часто используется в комбинации с ffill для заполнения начальных NaN
df_cleaned_bfill = handle_nan_values(df.copy(), method='bfill')
print("\nДанные после заполнения NaN методом 'bfill':")
print(df_cleaned_bfill.isnull().sum())

# Пример 3: Удаление строк с NaN (используйте осторожно, может привести к потере данных)
# df_cleaned_drop = handle_nan_values(df.copy(), method='drop')
# print("\nДанные после удаления строк с NaN:")
# print(df_cleaned_drop.isnull().sum())
# print(f"Размер DataFrame после удаления: {df_cleaned_drop.shape}")


# --- Обработка выбросов (Outliers) ---

# Метод 1: Межквартильный размах (IQR)
def cap_outliers_iqr(series, k=1.5):
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - k * IQR
    upper_bound = Q3 + k * IQR
    
    # Ограничиваем значения
    capped_series = series.clip(lower=lower_bound, upper=upper_bound)
    return capped_series

# Метод 2: Z-оценка
def cap_outliers_zscore(series, threshold=3):
    mean = series.mean()
    std = series.std()
    
    # Вычисляем Z-оценки
    z_scores = np.abs((series - mean) / std)
    
    # Заменяем выбросы на NaN, затем можно заполнить медианой или средним
    # Или просто ограничиваем значения до +/- threshold * std
    capped_series = series.copy()
    capped_series[z_scores > threshold] = np.nan # Можно заменить на медиану или среднее
    
    # Для демонстрации заполним медианой после выявления
    median_val = series[z_scores <= threshold].median()
    capped_series = capped_series.fillna(median_val)
    
    return capped_series

# Метод 3: Скользящая медиана/среднее (для сглаживания и выявления резких скачков)
def smooth_and_detect_outliers_rolling(series, window=5, threshold_multiplier=3):
    rolling_median = series.rolling(window=window, center=True).median()
    # Разница между значением и скользящей медианой
    deviation = np.abs(series - rolling_median)
    # Стандартное отклонение отклонений
    mad = deviation.rolling(window=window, center=True).median() * 1.4826 # Median Absolute Deviation
    
    # Выбросы - это значения, которые сильно отклоняются от скользящей медианы
    outlier_mask = deviation > (threshold_multiplier * mad)
    
    # Заменяем выбросы на скользящую медиану
    cleaned_series = series.copy()
    cleaned_series[outlier_mask] = rolling_median[outlier_mask]
    
    # Заполняем NaN, которые могли появиться из-за rolling window на краях
    cleaned_series = cleaned_series.fillna(method='ffill').fillna(method='bfill')
    return cleaned_series

# Применяем обработку NaN ко всему DataFrame перед обработкой выбросов
df_processed = handle_nan_values(df.copy(), method='ffill')
df_processed = handle_nan_values(df_processed, method='bfill') # Для заполнения начальных NaN

print("\nДанные после обработки NaN (ffill + bfill):")
print(df_processed.head(15))

# Применяем методы обработки выбросов
print("\nОбработка выбросов:")

# Обработка 'Close' цены с помощью IQR
df_processed['Close_IQR_Cleaned'] = cap_outliers_iqr(df_processed['Close'])
print("\n'Close' после IQR очистки:")
print(df_processed[['Close', 'Close_IQR_Cleaned']].head(15))
print(f"Выбросы 'Close' (оригинал): {df.loc[[10, 25], 'Close'].tolist()}")
print(f"Выбросы 'Close' (после IQR): {df_processed.loc[[10, 25], 'Close_IQR_Cleaned'].tolist()}")


# Обработка 'Volume' с помощью Z-оценки
df_processed['Volume_ZScore_Cleaned'] = cap_outliers_zscore(df_processed['Volume'])
print("\n'Volume' после Z-Score очистки:")
print(df_processed[['Volume', 'Volume_ZScore_Cleaned']].head(15))
print(f"Выброс 'Volume' (оригинал): {df.loc[40, 'Volume']}")
print(f"Выброс 'Volume' (после Z-Score): {df_processed.loc[40, 'Volume_ZScore_Cleaned']}")


# Обработка 'High' и 'Low' с помощью скользящей медианы
df_processed['High_Rolling_Cleaned'] = smooth_and_detect_outliers_rolling(df_processed['High'])
df_processed['Low_Rolling_Cleaned'] = smooth_and_detect_outliers_rolling(df_processed['Low'])
print("\n'High' и 'Low' после очистки скользящей медианой:")
print(df_processed[['High', 'High_Rolling_Cleaned', 'Low', 'Low_Rolling_Cleaned']].head(15))
print(f"Выброс 'High' (оригинал): {df.loc[60, 'High']}")
print(f"Выброс 'High' (после Rolling): {df_processed.loc[60, 'High_Rolling_Cleaned']}")
print(f"Выброс 'Low' (оригинал): {df.loc[75, 'Low']}")
print(f"Выброс 'Low' (после Rolling): {df_processed.loc[75, 'Low_Rolling_Cleaned']}")


# Финальный DataFrame с очищенными данными (можно выбрать, какие колонки использовать)
df_final_cleaned = df_processed.copy()
df_final_cleaned['Close'] = df_final_cleaned['Close_IQR_Cleaned']
df_final_cleaned['Volume'] = df_final_cleaned['Volume_ZScore_Cleaned']
df_final_cleaned['High'] = df_final_cleaned['High_Rolling_Cleaned']
df_final_cleaned['Low'] = df_final_cleaned['Low_Rolling_Cleaned']

# Удаляем временные колонки
df_final_cleaned = df_final_cleaned.drop(columns=[
    'Close_IQR_Cleaned', 'Volume_ZScore_Cleaned',
    'High_Rolling_Cleaned', 'Low_Rolling_Cleaned'
])

print("\nФинальные очищенные данные (первые 15 строк):")
print(df_final_cleaned.head(15))
print("\nКоличество NaN в финальных данных:")
print(df_final_cleaned.isnull().sum())