Привет, Хабр! Около года назад мне пришла странная идея: а что, если сделать новую версию ЕМИСС, хранилища российской статистики, чтобы наконец-то было удобно сводить данные. А то постоянно сопоставлять несколько показателей из множества 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 годов!):
суточные данные о температуре почвы по станциям в РФ;
суточные данные о температуре атмосферного воздуха и осадках по станциям в РФ;
суточные данные о снежном покрове по станциям в РФ.
Заключительные положения
Проект изначально был моим личным небольшим кодом, который помогал мне работать с ЕМИСС, но, как мне кажется, может он поможет и Вам.
Сейчас не все показатели есть у меня по сравнению с ЕМИСС, но я стараюсь добавлять с ЕМИСС новые данные, и, как я указывал в посте, если Вы напиши мне с указанием, какой показатель добавить (а еще лучше прямая ссылка на этот показатель в ЕМИСС), я его постараюсь оперативно добавить.
Еще раз положу здесь:
