Анализ временных рядов и прогнозирование: от сбора данных до развертывания с использованием Flask и Docker…

Анализ временных рядов и прогнозирование: от сбора данных до развертывания с использованием Flask и Docker (часть II)

Практическое руководство для начинающих по комплексному проекту машинного обучения по прогнозированию временных рядов.

В Части 1 вы узнали, как анализировать проблему, собирать для нее данные и выполнять важные визуализации для понимания данных.

Во второй части мы узнаем о следующих (и более интересных) вещах.

Создание модели и обучение
Оценка модели, настройка и улучшение
Создание API с помощью FLASK
Докеризация приложения Flask
Результаты обучения

1. Создание и обучение модели

Существует множество вариантов модели временных рядов. ARIMA и SARIMAX являются наиболее популярным выбором для моделей авторегрессии, а FBProphet (или Prophet только в более новых версиях) основаны на аддитивных моделях, но также имеют хорошую поддержку нелинейных тенденций через сезонность, пользовательская или встроенная, а также поддержка в праздничные дни.

В этом руководстве мы будем использовать fbprophet, так как он прост и удобен в использовании по сравнению с авторегрессионными моделями, такими как ARIMA или SARIMAX, которые требуют много настроек гиперпараметров и статистического тестирования перед подбором модели, а FBProphet хорошо работает с данными. с сезонными эффектами и сильными сессиями исторических данных (Ссылка на документацию).

Установка

Обратите внимание, что FBProphet — это старое имя, а Prophet — более новое имя библиотеки. Если вы устанавливаете fbprophet, то убедитесь, что используете fbprophet везде, а если вы устанавливаете Prophet, то убедитесь, что используете Prophet везде, иначе вы потратите день или два на отладку, как я.

В этой статье мы будем использовать fbprophet.

Если вы используете ОС Windows, я настоятельно рекомендую использовать для установки Conda-Forge вместо PIP.

$ conda activate time_series
(time_series) $ conda install -c conda-forge fbprophet

Если вы работаете в системе на базе Linux или macOS, вы можете использовать pip для установки fbprophet.

$ conda activate time_series
(time_series) $ pip install fbprophet

Вы можете подтвердить установку с помощью простого сеанса REPL.

(time_series) $ python
>>> import fbprophet

Если это работает нормально, это означает, что ваша установка в полном порядке. Давайте теперь создадим новый блокнот Jupyter в каталоге ноутбуков с именем training.ipynb.

Предварительная обработка

В FBProphet есть странное правило (это странно? Или мне одному так кажется?), то есть ваш фрейм данных должен иметь 2 столбца с именами "ds" для отметки даты и «y» для значения на этой конкретной метке даты. И "ds" должен быть столбцом (а не индексом).

Итак, давайте напишем код, чтобы получить кадр данных в нужном формате.

# renaming for fbprophet
df.rename_axis('ds', inplace=True)
df.rename(columns={'US dollar':'y'}, inplace=True)
df.reset_index(inplace=True)
df.head()

Давайте создадим простую модель с параметрами по умолчанию и подгоним ее.

from fbprophet import Prophet 
prophet_model = Prophet()
prophet_model.fit(df)

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

Мы можем проверить, как модель работает в будущем, создав кадр данных будущих дат. В FBProphet эта опция встроена.

future_dataset= prophet_model.make_future_dataframe(periods=15, freq='y') # Next 15 YEARS OF DATA
future_dataset.tail()

Теперь мы можем выполнять прогнозы, используя нашу модель.

pred = prophet_model.predict(future_dataset)
pred[['ds','yhat', 'yhat_lower', 'yhat_upper']].head() # only useful columns

В предсказаниях FBProphet много колонок, но мы будем заниматься только теми, которые нам сейчас пригодятся. Здесь мы выбираем 4 столбца: прогнозируемое значение (yhat), верхний диапазон прогнозируемого значения (yhat_upper) и нижний диапазон прогнозируемых значений (yhat_lower) и дату.

Мы можем просто вызвать функцию графика модели, чтобы увидеть поведение прогноза в будущем.

prophet_model.plot(pred);

