Поиск по сайту:

Если у вас есть медленные циклы в Python, вы можете это исправить… пока не сможете


автор Максим Мамаев

Давайте возьмем в качестве примера вычислительную задачу, напишем некоторый код и посмотрим, как мы можем улучшить время выполнения. Вот так.

Обстановка: проблема с рюкзаком

Это вычислительная задача, которую мы будем использовать в качестве примера:

Задача о рюкзаке — хорошо известная задача комбинаторной оптимизации. В этом разделе мы рассмотрим ее наиболее распространенную разновидность — задачу о рюкзаке 0–1 и ее решение с помощью динамического программирования. Если вы знакомы с предметом, можете пропустить эту часть.

Вам дан рюкзак вместимостью C и набор из N предметов. Каждый элемент имеет вес w[i] и значение v[i]. Ваша задача — упаковать рюкзак самыми ценными предметами. Другими словами, вы должны максимизировать общую ценность предметов, которые вы кладете в предмет-рюкзак, с ограничением: общий вес взятых предметов не может превышать вместимость рюкзака.

Как только вы получите решение, общий вес предметов в рюкзаке будет называться «весом решения», а их общая стоимость — «значением решения».

Проблема имеет множество практических приложений. Например, вы решили инвестировать 1600 долларов в знаменитые акции FAANG (общее название акций Facebook, Amazon, Apple, Netflix и Google, также известных как Alphabet). Каждая акция имеет текущую рыночную цену и оценку цены на один год. По состоянию на один день 2018 года они были следующими:

========= ======= ======= =========Company   Ticker  Price   Estimate========= ======= ======= =========Alphabet  GOOG    1030    1330Amazon    AMZN    1573    1675Apple     AAPL    162     193 Facebook  FB      174     216 Netflix   NFLX    312     327========= ======= ======= =========

Для простоты примера предположим, что вы никогда не клали все яйца в одну корзину. Вы готовы купить не более одной акции каждой акции. Какие акции вы покупаете, чтобы максимизировать свою прибыль?

Это проблема с рюкзаком. Ваш бюджет (1600 долларов США) — это вместимость мешка (C). Акции — это предметы, которые необходимо упаковать. Текущие цены представляют собой веса (w). Ориентировочные цены представляют собой значения. Проблема выглядит тривиальной. Однако решение неочевидно на первый взгляд: стоит ли вам покупать одну акцию Amazon или одну акцию Google плюс по одной акции Apple, Facebook или Netflix.

Конечно, в этом случае вы можете быстро посчитать вручную и прийти к решению: вам следует купить Google, Netflix и Facebook. Таким образом, вы потратите 1516 долларов и ожидаете получить 1873 доллара.

Теперь вы верите, что открыли Клондайк. Вы разбиваете свою копилку и получаете 10 000 долларов. Несмотря на ваше волнение, вы остаетесь непреклонны в правиле «одна акция — одна покупка». Следовательно, с таким большим бюджетом вам придется расширить свои возможности. Вы решаете рассмотреть все акции из списка NASDAQ 100 в качестве кандидатов на покупку.

Будущее никогда не было таким ярким, но вдруг вы понимаете, что для того, чтобы определить свой идеальный инвестиционный портфель, вам придется проверить около 2¹⁰⁰ комбинаций. Даже если вы очень оптимистичны в отношении неизбежности и повсеместного распространения цифровой экономики, любая экономика требует — как минимум — вселенной, в которой она работает. К сожалению, через несколько триллионов лет, когда ваши вычисления закончатся, наша Вселенная, вероятно, перестанет существовать.

Алгоритм динамического программирования

Нам нужно отказаться от подхода грубой силы и запрограммировать какое-нибудь умное решение. Небольшие задачи о рюкзаке (а наша, хотите верьте, хотите нет, небольшая) решаются с помощью динамического программирования. Основная идея состоит в том, чтобы начать с тривиальной проблемы, решение которой нам известно, а затем постепенно усложнять ее.

Если следующие объяснения кажутся вам слишком абстрактными, вот аннотированная иллюстрация решения очень маленькой задачи о рюкзаке. Это поможет вам визуализировать происходящее.

Предположим, что по первым i элементам коллекции мы знаем значения решения s(i, k) для всех вместимости ранца k в диапазон от 0 до C.

