Как вычислить профиль объема (VPVR) на 50 ГБ данных с Dask

import dask.dataframe as dd
import pandas as pd
import numpy as np
from dask.distributed import Client

def calculate_local_vpvr(partition: pd.DataFrame, bin_size: float) -> pd.DataFrame:
    """
    Локальная агрегация внутри одной партиции.
    Снижает объем передаваемых по сети данных в тысячи раз.
    """
    if partition.empty:
        return pd.DataFrame(columns=['volume', 'buy_volume', 'sell_volume'], dtype='float64')
    
    # Округляем цены до заданного шага (bin_size)
    partition['price_bin'] = (partition['price'] / bin_size).round() * bin_size
    
    # Разделяем объемы на покупку и продажу
    partition['buy_volume'] = partition['volume'].where(partition['side'] == 1, 0.0)
    partition['sell_volume'] = partition['volume'].where(partition['side'] == -1, 0.0)
    
    # Группируем локально
    local_profile = partition.groupby('price_bin')[['volume', 'buy_volume', 'sell_volume']].sum()
    return local_profile

def compute_global_vpvr(parquet_path: str, bin_size: float = 0.5) -> pd.DataFrame:
    """
    Основная функция для вычисления глобального VPVR без shuffle-эффекта.
    """
    # Загружаем только необходимые колонки для экономии RAM
    df = dd.read_parquet(
        parquet_path, 
        columns=['price', 'volume', 'side']
    )
    
    # Определяем метаданные для map_partitions
    meta = pd.DataFrame(
        columns=['volume', 'buy_volume', 'sell_volume'], 
        dtype='float64'
    )
    meta.index.name = 'price_bin'
    
    # Шаг 1: Локальный VPVR на каждой партиции
    local_profiles = df.map_partitions(
        calculate_local_vpvr,
        bin_size=bin_size,
        meta=meta
    )
    
    # Шаг 2: Финальное объединение локальных профилей
    # Так как размер локальных профилей ничтожно мал, groupby здесь мгновенный
    global_vpvr = local_profiles.groupby(local_profiles.index).sum()
    
    # Запуск вычислений
    result = global_vpvr.compute()
    return result.sort_index()

if __name__ == '__main__':
    # Инициализируем локальный Dask-кластер
    client = Client(n_workers=4, threads_per_worker=2, memory_limit='8GB')
    print(f"Dask Dashboard доступен по адресу: {client.dashboard_link}")
    
    path_to_data = './data/ticks_50gb.parquet'
    vpvr_profile = compute_global_vpvr(path_to_data, bin_size=0.1)
    print(vpvr_profile.head(20))