Мы видим, что модель работает посредственно на наблюдаемом наборе данных, и это просто прямая линия с 2012 года, что не кажется хорошим представлением нашего набора данных.

2. Оценка модели, настройка гиперпараметров и улучшение

Как мы видели в предыдущих разделах, среднее значение за долгий год действительно имеет некоторую закономерность или, по крайней мере, оно не очень неравномерно. Таким образом, мы можем добавить пользовательскую сезонность на 10 или 15 лет, чтобы увидеть, как работает наша модель.

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

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

Порядок Фурье по умолчанию для годовой сезонности равен 10 в соответствии с документацией FBProphet.

Итак, если мы собираемся добавить пользовательскую сезонность 10 лет, порядок Фурье должен быть 100, а для 15 лет он должен быть 150. Поскольку мы видели на графиках выше, что наши данные не имеют быстро меняющейся сезонности, поэтому мы собираемся использовать более низкое значение, чем рассчитанное, чтобы наша модель не была переоснащена.

Давайте создадим несколько вспомогательных функций, чтобы мы могли быстро изменить параметры и увидеть результаты модели.

def fb_prophet_function(data, future_years, seasonality_name, seasonality_val,seasonality_fourier, **params):
    """
    Trains a fb prophet model on given hyperparameters and custom
    seasonality, predicts on future dataset, plot the results and
    return the model.
    """
    start= time.time()
    prophet_model = Prophet(**params)
    
    prophet_model.add_seasonality(name=seasonality_name, period=seasonality_val, fourier_order=seasonality_fourier)
        
    prophet_model.fit(data)
    
    future_dataset = prophet_model.make_future_dataframe(periods=future_years, freq='y')
    
    pred = prophet_model.predict(future_dataset)
    
    prophet_model.plot(pred, figsize=(15,7));
    plt.ylim(-500, 3000)
    plt.title(f"fourier order{seasonality_fourier}, seasonality time {seasonality_name}")
    plt.show()
    
    end = time.time()
    print(f"Total Execution Time {end-start} seconds")
    return prophet_model

И вторая функция для построения прогнозов за последние несколько лет, заданных через параметр размера.

def plot_valid(validation_set, size, model):
    pred = model.predict(validation_set)
    temp = df[-size:].copy().reset_index()
    temp['pred']=pred['yhat']
    temp.set_index('ds')[['y', 'pred']].plot()
    plt.tight_layout();

Теперь давайте немного разделим проверочный набор, чтобы еще лучше оценить модель.

training_set = df[:-1000] 
validation_set = df[-1000:] #last 1000 rows, i.e from Jul 2018
# 15 years seasonlaity, additive, no other seasonality, less fourier value
fifteen_years = fb_prophet_function(data=training_set, future_years=6, seasonality_name='15_years', seasonality_val=365*15, seasonality_fourier=100,seasonality_mode='additive')

Значение сезонности определяет, какова продолжительность нашей сезонности. На 1 год это 365 дней, на 15 лет это 365*15 дней.

Сюжет на 15 лет нестандартной сезонности будет

Производительность на проверочном наборе будет следующей:

plot_valid(validation_set, 1000, fifteen_years)
plt.title("Hyp parameters: 15_years seasonality, seasonality_fourier=100, seasonality_mode=additive\n prediction from Jul2018-Apr2022(from training set i.e validation set)");

Давайте теперь попробуем с 10-летней индивидуальной сезонностью. Значение Фурье должно быть 100 (на основе 10 для одного года в качестве значения по умолчанию), но мы собираемся использовать 80, поскольку у нас нет сильно меняющегося графика сезонности за 10 лет.

# 10 years seasonlaity, no other seasonlaity, additive, less fourier
training_set = df[:-1000]
validation_set = df[-1000:]
ten_years_model = fb_prophet_function(data=training_set, future_years=6, seasonality_name='10_years', seasonality_val=365*10, seasonality_fourier=80,seasonality_mode='additive')

Сюжет:

Давайте построим прогнозы на проверочном наборе.

plot_valid(validation_set, 1000, ten_years_model)
plt.title("Hyp parameters: 10_years seasonality, seasonality_fourier=80, seasonality_mode=additive\n prediction from Jul2018-Apr2022(from training set i.e validation set)");

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

Точно так же есть и другие области улучшения модели, которые мы можем решить на основе наших визуализаций. Например, на первом графике мы видим, что тренд изменился после 2001 года, поэтому вместо того, чтобы вводить в модель полные данные, мы можем просто попробовать ввести в нее данные с 2001 года и далее с какой-либо другой подходящей пользовательской сезонностью и значением Фурье, но я оставил это как домашнее задание для вас!! Дайте мне знать в комментариях, если вы попробуете это.

Сохранение модели

Мы можем использовать библиотеку Pickle для сохранения модели, так как было бы также удобно загружать модель через pickle и выполнять на ней прогнозы.

import pickle
with open('../models/fbprophet.pckl', 'wb') as fout: # saving the model in models directory
    pickle.dump(ten_years_model, fout)

3. Создание API с помощью Flask

Flask — это микровеб-фреймворк, написанный на Python, который можно использовать для быстрого и простого создания веб-приложений и API-интерфейсов на Python.

Хватит использовать Jupyter Notebooks, давайте теперь перейдем к сценариям 🤗 для создания нашего API во Flask. Код будет структурирован таким образом, чтобы вы могли легко добавлять в него свои модели и сравнивать производительность всех моделей.

Создание служебных классов для нашего приложения

Давайте создадим несколько вспомогательных классов, которые помогут нам прогнозировать значения. Здесь мы будем следовать принципу открытого/закрытого дизайна, который гласит:

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

Это означает, что мы можем расширить наш класс, т. е. посредством наследования, без необходимости его модификации. Хороший способ добиться этого в Python — использовать Абстрактные классы. Давайте перейдем к файлу api/utils.py и выполним важные операции импорта.

import pickle
import datetime
from typing import List
import pandas as pd
from abc import ABC, abstractmethod

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

import pickle
import datetime
from typing import List
import pandas as pd
from abc import ABC, abstractmethod
class GoldPricePredictor(ABC):
    def __init__(self, model_name) -> None:
        """
        Loads the model from the given file
        """
        with open(f"models/{model_name}.pckl", "rb",) as fin:
            try:
                self.model = pickle.load(fin)
            except (OSError, FileNotFoundError, TypeError):
                print("wrong path/ model not available")
                exit(-1)
def calculate_next_date(self, prev_date):
        """
        Calculates next date
        date_format = yyyy-mm-dd
        """
        self.next_date = datetime.datetime(
            *list(map(lambda x: int(x), prev_date.split("-")))
        ) + datetime.timedelta(
            days=1
        )  # next date
def get_next_date(self, prev_date):
        try:
            return self.next_date.strftime("%y-%m-%d")
        except NameError:
            self.calculate_next_date(prev_date)
@abstractmethod
    def predict(self, prev_date) -> List:
        pass
@abstractmethod
    def preprocess_inputs(self, prev_date):
        pass
@abstractmethod
    def postprocess_outputs(self, output_from_model) -> List:
        pass

Мы создали несколько простых методов (которые говорят сами за себя) и 3 абстрактных метода, которые будут реализованы в базовых классах. Итак, если у нас есть 3 отдельные модели, то есть LSTM, FBProphet и Arima, мы собираемся создать 3 производных класса, каждый с 3 разными методами, основанными на требованиях их модели.

Давайте сейчас создадим класс модели FBProphet.

class FBProphetPredictor(GoldPricePredictor):
    def __init__(self,) -> None:
        """
        Load the Model from file models/fbprophet.pckl
        """
        super().__init__("fbprophet")
def preprocess_inputs(self, prev_date):
        """
        Model takes in an input as a pandas dataframe having index 
        as the day to be predicted
        """
        self.calculate_next_date(prev_date)  # get the self.next_date var
        next_date_series = pd.DataFrame(
            {"ds": pd.date_range(start=self.next_date, end=self.next_date)}
        )
        return next_date_series
def postprocess_outputs(self, output_from_model) -> List:
        """
        Return the yhat in the list format
        """
        return output_from_model["yhat"].tolist()
