/usr/lib/python3/dist-packages/pandas/stats/fama_macbeth.py is in python3-pandas 0.14.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | from pandas.core.base import StringMixin
from pandas.compat import StringIO, range
import numpy as np
from pandas.core.api import Series, DataFrame
import pandas.stats.common as common
from pandas.util.decorators import cache_readonly
def fama_macbeth(**kwargs):
"""Runs Fama-MacBeth regression.
Parameters
----------
Takes the same arguments as a panel OLS, in addition to:
nw_lags_beta: int
Newey-West adjusts the betas by the given lags
"""
window_type = kwargs.get('window_type')
if window_type is None:
klass = FamaMacBeth
else:
klass = MovingFamaMacBeth
return klass(**kwargs)
class FamaMacBeth(StringMixin):
def __init__(self, y, x, intercept=True, nw_lags=None,
nw_lags_beta=None,
entity_effects=False, time_effects=False, x_effects=None,
cluster=None, dropped_dummies={}, verbose=False):
self._nw_lags_beta = nw_lags_beta
from pandas.stats.plm import MovingPanelOLS
self._ols_result = MovingPanelOLS(
y=y, x=x, window_type='rolling', window=1,
intercept=intercept,
nw_lags=nw_lags, entity_effects=entity_effects,
time_effects=time_effects, x_effects=x_effects, cluster=cluster,
dropped_dummies=dropped_dummies, verbose=verbose)
self._cols = self._ols_result._x.columns
@cache_readonly
def _beta_raw(self):
return self._ols_result._beta_raw
@cache_readonly
def _stats(self):
return _calc_t_stat(self._beta_raw, self._nw_lags_beta)
@cache_readonly
def _mean_beta_raw(self):
return self._stats[0]
@cache_readonly
def _std_beta_raw(self):
return self._stats[1]
@cache_readonly
def _t_stat_raw(self):
return self._stats[2]
def _make_result(self, result):
return Series(result, index=self._cols)
@cache_readonly
def mean_beta(self):
return self._make_result(self._mean_beta_raw)
@cache_readonly
def std_beta(self):
return self._make_result(self._std_beta_raw)
@cache_readonly
def t_stat(self):
return self._make_result(self._t_stat_raw)
@cache_readonly
def _results(self):
return {
'mean_beta': self._mean_beta_raw,
'std_beta': self._std_beta_raw,
't_stat': self._t_stat_raw,
}
@cache_readonly
def _coef_table(self):
buffer = StringIO()
buffer.write('%13s %13s %13s %13s %13s %13s\n' %
('Variable', 'Beta', 'Std Err', 't-stat', 'CI 2.5%', 'CI 97.5%'))
template = '%13s %13.4f %13.4f %13.2f %13.4f %13.4f\n'
for i, name in enumerate(self._cols):
if i and not (i % 5):
buffer.write('\n' + common.banner(''))
mean_beta = self._results['mean_beta'][i]
std_beta = self._results['std_beta'][i]
t_stat = self._results['t_stat'][i]
ci1 = mean_beta - 1.96 * std_beta
ci2 = mean_beta + 1.96 * std_beta
values = '(%s)' % name, mean_beta, std_beta, t_stat, ci1, ci2
buffer.write(template % values)
if self._nw_lags_beta is not None:
buffer.write('\n')
buffer.write('*** The Std Err, t-stat are Newey-West '
'adjusted with Lags %5d\n' % self._nw_lags_beta)
return buffer.getvalue()
def __unicode__(self):
return self.summary
@cache_readonly
def summary(self):
template = """
----------------------Summary of Fama-MacBeth Analysis-------------------------
Formula: Y ~ %(formulaRHS)s
# betas : %(nu)3d
----------------------Summary of Estimated Coefficients------------------------
%(coefTable)s
--------------------------------End of Summary---------------------------------
"""
params = {
'formulaRHS': ' + '.join(self._cols),
'nu': len(self._beta_raw),
'coefTable': self._coef_table,
}
return template % params
class MovingFamaMacBeth(FamaMacBeth):
def __init__(self, y, x, window_type='rolling', window=10,
intercept=True, nw_lags=None, nw_lags_beta=None,
entity_effects=False, time_effects=False, x_effects=None,
cluster=None, dropped_dummies={}, verbose=False):
self._window_type = common._get_window_type(window_type)
self._window = window
FamaMacBeth.__init__(
self, y=y, x=x, intercept=intercept,
nw_lags=nw_lags, nw_lags_beta=nw_lags_beta,
entity_effects=entity_effects, time_effects=time_effects,
x_effects=x_effects, cluster=cluster,
dropped_dummies=dropped_dummies, verbose=verbose)
self._index = self._ols_result._index
self._T = len(self._index)
@property
def _is_rolling(self):
return self._window_type == 'rolling'
def _calc_stats(self):
mean_betas = []
std_betas = []
t_stats = []
# XXX
mask = self._ols_result._rolling_ols_call[2]
obs_total = mask.astype(int).cumsum()
start = self._window - 1
betas = self._beta_raw
for i in range(start, self._T):
if self._is_rolling:
begin = i - start
else:
begin = 0
B = betas[max(obs_total[begin] - 1, 0): obs_total[i]]
mean_beta, std_beta, t_stat = _calc_t_stat(B, self._nw_lags_beta)
mean_betas.append(mean_beta)
std_betas.append(std_beta)
t_stats.append(t_stat)
return np.array([mean_betas, std_betas, t_stats])
_stats = cache_readonly(_calc_stats)
def _make_result(self, result):
return DataFrame(result, index=self._result_index, columns=self._cols)
@cache_readonly
def _result_index(self):
mask = self._ols_result._rolling_ols_call[2]
# HACK XXX
return self._index[mask.cumsum() >= self._window]
@cache_readonly
def _results(self):
return {
'mean_beta': self._mean_beta_raw[-1],
'std_beta': self._std_beta_raw[-1],
't_stat': self._t_stat_raw[-1],
}
def _calc_t_stat(beta, nw_lags_beta):
N = len(beta)
B = beta - beta.mean(0)
C = np.dot(B.T, B) / N
if nw_lags_beta is not None:
for i in range(nw_lags_beta + 1):
cov = np.dot(B[i:].T, B[:(N - i)]) / N
weight = i / (nw_lags_beta + 1)
C += 2 * (1 - weight) * cov
mean_beta = beta.mean(0)
std_beta = np.sqrt(np.diag(C)) / np.sqrt(N)
t_stat = mean_beta / std_beta
return mean_beta, std_beta, t_stat
|