Другими словами, мы шили С+1 «вспомогательные» рюкзаки всех размеров от 0 до С. Затем мы отсортировали нашу коллекцию, взяли первый элемент i и временно отложили все остальные. И теперь мы предполагаем, что каким-то волшебством мы знаем, как оптимально упаковать каждый из мешков из этого рабочего набора предметов i. Предметы, которые мы выбираем из рабочего набора, могут быть разными для разных мешков, но на данный момент нас не интересует, какие предметы мы берем, а какие пропускаем. Это только значение решения s(i, k), которое мы записываем для каждого из наших вновь сшитых мешков.

Теперь мы извлекаем следующий, (i+1)-й элемент из коллекции и добавляем его в рабочий набор. Давайте найдем значения решения для всех вспомогательных рюкзаков с помощью этого нового рабочего набора. Другими словами, мы находим s(i+1, k) для всех k=0..C, учитывая s(i, k) .

Если k меньше веса нового предмета w[i+1], мы не можем взять этот предмет. Действительно, даже если бы мы взяли только этот предмет, он сам по себе не поместился бы в рюкзак. Следовательно, s(i+1, k)=s(i, k) для всех k < w[i+1].

Для значений k >= w[i+1] нам предстоит сделать выбор: либо берем новый предмет в рюкзак емкостиty k, или мы пропустим это. Нам нужно оценить эти два варианта, чтобы определить, какой из них принесет нам больше пользы.

Если мы возьмем (i+1)-й предмет, мы получим значение v[i+1] и израсходуем часть вместимости рюкзака, чтобы вместить вес w[i+1]. Это оставляет нам емкость k–w[i+1] , которую мы должны оптимально заполнить, используя (некоторые из) первых элементов i. . Это оптимальное заполнение имеет значение решения s(i, k–w[i+1]). Это число нам уже известно, поскольку по предположению мы знаем все значения решения для рабочего набора элементов i. Следовательно, значение возможного решения для рюкзака k с взятым предметом i+1 будет равно
s(i+1, k | i+1 взято)=v[i+1] + s(i, k–w[i+1]).

Другой вариант — пропустить элемент i+1. В этом случае в нашем рюкзаке ничего не изменится, и значение возможного решения будет таким же, как s(i, k).

Чтобы принять решение о лучшем выборе, мы сравниваем двух кандидатов на значения решения:
s(i+1, k | i+1 взято)=v[i+1] + s(i, k–w [i+1])
s(i+1, k | i+1 пропущено)=s(i, k)

Максимум из них становится решением s(i+1, k).

В итоге:

if k < w[i+1]:    s(i+1, k) = s(i, k)else:    s(i+1, k) = max( v[i+1] + s(i, k-w[i+1]), s(i, k) )

Теперь мы можем шаг за шагом решить задачу о рюкзаке. Начнем с пустого рабочего набора (i=0). Очевидно, s(0, k)=0 для любого k. Затем мы добавляем элементы в рабочий набор и находим значения решения s(i, k), пока не придем к s(i+1=N, k=C). > что является значением решения исходной задачи.

Обратите внимание, что таким образом мы построили сетку значений NxC .

Однако, несмотря на то, что мы узнали ценность решения, мы не знаем точно, какие предметы были взяты в рюкзак. Чтобы выяснить это, мы вернемся к сетке. Начиная с s(i=N, k=C), мы сравниваем s(i, k) с s(i–1, k).

Если s(i, k)=s(i–1, k), i-й элемент не был взят. Мы повторяем операцию с i=i–1, сохраняя значение k неизменным. В противном случае i-й предмет будет взят, и для следующего шага проверки мы уменьшим рюкзак на w[i] — мы установили i=i– 1, k=k–w[i].

Таким образом мы осматриваем все предметы с N-го по первый и определяем, какие из них положили в рюкзак. Это дает нам решение задачи о рюкзаке.

Код и анализ

Теперь, когда у нас есть алгоритм, сравним несколько реализаций, начиная с простой. Код доступен на GitHub.

Данные представляют собой список Nasdaq 100, содержащий текущие цены и оценки цен на сто акций (по состоянию на один день 2018 года). Наш инвестиционный бюджет составляет 10 000 долларов США.

Напомним, что цены на акции выражаются не в круглых долларах, а в центах. Поэтому, чтобы получить точное решение, нам нужно все считать в центах — мы определенно хотим избежать чисел с плавающей запятой. Следовательно, вместимость нашего рюкзака равна ($)10000 x 100 центов=($)1000000, а общий размер нашей задачи N x C=1 000 000.

Поскольку целое число занимает 4 байта памяти, мы ожидаем, что алгоритм будет использовать примерно 400 МБ ОЗУ. Таким образом, память не будет ограничением. Нам следует заботиться о времени выполнения.

