<s> """<STR_LIT>""" <EOL> from datetime import datetime <EOL> from warnings import ( <EOL> catch_warnings , <EOL> filterwarnings , <EOL> ) <EOL> from numpy import ( <EOL> broadcast , <EOL> busday_count , <EOL> datetime64 , <EOL> dtype , <EOL> empty , <EOL> nan , <EOL> where <EOL> ) <EOL> from numpy . lib . stride_tricks import as_strided <EOL> from toolz import flip <EOL> uint8_dtype = dtype ( '<STR_LIT>' ) <EOL> bool_dtype = dtype ( '<STR_LIT:bool>' ) <EOL> int64_dtype = dtype ( '<STR_LIT>' ) <EOL> float32_dtype = dtype ( '<STR_LIT>' ) <EOL> float64_dtype = dtype ( '<STR_LIT>' ) <EOL> complex128_dtype = dtype ( '<STR_LIT>' ) <EOL> datetime64D_dtype = dtype ( '<STR_LIT>' ) <EOL> datetime64ns_dtype = dtype ( '<STR_LIT>' ) <EOL> make_datetime64ns = flip ( datetime64 , '<STR_LIT>' ) <EOL> make_datetime64D = flip ( datetime64 , '<STR_LIT:D>' ) <EOL> NaTmap = { <EOL> dtype ( '<STR_LIT>' % unit ) : datetime64 ( '<STR_LIT>' , unit ) <EOL> for unit in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:s>' , '<STR_LIT:m>' , '<STR_LIT:D>' ) <EOL> } <EOL> NaT_for_dtype = NaTmap . __getitem__ <EOL> NaTns = NaT_for_dtype ( datetime64ns_dtype ) <EOL> NaTD = NaT_for_dtype ( datetime64D_dtype ) <EOL> _FILLVALUE_DEFAULTS = { <EOL> bool_dtype : False , <EOL> float32_dtype : nan , <EOL> float64_dtype : nan , <EOL> datetime64ns_dtype : NaTns , <EOL> } <EOL> class NoDefaultMissingValue ( Exception ) : <EOL> pass <EOL> def make_kind_check ( python_types , numpy_kind ) : <EOL> """<STR_LIT>""" <EOL> def check ( value ) : <EOL> if hasattr ( value , '<STR_LIT>' ) : <EOL> return value . dtype . kind == numpy_kind <EOL> return isinstance ( value , python_types ) <EOL> return check <EOL> is_float = make_kind_check ( float , '<STR_LIT:f>' ) <EOL> is_int = make_kind_check ( int , '<STR_LIT:i>' ) <EOL> is_datetime = make_kind_check ( datetime , '<STR_LIT:M>' ) <EOL> def coerce_to_dtype ( dtype , value ) : <EOL> """<STR_LIT>""" <EOL> name = dtype . name <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> if name == '<STR_LIT>' : <EOL> return make_datetime64D ( value ) <EOL> elif name == '<STR_LIT>' : <EOL> return make_datetime64ns ( value ) <EOL> else : <EOL> raise TypeError ( <EOL> "<STR_LIT>" % dtype <EOL> ) <EOL> return dtype . type ( value ) <EOL> def default_missing_value_for_dtype ( dtype ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return _FILLVALUE_DEFAULTS [ dtype ] <EOL> except KeyError : <EOL> raise NoDefaultMissingValue ( <EOL> "<STR_LIT>" % dtype <EOL> ) <EOL> def repeat_first_axis ( array , count ) : <EOL> """<STR_LIT>""" <EOL> return as_strided ( array , ( count , ) + array . shape , ( <NUM_LIT:0> , ) + array . strides ) <EOL> def repeat_last_axis ( array , count ) : <EOL> """<STR_LIT>""" <EOL> return as_strided ( array , array . shape + ( count , ) , array . strides + ( <NUM_LIT:0> , ) ) <EOL> def rolling_window ( array , length ) : <EOL> """<STR_LIT>""" <EOL> orig_shape = array . shape <EOL> if not orig_shape : <EOL> raise IndexError ( "<STR_LIT>" ) <EOL> elif orig_shape [ <NUM_LIT:0> ] <= length : <EOL> raise IndexError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> shape = orig_shape , <EOL> len = length , <EOL> ) <EOL> ) <EOL> num_windows = ( orig_shape [ <NUM_LIT:0> ] - length + <NUM_LIT:1> ) <EOL> new_shape = ( num_windows , length ) + orig_shape [ <NUM_LIT:1> : ] <EOL> new_strides = ( array . strides [ <NUM_LIT:0> ] , ) + array . strides <EOL> return as_strided ( array , new_shape , new_strides ) <EOL> _notNaT = make_datetime64D ( <NUM_LIT:0> ) <EOL> def busday_count_mask_NaT ( begindates , <EOL> enddates , <EOL> out = None ) : <EOL> """<STR_LIT>""" <EOL> if out is None : <EOL> out = empty ( broadcast ( begindates , enddates ) . shape , dtype = float ) <EOL> beginmask = ( begindates == NaTD ) <EOL> endmask = ( enddates == NaTD ) <EOL> out = busday_count ( <EOL> where ( beginmask , _notNaT , begindates ) , <EOL> where ( endmask , _notNaT , enddates ) , <EOL> out = out , <EOL> ) <EOL> out [ beginmask | endmask ] = nan <EOL> return out <EOL> class WarningContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * warning_specs ) : <EOL> self . _warning_specs = warning_specs <EOL> self . _catchers = [ ] <EOL> def __enter__ ( self ) : <EOL> catcher = catch_warnings ( ) <EOL> catcher . __enter__ ( ) <EOL> self . _catchers . append ( catcher ) <EOL> for args , kwargs in self . _warning_specs : <EOL> filterwarnings ( * args , ** kwargs ) <EOL> return self <EOL> def __exit__ ( self , * exc_info ) : <EOL> catcher = self . _catchers . pop ( ) <EOL> return catcher . __exit__ ( * exc_info ) <EOL> def ignore_nanwarnings ( ) : <EOL> """<STR_LIT>""" <EOL> return WarningContext ( <EOL> ( <EOL> ( '<STR_LIT:ignore>' , ) , <EOL> { '<STR_LIT>' : RuntimeWarning , '<STR_LIT>' : '<STR_LIT>' } , <EOL> ) <EOL> ) </s>
