<jupyter_start><jupyter_text>Cointegration Test  Engle-Granger Method<jupyter_code>infl = test_data['inflation']
mkt = test_data.loc[infl.index]['MktCap']
i = 0
while True:
    if adfuller(infl.values)[1] < 0.05:
        print('Order of the Inflation series is:', i)
        break
    else:
        infl = infl.diff().dropna()
    i += 1
    
j = 0
while True:
    if adfuller(mkt.values)[1] < 0.05:
        print('Order of the MktCap series is:', j)
        break
    else:
        mkt = mkt.diff().dropna()
    j += 1
    
infl = test_data['inflation']
mkt = test_data.loc[infl.index]['MktCap']
model = sm.OLS(infl.values, mkt.values)
results = model.fit()

residuals = infl.values - results.predict()
adfuller_pvalue = adfuller(residuals)[1]
print('Engle-Granger pvalue for cointegration:', adfuller_pvalue)

residuals = np.diff(residuals,1)
adfuller_pvalue = adfuller(residuals)[1]
print('Engle-Granger pvalue for cointegration order one:', adfuller_pvalue)
infl = infl.diff().dropna()
mkt = mkt.diff().dropna()

mkt = sm.add_constant(mkt)

model = sm.OLS(infl, mkt)
results = model.fit()

hypotheses = 'MktCap = 0'
print('Parameters')
print(results.params)
print('------------\n')


print('t-statistic')
print(results.tvalues)
print('------------\n')

print('Degrees of Freedom:', mkt.shape[0])
print('------------\n')

residuals = infl.values - results.predict()
print('Normality test result:', kstest(residuals, 'norm')[1])
adfuller_pvalue = adfuller(residuals)[1]
print('p-value for unit root:', adfuller_pvalue)
infl = test_data['inflation']
mkt = test_data.loc[infl.index]['MktCap']
infl = infl.diff().dropna()
mkt = mkt.diff().dropna()
plt.figure(figsize=(15,9))
plt.scatter(mkt.values, infl.values)
plt.plot(np.linspace(start = min(mkt.values), stop = max(mkt.values)), 0.925 + 0.120 * np.linspace(start = min(mkt.values), stop = max(mkt.values)))
plt.xlabel(r'$\Delta$MktCap')
plt.ylabel('Inflation')
plt.suptitle('Market Cap first difference regressed in Inflation')
plt.show()<jupyter_output><empty_output><jupyter_text>Granger Causality<jupyter_code>from io import StringIO 
import sys

class Capturing(list):
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio.getvalue().splitlines())
        del self._stringio    # free up some memory
        sys.stdout = self._stdout
a = np.log(granger_test_data['MktCap']) - np.log(granger_test_data['MktCap'].shift(1))
# a = (granger_test_data['MktCap']) - (granger_test_data['MktCap'].shift(1))
a.dropna(inplace = True)
a = pd.DataFrame(a)
a.columns = ['MktCap']
adfuller(a.values)[1]
granger_test_data.shape
data = np.log(granger_test_data) - (np.log(granger_test_data)).shift(1)
data.dropna(inplace = True)

data = a.join(granger_test_data['inflation'])
# data = data.diff()
# data.dropna(inplace = True)

for col in data.columns:
   print(col, adfuller(data[col].values)[1])

with Capturing() as output:
    grangercausalitytests(data[['inflation','MktCap']], 60)
# print(output)

f_pvalue = []
chi2_pvalue = []
likelihood_pvalue = []

for line in output:
    if line[0:16] == 'ssr based F test':
        f_pvalue.append(float(line[40:46]))
    elif line[0:19] == 'ssr based chi2 test':
        chi2_pvalue.append(float(line[40:46]))
    elif line[0:21] == 'likelihood ratio test':
        likelihood_pvalue.append(float(line[40:46]))

