Как из PostgreSQL и ClickHouse в Python много, быстро и сразу в numpy

Сломал много копий в поисках решения для быстрого получения длинных историй цен для большого количества активов в Python. Ещё имел смелость желать работать с ценами в numpy-массивах, а лучше сразу в pandas.

Стандартные подходы в лоб работали разочаровывающе, что приводило к выполнению запроса к БД в течение 30 секунд и более. Не желая мириться, я нашёл несколько решений, которые полностью меня удовлетворили.

Ноги растут из объектной природы Python. Ведь у него даже целые числа являются объектами, что крайне отрицательно влияет на скорость работы. Менять язык я категорически не хотел.

Первым решением была группировка истории цен силами PostgreSQL, что приводило к незначительной просадке производительности на стороне БД, но ускоряло задачу примерно в ~3 раза. Подробнее метод описан в другой статье.

Итогом появилось понимание, что в Python надо каким-то образом получить весь набор данных одним куском, хотя бы строкой. И разобрать по numpy-массивам или сразу в pandas.

Итоговые результаты:

Решение в лоб для PostgreSQL

Делаем группировку данных в sql-запрос. Пример:

SELECT
    string_agg(symbol::text, ',') AS symbol_list
    , string_agg(dt::text, ',') AS dt_list
    , string_agg(open::text, ',') AS open_list
    , string_agg(high::text, ',') AS high_list
    , string_agg(low::text, ',') AS low_list
    , string_agg("close"::text, ',') AS close_list
    , string_agg(volume::text, ',') AS volume_list
    , string_agg(adj::text, ',') AS adj_list
FROM v_prices_fast
WHERE symbol IN ('{symbols}')

Разобрать данные проще простого:

{
    'symbol': np.array(r[0].split(',')),  # str
    'dt': np.array(r[1].split(','), dtype='datetime64'),  # str w/type
    'open': np.fromstring(r[2], sep=','),  # numbers
    # ...
}

Производительность на ~1.7 млн. строк:

%timeit get_prices_fast(is_adj=False)  # 11.9s

Готовые пакеты Python

Python хорош своим сообществом, которое сталкивается со схожими проблемами. Для нашей цели подойдут следующие:

• odo — создан для оптимизации скорости передачи данных из одного источника в другой. Полностью на Python. С PostgreSQL взаимодействует через SQLAlchemy.
• warp_prism — C-расширение, используемое проектом Quantopian для получения данных из PostgreSQL. В основе заложен функционал odo.

Оба пакета используют возможность PostgreSQL копировать данные в CSV:

COPY {query} TO :path
    WITH (
        FORMAT CSV,
        HEADER :header,
        DELIMITER :delimiter,
        QUOTE :quotechar,
        NULL :na_value,
        ESCAPE :escapechar,
        ENCODING :encoding
    )

На выходе данные разбираются в pandas.DataFrame() или numpy.ndarray().

Так как warp_prism написан на C, он имеет существенное преимущество по скорости парсинга данных. Но одновременно с этим имеет существенный недостаток — ограниченную поддержку типов данных. То есть он парсит int, float, date и str, но не numeric. У odo подобных ограничений нет.

Для использования необходимо описать структуру таблицы и запрос с помощью пакета sqlalchemy:

tbl_prices = sa.Table(
    'prices', metadata,
    sa.Column('symbol', sa.String(16)),
    sa.Column('dt', sa.Date),
    sa.Column('open', sa.FLOAT),
    sa.Column('high', sa.FLOAT),
    sa.Column('low', sa.FLOAT),
    sa.Column('close', sa.FLOAT),
    sa.Column('volume', sa.BIGINT),
    sa.Column('adj', sa.FLOAT),
)
query = sa.select(tbl_prices.c).where(
    tbl_prices.c.symbol.in_(SYMBOLS)
).order_by('symbol', 'dt')

Тесты скорости:

%timeit odo(query, pd.DataFrame, bind=engine)  # 13.8s
%timeit warp_prism.to_dataframe(query, bind=engine)  # 8.4s
%timeit warp_prism.to_arrays(query, bind=engine)  # 8.0s

warp_prism.to_arrays() - подготовка python-словаря с numpy-массивами.

Что можно сделать с ClickHouse?

PostgreSQL всем хорош, кроме аппетита с размеру хранилища и необходимости настройки шардинга для больших таблиц. ClickHouse сам шардирует, хранит всё компактно, а работает молниеносно. Для примера таблица на PostgreSQL размером ~5Gb в ClickHouse умещается в ~1Gb. Использование ClickHouse для хранения цен описываю в другой статье.

К моему огорчению odo не помог, хоть для sqlalchemy есть расширение clickhouse. Воспоминания о скорости работы clickhouse в консоли меня навели на идею обращения к БД через создание отдельного процесса. Я знаю, что это долго и ресурсозатратно, но результаты оказались выше всяких похвал.

sql = 'SELECT days.symbol, days.date, days.open/10000, days.high/10000, days.low/10000, days.close/10000, days.volume FROM days ' 
      'WHERE days.symbol IN ('{0}') ORDER BY days.symbol, days.date;'.format("','".join(SYMBOLS))
cmd = 'clickhouse-client --query="{0}"'.format(sql)

def ch_pandas(cmd):
    p = subprocess.Popen([cmd], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    return pd.io.parsers.read_csv(p.stdout, sep="t", names=['symbol', 'date', 'open', 'high', 'low', 'close', 'volume'])

Результат:

%timeit ch_pandas(cmd)  # 1.6s

Запрос к HTTP-порту ClickHouse

Результаты чуть ухудшились при обращении непосредственно к порту 8123, где отвечает БД:

import urllib
%timeit pd.io.parsers.read_csv('http://localhost:8123/?{0}'.format(urllib.parse.urlencode({'query': sql})), sep="t", names=['symbol', 'date', 'open', 'high', 'low', 'close', 'volume'])
# 1.9s

Но не обошлось без ложки дёгтя.

Ложка дёгтя с ClickHouse

БД впечатлила на больших выборках, но на маленьких результаты разочаровали. В ~20 раз хуже odo. Но это издержки на дополнительный обвес с запуском процесса или обращением по HTTP.

Результаты:

Заключение

Данной статьёй погоня за ускорением взаимодействия между Python и базами данных закончена. Для PostgreSQL при стандартных полях и необходимости универсального доступа к ценам лучшим способом является использование пакета warp_prism от Quantopian. При необходимости хранить большие объёмы истории и высокой частоте запросов большого количества строк идеально подойдёт ClickHouse.

Автор: i0am0raa

Источник