Source code for statsmodels.tsa.statespace.kalman_filter
"""
State Space Representation and Kalman Filter
Author: Chad Fulton
License: Simplified-BSD
"""
import contextlib
from warnings import warn
import numpy as np
from .representation import OptionWrapper, Representation, FrozenRepresentation
from .tools import reorder_missing_matrix, reorder_missing_vector
from . import tools
from statsmodels.tools.sm_exceptions import ValueWarning
# Define constants
FILTER_CONVENTIONAL = 0x01 # Durbin and Koopman (2012), Chapter 4
FILTER_EXACT_INITIAL = 0x02 # ibid., Chapter 5.6
FILTER_AUGMENTED = 0x04 # ibid., Chapter 5.7
FILTER_SQUARE_ROOT = 0x08 # ibid., Chapter 6.3
FILTER_UNIVARIATE = 0x10 # ibid., Chapter 6.4
FILTER_COLLAPSED = 0x20 # ibid., Chapter 6.5
FILTER_EXTENDED = 0x40 # ibid., Chapter 10.2
FILTER_UNSCENTED = 0x80 # ibid., Chapter 10.3
FILTER_CONCENTRATED = 0x100 # Harvey (1989), Chapter 3.4
FILTER_CHANDRASEKHAR = 0x200 # Herbst (2015)
INVERT_UNIVARIATE = 0x01
SOLVE_LU = 0x02
INVERT_LU = 0x04
SOLVE_CHOLESKY = 0x08
INVERT_CHOLESKY = 0x10
STABILITY_FORCE_SYMMETRY = 0x01
MEMORY_STORE_ALL = 0
MEMORY_NO_FORECAST_MEAN = 0x01
MEMORY_NO_FORECAST_COV = 0x02
MEMORY_NO_FORECAST = MEMORY_NO_FORECAST_MEAN | MEMORY_NO_FORECAST_COV
MEMORY_NO_PREDICTED_MEAN = 0x04
MEMORY_NO_PREDICTED_COV = 0x08
MEMORY_NO_PREDICTED = MEMORY_NO_PREDICTED_MEAN | MEMORY_NO_PREDICTED_COV
MEMORY_NO_FILTERED_MEAN = 0x10
MEMORY_NO_FILTERED_COV = 0x20
MEMORY_NO_FILTERED = MEMORY_NO_FILTERED_MEAN | MEMORY_NO_FILTERED_COV
MEMORY_NO_LIKELIHOOD = 0x40
MEMORY_NO_GAIN = 0x80
MEMORY_NO_SMOOTHING = 0x100
MEMORY_NO_STD_FORECAST = 0x200
MEMORY_CONSERVE = (
MEMORY_NO_FORECAST_COV | MEMORY_NO_PREDICTED | MEMORY_NO_FILTERED |
MEMORY_NO_LIKELIHOOD | MEMORY_NO_GAIN | MEMORY_NO_SMOOTHING
)
TIMING_INIT_PREDICTED = 0
TIMING_INIT_FILTERED = 1
[docs]
class KalmanFilter(Representation):
r"""
State space representation of a time series process, with Kalman filter
Parameters
----------
k_endog : {array_like, int}
The observed time-series process :math:`y` if array like or the
number of variables in the process if an integer.
k_states : int
The dimension of the unobserved state process.
k_posdef : int, optional
The dimension of a guaranteed positive definite covariance matrix
describing the shocks in the transition equation. Must be less than
or equal to `k_states`. Default is `k_states`.
loglikelihood_burn : int, optional
The number of initial periods during which the loglikelihood is not
recorded. Default is 0.
tolerance : float, optional
The tolerance at which the Kalman filter determines convergence to
steady-state. Default is 1e-19.
results_class : class, optional
Default results class to use to save filtering output. Default is
`FilterResults`. If specified, class must extend from `FilterResults`.
**kwargs
Keyword arguments may be used to provide values for the filter,
inversion, and stability methods. See `set_filter_method`,
`set_inversion_method`, and `set_stability_method`.
Keyword arguments may be used to provide default values for state space
matrices. See `Representation` for more details.
See Also
--------
FilterResults
statsmodels.tsa.statespace.representation.Representation
Notes
-----
There are several types of options available for controlling the Kalman
filter operation. All options are internally held as bitmasks, but can be
manipulated by setting class attributes, which act like boolean flags. For
more information, see the `set_*` class method documentation. The options
are:
filter_method
The filtering method controls aspects of which
Kalman filtering approach will be used.
inversion_method
The Kalman filter may contain one matrix inversion: that of the
forecast error covariance matrix. The inversion method controls how and
if that inverse is performed.
stability_method
The Kalman filter is a recursive algorithm that may in some cases
suffer issues with numerical stability. The stability method controls
what, if any, measures are taken to promote stability.
conserve_memory
By default, the Kalman filter computes a number of intermediate
matrices at each iteration. The memory conservation options control
which of those matrices are stored.
filter_timing
By default, the Kalman filter follows Durbin and Koopman, 2012, in
initializing the filter with predicted values. Kim and Nelson, 1999,
instead initialize the filter with filtered values, which is
essentially just a different timing convention.
The `filter_method` and `inversion_method` options intentionally allow
the possibility that multiple methods will be indicated. In the case that
multiple methods are selected, the underlying Kalman filter will attempt to
select the optional method given the input data.
For example, it may be that INVERT_UNIVARIATE and SOLVE_CHOLESKY are
indicated (this is in fact the default case). In this case, if the
endogenous vector is 1-dimensional (`k_endog` = 1), then INVERT_UNIVARIATE
is used and inversion reduces to simple division, and if it has a larger
dimension, the Cholesky decomposition along with linear solving (rather
than explicit matrix inversion) is used. If only SOLVE_CHOLESKY had been
set, then the Cholesky decomposition method would *always* be used, even in
the case of 1-dimensional data.
"""
filter_methods = [
'filter_conventional', 'filter_exact_initial', 'filter_augmented',
'filter_square_root', 'filter_univariate', 'filter_collapsed',
'filter_extended', 'filter_unscented', 'filter_concentrated',
'filter_chandrasekhar'
]
filter_conventional = OptionWrapper('filter_method', FILTER_CONVENTIONAL)
"""
(bool) Flag for conventional Kalman filtering.
"""
filter_exact_initial = OptionWrapper('filter_method', FILTER_EXACT_INITIAL)
"""
(bool) Flag for exact initial Kalman filtering. Not implemented.
"""
filter_augmented = OptionWrapper('filter_method', FILTER_AUGMENTED)
"""
(bool) Flag for augmented Kalman filtering. Not implemented.
"""
filter_square_root = OptionWrapper('filter_method', FILTER_SQUARE_ROOT)
"""
(bool) Flag for square-root Kalman filtering. Not implemented.
"""
filter_univariate = OptionWrapper('filter_method', FILTER_UNIVARIATE)
"""
(bool) Flag for univariate filtering of multivariate observation vector.
"""
filter_collapsed = OptionWrapper('filter_method', FILTER_COLLAPSED)
"""
(bool) Flag for Kalman filtering with collapsed observation vector.
"""
filter_extended = OptionWrapper('filter_method', FILTER_EXTENDED)
"""
(bool) Flag for extended Kalman filtering. Not implemented.
"""
filter_unscented = OptionWrapper('filter_method', FILTER_UNSCENTED)
"""
(bool) Flag for unscented Kalman filtering. Not implemented.
"""
filter_concentrated = OptionWrapper('filter_method', FILTER_CONCENTRATED)
"""
(bool) Flag for Kalman filtering with concentrated log-likelihood.
"""
filter_chandrasekhar = OptionWrapper('filter_method', FILTER_CHANDRASEKHAR)
"""
(bool) Flag for filtering with Chandrasekhar recursions.
"""
inversion_methods = [
'invert_univariate', 'solve_lu', 'invert_lu', 'solve_cholesky',
'invert_cholesky'
]
invert_univariate = OptionWrapper('inversion_method', INVERT_UNIVARIATE)
"""
(bool) Flag for univariate inversion method (recommended).
"""
solve_lu = OptionWrapper('inversion_method', SOLVE_LU)
"""
(bool) Flag for LU and linear solver inversion method.
"""
invert_lu = OptionWrapper('inversion_method', INVERT_LU)
"""
(bool) Flag for LU inversion method.
"""
solve_cholesky = OptionWrapper('inversion_method', SOLVE_CHOLESKY)
"""
(bool) Flag for Cholesky and linear solver inversion method (recommended).
"""
invert_cholesky = OptionWrapper('inversion_method', INVERT_CHOLESKY)
"""
(bool) Flag for Cholesky inversion method.
"""
stability_methods = ['stability_force_symmetry']
stability_force_symmetry = (
OptionWrapper('stability_method', STABILITY_FORCE_SYMMETRY)
)
"""
(bool) Flag for enforcing covariance matrix symmetry
"""
memory_options = [
'memory_store_all', 'memory_no_forecast_mean',
'memory_no_forecast_cov', 'memory_no_forecast',
'memory_no_predicted_mean', 'memory_no_predicted_cov',
'memory_no_predicted', 'memory_no_filtered_mean',
'memory_no_filtered_cov', 'memory_no_filtered',
'memory_no_likelihood', 'memory_no_gain',
'memory_no_smoothing', 'memory_no_std_forecast', 'memory_conserve'
]
memory_store_all = OptionWrapper('conserve_memory', MEMORY_STORE_ALL)
"""
(bool) Flag for storing all intermediate results in memory (default).
"""
memory_no_forecast_mean = OptionWrapper(
'conserve_memory', MEMORY_NO_FORECAST_MEAN)
"""
(bool) Flag to prevent storing forecasts and forecast errors.
"""
memory_no_forecast_cov = OptionWrapper(
'conserve_memory', MEMORY_NO_FORECAST_COV)
"""
(bool) Flag to prevent storing forecast error covariance matrices.
"""
@property
def memory_no_forecast(self):
"""
(bool) Flag to prevent storing all forecast-related output.
"""
return self.memory_no_forecast_mean or self.memory_no_forecast_cov
@memory_no_forecast.setter
def memory_no_forecast(self, value):
if bool(value):
self.memory_no_forecast_mean = True
self.memory_no_forecast_cov = True
else:
self.memory_no_forecast_mean = False
self.memory_no_forecast_cov = False
memory_no_predicted_mean = OptionWrapper(
'conserve_memory', MEMORY_NO_PREDICTED_MEAN)
"""
(bool) Flag to prevent storing predicted states.
"""
memory_no_predicted_cov = OptionWrapper(
'conserve_memory', MEMORY_NO_PREDICTED_COV)
"""
(bool) Flag to prevent storing predicted state covariance matrices.
"""
@property
def memory_no_predicted(self):
"""
(bool) Flag to prevent storing predicted state and covariance matrices.
"""
return self.memory_no_predicted_mean or self.memory_no_predicted_cov
@memory_no_predicted.setter
def memory_no_predicted(self, value):
if bool(value):
self.memory_no_predicted_mean = True
self.memory_no_predicted_cov = True
else:
self.memory_no_predicted_mean = False
self.memory_no_predicted_cov = False
memory_no_filtered_mean = OptionWrapper(
'conserve_memory', MEMORY_NO_FILTERED_MEAN)
"""
(bool) Flag to prevent storing filtered states.
"""
memory_no_filtered_cov = OptionWrapper(
'conserve_memory', MEMORY_NO_FILTERED_COV)
"""
(bool) Flag to prevent storing filtered state covariance matrices.
"""
@property
def memory_no_filtered(self):
"""
(bool) Flag to prevent storing filtered state and covariance matrices.
"""
return self.memory_no_filtered_mean or self.memory_no_filtered_cov
@memory_no_filtered.setter
def memory_no_filtered(self, value):
if bool(value):
self.memory_no_filtered_mean = True
self.memory_no_filtered_cov = True
else:
self.memory_no_filtered_mean = False
self.memory_no_filtered_cov = False
memory_no_likelihood = (
OptionWrapper('conserve_memory', MEMORY_NO_LIKELIHOOD)
)
"""
(bool) Flag to prevent storing likelihood values for each observation.
"""
memory_no_gain = OptionWrapper('conserve_memory', MEMORY_NO_GAIN)
"""
(bool) Flag to prevent storing the Kalman gain matrices.
"""
memory_no_smoothing = OptionWrapper('conserve_memory', MEMORY_NO_SMOOTHING)
"""
(bool) Flag to prevent storing likelihood values for each observation.
"""
memory_no_std_forecast = (
OptionWrapper('conserve_memory', MEMORY_NO_STD_FORECAST))
"""
(bool) Flag to prevent storing standardized forecast errors.
"""
memory_conserve = OptionWrapper('conserve_memory', MEMORY_CONSERVE)
"""
(bool) Flag to conserve the maximum amount of memory.
"""
timing_options = [
'timing_init_predicted', 'timing_init_filtered'
]
timing_init_predicted = OptionWrapper('filter_timing',
TIMING_INIT_PREDICTED)
"""
(bool) Flag for the default timing convention (Durbin and Koopman, 2012).
"""
timing_init_filtered = OptionWrapper('filter_timing', TIMING_INIT_FILTERED)
"""
(bool) Flag for the alternate timing convention (Kim and Nelson, 2012).
"""
# Default filter options
filter_method = FILTER_CONVENTIONAL
"""
(int) Filtering method bitmask.
"""
inversion_method = INVERT_UNIVARIATE | SOLVE_CHOLESKY
"""
(int) Inversion method bitmask.
"""
stability_method = STABILITY_FORCE_SYMMETRY
"""
(int) Stability method bitmask.
"""
conserve_memory = MEMORY_STORE_ALL
"""
(int) Memory conservation bitmask.
"""
filter_timing = TIMING_INIT_PREDICTED
"""
(int) Filter timing.
"""
def __init__(self, k_endog, k_states, k_posdef=None,
loglikelihood_burn=0, tolerance=1e-19, results_class=None,
kalman_filter_classes=None, **kwargs):
# Extract keyword arguments to-be-used later
keys = ['filter_method'] + KalmanFilter.filter_methods
filter_method_kwargs = {key: kwargs.pop(key) for key in keys
if key in kwargs}
keys = ['inversion_method'] + KalmanFilter.inversion_methods
inversion_method_kwargs = {key: kwargs.pop(key) for key in keys
if key in kwargs}
keys = ['stability_method'] + KalmanFilter.stability_methods
stability_method_kwargs = {key: kwargs.pop(key) for key in keys
if key in kwargs}
keys = ['conserve_memory'] + KalmanFilter.memory_options
conserve_memory_kwargs = {key: kwargs.pop(key) for key in keys
if key in kwargs}
keys = ['alternate_timing'] + KalmanFilter.timing_options
filter_timing_kwargs = {key: kwargs.pop(key) for key in keys
if key in kwargs}
# Initialize the base class
super().__init__(
k_endog, k_states, k_posdef, **kwargs
)
# Setup the underlying Kalman filter storage
self._kalman_filters = {}
# Filter options
self.loglikelihood_burn = loglikelihood_burn
self.results_class = (
results_class if results_class is not None else FilterResults
)
# Options
self.prefix_kalman_filter_map = (
kalman_filter_classes
if kalman_filter_classes is not None
else tools.prefix_kalman_filter_map.copy())
self.set_filter_method(**filter_method_kwargs)
self.set_inversion_method(**inversion_method_kwargs)
self.set_stability_method(**stability_method_kwargs)
self.set_conserve_memory(**conserve_memory_kwargs)
self.set_filter_timing(**filter_timing_kwargs)
self.tolerance = tolerance
# Internal flags
# The _scale internal flag is used because we may want to
# use a fixed scale, in which case we want the flag to the Cython
# Kalman filter to indicate that the scale should not be concentrated
# out, so that self.filter_concentrated = False, but we still want to
# alert the results object that we are viewing the model as one in
# which the scale had been concentrated out for e.g. degree of freedom
# computations.
# This value should always be None, except within the fixed_scale
# context, and should not be modified by users or anywhere else.
self._scale = None
def _clone_kwargs(self, endog, **kwargs):
# See Representation._clone_kwargs for docstring
kwargs = super()._clone_kwargs(endog, **kwargs)
# Get defaults for options
kwargs.setdefault('filter_method', self.filter_method)
kwargs.setdefault('inversion_method', self.inversion_method)
kwargs.setdefault('stability_method', self.stability_method)
kwargs.setdefault('conserve_memory', self.conserve_memory)
kwargs.setdefault('alternate_timing', bool(self.filter_timing))
kwargs.setdefault('tolerance', self.tolerance)
kwargs.setdefault('loglikelihood_burn', self.loglikelihood_burn)
return kwargs
@property
def _kalman_filter(self):
prefix = self.prefix
if prefix in self._kalman_filters:
return self._kalman_filters[prefix]
return None
def _initialize_filter(self, filter_method=None, inversion_method=None,
stability_method=None, conserve_memory=None,
tolerance=None, filter_timing=None,
loglikelihood_burn=None):
if filter_method is None:
filter_method = self.filter_method
if inversion_method is None:
inversion_method = self.inversion_method
if stability_method is None:
stability_method = self.stability_method
if conserve_memory is None:
conserve_memory = self.conserve_memory
if loglikelihood_burn is None:
loglikelihood_burn = self.loglikelihood_burn
if filter_timing is None:
filter_timing = self.filter_timing
if tolerance is None:
tolerance = self.tolerance
# Make sure we have endog
if self.endog is None:
raise RuntimeError('Must bind a dataset to the model before'
' filtering or smoothing.')
# Initialize the representation matrices
prefix, dtype, create_statespace = self._initialize_representation()
# Determine if we need to (re-)create the filter
# (definitely need to recreate if we recreated the _statespace object)
create_filter = create_statespace or prefix not in self._kalman_filters
if not create_filter:
kalman_filter = self._kalman_filters[prefix]
create_filter = (
not kalman_filter.conserve_memory == conserve_memory or
not kalman_filter.loglikelihood_burn == loglikelihood_burn
)
# If the dtype-specific _kalman_filter does not exist (or if we need
# to re-create it), create it
if create_filter:
if prefix in self._kalman_filters:
# Delete the old filter
del self._kalman_filters[prefix]
# Setup the filter
cls = self.prefix_kalman_filter_map[prefix]
self._kalman_filters[prefix] = cls(
self._statespaces[prefix], filter_method, inversion_method,
stability_method, conserve_memory, filter_timing, tolerance,
loglikelihood_burn
)
# Otherwise, update the filter parameters
else:
kalman_filter = self._kalman_filters[prefix]
kalman_filter.set_filter_method(filter_method, False)
kalman_filter.inversion_method = inversion_method
kalman_filter.stability_method = stability_method
kalman_filter.filter_timing = filter_timing
kalman_filter.tolerance = tolerance
# conserve_memory and loglikelihood_burn changes always lead to
# re-created filters
return prefix, dtype, create_filter, create_statespace
[docs]
def set_filter_method(self, filter_method=None, **kwargs):
r"""
Set the filtering method
The filtering method controls aspects of which Kalman filtering
approach will be used.
Parameters
----------
filter_method : int, optional
Bitmask value to set the filter method to. See notes for details.
**kwargs
Keyword arguments may be used to influence the filter method by
setting individual boolean flags. See notes for details.
Notes
-----
The filtering method is defined by a collection of boolean flags, and
is internally stored as a bitmask. The methods available are:
FILTER_CONVENTIONAL
Conventional Kalman filter.
FILTER_UNIVARIATE
Univariate approach to Kalman filtering. Overrides conventional
method if both are specified.
FILTER_COLLAPSED
Collapsed approach to Kalman filtering. Will be used *in addition*
to conventional or univariate filtering.
FILTER_CONCENTRATED
Use the concentrated log-likelihood function. Will be used
*in addition* to the other options.
Note that only the first method is available if using a Scipy version
older than 0.16.
If the bitmask is set directly via the `filter_method` argument, then
the full method must be provided.
If keyword arguments are used to set individual boolean flags, then
the lowercase of the method must be used as an argument name, and the
value is the desired value of the boolean flag (True or False).
Note that the filter method may also be specified by directly modifying
the class attributes which are defined similarly to the keyword
arguments.
The default filtering method is FILTER_CONVENTIONAL.
Examples
--------
>>> mod = sm.tsa.statespace.SARIMAX(range(10))
>>> mod.ssm.filter_method
1
>>> mod.ssm.filter_conventional
True
>>> mod.ssm.filter_univariate = True
>>> mod.ssm.filter_method
17
>>> mod.ssm.set_filter_method(filter_univariate=False,
... filter_collapsed=True)
>>> mod.ssm.filter_method
33
>>> mod.ssm.set_filter_method(filter_method=1)
>>> mod.ssm.filter_conventional
True
>>> mod.ssm.filter_univariate
False
>>> mod.ssm.filter_collapsed
False
>>> mod.ssm.filter_univariate = True
>>> mod.ssm.filter_method
17
"""
if filter_method is not None:
self.filter_method = filter_method
for name in KalmanFilter.filter_methods:
if name in kwargs:
setattr(self, name, kwargs[name])
[docs]
def set_inversion_method(self, inversion_method=None, **kwargs):
r"""
Set the inversion method
The Kalman filter may contain one matrix inversion: that of the
forecast error covariance matrix. The inversion method controls how and
if that inverse is performed.
Parameters
----------
inversion_method : int, optional
Bitmask value to set the inversion method to. See notes for
details.
**kwargs
Keyword arguments may be used to influence the inversion method by
setting individual boolean flags. See notes for details.
Notes
-----
The inversion method is defined by a collection of boolean flags, and
is internally stored as a bitmask. The methods available are:
INVERT_UNIVARIATE
If the endogenous time series is univariate, then inversion can be
performed by simple division. If this flag is set and the time
series is univariate, then division will always be used even if
other flags are also set.
SOLVE_LU
Use an LU decomposition along with a linear solver (rather than
ever actually inverting the matrix).
INVERT_LU
Use an LU decomposition along with typical matrix inversion.
SOLVE_CHOLESKY
Use a Cholesky decomposition along with a linear solver.
INVERT_CHOLESKY
Use an Cholesky decomposition along with typical matrix inversion.
If the bitmask is set directly via the `inversion_method` argument,
then the full method must be provided.
If keyword arguments are used to set individual boolean flags, then
the lowercase of the method must be used as an argument name, and the
value is the desired value of the boolean flag (True or False).
Note that the inversion method may also be specified by directly
modifying the class attributes which are defined similarly to the
keyword arguments.
The default inversion method is `INVERT_UNIVARIATE | SOLVE_CHOLESKY`
Several things to keep in mind are:
- If the filtering method is specified to be univariate, then simple
division is always used regardless of the dimension of the endogenous
time series.
- Cholesky decomposition is about twice as fast as LU decomposition,
but it requires that the matrix be positive definite. While this
should generally be true, it may not be in every case.
- Using a linear solver rather than true matrix inversion is generally
faster and is numerically more stable.
Examples
--------
>>> mod = sm.tsa.statespace.SARIMAX(range(10))
>>> mod.ssm.inversion_method
1
>>> mod.ssm.solve_cholesky
True
>>> mod.ssm.invert_univariate
True
>>> mod.ssm.invert_lu
False
>>> mod.ssm.invert_univariate = False
>>> mod.ssm.inversion_method
8
>>> mod.ssm.set_inversion_method(solve_cholesky=False,
... invert_cholesky=True)
>>> mod.ssm.inversion_method
16
"""
if inversion_method is not None:
self.inversion_method = inversion_method
for name in KalmanFilter.inversion_methods:
if name in kwargs:
setattr(self, name, kwargs[name])
[docs]
def set_stability_method(self, stability_method=None, **kwargs):
r"""
Set the numerical stability method
The Kalman filter is a recursive algorithm that may in some cases
suffer issues with numerical stability. The stability method controls
what, if any, measures are taken to promote stability.
Parameters
----------
stability_method : int, optional
Bitmask value to set the stability method to. See notes for
details.
**kwargs
Keyword arguments may be used to influence the stability method by
setting individual boolean flags. See notes for details.
Notes
-----
The stability method is defined by a collection of boolean flags, and
is internally stored as a bitmask. The methods available are:
STABILITY_FORCE_SYMMETRY = 0x01
If this flag is set, symmetry of the predicted state covariance
matrix is enforced at each iteration of the filter, where each
element is set to the average of the corresponding elements in the
upper and lower triangle.
If the bitmask is set directly via the `stability_method` argument,
then the full method must be provided.
If keyword arguments are used to set individual boolean flags, then
the lowercase of the method must be used as an argument name, and the
value is the desired value of the boolean flag (True or False).
Note that the stability method may also be specified by directly
modifying the class attributes which are defined similarly to the
keyword arguments.
The default stability method is `STABILITY_FORCE_SYMMETRY`
Examples
--------
>>> mod = sm.tsa.statespace.SARIMAX(range(10))
>>> mod.ssm.stability_method
1
>>> mod.ssm.stability_force_symmetry
True
>>> mod.ssm.stability_force_symmetry = False
>>> mod.ssm.stability_method
0
"""
if stability_method is not None:
self.stability_method = stability_method
for name in KalmanFilter.stability_methods:
if name in kwargs:
setattr(self, name, kwargs[name])
[docs]
def set_conserve_memory(self, conserve_memory=None, **kwargs):
r"""
Set the memory conservation method
By default, the Kalman filter computes a number of intermediate
matrices at each iteration. The memory conservation options control
which of those matrices are stored.
Parameters
----------
conserve_memory : int, optional
Bitmask value to set the memory conservation method to. See notes
for details.
**kwargs
Keyword arguments may be used to influence the memory conservation
method by setting individual boolean flags. See notes for details.
Notes
-----
The memory conservation method is defined by a collection of boolean
flags, and is internally stored as a bitmask. The methods available
are:
MEMORY_STORE_ALL
Store all intermediate matrices. This is the default value.
MEMORY_NO_FORECAST_MEAN
Do not store the forecast or forecast errors. If this option is
used, the `predict` method from the results class is unavailable.
MEMORY_NO_FORECAST_COV
Do not store the forecast error covariance matrices.
MEMORY_NO_FORECAST
Do not store the forecast, forecast error, or forecast error
covariance matrices. If this option is used, the `predict` method
from the results class is unavailable.
MEMORY_NO_PREDICTED_MEAN
Do not store the predicted state.
MEMORY_NO_PREDICTED_COV
Do not store the predicted state covariance
matrices.
MEMORY_NO_PREDICTED
Do not store the predicted state or predicted state covariance
matrices.
MEMORY_NO_FILTERED_MEAN
Do not store the filtered state.
MEMORY_NO_FILTERED_COV
Do not store the filtered state covariance
matrices.
MEMORY_NO_FILTERED
Do not store the filtered state or filtered state covariance
matrices.
MEMORY_NO_LIKELIHOOD
Do not store the vector of loglikelihood values for each
observation. Only the sum of the loglikelihood values is stored.
MEMORY_NO_GAIN
Do not store the Kalman gain matrices.
MEMORY_NO_SMOOTHING
Do not store temporary variables related to Kalman smoothing. If
this option is used, smoothing is unavailable.
MEMORY_NO_STD_FORECAST
Do not store standardized forecast errors.
MEMORY_CONSERVE
Do not store any intermediate matrices.
If the bitmask is set directly via the `conserve_memory` argument,
then the full method must be provided.
If keyword arguments are used to set individual boolean flags, then
the lowercase of the method must be used as an argument name, and the
value is the desired value of the boolean flag (True or False).
Note that the memory conservation method may also be specified by
directly modifying the class attributes which are defined similarly to
the keyword arguments.
The default memory conservation method is `MEMORY_STORE_ALL`, so that
all intermediate matrices are stored.
Examples
--------
>>> mod = sm.tsa.statespace.SARIMAX(range(10))
>>> mod.ssm..conserve_memory
0
>>> mod.ssm.memory_no_predicted
False
>>> mod.ssm.memory_no_predicted = True
>>> mod.ssm.conserve_memory
2
>>> mod.ssm.set_conserve_memory(memory_no_filtered=True,
... memory_no_forecast=True)
>>> mod.ssm.conserve_memory
7
"""
if conserve_memory is not None:
self.conserve_memory = conserve_memory
for name in KalmanFilter.memory_options:
if name in kwargs:
setattr(self, name, kwargs[name])
[docs]
def set_filter_timing(self, alternate_timing=None, **kwargs):
r"""
Set the filter timing convention
By default, the Kalman filter follows Durbin and Koopman, 2012, in
initializing the filter with predicted values. Kim and Nelson, 1999,
instead initialize the filter with filtered values, which is
essentially just a different timing convention.
Parameters
----------
alternate_timing : int, optional
Whether or not to use the alternate timing convention. Default is
unspecified.
**kwargs
Keyword arguments may be used to influence the memory conservation
method by setting individual boolean flags. See notes for details.
"""
if alternate_timing is not None:
self.filter_timing = int(alternate_timing)
if 'timing_init_predicted' in kwargs:
self.filter_timing = int(not kwargs['timing_init_predicted'])
if 'timing_init_filtered' in kwargs:
self.filter_timing = int(kwargs['timing_init_filtered'])
[docs]
@contextlib.contextmanager
def fixed_scale(self, scale):
"""
fixed_scale(scale)
Context manager for fixing the scale when FILTER_CONCENTRATED is set
Parameters
----------
scale : numeric
Scale of the model.
Notes
-----
This a no-op if scale is None.
This context manager is most useful in models which are explicitly
concentrating out the scale, so that the set of parameters they are
estimating does not include the scale.
"""
# If a scale was provided, use it and do not concentrate it out of the
# loglikelihood
if scale is not None and scale != 1:
if not self.filter_concentrated:
raise ValueError('Cannot provide scale if filter method does'
' not include FILTER_CONCENTRATED.')
self.filter_concentrated = False
self._scale = scale
obs_cov = self['obs_cov']
state_cov = self['state_cov']
self['obs_cov'] = scale * obs_cov
self['state_cov'] = scale * state_cov
try:
yield
finally:
# If a scale was provided, reset the model
if scale is not None and scale != 1:
self['state_cov'] = state_cov
self['obs_cov'] = obs_cov
self.filter_concentrated = True
self._scale = None
def _filter(self, filter_method=None, inversion_method=None,
stability_method=None, conserve_memory=None,
filter_timing=None, tolerance=None, loglikelihood_burn=None,
complex_step=False):
# Initialize the filter
prefix, dtype, create_filter, create_statespace = (
self._initialize_filter(
filter_method, inversion_method, stability_method,
conserve_memory, filter_timing, tolerance, loglikelihood_burn
)
)
kfilter = self._kalman_filters[prefix]
# Initialize the state
self._initialize_state(prefix=prefix, complex_step=complex_step)
# Run the filter
kfilter()
return kfilter
[docs]
def filter(self, filter_method=None, inversion_method=None,
stability_method=None, conserve_memory=None, filter_timing=None,
tolerance=None, loglikelihood_burn=None, complex_step=False):
r"""
Apply the Kalman filter to the statespace model.
Parameters
----------
filter_method : int, optional
Determines which Kalman filter to use. Default is conventional.
inversion_method : int, optional
Determines which inversion technique to use. Default is by Cholesky
decomposition.
stability_method : int, optional
Determines which numerical stability techniques to use. Default is
to enforce symmetry of the predicted state covariance matrix.
conserve_memory : int, optional
Determines what output from the filter to store. Default is to
store everything.
filter_timing : int, optional
Determines the timing convention of the filter. Default is that
from Durbin and Koopman (2012), in which the filter is initialized
with predicted values.
tolerance : float, optional
The tolerance at which the Kalman filter determines convergence to
steady-state. Default is 1e-19.
loglikelihood_burn : int, optional
The number of initial periods during which the loglikelihood is not
recorded. Default is 0.
Notes
-----
This function by default does not compute variables required for
smoothing.
"""
# Handle memory conservation
if conserve_memory is None:
conserve_memory = self.conserve_memory | MEMORY_NO_SMOOTHING
conserve_memory_cache = self.conserve_memory
self.set_conserve_memory(conserve_memory)
# Run the filter
kfilter = self._filter(
filter_method, inversion_method, stability_method, conserve_memory,
filter_timing, tolerance, loglikelihood_burn, complex_step)
# Create the results object
results = self.results_class(self)
results.update_representation(self)
results.update_filter(kfilter)
# Resent memory conservation
self.set_conserve_memory(conserve_memory_cache)
return results
[docs]
def loglike(self, **kwargs):
r"""
Calculate the loglikelihood associated with the statespace model.
Parameters
----------
**kwargs
Additional keyword arguments to pass to the Kalman filter. See
`KalmanFilter.filter` for more details.
Returns
-------
loglike : float
The joint loglikelihood.
"""
kwargs.setdefault('conserve_memory',
MEMORY_CONSERVE ^ MEMORY_NO_LIKELIHOOD)
kfilter = self._filter(**kwargs)
loglikelihood_burn = kwargs.get('loglikelihood_burn',
self.loglikelihood_burn)
if not (kwargs['conserve_memory'] & MEMORY_NO_LIKELIHOOD):
loglike = np.sum(kfilter.loglikelihood[loglikelihood_burn:])
else:
loglike = np.sum(kfilter.loglikelihood)
# Need to modify the computed log-likelihood to incorporate the
# MLE scale.
if self.filter_method & FILTER_CONCENTRATED:
d = max(loglikelihood_burn, kfilter.nobs_diffuse)
nobs_k_endog = np.sum(
self.k_endog -
np.array(self._statespace.nmissing)[d:])
# In the univariate case, we need to subtract observations
# associated with a singular forecast error covariance matrix
nobs_k_endog -= kfilter.nobs_kendog_univariate_singular
if not (kwargs['conserve_memory'] & MEMORY_NO_LIKELIHOOD):
scale = np.sum(kfilter.scale[d:]) / nobs_k_endog
else:
scale = kfilter.scale[0] / nobs_k_endog
loglike += -0.5 * nobs_k_endog
# Now need to modify this for diffuse initialization, since for
# diffuse periods we only need to add in the scale value part if
# the diffuse forecast error covariance matrix element was singular
if kfilter.nobs_diffuse > 0:
nobs_k_endog -= kfilter.nobs_kendog_diffuse_nonsingular
loglike += -0.5 * nobs_k_endog * np.log(scale)
return loglike
[docs]
def loglikeobs(self, **kwargs):
r"""
Calculate the loglikelihood for each observation associated with the
statespace model.
Parameters
----------
**kwargs
Additional keyword arguments to pass to the Kalman filter. See
`KalmanFilter.filter` for more details.
Notes
-----
If `loglikelihood_burn` is positive, then the entries in the returned
loglikelihood vector are set to be zero for those initial time periods.
Returns
-------
loglike : array of float
Array of loglikelihood values for each observation.
"""
if self.memory_no_likelihood:
raise RuntimeError('Cannot compute loglikelihood if'
' MEMORY_NO_LIKELIHOOD option is selected.')
if not self.filter_method & FILTER_CONCENTRATED:
kwargs.setdefault('conserve_memory',
MEMORY_CONSERVE ^ MEMORY_NO_LIKELIHOOD)
else:
kwargs.setdefault(
'conserve_memory',
MEMORY_CONSERVE ^ (MEMORY_NO_FORECAST | MEMORY_NO_LIKELIHOOD))
kfilter = self._filter(**kwargs)
llf_obs = np.array(kfilter.loglikelihood, copy=True)
loglikelihood_burn = kwargs.get('loglikelihood_burn',
self.loglikelihood_burn)
# If the scale was concentrated out of the log-likelihood function,
# then the llf_obs above is:
# -0.5 * k_endog * log 2 * pi - 0.5 * log |F_t|
# and we need to add in the effect of the scale:
# -0.5 * k_endog * log scale - 0.5 v' F_t^{-1} v / scale
# and note that v' F_t^{-1} is in the _kalman_filter.scale array
# Also note that we need to adjust the nobs and k_endog in both the
# denominator of the scale computation and in the llf_obs adjustment
# to take into account missing values.
if self.filter_method & FILTER_CONCENTRATED:
d = max(loglikelihood_burn, kfilter.nobs_diffuse)
nmissing = np.array(self._statespace.nmissing)
nobs_k_endog = np.sum(self.k_endog - nmissing[d:])
# In the univariate case, we need to subtract observations
# associated with a singular forecast error covariance matrix
nobs_k_endog -= kfilter.nobs_kendog_univariate_singular
scale = np.sum(kfilter.scale[d:]) / nobs_k_endog
# Need to modify this for diffuse initialization, since for
# diffuse periods we only need to add in the scale value if the
# diffuse forecast error covariance matrix element was singular
nsingular = 0
if kfilter.nobs_diffuse > 0:
d = kfilter.nobs_diffuse
Finf = kfilter.forecast_error_diffuse_cov
singular = np.diagonal(Finf).real <= kfilter.tolerance_diffuse
nsingular = np.sum(~singular, axis=1)
scale_obs = np.array(kfilter.scale, copy=True)
llf_obs += -0.5 * (
(self.k_endog - nmissing - nsingular) * np.log(scale) +
scale_obs / scale)
# Set any burned observations to have zero likelihood
llf_obs[:loglikelihood_burn] = 0
return llf_obs
[docs]
def simulate(self, nsimulations, measurement_shocks=None,
state_shocks=None, initial_state=None,
pretransformed_measurement_shocks=True,
pretransformed_state_shocks=True,
pretransformed_initial_state=True,
simulator=None, return_simulator=False,
random_state=None):
r"""
Simulate a new time series following the state space model
Parameters
----------
nsimulations : int
The number of observations to simulate. If the model is
time-invariant this can be any number. If the model is
time-varying, then this number must be less than or equal to the
number
measurement_shocks : array_like, optional
If specified, these are the shocks to the measurement equation,
:math:`\varepsilon_t`. If unspecified, these are automatically
generated using a pseudo-random number generator. If specified,
must be shaped `nsimulations` x `k_endog`, where `k_endog` is the
same as in the state space model.
state_shocks : array_like, optional
If specified, these are the shocks to the state equation,
:math:`\eta_t`. If unspecified, these are automatically
generated using a pseudo-random number generator. If specified,
must be shaped `nsimulations` x `k_posdef` where `k_posdef` is the
same as in the state space model.
initial_state : array_like, optional
If specified, this is the state vector at time zero, which should
be shaped (`k_states` x 1), where `k_states` is the same as in the
state space model. If unspecified, but the model has been
initialized, then that initialization is used. If unspecified and
the model has not been initialized, then a vector of zeros is used.
Note that this is not included in the returned `simulated_states`
array.
pretransformed_measurement_shocks : bool, optional
If `measurement_shocks` is provided, this flag indicates whether it
should be directly used as the shocks. If False, then it is assumed
to contain draws from the standard Normal distribution that must be
transformed using the `obs_cov` covariance matrix. Default is True.
pretransformed_state_shocks : bool, optional
If `state_shocks` is provided, this flag indicates whether it
should be directly used as the shocks. If False, then it is assumed
to contain draws from the standard Normal distribution that must be
transformed using the `state_cov` covariance matrix. Default is
True.
pretransformed_initial_state : bool, optional
If `initial_state` is provided, this flag indicates whether it
should be directly used as the initial_state. If False, then it is
assumed to contain draws from the standard Normal distribution that
must be transformed using the `initial_state_cov` covariance
matrix. Default is True.
return_simulator : bool, optional
Whether or not to return the simulator object. Typically used to
improve performance when performing repeated sampling. Default is
False.
random_state : {None, int, Generator, RandomState}, optionall
If `seed` is None (or `np.random`), the `numpy.random.RandomState`
singleton is used.
If `seed` is an int, a new ``RandomState`` instance is used,
seeded with `seed`.
If `seed` is already a ``Generator`` or ``RandomState`` instance
then that instance is used.
Returns
-------
simulated_obs : ndarray
An (nsimulations x k_endog) array of simulated observations.
simulated_states : ndarray
An (nsimulations x k_states) array of simulated states.
simulator : SimulationSmoothResults
If `return_simulator=True`, then an instance of a simulator is
returned, which can be reused for additional simulations of the
same size.
"""
time_invariant = self.time_invariant
# Check for valid number of simulations
if not time_invariant and nsimulations > self.nobs:
raise ValueError('In a time-varying model, cannot create more'
' simulations than there are observations.')
return self._simulate(
nsimulations,
measurement_disturbance_variates=measurement_shocks,
state_disturbance_variates=state_shocks,
initial_state_variates=initial_state,
pretransformed_measurement_disturbance_variates=(
pretransformed_measurement_shocks),
pretransformed_state_disturbance_variates=(
pretransformed_state_shocks),
pretransformed_initial_state_variates=(
pretransformed_initial_state),
simulator=simulator, return_simulator=return_simulator,
random_state=random_state)
def _simulate(self, nsimulations, simulator=None, random_state=None,
**kwargs):
raise NotImplementedError('Simulation only available through'
' the simulation smoother.')
[docs]
def impulse_responses(self, steps=10, impulse=0, orthogonalized=False,
cumulative=False, direct=False):
r"""
Impulse response function
Parameters
----------
steps : int, optional
The number of steps for which impulse responses are calculated.
Default is 10. Note that the initial impulse is not counted as a
step, so if `steps=1`, the output will have 2 entries.
impulse : int or array_like
If an integer, the state innovation to pulse; must be between 0
and `k_posdef-1` where `k_posdef` is the same as in the state
space model. Alternatively, a custom impulse vector may be
provided; must be a column vector with shape `(k_posdef, 1)`.
orthogonalized : bool, optional
Whether or not to perform impulse using orthogonalized innovations.
Note that this will also affect custum `impulse` vectors. Default
is False.
cumulative : bool, optional
Whether or not to return cumulative impulse responses. Default is
False.
Returns
-------
impulse_responses : ndarray
Responses for each endogenous variable due to the impulse
given by the `impulse` argument. A (steps + 1 x k_endog) array.
Notes
-----
Intercepts in the measurement and state equation are ignored when
calculating impulse responses.
TODO: add note about how for time-varying systems this is - perhaps
counter-intuitively - returning the impulse response within the given
model (i.e. starting at period 0 defined by the model) and it is *not*
doing impulse responses after the end of the model. To compute impulse
responses from arbitrary time points, it is necessary to clone a new
model with the appropriate system matrices.
"""
# We need to add an additional step, since the first simulated value
# will always be zeros (note that we take this value out at the end).
steps += 1
# For time-invariant models, add an additional `step`. This is the
# default for time-invariant models based on the expected behavior for
# ARIMA and VAR models: we want to record the initial impulse and also
# `steps` values of the responses afterwards.
if (self._design.shape[2] == 1 and self._transition.shape[2] == 1 and
self._selection.shape[2] == 1):
steps += 1
# Check for what kind of impulse we want
if type(impulse) is int:
if impulse >= self.k_posdef or impulse < 0:
raise ValueError('Invalid value for `impulse`. Must be the'
' index of one of the state innovations.')
# Create the (non-orthogonalized) impulse vector
idx = impulse
impulse = np.zeros(self.k_posdef)
impulse[idx] = 1
else:
impulse = np.array(impulse)
if impulse.ndim > 1:
impulse = np.squeeze(impulse)
if not impulse.shape == (self.k_posdef,):
raise ValueError('Invalid impulse vector. Must be shaped'
' (%d,)' % self.k_posdef)
# Orthogonalize the impulses, if requested, using Cholesky on the
# first state covariance matrix
if orthogonalized:
state_chol = np.linalg.cholesky(self.state_cov[:, :, 0])
impulse = np.dot(state_chol, impulse)
# If we have time-varying design, transition, or selection matrices,
# then we can't produce more IRFs than we have time points
time_invariant_irf = (
self._design.shape[2] == self._transition.shape[2] ==
self._selection.shape[2] == 1)
# Note: to generate impulse responses following the end of a
# time-varying model, one should `clone` the state space model with the
# new time-varying model, and then compute the IRFs using the cloned
# model
if not time_invariant_irf and steps > self.nobs:
raise ValueError('In a time-varying model, cannot create more'
' impulse responses than there are'
' observations')
# Impulse responses only depend on the design, transition, and
# selection matrices. We set the others to zeros because they must be
# set in the call to `clone`.
# Note: we don't even need selection after the first point, because
# the state shocks will be zeros in every period except the first.
sim_model = self.clone(
endog=np.zeros((steps, self.k_endog), dtype=self.dtype),
obs_intercept=np.zeros(self.k_endog),
design=self['design', :, :, :steps],
obs_cov=np.zeros((self.k_endog, self.k_endog)),
state_intercept=np.zeros(self.k_states),
transition=self['transition', :, :, :steps],
selection=self['selection', :, :, :steps],
state_cov=np.zeros((self.k_posdef, self.k_posdef)))
# Get the impulse response function via simulation of the state
# space model, but with other shocks set to zero
measurement_shocks = np.zeros((steps, self.k_endog))
state_shocks = np.zeros((steps, self.k_posdef))
state_shocks[0] = impulse
initial_state = np.zeros((self.k_states,))
irf, _ = sim_model.simulate(
steps, measurement_shocks=measurement_shocks,
state_shocks=state_shocks, initial_state=initial_state)
# Get the cumulative response if requested
if cumulative:
irf = np.cumsum(irf, axis=0)
# Here we ignore the first value, because it is always zeros (we added
# an additional `step` at the top to account for this).
return irf[1:]
[docs]
class FilterResults(FrozenRepresentation):
"""
Results from applying the Kalman filter to a state space model.
Parameters
----------
model : Representation
A Statespace representation
Attributes
----------
nobs : int
Number of observations.
nobs_diffuse : int
Number of observations under the diffuse Kalman filter.
k_endog : int
The dimension of the observation series.
k_states : int
The dimension of the unobserved state process.
k_posdef : int
The dimension of a guaranteed positive definite
covariance matrix describing the shocks in the
measurement equation.
dtype : dtype
Datatype of representation matrices
prefix : str
BLAS prefix of representation matrices
shapes : dictionary of name,tuple
A dictionary recording the shapes of each of the
representation matrices as tuples.
endog : ndarray
The observation vector.
design : ndarray
The design matrix, :math:`Z`.
obs_intercept : ndarray
The intercept for the observation equation, :math:`d`.
obs_cov : ndarray
The covariance matrix for the observation equation :math:`H`.
transition : ndarray
The transition matrix, :math:`T`.
state_intercept : ndarray
The intercept for the transition equation, :math:`c`.
selection : ndarray
The selection matrix, :math:`R`.
state_cov : ndarray
The covariance matrix for the state equation :math:`Q`.
missing : array of bool
An array of the same size as `endog`, filled
with boolean values that are True if the
corresponding entry in `endog` is NaN and False
otherwise.
nmissing : array of int
An array of size `nobs`, where the ith entry
is the number (between 0 and `k_endog`) of NaNs in
the ith row of the `endog` array.
time_invariant : bool
Whether or not the representation matrices are time-invariant
initialization : str
Kalman filter initialization method.
initial_state : array_like
The state vector used to initialize the Kalamn filter.
initial_state_cov : array_like
The state covariance matrix used to initialize the Kalamn filter.
initial_diffuse_state_cov : array_like
Diffuse state covariance matrix used to initialize the Kalamn filter.
filter_method : int
Bitmask representing the Kalman filtering method
inversion_method : int
Bitmask representing the method used to
invert the forecast error covariance matrix.
stability_method : int
Bitmask representing the methods used to promote
numerical stability in the Kalman filter
recursions.
conserve_memory : int
Bitmask representing the selected memory conservation method.
filter_timing : int
Whether or not to use the alternate timing convention.
tolerance : float
The tolerance at which the Kalman filter
determines convergence to steady-state.
loglikelihood_burn : int
The number of initial periods during which
the loglikelihood is not recorded.
converged : bool
Whether or not the Kalman filter converged.
period_converged : int
The time period in which the Kalman filter converged.
filtered_state : ndarray
The filtered state vector at each time period.
filtered_state_cov : ndarray
The filtered state covariance matrix at each time period.
predicted_state : ndarray
The predicted state vector at each time period.
predicted_state_cov : ndarray
The predicted state covariance matrix at each time period.
forecast_error_diffuse_cov : ndarray
Diffuse forecast error covariance matrix at each time period.
predicted_diffuse_state_cov : ndarray
The predicted diffuse state covariance matrix at each time period.
kalman_gain : ndarray
The Kalman gain at each time period.
forecasts : ndarray
The one-step-ahead forecasts of observations at each time period.
forecasts_error : ndarray
The forecast errors at each time period.
forecasts_error_cov : ndarray
The forecast error covariance matrices at each time period.
llf_obs : ndarray
The loglikelihood values at each time period.
"""
_filter_attributes = [
'filter_method', 'inversion_method', 'stability_method',
'conserve_memory', 'filter_timing', 'tolerance', 'loglikelihood_burn',
'converged', 'period_converged', 'filtered_state',
'filtered_state_cov', 'predicted_state', 'predicted_state_cov',
'forecasts_error_diffuse_cov', 'predicted_diffuse_state_cov',
'tmp1', 'tmp2', 'tmp3', 'tmp4', 'forecasts',
'forecasts_error', 'forecasts_error_cov', 'llf', 'llf_obs',
'collapsed_forecasts', 'collapsed_forecasts_error',
'collapsed_forecasts_error_cov', 'scale'
]
_filter_options = (
KalmanFilter.filter_methods + KalmanFilter.stability_methods +
KalmanFilter.inversion_methods + KalmanFilter.memory_options
)
_attributes = FrozenRepresentation._model_attributes + _filter_attributes
def __init__(self, model):
super().__init__(model)
# Setup caches for uninitialized objects
self._kalman_gain = None
self._standardized_forecasts_error = None
[docs]
def update_representation(self, model, only_options=False):
"""
Update the results to match a given model
Parameters
----------
model : Representation
The model object from which to take the updated values.
only_options : bool, optional
If set to true, only the filter options are updated, and the state
space representation is not updated. Default is False.
Notes
-----
This method is rarely required except for internal usage.
"""
if not only_options:
super().update_representation(model)
# Save the options as boolean variables
for name in self._filter_options:
setattr(self, name, getattr(model, name, None))
[docs]
def update_filter(self, kalman_filter):
"""
Update the filter results
Parameters
----------
kalman_filter : statespace.kalman_filter.KalmanFilter
The model object from which to take the updated values.
Notes
-----
This method is rarely required except for internal usage.
"""
# State initialization
self.initial_state = np.array(
kalman_filter.model.initial_state, copy=True
)
self.initial_state_cov = np.array(
kalman_filter.model.initial_state_cov, copy=True
)
# Save Kalman filter parameters
self.filter_method = kalman_filter.filter_method
self.inversion_method = kalman_filter.inversion_method
self.stability_method = kalman_filter.stability_method
self.conserve_memory = kalman_filter.conserve_memory
self.filter_timing = kalman_filter.filter_timing
self.tolerance = kalman_filter.tolerance
self.loglikelihood_burn = kalman_filter.loglikelihood_burn
# Save Kalman filter output
self.converged = bool(kalman_filter.converged)
self.period_converged = kalman_filter.period_converged
self.univariate_filter = np.array(kalman_filter.univariate_filter,
copy=True)
self.filtered_state = np.array(kalman_filter.filtered_state, copy=True)
self.filtered_state_cov = np.array(
kalman_filter.filtered_state_cov, copy=True
)
self.predicted_state = np.array(
kalman_filter.predicted_state, copy=True
)
self.predicted_state_cov = np.array(
kalman_filter.predicted_state_cov, copy=True
)
# Reset caches
has_missing = np.sum(self.nmissing) > 0
if not (self.memory_no_std_forecast or self.invert_lu or
self.solve_lu or self.filter_collapsed):
if has_missing:
self._standardized_forecasts_error = np.array(
reorder_missing_vector(
kalman_filter.standardized_forecast_error,
self.missing, prefix=self.prefix))
else:
self._standardized_forecasts_error = np.array(
kalman_filter.standardized_forecast_error, copy=True)
else:
self._standardized_forecasts_error = None
# In the partially missing data case, all entries will
# be in the upper left submatrix rather than the correct placement
# Re-ordering does not make sense in the collapsed case.
if has_missing and (not self.memory_no_gain and
not self.filter_collapsed):
self._kalman_gain = np.array(reorder_missing_matrix(
kalman_filter.kalman_gain, self.missing, reorder_cols=True,
prefix=self.prefix))
self.tmp1 = np.array(reorder_missing_matrix(
kalman_filter.tmp1, self.missing, reorder_cols=True,
prefix=self.prefix))
self.tmp2 = np.array(reorder_missing_vector(
kalman_filter.tmp2, self.missing, prefix=self.prefix))
self.tmp3 = np.array(reorder_missing_matrix(
kalman_filter.tmp3, self.missing, reorder_rows=True,
prefix=self.prefix))
self.tmp4 = np.array(reorder_missing_matrix(
kalman_filter.tmp4, self.missing, reorder_cols=True,
reorder_rows=True, prefix=self.prefix))
else:
if not self.memory_no_gain:
self._kalman_gain = np.array(
kalman_filter.kalman_gain, copy=True)
self.tmp1 = np.array(kalman_filter.tmp1, copy=True)
self.tmp2 = np.array(kalman_filter.tmp2, copy=True)
self.tmp3 = np.array(kalman_filter.tmp3, copy=True)
self.tmp4 = np.array(kalman_filter.tmp4, copy=True)
self.M = np.array(kalman_filter.M, copy=True)
self.M_diffuse = np.array(kalman_filter.M_inf, copy=True)
# Note: use forecasts rather than forecast, so as not to interfer
# with the `forecast` methods in subclasses
self.forecasts = np.array(kalman_filter.forecast, copy=True)
self.forecasts_error = np.array(
kalman_filter.forecast_error, copy=True
)
self.forecasts_error_cov = np.array(
kalman_filter.forecast_error_cov, copy=True
)
# Note: below we will set self.llf, and in the memory_no_likelihood
# case we will replace self.llf_obs = None at that time.
self.llf_obs = np.array(kalman_filter.loglikelihood, copy=True)
# Diffuse objects
self.nobs_diffuse = kalman_filter.nobs_diffuse
self.initial_diffuse_state_cov = None
self.forecasts_error_diffuse_cov = None
self.predicted_diffuse_state_cov = None
if self.nobs_diffuse > 0:
self.initial_diffuse_state_cov = np.array(
kalman_filter.model.initial_diffuse_state_cov, copy=True)
self.predicted_diffuse_state_cov = np.array(
kalman_filter.predicted_diffuse_state_cov, copy=True)
if has_missing and not self.filter_collapsed:
self.forecasts_error_diffuse_cov = np.array(
reorder_missing_matrix(
kalman_filter.forecast_error_diffuse_cov,
self.missing, reorder_cols=True, reorder_rows=True,
prefix=self.prefix))
else:
self.forecasts_error_diffuse_cov = np.array(
kalman_filter.forecast_error_diffuse_cov, copy=True)
# If there was missing data, save the original values from the Kalman
# filter output, since below will set the values corresponding to
# the missing observations to nans.
self.missing_forecasts = None
self.missing_forecasts_error = None
self.missing_forecasts_error_cov = None
if np.sum(self.nmissing) > 0:
# Copy the provided arrays (which are as the Kalman filter dataset)
# into new variables
self.missing_forecasts = np.copy(self.forecasts)
self.missing_forecasts_error = np.copy(self.forecasts_error)
self.missing_forecasts_error_cov = (
np.copy(self.forecasts_error_cov)
)
# Save the collapsed values
self.collapsed_forecasts = None
self.collapsed_forecasts_error = None
self.collapsed_forecasts_error_cov = None
if self.filter_collapsed:
# Copy the provided arrays (which are from the collapsed dataset)
# into new variables
self.collapsed_forecasts = self.forecasts[:self.k_states, :]
self.collapsed_forecasts_error = (
self.forecasts_error[:self.k_states, :]
)
self.collapsed_forecasts_error_cov = (
self.forecasts_error_cov[:self.k_states, :self.k_states, :]
)
# Recreate the original arrays (which should be from the original
# dataset) in the appropriate dimension
dtype = self.collapsed_forecasts.dtype
self.forecasts = np.zeros((self.k_endog, self.nobs), dtype=dtype)
self.forecasts_error = np.zeros((self.k_endog, self.nobs),
dtype=dtype)
self.forecasts_error_cov = (
np.zeros((self.k_endog, self.k_endog, self.nobs), dtype=dtype)
)
# Fill in missing values in the forecast, forecast error, and
# forecast error covariance matrix (this is required due to how the
# Kalman filter implements observations that are either partly or
# completely missing)
# Construct the predictions, forecasts
can_compute_mean = not (self.memory_no_forecast_mean or
self.memory_no_predicted_mean)
can_compute_cov = not (self.memory_no_forecast_cov or
self.memory_no_predicted_cov)
if can_compute_mean or can_compute_cov:
for t in range(self.nobs):
design_t = 0 if self.design.shape[2] == 1 else t
obs_cov_t = 0 if self.obs_cov.shape[2] == 1 else t
obs_intercept_t = 0 if self.obs_intercept.shape[1] == 1 else t
# For completely missing observations, the Kalman filter will
# produce forecasts, but forecast errors and the forecast
# error covariance matrix will be zeros - make them nan to
# improve clarity of results.
if self.nmissing[t] > 0:
mask = ~self.missing[:, t].astype(bool)
# We can recover forecasts
# For partially missing observations, the Kalman filter
# will produce all elements (forecasts, forecast errors,
# forecast error covariance matrices) as usual, but their
# dimension will only be equal to the number of non-missing
# elements, and their location in memory will be in the
# first blocks (e.g. for the forecasts_error, the first
# k_endog - nmissing[t] columns will be filled in),
# regardless of which endogenous variables they refer to
# (i.e. the non- missing endogenous variables for that
# observation). Furthermore, the forecast error covariance
# matrix is only valid for those elements. What is done is
# to set all elements to nan for these observations so that
# they are flagged as missing. The variables
# missing_forecasts, etc. then provide the forecasts, etc.
# provided by the Kalman filter, from which the data can be
# retrieved if desired.
if can_compute_mean:
self.forecasts[:, t] = np.dot(
self.design[:, :, design_t],
self.predicted_state[:, t]
) + self.obs_intercept[:, obs_intercept_t]
self.forecasts_error[:, t] = np.nan
self.forecasts_error[mask, t] = (
self.endog[mask, t] - self.forecasts[mask, t])
# TODO: We should only fill in the non-masked elements of
# this array. Also, this will give the multivariate version
# even if univariate filtering was selected. Instead, we
# should use the reordering methods and then replace the
# masked values with NaNs
if can_compute_cov:
self.forecasts_error_cov[:, :, t] = np.dot(
np.dot(self.design[:, :, design_t],
self.predicted_state_cov[:, :, t]),
self.design[:, :, design_t].T
) + self.obs_cov[:, :, obs_cov_t]
# In the collapsed case, everything just needs to be rebuilt
# for the original observed data, since the Kalman filter
# produced these values for the collapsed data.
elif self.filter_collapsed:
if can_compute_mean:
self.forecasts[:, t] = np.dot(
self.design[:, :, design_t],
self.predicted_state[:, t]
) + self.obs_intercept[:, obs_intercept_t]
self.forecasts_error[:, t] = (
self.endog[:, t] - self.forecasts[:, t]
)
if can_compute_cov:
self.forecasts_error_cov[:, :, t] = np.dot(
np.dot(self.design[:, :, design_t],
self.predicted_state_cov[:, :, t]),
self.design[:, :, design_t].T
) + self.obs_cov[:, :, obs_cov_t]
# Note: if we concentrated out the scale, need to adjust the
# loglikelihood values and all of the covariance matrices and the
# values that depend on the covariance matrices
# Note: concentrated computation is not permitted with collapsed
# version, so we do not need to modify collapsed arrays.
self.scale = 1.
if self.filter_concentrated and self.model._scale is None:
d = max(self.loglikelihood_burn, self.nobs_diffuse)
# Compute the scale
nmissing = np.array(kalman_filter.model.nmissing)
nobs_k_endog = np.sum(self.k_endog - nmissing[d:])
# In the univariate case, we need to subtract observations
# associated with a singular forecast error covariance matrix
nobs_k_endog -= kalman_filter.nobs_kendog_univariate_singular
scale_obs = np.array(kalman_filter.scale, copy=True)
if not self.memory_no_likelihood:
self.scale = np.sum(scale_obs[d:]) / nobs_k_endog
else:
self.scale = scale_obs[0] / nobs_k_endog
# Need to modify this for diffuse initialization, since for
# diffuse periods we only need to add in the scale value if the
# diffuse forecast error covariance matrix element was singular
nsingular = 0
if kalman_filter.nobs_diffuse > 0:
Finf = kalman_filter.forecast_error_diffuse_cov
singular = (np.diagonal(Finf).real <=
kalman_filter.tolerance_diffuse)
nsingular = np.sum(~singular, axis=1)
# Adjust the loglikelihood obs (see `KalmanFilter.loglikeobs` for
# defaults on the adjustment)
if not self.memory_no_likelihood:
self.llf_obs += -0.5 * (
(self.k_endog - nmissing - nsingular) * np.log(self.scale)
+ scale_obs / self.scale)
else:
self.llf_obs[0] += -0.5 * np.squeeze(
np.sum(
(self.k_endog - nmissing - nsingular)
* np.log(self.scale)
)
+ scale_obs / self.scale
)
# Scale the filter output
self.obs_cov = self.obs_cov * self.scale
self.state_cov = self.state_cov * self.scale
self.initial_state_cov = self.initial_state_cov * self.scale
self.predicted_state_cov = self.predicted_state_cov * self.scale
self.filtered_state_cov = self.filtered_state_cov * self.scale
self.forecasts_error_cov = self.forecasts_error_cov * self.scale
if self.missing_forecasts_error_cov is not None:
self.missing_forecasts_error_cov = (
self.missing_forecasts_error_cov * self.scale)
# Note: do not have to adjust the Kalman gain or tmp4
self.tmp1 = self.tmp1 * self.scale
self.tmp2 = self.tmp2 / self.scale
self.tmp3 = self.tmp3 / self.scale
if not (self.memory_no_std_forecast or
self.invert_lu or
self.solve_lu or
self.filter_collapsed):
self._standardized_forecasts_error = (
self._standardized_forecasts_error / self.scale**0.5)
# The self.model._scale value is only not None within a fixed_scale
# context, in which case it is set and indicates that we should
# generally view this results object as using a concentrated scale
# (e.g. for d.o.f. computations), but because the fixed scale was
# actually applied to the model prior to filtering, we do not need to
# make any adjustments to the filter output, etc.
elif self.model._scale is not None:
self.filter_concentrated = True
self.scale = self.model._scale
# Now, save self.llf, and handle the memory_no_likelihood case
if not self.memory_no_likelihood:
self.llf = np.sum(self.llf_obs[self.loglikelihood_burn:])
else:
self.llf = self.llf_obs[0]
self.llf_obs = None
@property
def kalman_gain(self):
"""
Kalman gain matrices
"""
if self._kalman_gain is None:
# k x n
self._kalman_gain = np.zeros(
(self.k_states, self.k_endog, self.nobs), dtype=self.dtype)
for t in range(self.nobs):
# In the case of entirely missing observations, let the Kalman
# gain be zeros.
if self.nmissing[t] == self.k_endog:
continue
design_t = 0 if self.design.shape[2] == 1 else t
transition_t = 0 if self.transition.shape[2] == 1 else t
if self.nmissing[t] == 0:
self._kalman_gain[:, :, t] = np.dot(
np.dot(
self.transition[:, :, transition_t],
self.predicted_state_cov[:, :, t]
),
np.dot(
np.transpose(self.design[:, :, design_t]),
np.linalg.inv(self.forecasts_error_cov[:, :, t])
)
)
else:
mask = ~self.missing[:, t].astype(bool)
F = self.forecasts_error_cov[np.ix_(mask, mask, [t])]
self._kalman_gain[:, mask, t] = np.dot(
np.dot(
self.transition[:, :, transition_t],
self.predicted_state_cov[:, :, t]
),
np.dot(
np.transpose(self.design[mask, :, design_t]),
np.linalg.inv(F[:, :, 0])
)
)
return self._kalman_gain
@property
def standardized_forecasts_error(self):
r"""
Standardized forecast errors
Notes
-----
The forecast errors produced by the Kalman filter are
.. math::
v_t \sim N(0, F_t)
Hypothesis tests are usually applied to the standardized residuals
.. math::
v_t^s = B_t v_t \sim N(0, I)
where :math:`B_t = L_t^{-1}` and :math:`F_t = L_t L_t'`; then
:math:`F_t^{-1} = (L_t')^{-1} L_t^{-1} = B_t' B_t`; :math:`B_t`
and :math:`L_t` are lower triangular. Finally,
:math:`B_t v_t \sim N(0, B_t F_t B_t')` and
:math:`B_t F_t B_t' = L_t^{-1} L_t L_t' (L_t')^{-1} = I`.
Thus we can rewrite :math:`v_t^s = L_t^{-1} v_t` or
:math:`L_t v_t^s = v_t`; the latter equation is the form required to
use a linear solver to recover :math:`v_t^s`. Since :math:`L_t` is
lower triangular, we can use a triangular solver (?TRTRS).
"""
if (self._standardized_forecasts_error is None
and not self.memory_no_forecast):
if self.k_endog == 1:
self._standardized_forecasts_error = (
self.forecasts_error /
self.forecasts_error_cov[0, 0, :]**0.5)
else:
from scipy import linalg
self._standardized_forecasts_error = np.zeros(
self.forecasts_error.shape, dtype=self.dtype)
for t in range(self.forecasts_error_cov.shape[2]):
if self.nmissing[t] > 0:
self._standardized_forecasts_error[:, t] = np.nan
if self.nmissing[t] < self.k_endog:
mask = ~self.missing[:, t].astype(bool)
F = self.forecasts_error_cov[np.ix_(mask, mask, [t])]
try:
upper, _ = linalg.cho_factor(F[:, :, 0])
self._standardized_forecasts_error[mask, t] = (
linalg.solve_triangular(
upper, self.forecasts_error[mask, t],
trans=1))
except linalg.LinAlgError:
self._standardized_forecasts_error[mask, t] = (
np.nan)
return self._standardized_forecasts_error
[docs]
def predict(self, start=None, end=None, dynamic=None, **kwargs):
r"""
In-sample and out-of-sample prediction for state space models generally
Parameters
----------
start : int, optional
Zero-indexed observation number at which to start prediction, i.e.,
the first prediction will be at start.
end : int, optional
Zero-indexed observation number at which to end prediction, i.e.,
the last prediction will be at end.
dynamic : int, optional
Offset relative to `start` at which to begin dynamic prediction.
Prior to this observation, true endogenous values will be used for
prediction; starting with this observation and continuing through
the end of prediction, predicted endogenous values will be used
instead.
**kwargs
If the prediction range is outside of the sample range, any
of the state space representation matrices that are time-varying
must have updated values provided for the out-of-sample range.
For example, of `obs_intercept` is a time-varying component and
the prediction range extends 10 periods beyond the end of the
sample, a (`k_endog` x 10) matrix must be provided with the new
intercept values.
Returns
-------
results : kalman_filter.PredictionResults
A PredictionResults object.
Notes
-----
All prediction is performed by applying the deterministic part of the
measurement equation using the predicted state variables.
Out-of-sample prediction first applies the Kalman filter to missing
data for the number of periods desired to obtain the predicted states.
"""
# Get the start and the end of the entire prediction range
if start is None:
start = 0
elif start < 0:
raise ValueError('Cannot predict values previous to the sample.')
if end is None:
end = self.nobs
# Prediction and forecasting is performed by iterating the Kalman
# Kalman filter through the entire range [0, end]
# Then, everything is returned corresponding to the range [start, end].
# In order to perform the calculations, the range is separately split
# up into the following categories:
# - static: (in-sample) the Kalman filter is run as usual
# - dynamic: (in-sample) the Kalman filter is run, but on missing data
# - forecast: (out-of-sample) the Kalman filter is run, but on missing
# data
# Short-circuit if end is before start
if end <= start:
raise ValueError('End of prediction must be after start.')
# Get the number of forecasts to make after the end of the sample
nforecast = max(0, end - self.nobs)
# Get the number of dynamic prediction periods
# If `dynamic=True`, then assume that we want to begin dynamic
# prediction at the start of the sample prediction.
if dynamic is True:
dynamic = 0
# If `dynamic=False`, then assume we want no dynamic prediction
if dynamic is False:
dynamic = None
# Check validity of dynamic and warn or error if issues
dynamic, ndynamic = _check_dynamic(dynamic, start, end, self.nobs)
# Get the number of in-sample static predictions
if dynamic is None:
nstatic = min(end, self.nobs) - min(start, self.nobs)
else:
# (use max(., 0), since dynamic can be prior to start)
nstatic = max(dynamic - start, 0)
# Cannot do in-sample prediction if we do not have appropriate
# arrays (we can do out-of-sample forecasting, however)
if nstatic > 0 and self.memory_no_forecast_mean:
raise ValueError('In-sample prediction is not available if memory'
' conservation has been used to avoid storing'
' forecast means.')
# Cannot do dynamic in-sample prediction if we do not have appropriate
# arrays (we can do out-of-sample forecasting, however)
if ndynamic > 0 and self.memory_no_predicted:
raise ValueError('In-sample dynamic prediction is not available if'
' memory conservation has been used to avoid'
' storing forecasted or predicted state means'
' or covariances.')
# Construct the predicted state and covariance matrix for each time
# period depending on whether that time period corresponds to
# one-step-ahead prediction, dynamic prediction, or out-of-sample
# forecasting.
# If we only have simple prediction, then we can use the already saved
# Kalman filter output
if ndynamic == 0 and nforecast == 0:
results = self
oos_results = None
# If we have dynamic prediction or forecasting, then we need to
# re-apply the Kalman filter
else:
# Figure out the period for which we need to run the Kalman filter
if dynamic is not None:
kf_start = min(dynamic, self.nobs)
else:
kf_start = self.nobs
kf_end = end
# Make start, end consistent with the results that we're generating
# start = max(start - kf_start, 0)
# end = kf_end - kf_start
# We must at least store forecasts and predictions
kwargs['conserve_memory'] = (
self.conserve_memory & ~MEMORY_NO_FORECAST &
~MEMORY_NO_PREDICTED)
# Can't use Chandrasekhar recursions for prediction
kwargs['filter_method'] = (
self.model.filter_method & ~FILTER_CHANDRASEKHAR)
# TODO: there is a corner case here when the filter has not
# exited the diffuse filter, in which case this known
# initialization is not correct.
# Even if we have not stored all predicted values (means and covs),
# we can still do pure out-of-sample forecasting because we will
# always have stored the last predicted values. In this case, we
# will initialize the forecasting filter with these values
if self.memory_no_predicted:
constant = self.predicted_state[..., -1]
stationary_cov = self.predicted_state_cov[..., -1]
# Otherwise initialize with the predicted state / cov from the
# existing results, at index kf_start (note that the time
# dimension of predicted_state and predicted_state_cov is
# self.nobs + 1; so e.g. in the case of pure forecasting we should
# be using the very last predicted state and predicted state cov
# elements, and kf_start will equal self.nobs which is correct)
else:
constant = self.predicted_state[..., kf_start]
stationary_cov = self.predicted_state_cov[..., kf_start]
kwargs.update({'initialization': 'known',
'constant': constant,
'stationary_cov': stationary_cov})
# Construct the new endogenous array.
endog = np.zeros((nforecast, self.k_endog)) * np.nan
model = self.model.extend(
endog, start=kf_start, end=kf_end - nforecast, **kwargs)
# Have to retroactively modify the model's endog
if ndynamic > 0:
model.endog[:, -(ndynamic + nforecast):] = np.nan
with model.fixed_scale(self.scale):
oos_results = model.filter()
results = self
return PredictionResults(results, start, end, nstatic, ndynamic,
nforecast, oos_results=oos_results)
[docs]
class PredictionResults(FilterResults):
r"""
Results of in-sample and out-of-sample prediction for state space models
generally
Parameters
----------
results : FilterResults
Output from filtering, corresponding to the prediction desired
start : int
Zero-indexed observation number at which to start forecasting,
i.e., the first forecast will be at start.
end : int
Zero-indexed observation number at which to end forecasting, i.e.,
the last forecast will be at end.
nstatic : int
Number of in-sample static predictions (these are always the first
elements of the prediction output).
ndynamic : int
Number of in-sample dynamic predictions (these always follow the static
predictions directly, and are directly followed by the forecasts).
nforecast : int
Number of in-sample forecasts (these always follow the dynamic
predictions directly).
Attributes
----------
npredictions : int
Number of observations in the predicted series; this is not necessarily
the same as the number of observations in the original model from which
prediction was performed.
start : int
Zero-indexed observation number at which to start prediction,
i.e., the first predict will be at `start`; this is relative to the
original model from which prediction was performed.
end : int
Zero-indexed observation number at which to end prediction,
i.e., the last predict will be at `end`; this is relative to the
original model from which prediction was performed.
nstatic : int
Number of in-sample static predictions.
ndynamic : int
Number of in-sample dynamic predictions.
nforecast : int
Number of in-sample forecasts.
endog : ndarray
The observation vector.
design : ndarray
The design matrix, :math:`Z`.
obs_intercept : ndarray
The intercept for the observation equation, :math:`d`.
obs_cov : ndarray
The covariance matrix for the observation equation :math:`H`.
transition : ndarray
The transition matrix, :math:`T`.
state_intercept : ndarray
The intercept for the transition equation, :math:`c`.
selection : ndarray
The selection matrix, :math:`R`.
state_cov : ndarray
The covariance matrix for the state equation :math:`Q`.
filtered_state : ndarray
The filtered state vector at each time period.
filtered_state_cov : ndarray
The filtered state covariance matrix at each time period.
predicted_state : ndarray
The predicted state vector at each time period.
predicted_state_cov : ndarray
The predicted state covariance matrix at each time period.
forecasts : ndarray
The one-step-ahead forecasts of observations at each time period.
forecasts_error : ndarray
The forecast errors at each time period.
forecasts_error_cov : ndarray
The forecast error covariance matrices at each time period.
Notes
-----
The provided ranges must be conformable, meaning that it must be that
`end - start == nstatic + ndynamic + nforecast`.
This class is essentially a view to the FilterResults object, but
returning the appropriate ranges for everything.
"""
representation_attributes = [
'endog', 'design', 'obs_intercept',
'obs_cov', 'transition', 'state_intercept', 'selection',
'state_cov'
]
filter_attributes = [
'filtered_state', 'filtered_state_cov',
'predicted_state', 'predicted_state_cov',
'forecasts', 'forecasts_error', 'forecasts_error_cov'
]
smoother_attributes = [
'smoothed_state', 'smoothed_state_cov',
]
def __init__(self, results, start, end, nstatic, ndynamic, nforecast,
oos_results=None):
# Save the filter results object
self.results = results
self.oos_results = oos_results
# Save prediction ranges
self.npredictions = start - end
self.start = start
self.end = end
self.nstatic = nstatic
self.ndynamic = ndynamic
self.nforecast = nforecast
self._predicted_signal = None
self._predicted_signal_cov = None
self._filtered_signal = None
self._filtered_signal_cov = None
self._smoothed_signal = None
self._smoothed_signal_cov = None
self._filtered_forecasts = None
self._filtered_forecasts_error_cov = None
self._smoothed_forecasts = None
self._smoothed_forecasts_error_cov = None
[docs]
def clear(self):
attributes = (['endog'] + self.representation_attributes
+ self.filter_attributes)
for attr in attributes:
_attr = '_' + attr
if hasattr(self, _attr):
delattr(self, _attr)
def __getattr__(self, attr):
"""
Provide access to the representation and filtered output in the
appropriate range (`start` - `end`).
"""
# Prevent infinite recursive lookups
if attr[0] == '_':
raise AttributeError("'%s' object has no attribute '%s'" %
(self.__class__.__name__, attr))
_attr = '_' + attr
# Cache the attribute
if not hasattr(self, _attr):
if attr == 'endog' or attr in self.filter_attributes:
# Get a copy
value = getattr(self.results, attr).copy()
if self.ndynamic > 0:
end = self.end - self.ndynamic - self.nforecast
value = value[..., :end]
if self.oos_results is not None:
oos_value = getattr(self.oos_results, attr).copy()
# Note that the last element of the results predicted state
# and state cov will overlap with the first element of the
# oos predicted state and state cov, so eliminate the
# last element of the results versions
# But if we have dynamic prediction, then we have already
# eliminated the last element of the predicted state, so
# we do not need to do it here.
if self.ndynamic == 0 and attr[:9] == 'predicted':
value = value[..., :-1]
value = np.concatenate([value, oos_value], axis=-1)
# Subset to the correct time frame
value = value[..., self.start:self.end]
elif attr in self.smoother_attributes:
if self.ndynamic > 0:
raise NotImplementedError(
'Cannot retrieve smoothed attributes when using'
' dynamic prediction, since the information set used'
' to compute the smoothed results differs from the'
' information set implied by the dynamic prediction.')
# Get a copy
value = getattr(self.results, attr).copy()
# The oos_results object is only dynamic or out-of-sample,
# so filtered == smoothed
if self.oos_results is not None:
filtered_attr = 'filtered' + attr[8:]
oos_value = getattr(self.oos_results, filtered_attr).copy()
value = np.concatenate([value, oos_value], axis=-1)
# Subset to the correct time frame
value = value[..., self.start:self.end]
elif attr in self.representation_attributes:
value = getattr(self.results, attr).copy()
# If a time-invariant matrix, return it. Otherwise, subset to
# the correct period.
if value.shape[-1] == 1:
value = value[..., 0]
else:
if self.ndynamic > 0:
end = self.end - self.ndynamic - self.nforecast
value = value[..., :end]
if self.oos_results is not None:
oos_value = getattr(self.oos_results, attr).copy()
value = np.concatenate([value, oos_value], axis=-1)
value = value[..., self.start:self.end]
else:
raise AttributeError("'%s' object has no attribute '%s'" %
(self.__class__.__name__, attr))
setattr(self, _attr, value)
return getattr(self, _attr)
def _compute_forecasts(self, states, states_cov, signal_only=False):
d = self.obs_intercept
Z = self.design
H = self.obs_cov
if d.ndim == 1:
d = d[:, None]
if Z.ndim == 2:
if not signal_only:
forecasts = d + Z @ states
forecasts_error_cov = (
Z[None, ...] @ states_cov.T @ Z.T[None, ...] + H.T).T
else:
forecasts = Z @ states
forecasts_error_cov = (
Z[None, ...] @ states_cov.T @ Z.T[None, ...]).T
else:
if not signal_only:
forecasts = d + (Z * states[None, :, :]).sum(axis=1)
tmp = Z[:, None, ...] * states_cov[None, ...]
tmp = (tmp[:, :, :, None, :]
* Z.transpose(1, 0, 2)[None, :, None, ...])
forecasts_error_cov = (tmp.sum(axis=1).sum(axis=1).T + H.T).T
else:
forecasts = (Z * states[None, :, :]).sum(axis=1)
tmp = Z[:, None, ...] * states_cov[None, ...]
tmp = (tmp[:, :, :, None, :]
* Z.transpose(1, 0, 2)[None, :, None, ...])
forecasts_error_cov = tmp.sum(axis=1).sum(axis=1)
return forecasts, forecasts_error_cov
@property
def predicted_signal(self):
if self._predicted_signal is None:
self._predicted_signal, self._predicted_signal_cov = (
self._compute_forecasts(self.predicted_state,
self.predicted_state_cov,
signal_only=True))
return self._predicted_signal
@property
def predicted_signal_cov(self):
if self._predicted_signal_cov is None:
self._predicted_signal, self._predicted_signal_cov = (
self._compute_forecasts(self.predicted_state,
self.predicted_state_cov,
signal_only=True))
return self._predicted_signal_cov
@property
def filtered_signal(self):
if self._filtered_signal is None:
self._filtered_signal, self._filtered_signal_cov = (
self._compute_forecasts(self.filtered_state,
self.filtered_state_cov,
signal_only=True))
return self._filtered_signal
@property
def filtered_signal_cov(self):
if self._filtered_signal_cov is None:
self._filtered_signal, self._filtered_signal_cov = (
self._compute_forecasts(self.filtered_state,
self.filtered_state_cov,
signal_only=True))
return self._filtered_signal_cov
@property
def smoothed_signal(self):
if self._smoothed_signal is None:
self._smoothed_signal, self._smoothed_signal_cov = (
self._compute_forecasts(self.smoothed_state,
self.smoothed_state_cov,
signal_only=True))
return self._smoothed_signal
@property
def smoothed_signal_cov(self):
if self._smoothed_signal_cov is None:
self._smoothed_signal, self._smoothed_signal_cov = (
self._compute_forecasts(self.smoothed_state,
self.smoothed_state_cov,
signal_only=True))
return self._smoothed_signal_cov
@property
def filtered_forecasts(self):
if self._filtered_forecasts is None:
self._filtered_forecasts, self._filtered_forecasts_cov = (
self._compute_forecasts(self.filtered_state,
self.filtered_state_cov))
return self._filtered_forecasts
@property
def filtered_forecasts_error_cov(self):
if self._filtered_forecasts_cov is None:
self._filtered_forecasts, self._filtered_forecasts_cov = (
self._compute_forecasts(self.filtered_state,
self.filtered_state_cov))
return self._filtered_forecasts_cov
@property
def smoothed_forecasts(self):
if self._smoothed_forecasts is None:
self._smoothed_forecasts, self._smoothed_forecasts_cov = (
self._compute_forecasts(self.smoothed_state,
self.smoothed_state_cov))
return self._smoothed_forecasts
@property
def smoothed_forecasts_error_cov(self):
if self._smoothed_forecasts_cov is None:
self._smoothed_forecasts, self._smoothed_forecasts_cov = (
self._compute_forecasts(self.smoothed_state,
self.smoothed_state_cov))
return self._smoothed_forecasts_cov
def _check_dynamic(dynamic, start, end, nobs):
"""
Verify dynamic and warn or error if issues
Parameters
----------
dynamic : {int, None}
The offset relative to start of the dynamic forecasts. None if no
dynamic forecasts are required.
start : int
The location of the first forecast.
end : int
The location of the final forecast (inclusive).
nobs : int
The number of observations in the time series.
Returns
-------
dynamic : {int, None}
The start location of the first dynamic forecast. None if there
are no in-sample dynamic forecasts.
ndynamic : int
The number of dynamic forecasts
"""
if dynamic is None:
return dynamic, 0
# Replace the relative dynamic offset with an absolute offset
dynamic = start + dynamic
# Validate the `dynamic` parameter
if dynamic < 0:
raise ValueError('Dynamic prediction cannot begin prior to the'
' first observation in the sample.')
elif dynamic > end:
warn('Dynamic prediction specified to begin after the end of'
' prediction, and so has no effect.', ValueWarning)
return None, 0
elif dynamic > nobs:
warn('Dynamic prediction specified to begin during'
' out-of-sample forecasting period, and so has no'
' effect.', ValueWarning)
return None, 0
# Get the total size of the desired dynamic forecasting component
# Note: the first `dynamic` periods of prediction are actually
# *not* dynamic, because dynamic prediction begins at observation
# `dynamic`.
ndynamic = max(0, min(end, nobs) - dynamic)
return dynamic, ndynamic
Last update:
Nov 14, 2024