Справляемся с нехваткой открытых данных для обучения ML-моделей

Привет! Меня зовут Илларион, я аспирант ИТМО и член команды, которая занимается предсказанием временных рядов, порожденных графовыми структурами, и другими исследованиями. Однако в разработке новых методов для прогнозирования есть существенное препятствие — нехватка открытых данных для обучения и тестирования моделей. Для решения проблемы мы создали открытый инструмент Time Series Generator. Я рассказал о нем на митапе, посвященном open source-разработке для научных задач.

О задаче

Динамические графы можно использовать при прогнозировании временных рядов дорожного и интернет-трафика или нагрузки на микросервисы в приложении. Справедливо ожидать, что объекты со связями будут генерировать схожие временные ряды.

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

Показания датчика 3 в основном зависят от времени суток, а предыдущие данные менее важны для предсказания значений ряда. В пространственно скоррелированных датчиках 1 и 2, расположенных на одном перекрестке, величина трафика должна быть схожей в каждый момент времени. Но степень связи оказывается различной в зависимости от фазы светофора.

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

Даже существующие наборы, такие как PEMS-BAY и METR-LA, не соответствуют всем требованиям. Они содержат только статические графы без добавления вершин и ребер (пример на рисунке ниже). Однако предсказание рядов для только что появившегося объекта — это важная способность предсказательных моделей.

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

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

Что нам не подошло

Для генерации временных рядов существуют готовые решения. К примеру, Python-библиотеки timeseries-generator и mockseries. Они позволяют создавать синтетические временные данные, но в них нет инструментов для разработки набора коррелирующих рядов. Чтобы применить библиотеки к решению задачи, пришлось бы сильно расширить их исходную функциональность. 

В частности, пакет timeseries-generator не позволяет параметризировать стохастические процессы с помощью данных извне. Для генерации используется только одна модель, заданная пользователем. Это означает, что исследование полученных синтетических временных рядов будет достаточно тривиальной задачей, так как отсутствуют точки изменения характеристик.

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

Кроме того, я изучил статью «Correlated synthetic time series generation for energy system simulations using Fourier and ARMA signal processing». В ней для генерации временных рядов используют ряды Фурье и ARMA-модели, обученные на реальных измерениях. Описанный метод предлагает способ получения коррелирующих временных данных на основе входного датасета с корреляцией.  Работа наиболее близка к нашей задаче, однако все сгенерированные таким методом ряды будут связаны. Мы же хотим контролировать степень их близости.

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

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

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

Что мы сделали

После долгого изучения статей и программных реализаций мы утвердили список желаемых функций и архитектуру приложения. Две ключевых части генератора — модель генерации временных рядов и модуль сэмплирования исходных данных. 

Метод, который мы разрабатываем, включает в себя набор разных процессов: белый шум, случайное блуждание, авторегрессионные модели (значения временного ряда линейно зависят от предыдущих) и т. д. Идеи реализации авторегрессионных процессов для генерации временных значений взяли из пакетов timeseries-generator и mockseries.

У всех реализаций процессов общий интерфейс. Это упрощает увеличение их числа. Затем мы реализовали комплексную модель временного ряда — «расписание»: взяли последовательность случайных процессов и распределили между ними количество наблюдений. 

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

В качестве исходных данных, которые порождают временные ряды, мы решили взять случайные точки на произвольной поверхности второго порядка. Их характеристиками выступают координаты — то есть, чем ближе точки в пространстве, тем сильнее временные ряды коррелируют. Расстояние между здесь не обязательно должно быть евклидовым: для нашей задачи также полезно считать геодезическое расстояние (расстояние на поверхности).

Сейчас мы реализовали частный случай, генерацию точек на единичной сфере с центром в начале координат. Для определения близости точек использовали k-means кластеризацию. Она позволила генерировать для каждого кластера собственное случайное расписание. Все объекты порождают временные данные, используя распорядок своего кластера. 

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

