Этот проект исследует потенциальные причинно-следственные связи между погодными условиями (среднемесячная температура) в Москве и другими социально-значимыми показателями (дорожно-транспортные происшествия или смертность) за период с 2010 по 2024 год, используя тест причинности по Грейнджеру и векторные авторегрессионные (VAR) модели.
Основная цель - определить, помогает ли знание прошлых значений температуры "предсказать" будущие значения выбранного показателя (ДТП или смертность) лучше, чем предсказание, основанное только на прошлых значениях самого этого показателя. И наоборот, влияет ли выбранный показатель на будущую температуру.
Важно: Тест Грейнджера выявляет предсказательную причинность, а не обязательно философскую или физическую причинно-следственную связь. Он показывает, улучшает ли один временной ряд прогноз другого.
- Загрузка данных: Загружает данные о температуре, ДТП и смертности из CSV-файлов с различными форматами и кодировками.
- Предобработка: Очищает данные, унифицирует временные метки к месячному формату, агрегирует ежедневные данные (температура) до месячных, выполняет нормализацию (Z-score, логарифмирование).
- Анализ стационарности: Проверяет временные ряды на стационарность с помощью тестов ADF (Augmented Dickey-Fuller) и KPSS (Kwiatkowski-Phillips-Schmidt-Shin), применяет дифференцирование при необходимости.
- Моделирование VAR: Подбирает оптимальный порядок лага для Векторной Авторегрессии (VAR) с использованием информационных критериев (AIC, BIC), строит и проверяет стабильность VAR-модели.
- Тест Грейнджера: Выполняет тест причинности по Грейнджеру для оценки направленных предсказательных связей между рядами.
- Визуализация: Генерирует графики временных рядов, автокорреляционных функций (ACF/PACF), кросс-корреляций, а также диагностические графики VAR-модели (функции импульсного отклика - IRF, разложение дисперсии ошибки прогноза - FEVD).
- Конфигурация: Позволяет легко настраивать пути к данным и параметры анализа через файл
src/config.py. - Логирование: Ведет журнал процесса анализа для отладки и мониторинга.
- Тестирование: Включает набор модульных тестов для проверки корректности основных функций.
Временной ряд - это последовательность данных, измеренных через равные промежутки времени (например, среднемесячная температура, количество ДТП в месяц). Анализ временных рядов фокусируется на выявлении закономерностей, трендов, сезонности и взаимосвязей в таких данных.
Стационарный временной ряд - это ряд, статистические свойства которого (среднее значение, дисперсия, автокорреляция) не изменяются со временем. Большинство стандартных методов анализа временных рядов (включая VAR и тест Грейнджера) требуют, чтобы ряды были стационарными.
- Почему это важно? Если среднее или дисперсия меняются со временем (например, есть тренд роста или падения), то корреляции между прошлыми и будущими значениями будут зависеть от того, какой временной интервал мы рассматриваем. Это затрудняет построение надежных моделей.
- Проверка:
- Тест ADF (Расширенный тест Дики-Фуллера): Проверяет нулевую гипотезу о том, что ряд нестационарен (имеет "единичный корень"). Если p-value < уровня значимости (обычно 0.05), мы отвергаем H0 и считаем ряд стационарным.
- Тест KPSS: Проверяет нулевую гипотезу о том, что ряд стационарен (вокруг среднего или тренда). Если p-value < уровня значимости, мы отвергаем H0 и считаем ряд нестационарным. Интерпретация противоположна ADF.
- Преобразование: Если ряд нестационарен, часто применяют дифференцирование - вычисление разницы между соседними наблюдениями (
value[t] - value[t-1]). Иногда требуется дифференцирование второго порядка. Логарифмирование также может помочь стабилизировать дисперсию.
VAR - это модель для анализа нескольких взаимосвязанных временных рядов. Она расширяет идею авторегрессии (AR), где будущее значение ряда зависит от его прошлых значений. В VAR-модели будущее значение каждой переменной зависит не только от её собственных прошлых значений, но и от прошлых значений всех других переменных в системе.
-
Пример (2 переменные, лаг 1):
Temp[t] = c1 + a11*Temp[t-1] + a12*Event[t-1] + error1[t] Event[t] = c2 + a21*Temp[t-1] + a22*Event[t-1] + error2[t]Здесь
Temp[t]иEvent[t]- значения температуры и события (ДТП/Смертность) в моментt,c1, c2- константы,aXX- коэффициенты, показывающие силу влияния прошлых значений,errorX[t]- случайные ошибки. -
Порядок лага (p): Определяет, сколько прошлых периодов (
t-1,t-2, ...,t-p) включаются в модель. Выбирается обычно с помощью информационных критериев (AIC, BIC), которые балансируют между качеством подгонки модели и её сложностью (количеством параметров).
Тест проверяет, помогает ли знание прошлых значений одного временного ряда (X) предсказать будущие значения другого временного ряда (Y) лучше, чем предсказание Y только на основе его собственных прошлых значений.
- Идея: Если прошлые значения X содержат уникальную информацию, полезную для прогноза Y (которой нет в прошлых значениях Y), то говорят, что X "является причиной Y по Грейнджеру".
- Как работает (упрощенно): Строятся две модели для Y:
- Модель, использующая только прошлые значения Y.
- Модель, использующая прошлые значения Y и прошлые значения X. Сравнивается точность предсказаний этих двух моделей (обычно через F-тест на значимость коэффициентов при прошлых значениях X во второй модели).
- Интерпретация:
- Если F-тест статистически значим (p-value < уровня значимости), то мы отвергаем нулевую гипотезу об отсутствии причинности и заключаем, что X является причиной Y по Грейнджеру.
- Тест выполняется в обе стороны (X -> Y и Y -> X).
qwinty-grangercasuality/
├── data/ # Исходные и обработанные данные
│ ├── Moscow_Temp (2010-2024).csv # Ежедневная температура и осадки
│ └── Moscow/ # Данные по ДТП и смертности
│ ├── convert.py # Скрипт для конвертации формата данных смертности
│ ├── moscow_dtp_transformed.csv # Обработанные данные ДТП (помесячно)
│ └── moscow_mortality.csv # Исходные данные смертности (помесячно)
├── docs/ # Документация
│ └── project_plan.md # План проекта
├── src/ # Исходный код
│ ├── config.py # Файл конфигурации
│ ├── Granger_Causality.ipynb # Jupyter Notebook для интерактивного анализа
│ ├── analysis/ # Модули анализа
│ │ ├── granger.py # Тест Грейнджера
│ │ ├── stationarity.py # Тесты стационарности
│ │ └── var_model.py # Построение и анализ VAR моделей
│ ├── data_processing/ # Модули обработки данных
│ │ ├── cleaner.py # Очистка, нормализация, агрегация
│ │ ├── loader.py # Загрузка данных
│ │ └── merger.py # Объединение данных
│ ├── utils/ # Вспомогательные утилиты
│ │ ├── helpers.py # Общие функции
│ │ └── logger.py # Настройка логирования
│ └── visualization/ # Модули визуализации
│ ├── diagnostics.py # Диагностические графики VAR (IRF, FEVD)
│ └── time_series.py # Графики временных рядов, ACF/PACF и др.
└── tests/ # Модульные тесты
├── analysis/
├── data_processing/
├── fixtures/ # Фиктивные данные для тестов
└── utils/
-
Создайте и активируйте виртуальное окружение (рекомендуется):
python -m venv venv # Windows venv\Scripts\activate # macOS/Linux source venv/bin/activate
-
Установите зависимости:
pip install -r requirements.txt
Основные параметры анализа настраиваются в файле src/config.py.
# src/config.py (Пример)
# Пути к данным
TEMP_DATA_PATH = "data/Moscow_Temp (2010-2024).csv"
# Выберите один из вторичных наборов данных:
# SECONDARY_DATA_PATH = "data/Moscow/moscow_dtp_transformed.csv"
SECONDARY_DATA_PATH = "data/Moscow/moscow_mortality.csv"
SECONDARY_DATA_NAME = "Mortality" # Или "DTP"
# Настройки предварительной обработки
DATE_FORMAT = "%Y-%m" # Формат для месячного индекса
NORMALIZATION_METHOD_TEMP = "z-score" # Метод нормализации температуры
NORMALIZATION_METHOD_SECONDARY = "log" # Метод нормализации втор. данных
AGGREGATION_TEMP = "mean" # Агрегация температуры (среднее)
AGGREGATION_SECONDARY = "sum" # Агрегация втор. данных (сумма)
# Настройки VAR модели
MAX_LAG_ORDER = 12 # Макс. лаг для подбора VAR
LAG_SELECTION_CRITERIA = ["aic", "bic"] # Критерии выбора лага
# Настройки причинности Грейнджера
GRANGER_SIGNIFICANCE_LEVEL = 0.05 # Уровень значимости
# Настройки логгера
LOG_LEVEL = "INFO"
LOG_FILE = "analysis.log"Перед запуском:
- Убедитесь, что пути
TEMP_DATA_PATHиSECONDARY_DATA_PATHвsrc/config.pyуказаны относительно директорииsrc/, так как они используются в модулях внутриsrc. Например, если скрипт запускается из корня проекта, а используетconfig.py, пути в конфиге должны быть вида"../data/...". Если же главный скрипт находится вsrc/, то пути должны быть"../data/...". Судя по кодуloader.pyиconfig.pyв вашем примере, ожидается, что скрипт запускается изsrc/, поэтому пути../data/...корректны. - Выберите, какой вторичный набор данных использовать (
moscow_dtp_transformed.csvилиmoscow_mortality.csv), закомментировав ненужную строкуSECONDARY_DATA_PATHи установив соответствующееSECONDARY_DATA_NAME.
Предполагается наличие основного скрипта (например, src/main.py, который не был предоставлен) или использование Jupyter Notebook (src/Granger_Causality.ipynb).
Общий рабочий процесс (как он должен быть реализован в main.py или Notebook):
- Загрузка данных: Использование
loader.load_all_data. - Очистка и Агрегация:
- Для температуры: агрегация ежедневных данных до среднемесячных (
cleaner.aggregate_monthly). - Для ДТП/Смертности: данные уже помесячные, но нужно унифицировать временные метки (
cleaner.unify_timestamps). - Применение нормализации (
cleaner.normalize_data).
- Для температуры: агрегация ежедневных данных до среднемесячных (
- Объединение данных: Слияние временных рядов температуры и второго показателя (
merger.merge_dataframes). Проверка полноты (merger.check_completeness). - Проверка стационарности: Применение тестов ADF и KPSS (
stationarity.check_stationarity_on_dataframe). При необходимости, применение дифференцирования (stationarity.apply_differencing) и повторная проверка. - Выбор лага VAR: Определение оптимального лага для VAR-модели (
var_model.select_optimal_lag). - Построение VAR: Подгонка VAR-модели с выбранным лагом (
var_model.fit_var_model). - Проверка стабильности VAR: Убедиться, что модель стабильна (
var_model.check_model_stability). - Тест Грейнджера: Выполнение теста причинности (
granger.perform_granger_causality_test). - Визуализация: Построение графиков исходных и обработанных рядов, ACF/PACF, кросс-корреляций, IRF, FEVD (
visualizationмодули). - Интерпретация результатов: Анализ p-значений теста Грейнджера и диагностических графиков.
Пример запуска (если бы был main.py в src/):
cd src
python main.pyИли: Откройте и выполните ячейки в src/Granger_Causality.ipynb.
Для проверки корректности работы модулей можно запустить модульные тесты:
# Находясь в корневой директории проекта (qwinty-grangercasuality)
python -m unittest discover testsdata/Moscow_Temp (2010-2024).csv:- Содержит ежедневные данные о средней температуре воздуха и количестве осадков в Москве.
- Разделитель:
; - Кодировка:
utf-8 - Заголовки на русском языке.
data/Moscow/moscow_mortality.csv:- Содержит ежемесячные данные о регистрации смертей, рождений и т.д. в Москве.
- Разделитель:
, - Кодировка:
utf-8 - Требует преобразования столбцов 'Year' и 'Month' в дату (см.
convert.py).
data/Moscow/moscow_dtp_transformed.csv:- Содержит ежемесячные данные о количестве ДТП, погибших и раненых в Москве.
- Разделитель:
; - Кодировка:
utf-8 - Столбец даты уже в формате
MM.YYYY.
Загрузка данных (src/data_processing/loader.py):
def load_temperature_data(filepath: str) -> pd.DataFrame:
"""Загружает данные о температуре из текстового файла."""
print(f"Загрузка данных о температуре из: {filepath}")
try:
df = pd.read_csv(filepath, delimiter=';', encoding='utf-8')
df = df.rename(columns=TEMP_COL_MAP)
df['Date'] = pd.to_datetime(df[['Year', 'Month', 'Day']])
df = df[['Date', 'Temperature', 'Precipitation']]
print("Данные о температуре успешно загружены.")
return df
# ... обработка ошибок ...Проверка стационарности (src/analysis/stationarity.py):
def check_stationarity_adf(series: pd.Series, significance_level: float = 0.05, regression: str = 'c') -> Tuple[bool, float]:
"""Выполняет расширенный тест Дики-Фуллера (ADF) на стационарность."""
# ... (печать информации) ...
try:
result = adfuller(series.dropna(), regression=regression)
p_value = result[1]
is_stationary = p_value < significance_level
# ... (печать результатов) ...
return is_stationary, p_value
# ... обработка ошибок ...Подгонка VAR модели (src/analysis/var_model.py):
def fit_var_model(data: pd.DataFrame, lag_order: int) -> Optional[VARResults]:
"""Подгоняет VAR модель к данным с указанным порядком лага."""
print(f"Подгонка VAR модели с порядком лагов: {lag_order}")
# ... (проверки) ...
try:
model = VAR(data)
results = model.fit(lag_order)
print("\nРезультаты подгонки VAR модели:")
print(results.summary())
return results
# ... обработка ошибок ...Тест Грейнджера (src/analysis/granger.py):
def perform_granger_causality_test(results: VARResults, max_lag: int, significance_level: float = 0.05) -> Optional[Dict[Tuple[str, str], Dict[str, Any]]]:
"""Выполняет тесты причинности по Грейнджеру для всех пар переменных."""
# ... (проверки и инициализация) ...
variables = results.names
data = pd.DataFrame(results.model.y, columns=variables)
test_results = {}
for caused_var in variables:
for causing_var in variables:
if caused_var == causing_var: continue
# ... (выбор данных) ...
try:
gc_results = grangercausalitytests(test_data, [max_lag], verbose=False)
lag_result = gc_results[max_lag][0]
p_value = lag_result['ssr_ftest'][1]
significant = p_value < significance_level
# ... (сохранение и печать результатов) ...
# ... обработка ошибок ...
return test_resultsВизуализация (src/visualization/time_series.py):
def plot_time_series(df: pd.DataFrame, columns: Optional[List[str]] = None, title: str = "График временного ряда", ..., save_path: Optional[str] = None):
"""Отображает один или несколько временных рядов из DataFrame."""
# ... (настройка графика) ...
for col in columns:
if col in df.columns:
plot_index = df.index.to_timestamp() if isinstance(df.index, pd.PeriodIndex) else df.index
plt.plot(plot_index, df[col], label=col)
# ... (настройка легенды, сетки, заголовка) ...
if save_path:
plt.savefig(save_path)
plt.close()
else:
plt.show()- Включить в анализ данные об осадках (
Precipitation). - Добавить другие потенциально влияющие факторы (праздники, экономические показатели).
- Использовать более сложные модели (VARMA, SVAR - структурная VAR).
- Реализовать более строгие методы валидации (например, скользящее окно для проверки стабильности результатов во времени).
- Создать полноценный скрипт
src/main.pyдля автоматизации всего процесса анализа.