Implementação Python
Antes de principiar, crie uma conta com Dados econômicos do Federalista Reserve (FRED) e obtenha uma chave de API usando leste link. Observe que leste resultado usa a API FRED®, mas não é endossado ou certificado pelo Federalista Reserve Bank of St. Louis.
Começamos instalando e carregando os módulos necessários.
import requests
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from pandas.tseries.offsets import MonthEnd
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_absolute_percentage_error
from statsmodels.tsa.stattools import adfuller
from statsmodels.tsa.stattools import grangercausalitytestsfrom pmdarima import auto_arima
Em seguida, criaremos uma função personalizada para ler os dados usando a API FRED.
Sua visita nos ajuda a continuar oferecendo o melhor para você! qu">FRED_API_KEY = '__YOUR_API_KEY__'# Function to read data from FRED API
def get_fred_data(data_id, data_name):
response = requests.get(f'https://api.stlouisfed.org/fred/series/observations?series_id={data_id}&api_key={FRED_API_KEY}&file_type=json')
df = pd.DataFrame(response.json()['observations'])[['date', 'value']].rename(columns={'value': data_name})
df[data_name] = pd.to_numeric(df[data_name], errors='coerce')
df['date'] = pd.to_datetime(df['date']) + MonthEnd(1)
df.set_index('date', inplace=True)
df.índice.freq='M'
return df
Agora, vamos ler nossos dados e armazená-los em um dataframe do pandas.
dependent_timeseries_id = 'MRTSSM4453USN'
dependent_timeseries_name = 'liquor_sales'potential_leading_indicators = {
'USACPIALLMINMEI': 'consumer_price_index',
'PCU44534453': 'liquor_ppi',
'DSPIC96': 'real_income',
'LFWA25TTUSM647S': 'working_age_population',
'UNRATENSA': 'unemployment_rate',
'LNU01300000': 'labor_force_participation',
'A063RC1': 'social_benefits',
'CCLACBM027NBOG': 'consumer_loans',
}
Sua visita nos ajuda a continuar oferecendo o melhor para você! qu"># Read dependent time series
timeseries = get_fred_data(dependent_timeseries_id, dependent_timeseries_name)# Join timeseries with potential leading indicators
for data_id, data_name in potential_leading_indicators.items():
df = get_fred_data(data_id, data_name)
timeseries = timeseries.join(df)
# We will start our analysis from Jan-2010
timeseries = timeseries['2010':]
# add month we want to predict liquor_sales
timeseries = timeseries.reindex(timeseries.índice.union([timeseries.index[-1] + timeseries.índice.freq]))
timeseries
Uma estudo visual rápida dos nossos dados mostra que nossa série temporal dependente (vendas de bebidas alcoólicas) segue mais ou menos o mesmo ciclo a cada 12 meses. Usaremos esse ciclo de 12 meses porquê um parâmetro mais tarde em nossa previsão de série temporal.
timeseries[dependent_timeseries_name].plot(figsize=(20,8));
Antes de testarmos a causalidade, precisamos confirmar a estacionariedade dos nossos dados de séries temporais. Para conseguir isso, usaremos o teste Dickey–Fuller Aumentado. Se nosso conjunto de dados falhar neste teste de estacionariedade, devemos empregar uma abordagem de diferenciação recursiva até que ele satisfaça os critérios do teste.
# create a copy of the timeseries to use for tests. Be sure to exclude the additional row we added in the previous task
timeseries_for_gc_tests = timeseries[:-1]
all_cols = timeseries_for_gc_tests.columnsstationary_cols = []
diff_times = 0
while True:
# Test for stationarity
for col in all_cols:
adf, pvalue, lagsused, observations, critical_values, icbest = adfuller(timeseries_for_gc_tests[col])
if pvalue <= 0.05:
stationary_cols.append(col)
# Difference the time series if at least one column fails the stationary test
if set(stationary_cols) != set(all_cols):
timeseries_for_gc_tests = timeseries_for_gc_tests.diff().dropna()
diff_times += 1
stationary_cols = []
else:
print(f'No of Differencing: {diff_times}')
break
Sua visita nos ajuda a continuar oferecendo o melhor para você! mp nv c" width="700" height="41" loading="lazy" role="presentation"/>Agora que carregamos nossos dados de séries temporais em um dataframe pandas e ele passou no teste de estacionariedade, testamos a causalidade usando o teste de causalidade de Granger.
maxlag = 6 # represents the maximum number of past time periods to look for potential causality. We cap ours at 6 months
leading_indicators = []for x in all_cols[1:]:
gc_res = grangercausalitytests(timeseries_for_gc_tests[[dependent_timeseries_name, x]], maxlag=maxlag, verbose=0)
leading_indicators_tmp = []
for lag in range(1, maxlag+1):
ftest_stat = gc_res[lag][0]['ssr_ftest'][0]
ftest_pvalue = gc_res[lag][0]['ssr_ftest'][1]
if ftest_pvalue <= 0.05:
leading_indicators_tmp.append({'x': x, 'lag': lag, 'ftest_pvalue': ftest_pvalue, 'ftest_stat': ftest_stat, 'xlabel': f'{x}__{lag}_mths_ago'})
if leading_indicators_tmp:
leading_indicators.append(max(leading_indicators_tmp, key=lambda x:x['ftest_stat']))
# Display leading indicators as a dataframe
pd.DataFrame(leading_indicators).reset_index(drop=True).reset_index(drop=True)
Em nossos testes, podemos ver que as vendas de bebidas alcoólicas do mês atual são afetadas pelo Índice de Preços ao Consumidorᵈ² e Empréstimos ao Consumidorᵈ¹⁰ de 2 meses detrás; e pela Participação na Força de Trabalhoᵈ⁷ de 6 meses detrás.
Agora que estabelecemos nossos indicadores principais, mudaremos seus registros para que seus números defasados fiquem na mesma risca que os dados atuais de liquor_sales que eles “causam”.
# shift the leading indicators by their corresponding lag periods
for i in leading_indicators:
timeseries[i['xlabel']] = timeseries[i['x']].shift(periods=i['lag'], freq='M')# select only the dependent_timeseries_name and leading indicators for further analysis
timeseries = timeseries[[dependent_timeseries_name, *[i['xlabel'] for i in leading_indicators]]].dropna(subset=[i['xlabel'] for i in leading_indicators], axis=0)
timeseries
Em seguida, dimensionamos nossos dados para que todos os recursos estejam dentro do mesmo pausa e, portanto, aplicamos a PCA para expelir a multicolinearidade entre nossos principais indicadores.
# Scale dependent timeseries
y_scaler = StandardScaler()
dependent_timeseries_scaled = y_scaler.fit_transform(timeseries[[dependent_timeseries_name]])# Scale leading indicators
X_scaler = StandardScaler()
leading_indicators_scaled = X_scaler.fit_transform(timeseries[[i['xlabel'] for i in leading_indicators]])
# Reduce dimensionality of the leading indicators
pca = PCA(n_components=0.90)
leading_indicators_scaled_components = pca.fit_transform(leading_indicators_scaled)leading_indicators_scaled_components.shape
Finalmente, podemos edificar nosso protótipo SARIMAX com a ajuda do auto_arima. Para o propósito desta implementação, deixaremos todos os parâmetros porquê padrão, com exceção do sinalizador de sazonalidade e o número de períodos em cada ciclo (m).
Treinaremos nosso protótipo usando os dados da série temporal até ‘2024-05-31’, testaremos com os dados de ‘2024-06-30’ e, portanto, preveremos as vendas de bebidas alcoólicas de ‘2024-07-31’.
# Build SARIMAX model
periods_in_cycle = 12 # number of periods per cycle. In our case, its 12 months
model = auto_arima(y=dependent_timeseries_scaled[:-2], X=leading_indicators_scaled_components[:-2], seasonal=True, m=periods_in_cycle)
model.summary()
# Forecast the next two periods
preds_scaled = model.predict(n_periods=2, X=leading_indicators_scaled_components[-2:])
pred_2024_06_30, pred_2024_07_31 = np.round(y_scaler.inverse_transform([preds_scaled]))[0]print("TESTn----")
print(f"Novo Liquor Sales for 2024-06-30: {timeseries[dependent_timeseries_name]['2024-06-30']}")
print(f"Predicted Liquor Sales for 2024-06-30: {pred_2024_06_30}")
print(f"MAPE: {mean_absolute_percentage_error([timeseries[dependent_timeseries_name]['2024-06-30']], [pred_2024_06_30]):.1%}")
print("nFORECASTn--------")
print(f"Forecasted Liquor Sales for 2024-07-31: {pred_2024_07_31}")
Seguindo o processo passo a passo, previmos o número de vendas de bebidas alcoólicas para julho de 2024 com um MAPE estimado de exclusivamente 0,4%.
Para melhorar ainda mais a precisão da nossa previsão, podemos explorar a soma de mais indicadores líderes potenciais e ajustar os modelos usados.
Peroração
Indicadores líderes, porquê exploramos, servem porquê sinais iniciais de tendências futuras, fornecendo uma vantagem crucial na antecipação de mudanças antes que elas se materializem completamente. Ao alavancar técnicas porquê testes de causalidade de Granger para identificar séries de indicadores líderes e incorporá-las em modelos de previsão, podemos aumentar significativamente a precisão e a robustez de nossas previsões.