Как мы хотим это использовать

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

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

Сlass RWParametersGenerator(ParametersGenerator):

        def __init__(
        self,
        lag: int,
        linspace_info: LinspaceInfo,
        parameters_generation_method: ParametersGenerationMethod,
        parameters_required: list[ParameterType],
        init_values_coeff: float = 0.5,
    ) -> None:
        super().__init__(
            lag=lag,
            linspace_info=linspace_info,
            parameters_generation_method=parameters_generation_method,
            parameters_required=parameters_required,
        )
        self.init_values_coeff = init_values_coeff

    def generate_parameters(
        self, source_data: NDArray | None = None
    ) -> NDArrayFloat64T:
        return self.parameters_generation_method.generate_all_parameters(
            parameters_required=self.parameters_required,
            source_data=source_data,
        )

    def generate_init_values(
        self, source_data: NDArray | None = None
    ) -> NDArrayFloat64T:
        return np.array(
            [
                self.parameters_generation_method.get_mean_value(source_data)
                * self.init_values_coeff
            ]
        )

Далее определяется класс процесса с авторегрессионной формой генерации временного ряда.

    class RandomWalk(Process):
    def __init__(
        self,
        linspace_info: LinspaceInfo,
        parameters_generation_method: ParametersGenerationMethod,
        init_values_coeff: float = 0.5,
    ) -> None:
        super().__init__(
            linspace_info=linspace_info,
            parameters_generation_method=parameters_generation_method,
        )
        self._parameters_generator = RWParametersGenerator(
            lag=self.lag,
            linspace_info=self.linspace_info,
            parameters_generation_method=parameters_generation_method,
            parameters_required=self.parameters,
            init_values_coeff=init_values_coeff,
        )

    def generate_time_series(
        self,
        data: tuple[int, NDArrayFloat64T],
        previous_values: NDArrayFloat64T | None = None,
        source_data: NDArrayFloat64T | None = None,
    ) -> tuple[TimeSeries, dict]:
        values = np.array([0.0 for _ in range(0, data[0])])
        values_to_add = data[0]
        if previous_values is None:
            values[0] = self.parameters_generator.generate_init_values(
                source_data=source_data
            )[0]
            previous_value = values[0]
            values_to_add -= 1
        else:
            previous_value = previous_values[-1]
        for i in range(data[0] - values_to_add, data[0]):
            previous_value += np.random.normal(0.0, data[1][0])
            values[i] = previous_value
        rw_time_series = TimeSeries(data[0])
        rw_time_series.add_values(values, (self.name, data))
        if previous_values is None:
            return rw_time_series, self.get_info(
                data, np.array(values[0 : data[0] - values_to_add])
            )
        else:
           return rw_time_series, self.get_info(data, np.array([previous_values[-1]]))

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

Результаты создания нового процесса для симуляции реальных данных.

Используемые модели — графовые нейросети для динамических данных: torch geometric temporal data, torch geometric temporal, tsl. Кроме того, в команде разрабатывается собственное решение для моделирования графов произвольного размера в дискретные моменты времени.

Нейросетевая модель предобучается на искусственных данных. Далее — дообучается на реальных для достижения большей точности. Впоследствии модель можно использовать  для предсказания временных рядов и на других графах (при условии нахождения в той же предметной области).

Заключение

Наша команда исследовала существующие решения для генерации временных рядов и разработала собственное приложение на Python. Оно позволяет успешно создавать синтетические ряды с пространственными взаимосвязями. Сейчас планируем расширить список фичей. В числе задач на будущее:

• добавить разные способы создания исходных данных (например, сэмплирование точек на произвольных поверхностях, генерацию случайного графа);
• внедрить функции для аппроксимации геодезических расстояний на поверхностях;
• реализовать методы изменения параметров процессов в некоторые моменты времени;
• применить результаты генерации для предобучения нейросетевой модели предсказания временных рядов.