x = [i + 1 for i in range(len(f_pvalue))] 
plt.figure(figsize=(15,9))
plt.plot(x,f_pvalue, label = 'F Statistic')
plt.plot(x,chi2_pvalue, label = r'$\chi^2$ Statistic')
plt.plot(x,likelihood_pvalue, label = 'Likelihood Statistic')
plt.axhline(0.05, color = 'gray', linestyle='--')
plt.axhline(0.1, color = 'gray', linestyle='--')
plt.legend()
plt.xlabel('Time lag in months')
plt.ylabel('p-Value')
plt.suptitle('p-Value of Granger Causality Test MktCap on Inflation for 3 statistics')
plt.show()
df_granger = pd.DataFrame({'Lag':x,'F-statistic p-value':f_pvalue, 'Chi2 statistic p-value':chi2_pvalue})
df_granger = df_granger.set_index('Lag')
display(df_granger)
grangercausalitytests(data[['inflation','MktCap']], 15)
data_inv = data[['MktCap', 'inflation']]
gc_res_inv = grangercausalitytests(data_inv, 48)
gc_res_inv
model = VAR(data[['inflation', 'MktCap']])
results = model.fit(36)
results.summary()
data = test_data_log_diff[['inflation', 'MktCap']]
gc_test = grangercausalitytests(data,12)
gc_test<jupyter_output>Granger Causality
number of lags (no zero) 1
ssr based F test:         F=1.0137  , p=0.3157  , df_denom=143, df_num=1
ssr based chi2 test:   chi2=1.0350  , p=0.3090  , df=1
likelihood ratio test: chi2=1.0313  , p=0.3098  , df=1
parameter F test:         F=1.0137  , p=0.3157  , df_denom=143, df_num=1

Granger Causality
number of lags (no zero) 2
ssr based F test:         F=0.2931  , p=0.7464  , df_denom=140, df_num=2
ssr based chi2 test:   chi2=0.6070  , p=0.7382  , df=2
likelihood ratio test: chi2=0.6058  , p=0.7387  , df=2
parameter F test:         F=0.2931  , p=0.7464  , df_denom=140, df_num=2

Granger Causality
number of lags (no zero) 3
ssr based F test:         F=0.1082  , p=0.9552  , df_denom=137, df_num=3
ssr based chi2 test:   chi2=0.3412  , p=0.9521  , df=3
likelihood ratio test: chi2=0.3408  , p=0.9522  , df=3
parameter F test:         F=0.1082  , p=0.9552  , df_denom=137, df_num=3

Granger Causality
number of lags (no zero) 4
ssr based F test:         F=0.1616  , p=0.9574  [...]<jupyter_text>VAR ANALYSIS  VAR: inflation, MktCap<jupyter_code>def total_squared_error(results, test_data, columns, step):
    observed_inflation = test_data[['inflation']]
    error = []
    for i in range(observed_inflation.shape[0] - step - 1):
        initial_data = (test_data[columns].iloc[i:(i+step),:]).values
        forecast_results = results.forecast(initial_data, 1)[0,0]
        observed_inflation_i = observed_inflation.iloc[i + step + 1].values
        error.append(observed_inflation_i-forecast_results)

    error_norm = np.linalg.norm(error)
    return error_norm

def predict(results, test_data, columns, step):
    observed_inflation = test_data[['inflation']]
    predictions = []
    for i in range(observed_inflation.shape[0] - step - 1):
        initial_data = (test_data[columns].iloc[i:(i+step),:]).values
        forecast_results = results.forecast(initial_data, 1)[0,0]
        predictions.append(forecast_results)
    pred_df = pd.DataFrame(predictions)
    pred_df.index = test_data.index[:observed_inflation.shape[0] - step - 1]
    pred_df.columns = ['prediction']
    return pred_df
model = VAR(test_data_log_diff[['inflation', 'MktCap']])
results = model.fit(4)
results.summary()
observed_inflation = test_data_log_diff[['inflation']]
step = 4
error = total_squared_error(results,test_data_log_diff, ['inflation', 'MktCap'], step)
print('Total Error:', error)


forecast = predict(results, test_data_log_diff, ['inflation', 'MktCap'], step)
plt.plot(forecast, label = 'forecast')
plt.plot(observed_inflation, label = 'observed')
plt.legend()
plt.show()

residuals = (observed_inflation.values[:observed_inflation.shape[0] - step - 1] - forecast.values)
print('Normality test result:', kstest(residuals, 'norm')[1])<jupyter_output>Total Error: 0.07333334403685315<jupyter_text>VAR: inflation, M2, MktCap<jupyter_code>model = VAR(test_data_log_diff[['inflation','M2', 'MktCap']])
results = model.fit(4)
results.summary()
observed_inflation = test_data_log_diff[['inflation']]
step = 4
error = total_squared_error(results,test_data_log_diff, ['inflation','M2', 'MktCap'], step)
print('Total Error:', error)


forecast = predict(results, test_data_log_diff,['inflation','M2', 'MktCap'], step)
plt.plot(forecast, label = 'forecast')
plt.plot(observed_inflation, label = 'observed')
plt.legend()
plt.show()

residuals = (observed_inflation.values[:observed_inflation.shape[0] - step - 1] - forecast.values)
print('Normality test result:', kstest(residuals, 'norm')[1])<jupyter_output>Total Error: 0.07311603053838961<jupyter_text>VAR: inflation, M2, GDP, MktCap<jupyter_code>model = VAR(test_data_log_diff[['inflation','M2','GDP', 'MktCap']])
results = model.fit(4)
results.summary()
observed_inflation = test_data_log_diff[['inflation']]
step = 4
error = total_squared_error(results,test_data_log_diff, ['inflation','M2','GDP', 'MktCap'], step)
print('Total Error:', error)


forecast = predict(results, test_data_log_diff,['inflation','M2','GDP', 'MktCap'], step)
plt.plot(forecast, label = 'forecast')
plt.plot(observed_inflation, label = 'observed')
plt.legend()
plt.show()

residuals = (observed_inflation.values[:observed_inflation.shape[0] - step - 1] - forecast.values)
print('Normality test result:', kstest(residuals, 'norm')[1])<jupyter_output>Total Error: 0.07566126739477709<jupyter_text>VAR: inlfation, M2, GDP, MV2, MktCap<jupyter_code>model = VAR(test_data_log_diff[['inflation','M2','GDP', 'M2V', 'MktCap']])
results = model.fit(4)
results.summary()
irf = results.irf(12)
irf.plot(impulse = 'MktCap',response = 'inflation')
observed_inflation = test_data_log_diff[['inflation']]
step = 4
error = total_squared_error(results,test_data_log_diff, ['inflation','M2','GDP', 'M2V', 'MktCap'], step)
print('Total Error:', error)


forecast = predict(results, test_data_log_diff,['inflation','M2','GDP', 'M2V', 'MktCap'], step)
plt.plot(forecast, label = 'forecast')
plt.plot(observed_inflation, label = 'observed')
plt.legend()
plt.ylabel('Log difference of CPI')
plt.show()

residuals = (observed_inflation.values[:observed_inflation.shape[0] - step - 1] - forecast.values)
print('Normality test result:', kstest(residuals, 'norm')[1])
model = VAR(test_data_log_diff[['inflation','M2','GDP', 'M2V']])
results = model.fit(4)
results.summary()
observed_inflation = test_data_log_diff[['inflation']]
step = 4
error = total_squared_error(results,test_data_log_diff, ['inflation','M2','GDP', 'M2V'], step)
print('Total Error:', error)


forecast = predict(results, test_data_log_diff,['inflation','M2','GDP', 'M2V'], step)
plt.plot(forecast, label = 'forecast')
plt.plot(observed_inflation, label = 'observed')
plt.legend()
plt.ylabel('Log difference of CPI')
plt.show()

residuals = (observed_inflation.values[:observed_inflation.shape[0] - step - 1] - forecast.values)
print('Normality test result:', kstest(residuals, 'norm')[1])
data_shifted = test_data_log_diff[['M2', 'GDP', 'M2V']].shift(-1)
data_reg = data_shifted.join(test_data_log_diff['inflation'], how = 'inner').dropna()
model = ols("inflation ~ M2 + M2V + GDP", data_reg).fit()
model.summary()<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup