pandas и pyarrow для обработки данных чанками, предотвращая переполнение оперативной памяти.Почему кванты переходят с HDF5 на Parquet?
В количественном анализе (Quantitative Finance) скорость доступа к историческим тикам определяет общую производительность бэктестера. Исторически формат HDF5 (.h5) пользовался популярностью благодаря высокой скорости последовательного чтения на локальных машинах. Однако при работе с распределенными хранилищами или облачной инфраструктурой HDF5 показывает себя не с лучшей стороны: он не поддерживает эффективное колоночное сжатие на уровне метаданных и не позволяет выполнять фильтрацию на стороне хранилища (predicate pushdown).
Формат Parquet с партиционированием по датам решает эти проблемы. Он позволяет считывать с диска только те дни и те колонки, которые необходимы для текущего расчета. Это критически важно, когда вы проводите ресурсоемкие вычисления, такие как Оптимизация торговых стратегий с помощью генетических алгоритмов DEAP на Python, или когда вам нужно рассчитать сложные производные параметры, выполняя Расчет и Визуализация Поверхности Подразумеваемой Волатильности (IV Surface) Опционов в Python Matplotlib.
Исходный код
Ниже представлен готовый к использованию скрипт. Он считывает данные из HDF5 порциями (чанками), извлекает дату из временной метки, преобразует данные в формат PyArrow и записывает их в структурированную директорию Parquet.
import os
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
def convert_hdf5_to_parquet(hdf_file_path, hdf_key, output_dir, chunksize=1000000, compression='snappy'):
'''
Конвертирует HDF5 файл с тиками в партиционированный Parquet.
:param hdf_file_path: Путь к исходному файлу HDF5.
:param hdf_key: Ключ (датасет) внутри HDF5.
:param output_dir: Директория для сохранения Parquet партиций.
:param chunksize: Размер чанка для итеративного чтения (избегает OOM).
:param compression: Алгоритм сжатия (snappy, gzip, zstd).
'''
if not os.path.exists(output_dir):
os.makedirs(output_dir)
print(f'Начало конвертации {hdf_file_path} (ключ: {hdf_key}) -> {output_dir}')
# Открываем HDF5 в режиме чтения
with pd.HDFStore(hdf_file_path, mode='r') as store:
if hdf_key not in store:
raise KeyError(f'Ключ {hdf_key} не найден в файле {hdf_file_path}')
# Получаем общее количество строк для мониторинга прогресса
nrows = store.get_storer(hdf_key).nrows
print(f'Всего строк для обработки: {nrows}')
processed_rows = 0
# Читаем HDF5 чанками для экономии RAM
for chunk in store.select(hdf_key, chunksize=chunksize):
# Проверяем структуру индекса
if not isinstance(chunk.index, pd.DatetimeIndex):
if 'timestamp' in chunk.columns:
chunk['timestamp'] = pd.to_datetime(chunk['timestamp'])
chunk.set_index('timestamp', inplace=True)
else:
raise ValueError('Данные должны содержать DatetimeIndex или колонку timestamp')
# Сбрасываем индекс, чтобы сохранить timestamp как колонку в Parquet
df_flat = chunk.reset_index()
# Создаем строковую колонку даты для партиционирования
df_flat['date'] = df_flat['timestamp'].dt.date.astype(str)
# Конвертируем pandas DataFrame в таблицу PyArrow
table = pa.Table.from_pandas(df_flat)
# Записываем чанк в партиционированную директорию
pq.write_to_dataset(
table,
root_path=output_dir,
partition_cols=['date'],
compression=compression,
use_dictionary=True
)
processed_rows += len(chunk)
print(f'Обработано строк: {processed_rows}/{nrows} ({processed_rows/nrows*100:.2f}%)')
print('Конвертация успешно завершена!')
if __name__ == '__main__':
# Конфигурация путей
HDF5_FILE = 'ticks.h5'
HDF5_KEY = '/BTCUSDT'
OUTPUT_DIRECTORY = 'parquet_dataset'
# Генерация тестового HDF5 файла, если он отсутствует
if not os.path.exists(HDF5_FILE):
print('Генерация тестового HDF5 файла...')
date_range = pd.date_range(start='2023-10-01', end='2023-10-05', freq='ms')
test_df = pd.DataFrame({
'bid': [27000.0 + i * 0.1 for i in range(len(date_range))],
'ask': [27000.5 + i * 0.1 for i in range(len(date_range))],
'volume': [1.0] * len(date_range)
}, index=date_range)
test_df.index.name = 'timestamp'
test_df.to_hdf(HDF5_FILE, key=HDF5_KEY, format='table')
print('Тестовый файл создан.')
# Запуск конвертации
convert_hdf5_to_parquet(
hdf_file_path=HDF5_FILE,
hdf_key=HDF5_KEY,
output_dir=OUTPUT_DIRECTORY,
chunksize=500000
)Разбор параметров
hdf_file_path: Путь к исходному файлу HDF5 на вашем диске. Поддерживаются форматы.h5и.hdf5.hdf_key: Внутренний путь к таблице внутри HDF5-файла (например, идентификатор торговой пары или тикера).output_dir: Корневая директория, куда будет записана структура партиций Parquet.chunksize: Размер порции данных, считываемой за один шаг. Настройка этого параметра позволяет обрабатывать файлы, размер которых превышает объем доступной оперативной памяти.compression: Метод сжатия данных. Опцияsnappyобеспечивает высокую скорость декомпрессии при умеренном сжатии, что идеально подходит для локального бэктестинга.
Как запустить
Для работы скрипта вам понадобятся библиотеки для работы с таблицами HDF5 и движок PyArrow. Установите их с помощью пакетного менеджера pip:
pip install pandas tables pyarrowСохраните приведенный выше код в файл convert_ticks.py и выполните его в терминале:
python convert_ticks.pyПосле завершения работы скрипта в целевой директории parquet_dataset будет создана структура папок, разделенная по датам:
parquet_dataset/
├── date=2023-10-01/
│ └── xxxxxxxx.parquet
├── date=2023-10-02/
│ └── xxxxxxxx.parquet
└── date=2023-10-03/
└── xxxxxxxx.parquetПолученный датасет готов к интеграции в любые современные бэктест-платформы. Вы можете использовать эти структурированные данные для симуляции торгов, предварительно изучив материал по теме Настройка комиссий и проскальзывания в Zipline-Reloaded для крипты, чтобы ваши тесты максимально соответствовали реальным рыночным условиям.
