Новый ЕМИСС 2.0 со сводными таблицами, API и погодой?

Привет, Хабр! Около года назад мне пришла странная идея: а что, если сделать новую версию ЕМИСС, хранилища российской статистики, чтобы наконец-то было удобно сводить данные. А то постоянно сопоставлять несколько показателей из множества Excel файлов – сущий ад. И вот уже год прошел с момента создания и написания первой версии и сайта, и статьи (недавно был небольшой пост).

Что нового теперь в созданном приложении StatKit?

Обновление поиска: теперь поиск понимает и семантику за счет векторизации и сортировки по косинусной близости:

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

Свободные таблицы теперь позволяют объединить несколько показателей (я для примера выбрал население и число преступлений в целом и среди несовершеннолетних)

по субъектам РФ:

по одному субъекту в динамике:

все субъекты в динамике:

И для меня менее интересный вид, как в ЕМИСС, для одного показателя:

И, наверное, самое интересное теперь появился API для доступа к данным. 

Как начать работу с API StatKit?

1. Установка зависимостей

Откройте Jupyter Notebook. Убедитесь, что у вас установлены библиотеки requests и pandas: 

# установка и импорт зависимостей !pip install requests pandas  import pandas as pd import requests

2. Подготовка авторизации

Замените username и token на свои учетные данные API:

# получить имя пользователя и токен можно по адресу: https://go.statkit.ru/profile username = "your_username"  # замените на Ваш username API token = "your_token"        # замените на Ваш токен доступа

3. Настройка заголовков запроса

Добавьте headers для авторизации:

headers = {     "X-API-User": username,              # username API     "Authorization": f"Bearer {token}",  # токен     "Content-Type": "application/json"   # Формат данных (JSON) }

Получим список всех статистических показателей из API, преобразуем их в таблицу pandas:

api_url = "https://api.alexstat.ru/api/v1/indicators"  try:     # Выполняем GET-запрос к API     response = requests.get(api_url, headers=headers)      data = None          # Проверяем статус ответа сервера     if response.status_code == 200:         data = response.json()         print("Данные успешно получены!")     else:         print(f"Ошибка запроса. Код состояния: {response.status_code}")         print(f"Текст ответа сервера: {response.text}")  except requests.exceptions.RequestException as e:     print(f"Произошла ошибка при выполнении запроса: {e}")  # Преобразуем полученные данные в DataFrame df = pd.DataFrame(data) # Конвертируем строковые даты в формат datetime df['add_date'] = pd.to_datetime(df['add_date']) df['lastmod_date'] = pd.to_datetime(df['lastmod_date']) df

Можно сузить поиск, указав ключевые слова в запросе. Найдем показатели по ID или ключевому слову доля:

# URL API для получения данных о показателе "доля" api_url = "https://api.alexstat.ru/api/v1/indicators/доля"  try:     # Отправляем GET-запрос к API с указанными заголовками     response = requests.get(api_url, headers=headers)      # Инициализируем переменную для хранения данных     data = None          # Проверяем код статуса ответа     if response.status_code == 200:         # Если запрос успешен (код 200), преобразуем ответ в JSON         data = response.json()         print("Успешный запрос!")     else:         # Если сервер вернул ошибку, выводим код статуса и текст ответа         print(f"Ошибка запроса. Код статуса: {response.status_code}")         print(f"Ответ сервера: {response.text}")  # Обрабатываем возможные исключения при выполнении запроса except requests.exceptions.RequestException as e:     print(f"Ошибка при выполнении запроса: {e}")  # Преобразуем полученные данные в DataFrame pandas df = pd.DataFrame(data) df

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

# URL API для получения данных показателя с ID = 100 api_url = "https://api.alexstat.ru/api/v1/get_indicator/100"  try:     # Отправляем GET-запрос к API с указанными заголовками     response = requests.get(api_url, headers=headers)      # Инициализируем переменную для хранения данных     data = None          # Проверяем статус ответа сервера     if response.status_code == 200:         # При успешном ответе преобразуем в JSON         data = response.json()         print("Данные успешно получены!")     else:         # Выводим информацию об ошибке, если статус не 200         print(f"Ошибка запроса. Код статуса: {response.status_code}")         print(f"Текст ответа: {response.text}")  # Обрабатываем возможные исключения при выполнении запроса except requests.exceptions.RequestException as e:     print(f"Ошибка при выполнении запроса: {e}")  # Преобразуем данные в pandas DataFrame df = pd.DataFrame(data)  # Преобразуем значения показателя в числовой формат (float) df['ObsValue'] = df['ObsValue'].astype(float)  df

Полный код и готовый Jupyter Notebook в интерактивном Colab размещены по ссылке: https://go.statkit.ru/api

Также там в коде по ссылке я предлагаю вариант, как из нескольких эндпоинтов API провести мини-исследование и построить регрессию: исследовать зависимость между показателями населения и преступности по субъектам в разрезе ОКАТО:

Найдем интересующие нас показатели для исследования:

# URL API для получения показателей по ключевому слову "численность" api_url = "https://api.alexstat.ru/api/v1/indicators/численность"  try:     # Отправляем GET-запрос к API с указанными заголовками     response = requests.get(api_url, headers=headers)      # Инициализируем переменную для хранения данных ответа     data = None      # Проверяем статус ответа сервера     if response.status_code == 200:         # При успешном ответе преобразуем в JSON         data = response.json()         print("Данные успешно получены!")     else:         # Выводим информацию об ошибке, если статус не 200         print(f"Ошибка запроса. Код статуса: {response.status_code}")         print(f"Текст ответа сервера: {response.text}")  # Обрабатываем возможные исключения при выполнении запроса except requests.exceptions.RequestException as e:     print(f"Произошла ошибка при выполнении запроса: {e}")  # Преобразуем полученные данные в DataFrame df = pd.DataFrame(data) df

Численность постоянного населения имеет ID 1278.

# URL API для получения показателей по ключевому слову "преступ" api_url = "https://api.alexstat.ru/api/v1/indicators/преступ"  try:     # Отправляем GET-запрос с заголовками     response = requests.get(api_url, headers=headers)      # Переменная для хранения данных     data = None      # Проверяем статус ответа     if response.status_code == 200:         # Если успешно, преобразуем JSON         data = response.json()         print("Успешно!")     else:         # Выводим ошибку, если статус не 200         print(f"Ошибка запроса. Код: {response.status_code}")         print(f"Ответ: {response.text}")  # Обрабатываем ошибки запроса except requests.exceptions.RequestException as e:     print(f"Ошибка запроса: {e}")  # Создаем DataFrame из полученных данных df = pd.DataFrame(data) df

Далее, возьмем доля лиц, совершивших преступления, с ID = 232. Получим данные по показателю численность населения:

# URL для получения показателя с ID 1278 (численность постоянного населения) api_url = "https://api.alexstat.ru/api/v1/get_indicator/1278"  try:     # Отправка GET-запроса к API с указанными заголовками     response = requests.get(api_url, headers=headers)      # Инициализация переменной для хранения данных     data = None      # Проверка статуса ответа сервера     if response.status_code == 200:         # Успешный ответ - преобразуем в JSON         data = response.json()         print("Данные успешно получены!")     else:         # Вывод информации об ошибке, если статус не 200         print(f"Ошибка запроса. Код статуса: {response.status_code}")         print(f"Текст ответа: {response.text}")  # Обработка возможных ошибок при выполнении запроса except requests.exceptions.RequestException as e:     print(f"Ошибка выполнения запроса: {e}")  # Создание DataFrame из полученных данных population = pd.DataFrame(data)  # Преобразование значений показателя в числовой формат float population['ObsValue'] = population['ObsValue'].astype(float)  population

Обратимся к справочникам ОКАТО (1) и Тип поселения (17):

# Находим все колонки, содержащие справочники (начинающиеся с index_) print("Идентификация колонок со справочниками...") index_columns = [col for col in population.columns if col.startswith('index_')] print(f"Найдены колонки со справочниками: {index_columns}")  # Извлекаем ID справочников из названий колонок index_ids = [col.split('_')[1] for col in index_columns] print(f"Извлеченные ID справочников: {index_ids}")  # Функция для получения соответствия между кодами и их текстовыми описаниями def get_index_mapping(index_id):     """Получает маппинг кодов в текстовые описания для указанного справочника"""     api_url = f"https://api.alexstat.ru/api/v1/get_index/{index_id}"     try:         print(f"Запрашиваем маппинг для справочника {index_id}...")         response = requests.get(api_url, headers=headers)          if response.status_code == 200:             data = response.json()             # Создаем словарь {код: описание}             mapping = {str(item['code_value']): item['code_text'] for item in data}             print(f"Найдено {len(mapping)} соответствий для справочника {index_id}")             return mapping         else:             print(f"Ошибка получения справочника {index_id}: статус {response.status_code}")             return None     except requests.exceptions.RequestException as e:         print(f"Ошибка запроса для справочника {index_id}: {e}")         return None     except KeyError as e:         print(f"Неожиданный формат данных для справочника {index_id}: {e}")         return None  # Получаем маппинги для всех найденных справочников index_mappings = {} for index_id in index_ids:     mapping = get_index_mapping(index_id)     if mapping:         index_mappings[index_id] = mapping  print(f"Созданы маппинги для {len(index_mappings)} из {len(index_ids)} справочников")  # Заменяем коды на текстовые описания в основном DataFrame for col in index_columns:     index_id = col.split('_')[1]     if index_id in index_mappings:         mapping = index_mappings[index_id]         # Преобразуем к строковому типу, применяем маппинг, оставляем оригинальные значения при отсутствии маппинга         population[col] = population[col].astype(str).map(mapping).fillna(population[col])         print(f"Обработана колонка {col} (справочник {index_id})")     else:         print(f"Не найден маппинг для колонки {col} (справочник {index_id})")  # Выводим результат population

Исследуем показатель, что он содержит:

# Проходим по всем колонкам в DataFrame population for col in population.columns:     # Пропускаем числовые колонки и идентификаторы     if col in ['ObsValue', 'id']:         continue      # Выводим название колонки     print(f'Уникальные значения в колонке {col}:')      # Выводим все уникальные значения в колонке     print(population[col].unique())      # Пустая строка для разделения     print()

Отфильтруем датафрейм со значениями показателя по населению. И возьмем статистику только с одного уровня – субъектного, исключив дублирования в виде РФ, федеральных округов, частей, входящих в состав субъектов:

OKATO_to_be_excluded = ['Российская Федерация',                         'Российская Федерация в границах до 04.10.2022',                         'Центральный федеральный округ',                         'Северо-Западный федеральный округ',                         'Ненецкий автономный округ (Архангельская область)',                         'Архангельская область (без АО)',                         'Архангельская область (кроме Ненецкого автономного округа)',                         'Южный федеральный округ (с 2010 года)',                         'Южный федеральный округ (с 29.07.2016)',                         'Северо-Кавказский федеральный округ',                         'Приволжский федеральный округ',                         'Коми-Пермяцкий округ, входящий в состав Пермского края',                         'Ханты-Мансийский автономный округ - Югра (Тюменская область)',                         'Ямало-Ненецкий автономный округ (Тюменская область)',                         'Тюменская область (без АО)',                         'Тюменская область (кроме Ханты-Мансийского автономного округа-Югры и Ямало-Ненецкого автономного округа)',                         'Сибирский федеральный округ',                         'Таймырский (Долгано-Ненецкий) автономный округ (Красноярский край)',                         'Эвенкийский автономный округ (Красноярский край)',                         'Усть-Ордынский Бурятский округ' ,                         'Агинский Бурятский округ (Забайкальский край)',                         'Корякский округ, входящий в состав Камчатского края',                         'Чеченская и Ингушская Республики',                         'Северный район' 'Северо-Западный район' 'Центральный район',                         'Волго-Вятский район',                         'Центрально-Черноземный район',                         'Поволжский район',                         'Северо-Кавказский район',                         'Уральский район',                         'Западно-Сибирский район',                         'Восточно-Сибирский район',                         'Дальневосточный район']  # Фильтрация данных о населении по нескольким условиям: population_filtered = population[     # 1. Выбираем только данные за 2024 год     (population['Time'] == 2024) &      # 2. Выбираем только записи, где указано "все население" (из справочника index_17)     (population['index_17'] == 'все население') &      # 3. Исключаем записи, где код региона (index_1) находится в списке исключений OKATO_to_be_excluded     # Оператор ~ означает отрицание (NOT IN)     (~population['index_1'].isin(OKATO_to_be_excluded)) ]  # Результат фильтрации сохраняется в новый DataFrame population_filtered population_filtered

Сделаем тоже самое для показателя по числу преступлений:

Делай раз:

api_url = "https://api.alexstat.ru/api/v1/get_indicator/232"  try:     response = requests.get(api_url, headers=headers)      data = None      if response.status_code == 200:         data = response.json()         print("Success!")     else:         print(f"Request failed with status code: {response.status_code}")         print(f"Response: {response.text}")  except requests.exceptions.RequestException as e:     print(f"Request failed: {e}")  crime = pd.DataFrame(data) crime['ObsValue'] = crime['ObsValue'].astype(float) crime

Делай два:

for col in crime.columns:     if col in ['ObsValue', 'id']:         continue     print(f'{col} unique values:')     print(crime[col].unique())     print()

Делай три:

for col in crime.columns:     if col in ['ObsValue', 'id']:         continue     print(f'{col} unique values:')     print(crime[col].unique())     print()

Делай четыре:

crime_filtered = crime[     (crime['Time'] == 2024) &     (crime['PERIOD'] == 'январь-декабрь') &     (crime['index_1'].isin(population_filtered['index_1'].tolist() ))] crime_filtered

И готово:

Объединим два показателя по ОКАТО (субъектам РФ):

# Объединение данных о населении и преступности в один DataFrame population_crime = pd.merge(     # Левый DataFrame - данные о населении:     population_filtered[['ObsValue', 'Time', 'index_1']].rename(columns={'ObsValue': 'population'}),      # Правый DataFrame - данные о преступности:     crime_filtered[['ObsValue', 'Time', 'index_1']].rename(columns={'ObsValue': 'crime'}),      # Параметры объединения:     on=['Time', 'index_1'],  # Ключи для объединения (год и код региона)     how='outer'              # Тип объединения (внешнее) )  # Результат объединения: population_crime

Рассмотрим зависимость между численностью населения и преступлениями:

# раскомментируйте строчку ниже, чтобы установить библиотеку для рисования #!pip install matplotlib import matplotlib.pyplot as plt  plt.scatter(population_crime['population'], population_crime['crime']) plt.xlabel('Population') plt.ylabel('Crime') plt.show()

Регрессия:

# раскомментируйте строчку ниже, чтобы установить библиотеку для построения стат. моделей #!pip install statsmodels  import statsmodels.api as sm import matplotlib.pyplot as plt import numpy as np  X = population_crime['population']  # независимая переменная y = population_crime['crime']       # зависимая переменная  # добавим константу в уравнение X = sm.add_constant(X)  # регрессионная модель model = sm.OLS(y, X).fit()  # статистики модели print(model.summary()) print("\nCoefficients:") print(f"Intercept: {model.params[0]:.2f}") print(f"Population Coefficient: {model.params[1]:.4f}") print(f"R-squared: {model.rsquared:.3f}") print(f"P-value for Population: {model.pvalues[1]:.4f}")  # построим график plt.figure(figsize=(10, 6))  # график разброса наблюдаемых данных plt.scatter(     population_crime['population'],     population_crime['crime'],     color='blue',     alpha=0.6,     label='Фактические данные' )  # линия регрессии regression_line = model.params[0] + model.params[1] * population_crime['population'] plt.plot(     population_crime['population'],     regression_line,     color='red',     linewidth=2,     label=f'Regression Line\ny = {model.params[0]:.2f} + {model.params[1]:.4f}x' )  # уравнение регрессии equation_text = f'y = {model.params[0]:.2f} + {model.params[1]:.4f}x\nR² = {model.rsquared:.3f}' plt.text(     0.05, 0.95,     equation_text,     transform=plt.gca().transAxes,     fontsize=12,     verticalalignment='top',     bbox=dict(boxstyle='round', facecolor='white', alpha=0.8) )  plt.xlabel('Population', fontsize=12) plt.ylabel('Crime', fontsize=12) plt.title('Линейная регрессия', fontsize=14) plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show()

Модель показывает сильную зависимость уровня преступности от численности населения с R² = 0,829, что означает, что 82,9 % дисперсии преступности объясняется показателем населения. Коэффициент регрессии статистически значим: при росте населения на 1 000 человек число преступников увеличивается в среднем на 4 человека (как бы 4 человека из 1 000 значимо статистически могут стать преступниками). Однако, константа незначима (при «нулевом» населении). Такая вот псевдо-зависимость, но как пример использования API StatKit вполне себе подойдет.

И картинка:

И причем здесь погода скажите Вы? Хочу сообщить о запуске разработки нового формата – больших погодных массивов по Российской Федерации от ВНИИГМИ-МЦД в совместимом с ЕМИСС формате SDMX. Что скоро будет внедрено (по некоторым городам РФ данные доступны аж с конца 1800 годов!):

— суточные данные о температуре почвы по станциям в РФ;

— суточные данные о температуре атмосферного воздуха и осадках по станциям в РФ;

— суточные данные о снежном покрове по станциям в РФ.

Заключительные положения

Проект изначально был моим личным небольшим кодом, который помогал мне работать с ЕМИСС, но, как мне кажется, может он поможет и Вам.

Сейчас не все показатели есть у меня по сравнению с ЕМИСС, но я стараюсь добавлять с ЕМИСС новые данные, и, как я указывал в посте, если Вы напиши мне с указанием, какой показатель добавить (а еще лучше прямая ссылка на этот показатель в ЕМИСС), я его постараюсь оперативно добавить.

Еще раз положу здесь:

• ссылка на проект: https://go.statkit.ru/

• ссылки на Colab: https://colab.research.google.com/drive/1yp37jkFGpFpphehHxakxd0dgCtvFkAPH?usp=sharing и https://colab.research.google.com/drive/1TRy3vILhplamlXnC_K_0TdlkGOZ0U46K?usp=sharing

• ссылки на часть 1 и часть 2 на Хабре.