Кластеризация Volume Profile с DBSCAN для POC уровней | Python scikit-learn

import pandas as pd
import numpy as np
from sklearn.cluster import DBSCAN
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker

def calculate_volume_profile(df: pd.DataFrame, price_bins: int = 100) -> pd.DataFrame:
    """
    Рассчитывает Volume Profile для заданного DataFrame OHLCV данных.

    Args:
        df (pd.DataFrame): DataFrame с колонками 'High', 'Low', 'Close', 'Volume'.
        price_bins (int): Количество ценовых интервалов (бинов) для Volume Profile.

    Returns:
        pd.DataFrame: DataFrame с колонками 'price_mid' (середина бина) и 'volume' (объем в бине).
    """
    min_price = df['Low'].min()
    max_price = df['High'].max()
    
    # Создаем бины для цен
    bins = np.linspace(min_price, max_price, price_bins + 1)
    
    # Инициализируем Volume Profile
    volume_profile = pd.DataFrame(0.0, index=range(price_bins), columns=['volume'])
    volume_profile['price_mid'] = [(bins[i] + bins[i+1]) / 2 for i in range(price_bins)]
    
    # Распределяем объем по бинам
    for _, row in df.iterrows():
        high, low, volume = row['High'], row['Low'], row['Volume']
        
        # Определяем диапазон цен для текущей свечи
        price_range_bins = np.where((bins[:-1] < high) & (bins[1:] > low))[0]
        
        if len(price_range_bins) == 0:
            continue
            
        # Упрощенное распределение: равномерно по всем затронутым бинам
        volume_per_bin = volume / len(price_range_bins)
        for bin_idx in price_range_bins:
            volume_profile.loc[bin_idx, 'volume'] += volume_per_bin
            
    return volume_profile[volume_profile['volume'] > 0].reset_index(drop=True)

def find_poc_clusters(volume_profile_df: pd.DataFrame, eps: float, min_samples: int, 
                      volume_threshold_percentile: float = 75) -> list:
    """
    Находит POC (Point of Control) с помощью DBSCAN кластеризации Volume Profile.

    Args:
        volume_profile_df (pd.DataFrame): DataFrame Volume Profile с 'price_mid' и 'volume'.
        eps (float): Максимальное расстояние между двумя выборками, чтобы одна считалась
                     находящейся в окрестности другой.
        min_samples (int): Количество выборок (или общая масса весов) в окрестности,
                           чтобы точка считалась основной.
        volume_threshold_percentile (float): Процентный порог для фильтрации бинов по объему.
                                             Только бины с объемом выше этого порога будут
                                             использованы для кластеризации.

    Returns:
        list: Список словарей, каждый из которых содержит 'price' и 'volume' для найденных POC.
    """
    # Фильтруем бины с низким объемом, чтобы сосредоточиться на значимых областях
    volume_threshold = volume_profile_df['volume'].quantile(volume_threshold_percentile / 100)
    significant_volume_df = volume_profile_df[volume_profile_df['volume'] >= volume_threshold].copy()

    if significant_volume_df.empty:
        return []

    # Подготавливаем данные для DBSCAN: кластеризуем по ценовым уровням
    # Reshape для scikit-learn, так как DBSCAN ожидает 2D массив
    X = significant_volume_df[['price_mid']].values

    # Применяем DBSCAN
    db = DBSCAN(eps=eps, min_samples=min_samples).fit(X)
    labels = db.labels_

    n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
    pocs = []

    for i in range(n_clusters):
        cluster_indices = significant_volume_df[labels == i].index
        cluster_data = significant_volume_df.loc[cluster_indices]
        
        # Находим POC внутри кластера (цена с максимальным объемом)
        poc_row = cluster_data.loc[cluster_data['volume'].idxmax()]
        pocs.append({'price': poc_row['price_mid'], 'volume': poc_row['volume']})
        
    return pocs

def plot_volume_profile_with_pocs(volume_profile_df: pd.DataFrame, pocs: list, title: str = 'Volume Profile with DBSCAN POCs'):
    """
    Визуализирует Volume Profile и найденные POC.

    Args:
        volume_profile_df (pd.DataFrame): DataFrame Volume Profile.
        pocs (list): Список найденных POC.
        title (str): Заголовок графика.
    """
    fig, ax = plt.subplots(figsize=(12, 8))

    # График Volume Profile
    # Высота бара должна соответствовать размеру бина
    bin_height = volume_profile_df['price_mid'].diff().mean() if len(volume_profile_df) > 1 else 1.0
    ax.barh(volume_profile_df['price_mid'], volume_profile_df['volume'], height=bin_height, 
            color='skyblue', edgecolor='gray', alpha=0.7, label='Volume Profile')

    # Отмечаем POCs
    for poc in pocs:
        ax.axhline(y=poc['price'], color='red', linestyle='--', linewidth=2, label=f'POC: {poc["price"]:.2f} (Vol: {poc["volume"]:.0f})')
        ax.text(ax.get_xlim()[1] * 0.95, poc['price'], f'POC {poc["price"]:.2f}', 
                verticalalignment='center', horizontalalignment='right', color='red', fontsize=10)

    ax.set_title(title)
    ax.set_xlabel('Volume')
    ax.set_ylabel('Price')
    ax.grid(True, linestyle=':', alpha=0.6)
    ax.legend()
    plt.tight_layout()
    plt.show()

