<s> from statsmodels . compat . python import iteritems , string_types , range <EOL> import numpy as np <EOL> from statsmodels . tools . decorators import cache_readonly <EOL> import pandas as pd <EOL> from . import var_model as _model <EOL> from . import util <EOL> from . import plotting <EOL> FULL_SAMPLE = <NUM_LIT:0> <EOL> ROLLING = <NUM_LIT:1> <EOL> EXPANDING = <NUM_LIT:2> <EOL> def _get_window_type ( window_type ) : <EOL> if window_type in ( FULL_SAMPLE , ROLLING , EXPANDING ) : <EOL> return window_type <EOL> elif isinstance ( window_type , string_types ) : <EOL> window_type_up = window_type . upper ( ) <EOL> if window_type_up in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> return FULL_SAMPLE <EOL> elif window_type_up == '<STR_LIT>' : <EOL> return ROLLING <EOL> elif window_type_up == '<STR_LIT>' : <EOL> return EXPANDING <EOL> raise Exception ( '<STR_LIT>' % window_type ) <EOL> class DynamicVAR ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data , lag_order = <NUM_LIT:1> , window = None , window_type = '<STR_LIT>' , <EOL> trend = '<STR_LIT:c>' , min_periods = None ) : <EOL> self . lag_order = lag_order <EOL> self . names = list ( data . columns ) <EOL> self . neqs = len ( self . names ) <EOL> self . _y_orig = data <EOL> self . _x_orig = _make_lag_matrix ( data , lag_order ) <EOL> self . _x_orig [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> ( self . y , self . x , self . x_filtered , self . _index , <EOL> self . _time_has_obs ) = _filter_data ( self . _y_orig , self . _x_orig ) <EOL> self . lag_order = lag_order <EOL> self . trendorder = util . get_trendorder ( trend ) <EOL> self . _set_window ( window_type , window , min_periods ) <EOL> def _set_window ( self , window_type , window , min_periods ) : <EOL> self . _window_type = _get_window_type ( window_type ) <EOL> if self . _is_rolling : <EOL> if window is None : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if min_periods is None : <EOL> min_periods = window <EOL> else : <EOL> window = len ( self . x ) <EOL> if min_periods is None : <EOL> min_periods = <NUM_LIT:1> <EOL> self . _window = int ( window ) <EOL> self . _min_periods = min_periods <EOL> @ cache_readonly <EOL> def T ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . result_index ) <EOL> @ property <EOL> def nobs ( self ) : <EOL> data = dict ( ( eq , r . nobs ) for eq , r in iteritems ( self . equations ) ) <EOL> return pd . DataFrame ( data ) <EOL> @ cache_readonly <EOL> def equations ( self ) : <EOL> eqs = { } <EOL> for col , ts in iteritems ( self . y ) : <EOL> model = pd . ols ( y = ts , x = self . x , window = self . _window , <EOL> window_type = self . _window_type , <EOL> min_periods = self . _min_periods ) <EOL> eqs [ col ] = model <EOL> return eqs <EOL> @ cache_readonly <EOL> def coefs ( self ) : <EOL> """<STR_LIT>""" <EOL> data = { } <EOL> for eq , result in iteritems ( self . equations ) : <EOL> data [ eq ] = result . beta <EOL> panel = pd . WidePanel . fromDict ( data ) <EOL> return panel . swapaxes ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ property <EOL> def result_index ( self ) : <EOL> return self . coefs . major_axis <EOL> @ cache_readonly <EOL> def _coefs_raw ( self ) : <EOL> """<STR_LIT>""" <EOL> coef_panel = self . coefs . copy ( ) <EOL> del coef_panel [ '<STR_LIT>' ] <EOL> coef_values = coef_panel . swapaxes ( '<STR_LIT>' , '<STR_LIT>' ) . values <EOL> coef_values = coef_values . reshape ( ( len ( coef_values ) , <EOL> self . lag_order , <EOL> self . neqs , self . neqs ) ) <EOL> return coef_values <EOL> @ cache_readonly <EOL> def _intercepts_raw ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . coefs [ '<STR_LIT>' ] . values <EOL> @ cache_readonly <EOL> def resid ( self ) : <EOL> data = { } <EOL> for eq , result in iteritems ( self . equations ) : <EOL> data [ eq ] = result . resid <EOL> return pd . DataFrame ( data ) <EOL> def forecast ( self , steps = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> output = np . empty ( ( self . T - steps , self . neqs ) ) <EOL> y_values = self . y . values <EOL> y_index_map = dict ( ( d , idx ) for idx , d in enumerate ( self . y . index ) ) <EOL> result_index_map = dict ( ( d , idx ) for idx , d in enumerate ( self . result_index ) ) <EOL> coefs = self . _coefs_raw <EOL> intercepts = self . _intercepts_raw <EOL> forc_index = self . result_index [ steps : ] <EOL> for i , date in enumerate ( forc_index ) : <EOL> idx = y_index_map [ date ] - steps <EOL> result_idx = result_index_map [ date ] - steps <EOL> y_slice = y_values [ : idx ] <EOL> forcs = _model . forecast ( y_slice , coefs [ result_idx ] , <EOL> intercepts [ result_idx ] , steps ) <EOL> output [ i ] = forcs [ - <NUM_LIT:1> ] <EOL> return pd . DataFrame ( output , index = forc_index , columns = self . names ) <EOL> def plot_forecast ( self , steps = <NUM_LIT:1> , figsize = ( <NUM_LIT:10> , <NUM_LIT:10> ) ) : <EOL> """<STR_LIT>""" <EOL> import matplotlib . pyplot as plt <EOL> fig , axes = plt . subplots ( figsize = figsize , nrows = self . neqs , <EOL> sharex = True ) <EOL> forc = self . forecast ( steps = steps ) <EOL> dates = forc . index <EOL> y_overlay = self . y . reindex ( dates ) <EOL> for i , col in enumerate ( forc . columns ) : <EOL> ax = axes [ i ] <EOL> y_ts = y_overlay [ col ] <EOL> forc_ts = forc [ col ] <EOL> y_handle = ax . plot ( dates , y_ts . values , '<STR_LIT>' , ms = <NUM_LIT:2> ) <EOL> forc_handle = ax . plot ( dates , forc_ts . values , '<STR_LIT>' ) <EOL> fig . legend ( ( y_handle , forc_handle ) , ( '<STR_LIT:Y>' , '<STR_LIT>' ) ) <EOL> fig . autofmt_xdate ( ) <EOL> fig . suptitle ( '<STR_LIT>' % steps ) <EOL> plotting . adjust_subplots ( bottom = <NUM_LIT> , left = <NUM_LIT> ) <EOL> plt . draw_if_interactive ( ) <EOL> @ property <EOL> def _is_rolling ( self ) : <EOL> return self . _window_type == ROLLING <EOL> @ cache_readonly <EOL> def r2 ( self ) : <EOL> """<STR_LIT>""" <EOL> data = dict ( ( eq , r . r2 ) for eq , r in iteritems ( self . equations ) ) <EOL> return pd . DataFrame ( data ) <EOL> class DynamicPanelVAR ( DynamicVAR ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data , lag_order = <NUM_LIT:1> , window = None , window_type = '<STR_LIT>' , <EOL> trend = '<STR_LIT:c>' , min_periods = None ) : <EOL> self . lag_order = lag_order <EOL> self . neqs = len ( data . columns ) <EOL> self . _y_orig = data <EOL> self . _x_orig = _make_lag_matrix ( data , lag_order ) <EOL> self . _x_orig [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> ( self . y , self . x , self . x_filtered , self . _index , <EOL> self . _time_has_obs ) = _filter_data ( self . _y_orig , self . _x_orig ) <EOL> self . lag_order = lag_order <EOL> self . trendorder = util . get_trendorder ( trend ) <EOL> self . _set_window ( window_type , window , min_periods ) <EOL> def _filter_data ( lhs , rhs ) : <EOL> """<STR_LIT>""" <EOL> def _has_all_columns ( df ) : <EOL> return np . isfinite ( df . values ) . sum ( <NUM_LIT:1> ) == len ( df . columns ) <EOL> rhs_valid = _has_all_columns ( rhs ) <EOL> if not rhs_valid . all ( ) : <EOL> pre_filtered_rhs = rhs [ rhs_valid ] <EOL> else : <EOL> pre_filtered_rhs = rhs <EOL> index = lhs . index . union ( rhs . index ) <EOL> if not index . equals ( rhs . index ) or not index . equals ( lhs . index ) : <EOL> rhs = rhs . reindex ( index ) <EOL> lhs = lhs . reindex ( index ) <EOL> rhs_valid = _has_all_columns ( rhs ) <EOL> lhs_valid = _has_all_columns ( lhs ) <EOL> valid = rhs_valid & lhs_valid <EOL> if not valid . all ( ) : <EOL> filt_index = rhs . index [ valid ] <EOL> filtered_rhs = rhs . reindex ( filt_index ) <EOL> filtered_lhs = lhs . reindex ( filt_index ) <EOL> else : <EOL> filtered_rhs , filtered_lhs = rhs , lhs <EOL> return filtered_lhs , filtered_rhs , pre_filtered_rhs , index , valid <EOL> def _make_lag_matrix ( x , lags ) : <EOL> data = { } <EOL> columns = [ ] <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:1> + lags ) : <EOL> lagstr = '<STR_LIT>' % i <EOL> lag = x . shift ( i ) . rename ( columns = lambda c : lagstr + c ) <EOL> data . update ( lag . _series ) <EOL> columns . extend ( lag . columns ) <EOL> return pd . DataFrame ( data , columns = columns ) <EOL> class Equation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , y , x ) : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import pandas . util . testing as ptest <EOL> ptest . N = <NUM_LIT> <EOL> data = ptest . makeTimeDataFrame ( ) . cumsum ( <NUM_LIT:0> ) <EOL> var = DynamicVAR ( data , lag_order = <NUM_LIT:2> , window_type = '<STR_LIT>' ) <EOL> var2 = DynamicVAR ( data , lag_order = <NUM_LIT:2> , window = <NUM_LIT:10> , <EOL> window_type = '<STR_LIT>' ) </s>