Конечно, все наши реализации дадут одно и то же решение. Для справки: инвестиции (вес решения) составляют 999930 (9999,30 долларов США), а ожидаемый доход (значение решения) составляет 1219475 (12194,75 долларов США). Список акций для покупки довольно длинный (80 из 100 позиций). Вы можете получить его, запустив код.

И помните, что это упражнение по программированию, а не инвестиционный совет. К тому времени, как вы прочитаете эту статью, цены и оценки изменятся по сравнению с тем, что используется здесь в качестве примера.

Обычные старые циклы «for»

Ниже представлена непосредственная реализация алгоритма.

Есть две части.

В первой части (строки 3–7 выше) два вложенных цикла for используются для построения сетки решений.

Внешний цикл добавляет элементы в рабочий набор до тех пор, пока мы не достигнем N (значение N передается в параметр items). Строка значений решения для каждого нового рабочего набора инициализируется значениями, вычисленными для предыдущего рабочего набора.

Внутренний цикл для каждого рабочего набора перебирает значения k от веса вновь добавленного item до C (значение C передается в параметре capacity).

Обратите внимание, что нам не нужно начинать цикл с k=0. Когда k меньше веса item, значения решения всегда совпадают со значениями, вычисленными для предыдущего рабочего набора. , и эти числа уже были скопированы в текущую строку при инициализации.

Когда циклы завершены, у нас есть сетка решения и значение решения.

Вторая часть (строки 9–17) представляет собой одиночный цикл for из N итераций. Он возвращается к сетке, чтобы определить, какие предметы были взяты в рюкзак.

Далее мы сосредоточимся исключительно на первой части алгоритма, поскольку она имеет O(N*C) временную и пространственную сложность. Часть обратного отслеживания требует всего O(N) времени и не затрачивает никакой дополнительной памяти — ее потребление ресурсов относительно незначительно.

Простая реализация решения проблемы с рюкзаком Nasdaq 100 на моем компьютере занимает 180 секунд.

Насколько плохо? С одной стороны, при нынешних скоростях мы не привыкли тратить три минуты на ожидание, пока компьютер что-то сделает. С другой стороны, размер проблемы — сто миллионов — выглядит действительно устрашающе, так что, может быть, три минуты — это нормально?

Чтобы получить некий эталон, давайте запрограммируем тот же алгоритм на другом языке. Нам нужен статически типизированный компилируемый язык, чтобы обеспечить скорость вычислений. Нет, не C. Это не фантастика. Будем придерживаться моды и напишем на Go:

Как видите, код Go очень похож на код Python. Я даже скопировал одну строчку, самую длинную, как есть.

Каково время работы? 400 миллисекунд! Другими словами, Python оказался в 500 раз медленнее Go. Разрыв, вероятно, будет еще больше, если мы попробуем это на C. Это определенно катастрофа для Python.

Чтобы узнать, что замедляет работу кода Python, давайте запустим его с помощью профилировщика строк. Вы можете найти выходные данные профилировщика для этой и последующих реализаций алгоритма на GitHub.

В прямом решателе 99,7% времени работы тратится на две строки. Эти две строки составляют внутренний цикл, который выполняется 98 миллионов раз:

Прошу прощения за слишком длинные строки, но профилировщик строк не может правильно обрабатывать разрывы строк в одном операторе.

Я слышал, что оператор for в Python работает медленно, но, что интересно, большая часть времени тратится не на строку for, а на тело цикла.

Мы можем разбить тело цикла на отдельные операции, чтобы увидеть, не является ли какая-либо конкретная операция слишком медленной:

Похоже, что ни одна конкретная операция не выделяется. Время выполнения отдельных операций внутри внутреннего цикла практически такое же, как и время выполнения аналогичных операций в других частях кода.

Обратите внимание, как разрушение кода увеличило общее время работы. Внутренний цикл теперь занимает 99,9% времени работы. Чем тупее ваш код Python, тем медленнее он становится. Интересно, не так ли?

Встроенная функция карты

Давайте сделаем код более оптимизированным и заменим внутренний цикл for встроенной функцией map():

Время выполнения этого кода составляет 102 секунды, что на 78 секунд меньше, чем у простой реализации. Действительно, map() работает заметно, но не намного быстрее.

Понимание списка

Возможно, вы заметили, что при каждом запуске внутреннего цикла создается список (который добавляется в сетку решений как новая строка). Pythonic способ создания списков — это, конечно, понимание списков. Давайте попробуем это вместо map().

Это завершилось через 81 секунду. Мы добились еще одного улучшения и сократили время работы вдвое по сравнению с простой реализацией (180 секунд). Вне контекста это можно было бы расценить как значительный прогресс. Увы, мы все еще находимся на расстоянии нескольких световых лет от нашей контрольной отметки в 0,4 секунды.

Массивы NumPy

Наконец-то мы исчерпали встроенные инструменты Python. Да, я слышу рев публики, скандирующей «NumPy! НумПи! Но чтобы оценить эффективность NumPy, нам следовало бы поместить его в контекст, предварительно попробовав for, map() и понимание списка.

Хорошо, теперь пришло время NumPy. Итак, отказываемся от списков и помещаем наши данные в numpy-массивы:

Внезапно результат разочаровывает. Этот код работает в 1,5 раза медленнее, чем стандартный алгоритм распознавания списков (123 секунды против 81 секунды). Как это может быть?

Давайте рассмотрим профили линий для обоих решателей.

Инициализация grid[0] как массива numpy (строка 274) происходит в три раза быстрее, чем если бы это был список Python (строка 245). Внутри внешнего цикла инициализация grid[item+1] для массива NumPy (строка 276) происходит в 4,5 раза быстрее, чем для списка (строка 248). Все идет нормально.

Однако выполнение строки 279 в 1,5 раза медленнее, чем ее аналог без numpy в строке 252. Проблема в том, что понимание списка создает список значений, но мы сохраняем эти значения в Массив NumPy, который находится в левой части выражения. Следовательно, эта строка неявно добавляет затраты на преобразование списка в массив NumPy. Поскольку на строку 279 приходится 99,9% времени выполнения, все ранее отмеченные преимущества numpy становятся незначительными.

Но нам по-прежнему нужны средства для перебора массивов для выполнения вычислений. Мы уже узнали, что понимание списков — самый быстрый инструмент итерации. (Кстати, если вы попытаетесь построить массивы NumPy в старом простом цикле for, избегая преобразования списка в массив NumPy, вы получите колоссальные 295 секунд времени выполнения.) Итак, мы застряли, и NumPy бесполезен? Конечно, нет.

Правильное использование NumPy

Простое хранение данных в массивах NumPy не поможет. Настоящая мощь NumPy заключается в функциях, которые выполняют вычисления над массивами NumPy. Они принимают массивы в качестве параметров и возвращают массивы в качестве результатов.

Например, есть функция where(), которая принимает в качестве параметров три массива: condition, x и y. и возвращает массив, построенный путем выбора элементов из x или из y. Первый параметр, condition, представляет собой массив логических значений. Он сообщает, из чего выбирать: если элемент condition оценивается как True, соответствующий элемент x отправляется на выход, в противном случае берется элемент из y.

Обратите внимание, что функция NumPy делает все это за один вызов. Циклическое перебор массивов убран под капот.

Вот как мы используем where() вместо внутреннего цикла for в первом решателе или, соответственно, в понимании списка последнего:

Есть три интересных фрагмента кода: строка 8, строка 9 и строки 10–13, пронумерованные выше. Вместе они заменяют внутренний цикл, который перебирает рюкзаки всех возможных размеров, чтобы найти значения решения.

Пока вместимость рюкзака не достигнет веса вновь добавленного в рабочий набор предмета (this_weight), мы должны игнорировать этот предмет и устанавливать значения решения равными значениям из предыдущего рабочего набора. Это довольно просто (строка 8):

grid[item+1, :this_weight] = grid[item, :this_weight]

Затем строим вспомогательный массив temp (строка 9):

temp = grid[item, :-this_weight] + this_value

Этот код аналогичен, но намного быстрее:

[grid[item, k — this_weight] + this_value  for k in range(this_weight, capacity+1)]

Он вычисляет возможные значения решения, если новый предмет будет помещен в каждый из рюкзаков, в которых может поместиться этот предмет.

Обратите внимание, как массив temp создается путем добавления к массиву скаляра . Это еще одна мощная функция NumPy, называемая «трансляция». Когда NumPy видит операнды с разными размерностями, он пытается расширить (то есть «транслировать») низкоразмерный операнд, чтобы он соответствовал размерам другого. В нашем случае скаляр расширяется до массива того же размера, что и grid[item, :-this_weight], и эти два массива складываются вместе. В результате значение this_value добавляется к каждому элементу grid[item, :-this_weight] — цикл не требуется.

В следующем фрагменте (строки 10–13) мы используем функцию where(), которая делает именно то, что требуется алгоритму: сравнивает два возможных значения решения для каждого размера рюкзака и выбирает тот, который больше.