def predict(self, prev_date) -> List:
        next_date_series = self.preprocess_inputs(prev_date)  # preprocess inp
pred = self.model.predict(next_date_series)  # prediction
pred = self.postprocess_outputs(pred)  # postprocess prediction
        return pred  # return prediction

Теперь давайте заглянем внутрь кода. Как мы видели, модель fbprophet принимает входные данные, которые представляют собой фрейм данных со столбцом «ds» с прогнозируемой датой, поэтому мы создали фрейм данных, используя Pandas со следующей датой, указанной в preprocess_inputs функция. В функции postprocess_outputs мы взяли нужное значение из фрейма данных и преобразовали его в список. Функция predict принимает входные данные, которые являются датой, предшествующей дате, которую мы должны предсказать, мы предварительно обрабатываем этот ввод с помощью функции ввода предварительной обработки, получаем прогнозы из моделей и выполняем их постобработку.

Файл конфигурации для наших моделей

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

import utils
models = {
    "fbprophet": utils.FBProphetPredictor,
    "auto_arima": utils.ArimaPredictor,
}

Приложение Flask

Как мы все знаем, у каждого приложения flask есть файл app.py, в котором вы описываете все маршруты и прочее. Давайте создадим это.

import flask
from cfg import models as model_list
from flask import jsonify, render_template, request
models = model_list.models  # dict of all models from which to select
app = flask.Flask(__name__)
app.config["debug"] = True

Давайте теперь определим маршрут, который имеет базовый пользовательский интерфейс, чтобы мы могли быстро протестировать производительность модели.

@app.route("/")
def home():
    return render_template("index.html", models=list(models.keys()))

Этот файл index.html, возвращаемый через render_template, находится в каталоге api/templates. Добавим к нему немного HTML и JINJA.

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Home Page</title>
</head>
<body>
<h2>
        Available Models
    </h2>
<form action="{{ url_for('predict_gold') }}" method="post">
        <label for="model_val">Choose a model</label>
        <select id="model_val" name="model_name">
            {% for model in models %}
<option value="{{model}}">{{model}}</option>
            {% endfor %}
        </select>
<input type="text" name="date" , placeholder="DATE:YYYY-MM-DD">
        <input type="submit" value="Submit">
    </form>
</body>
</html>

Мы просто добавили форму, в которой вы можете выбрать модель среди всех доступных моделей из файла конфигурации, выбрать дату и через нее получить прогнозы. Простой HTML выглядит следующим образом при запуске сервера.

Теперь вернемся к api/app.py и добавим функциональность для прогнозов.

@app.route("/predict", methods=["POST"])
def predict_gold():
    """
    Given the date, predict the gold price for next date
    """
    model_name = request.form.get("model_name")
    date_given = request.form.get("date")
    model = models[model_name]() # get and initialize the model class from dictionary
    pred = model.predict(date_given)
return jsonify(
        {
            "given_date": date_given,
            "next_date": model.get_next_date(date_given),
            "price": pred,
        },
    )

Теперь вы можете протестировать API и маршрут прогнозирования для вашей модели, чтобы давать прогнозы в веб-приложении через пользовательский интерфейс.

Обработка других типов запросов с помощью нашего API

Как вам хорошо известно, большинство API используются через запросы либо с использованием Postman или Curl, либо через запросы в языке программирования, таком как модуль requests в Python.

Поэтому нам нужно убедиться, что наш API справляется с этим.

@app.route("/predict", methods=["POST"])
def predict_gold():
    """
    Given the date, predict the gold price for next date
    """
    # print(request.form.get)
    try:
        model_name = request.form.get("model_name")
        date_given = request.form.get("date")
        model = models[model_name]()
        pred = model.predict(date_given)
    except KeyError:  # get value from curl header
        model_name = request.headers.get("model_name")
        date_given = request.headers.get("date")
        model = models[model_name]()
        pred = model.predict(date_given)
return jsonify(
        {
            "given_date": date_given,
            "next_date": model.get_next_date(date_given),
            "price": pred,
        },
    )