if __name__ == '__main__':
    # --- 1. Генерация синтетических данных для примера ---
    # В реальном сценарии вы бы загружали данные из файла или API
    # Например, с помощью pandas.read_csv или yfinance
    np.random.seed(42)
    num_candles = 500
    prices = np.cumsum(np.random.normal(0, 0.5, num_candles)) + 100 # Базовая цена
    volumes = np.random.randint(100, 1000, num_candles) # Базовый объем

    # Добавим несколько областей с повышенным объемом для демонстрации POC
    prices[100:150] += np.random.normal(0, 0.1, 50) # Флэт в районе 100-150 свечи
    volumes[100:150] *= 3 # Увеличиваем объем в этом флэте
    prices[300:350] += np.random.normal(0, 0.1, 50) # Еще один флэт
    volumes[300:350] *= 2.5 # Увеличиваем объем

    df_data = pd.DataFrame({
        'Open': prices,
        'High': prices + np.random.uniform(0.1, 0.5, num_candles),
        'Low': prices - np.random.uniform(0.1, 0.5, num_candles),
        'Close': prices + np.random.uniform(-0.2, 0.2, num_candles),
        'Volume': volumes
    })

    # --- 2. Расчет Volume Profile ---
    price_bins = 100 # Количество ценовых бинов
    volume_profile = calculate_volume_profile(df_data, price_bins=price_bins)
    print("\nVolume Profile Head:")
    print(volume_profile.head())

    # --- 3. Поиск POC с помощью DBSCAN ---
    # Параметры DBSCAN:
    # eps: Максимальное расстояние между точками для формирования кластера. 
    #      Должен быть подобран относительно размера ценового бина.
    #      Например, если бин ~0.5, то eps может быть 0.5-1.0.
    # min_samples: Минимальное количество бинов, чтобы сформировать кластер.
    #              Зависит от детализации Volume Profile и ожидаемого размера POC.
    
    # Пример подбора eps: можно взять средний размер бина или его часть.
    # Если price_bins = 100, а диапазон цен ~100, то размер бина ~1.
    price_range = df_data['High'].max() - df_data['Low'].min()
    avg_bin_size = price_range / price_bins
    
    dbscan_eps = avg_bin_size * 1.5 # Пример: 1.5x от среднего размера бина
    dbscan_min_samples = 3 # Минимум 3 бина для формирования кластера
    volume_filter_percentile = 70 # Фильтруем бины с объемом ниже 70-го перцентиля

    pocs = find_poc_clusters(volume_profile, 
                             eps=dbscan_eps, 
                             min_samples=dbscan_min_samples, 
                             volume_threshold_percentile=volume_filter_percentile)

    print("\nFound POCs:")
    for poc in pocs:
        print(f"  Price: {poc['price']:.2f}, Volume: {poc['volume']:.0f}")

    # --- 4. Визуализация результатов ---
    plot_volume_profile_with_pocs(volume_profile, pocs, title='Volume Profile with DBSCAN POCs (Synthetic Data)')

    # --- Пример использования с реальными данными (закомментировано) ---
    # try:
    #     import yfinance as yf
    #     ticker = "SPY"
    #     data = yf.download(ticker, start="2023-01-01", end="2023-03-01")
    #     data.columns = [col.capitalize() for col in data.columns] # Приводим к 'High', 'Low', 'Close', 'Volume'
    #     
    #     real_volume_profile = calculate_volume_profile(data, price_bins=150)
    #     
    #     real_price_range = data['High'].max() - data['Low'].min()
    #     real_avg_bin_size = real_price_range / 150
    #     
    #     real_dbscan_eps = real_avg_bin_size * 2.0 # Подбираем параметры для реальных данных
    #     real_dbscan_min_samples = 5
    #     real_volume_filter_percentile = 80
    #     
    #     real_pocs = find_poc_clusters(real_volume_profile, 
    #                                   eps=real_dbscan_eps, 
    #                                   min_samples=real_dbscan_min_samples, 
    #                                   volume_threshold_percentile=real_volume_filter_percentile)
    #     
    #     print("\nFound POCs for SPY:")
    #     for poc in real_pocs:
    #         print(f"  Price: {poc['price']:.2f}, Volume: {poc['volume']:.0f}")
    #         
    #     plot_volume_profile_with_pocs(real_volume_profile, real_pocs, title=f'Volume Profile with DBSCAN POCs ({ticker})')
    # except ImportError:
    #     print("\nБиблиотека yfinance не установлена. Пропустили пример с реальными данными.")
    #     print("Установите ее командой: pip install yfinance")