grid[item + 1, this_weight:] =                 np.where(temp > grid[item, this_weight:],             temp,             grid[item, this_weight:])

Сравнение осуществляется с помощью параметра condition, который рассчитывается как temp >grid[item, this_weight:]. Это поэлементная операция, которая создает массив логических значений, по одному для каждого размера вспомогательного рюкзака. Значение True означает, что соответствующий элемент должен быть упаковал в рюкзак. Следовательно, значение решения, взятое из массива, является вторым аргументом функции n, temp. В противном случае элемент пропускается, а значение решения копируется из предыдущей строки сетки — третьего аргумента функции the wheree() .

Наконец включился варп-двигатель! Этот решатель выполняется за 0,55 секунды. Это в 145 раз быстрее, чем решатель на основе списка, и в 329 раз быстрее, чем код, использующий цикл for. Хотя мы и не обогнали решатель, написанный на Go (0,4 секунды), но подошли к нему достаточно близко.

Некоторые петли должны остаться

Подождите, а как насчет внешнего цикла for?

В нашем примере код внешнего цикла, который не является частью внутреннего цикла, выполняется только 100 раз, поэтому мы можем обойтись без каких-либо изменений. Однако в других случаях внешний цикл может оказаться такой же длины, как внутренний.

Можем ли мы переписать внешний цикл с помощью функции NumPy так же, как мы сделали с внутренним циклом? Ответ - нет.

Несмотря на то, что оба цикла являются циклами for, внешний и внутренний циклы совершенно различаются по тому, что они делают.

Внутренний цикл создает одномерный массив на основе другого одномерного массива, элементы которого все известны на момент запуска цикла. Именно эта предварительная доступность входных данных позволила нам заменить внутренний цикл на map(), понимание списка или функцию NumPy.

Внешний цикл создает 2D-массив из 1D-массивов, элементы которого не известны на момент запуска цикла. Более того, эти массивы компонентов вычисляются с помощью рекурсивного алгоритма: мы можем найти элементы (i+1)-го массива только после того, как мы нашли i-й.

Предположим, что внешний цикл может быть представлен как функция:
grid=g(row0, row1, … rowN)
Все параметры функции должны быть вычислены до вызова функции, но только row0 известен заранее. Поскольку вычисление (i+1)-й строки зависит от доступности i-й, нам нужен цикл, идущий от 1 к N для вычисления всех параметров row . Следовательно, чтобы заменить внешний цикл функцией, нам нужен еще один цикл, который оценивает параметры этой функции. Этот другой цикл — это именно тот цикл, который мы пытаемся заменить.

Другой способ избежать внешнего цикла for — использовать рекурсию. Можно легко написать рекурсивную функцию calculate(i), которая создает i-ю строку сетки. Чтобы выполнить задание, функции необходимо знать (i-1)-ю строку, поэтому она вызывает себя как calculate(i-1), а затем вычисляет i-я строка, используя функции NumPy, как мы делали раньше. Затем весь внешний цикл можно заменить на calculate(N). Для полноты картины в исходном коде этой статьи на GitHub можно найти рекурсивный решатель ранцев.

Однако рекурсивный подход явно не масштабируем. Python не оптимизирован по хвосту. Глубина стека рекурсии по умолчанию ограничена примерно тысячей. Этот предел, безусловно, консервативен, но когда нам требуется глубина в миллионы, весьма вероятно переполнение стека. Более того, эксперимент показывает, что рекурсия даже не дает преимущества в производительности по сравнению с решателем на основе NumPy с внешним циклом for.

Именно здесь у нас заканчиваются инструменты, предоставляемые Python и его библиотеками (насколько мне известно). Если вам абсолютно необходимо ускорить цикл, реализующий рекурсивный алгоритм, вам придется прибегнуть к Cython, JIT-скомпилированной версии Python или другому языку.

Вынос

  • Выполняйте численные вычисления с помощью функций NumPy. Они на два порядка быстрее встроенных инструментов Python.
  • Из встроенных инструментов Python понимание списка выполняется быстрее, чем map() , что значительно быстрее, чем for.
  • Для глубоко рекурсивных алгоритмов циклы более эффективны, чем рекурсивные вызовы функций.
  • Вы не можете заменить рекурсивные циклы map(), пониманием списка или функцией NumPy.
  • «Тупой» код (разбитый на элементарные операции) — самый медленный. Используйте встроенные функции и инструменты.

Статьи по данной тематике