Мы поймали исключение KeyError, которое сообщает нам, что указанные ключи (имя_модели, дата) недоступны в форме, поэтому они должны быть в заголовках, т.е. через CURL, Postman или связанный инструмент. Мы использовали request.headers.get из Flask, чтобы получить значение заголовков.

Теперь мы можем отправить запрос через curl, чтобы протестировать его.

$ curl -XPOST -H 'model_name: fbprophet' -H 'date: 2022-04-21' 'http://127.0.0.1:5000/predict'

Вы можете использовать модуль запросов Python для доступа к API.

import requests
response = requests.post('http://127.0.0.1:5000/predict', headers={'model_name':'fbprophet', 'date': '2022-04-21'})
print(response.json())

Выход

{'given_date': '2022-04-21', 'next_date': '22-04-22', 'price': [1922.3042464629968]}

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

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

if __name__ == "__main__":
    app.run(host="0.0.0.0", debug=True)

4. Докеризация приложения Flask

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

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

Требования.txt

Чтобы контейнеризировать ваше приложение, вам нужно убедиться, что при запуске вашего контейнера в нем установлены все необходимые пакеты. Наличие файла requirements.txt — лучший вариант в Python.

numpy
flask
wheel
pystan==2.19.1.1
fbprophet
scikit-learn
openpyxl
pandas
pmdarima
statsmodels
seaborn
matplotlib
gunicorn

Одна из ошибок, с которой я столкнулся, заключалась в том, что при установке fbprophet через требования.txt он не мог собраться через pip, так как предыдущие пакеты не были собраны в то время. Таким образом, хорошим решением была установка всех необходимых пакетов, которые ранее требовались для fbprophet. Вы можете найти prophet requirements.txt в оригинальном репозитории Facebook.

Кроме того, если вы используете fbprophet, обязательно добавьте его как fbprophet в файл требований, а если вы используете пророка, обязательно добавьте его как пророка в файл требований.

Я просто скопировал его содержимое и создал новый файл требований с именем requirements_proph.txt.

Для создания образа нам понадобится Dockerfile. Давайте создадим новый файл с именем Dockerfile (без расширения).

FROM python:3.7
RUN mkdir /app
WORKDIR /app
COPY . /app
RUN python -m pip install --upgrade pip setuptools wheel
RUN python -m pip install -r requirements_proph.txt
RUN python -m pip install -r requirements.txt
CMD ["python", "api/app.py"]

Давайте проанализируем это шаг за шагом.

Мы указали, что будем использовать образ Python:3.7 в качестве базового образа. Затем мы запустили команду mkdir, чтобы создать каталог с именем app, изменили рабочий каталог на app и скопировали все содержимое нашего локального каталога в /app в контейнере. Затем мы обновили пип и колесо и установили файл требований, который я скопировал из репозитория FBProphet, и наш файл requirements.txt. После этого я только что указал команду для запуска файла app.py (в котором уже есть app.run(), если __name__==”__main__”.

Теперь вы можете создать образ с помощью

$ sudo docker image build -t gold_price_prediction_api .

Теперь мы можем запустить образ докера с помощью следующих команд.

$ sudo docker run gold_price_prediction_api

Вот и все, теперь вы можете зайти на свой http://localhost:5000/ и увидеть, как там запускается ваш docker-контейнер. Теперь вы можете легко развернуть свой док-контейнер в любом месте, о чем мы поговорим в следующей статье.

5. Результаты обучения

Из этой статьи вы узнали следующее:

Моделирование временных рядов с помощью FBProphet
Пользовательская сезонность с FBProphet
Создание API с помощью Flask
Принцип ОТКРЫТОГО/ЗАКРЫТОГО дизайна в Python
Докеризация вашего приложения Flask для развертывания

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

Полный исходный код: Github

Надеюсь, вам понравилась статья, дайте мне знать в комментариях, что вы думаете об этом. Я надеюсь, что вы реализуете некоторые другие модели, такие как LSTM и ARIMA, и сообщите мне в комментариях, как они работают.

Подпишитесь на меня в Medium, чтобы узнать больше статей, и свяжитесь со мной в LinkedIn (в основном я активен в LinkedIn).

Анализ временных рядов и прогнозирование: от сбора данных до развертывания с использованием Flask и Docker…