#NEWLINE# def _load_dataset( self , dates , assets , mask , columns ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# ( dataset , ) = set ( map ( getdataset , columns ) ) #NEWLINE# #UNINDENT# except ValueError : #NEWLINE# #INDENT# raise AssertionError ( 'all columns must come from the same dataset' ) #NEWLINE# #UNINDENT# ( expr , deltas , odo_kwargs ) = self [ dataset ] #NEWLINE# have_sids = ( SID_FIELD_NAME in expr . fields ) #NEWLINE# asset_idx = pd . Series ( index = assets , data = np . arange ( len ( assets ) ) ) #NEWLINE# assets = list ( map ( int , assets ) ) #NEWLINE# added_query_fields = ( [ AD_FIELD_NAME , TS_FIELD_NAME ] + ( [ SID_FIELD_NAME ] if have_sids else [ ] ) ) #NEWLINE# data_query_time = self . _data_query_time #NEWLINE# data_query_tz = self . _data_query_tz #NEWLINE# ( lower_dt , upper_dt ) = normalize_data_query_bounds ( dates [ 0 ] , dates [ ( - 1 ) ] , data_query_time , data_query_tz ) #NEWLINE# def where( e ) : #NEWLINE# #INDENT# 'Create the query to run against the resources.\n\n            Parameters\n            ----------\n            e : Expr\n                The baseline or deltas expression.\n\n            Returns\n            -------\n            q : Expr\n                The query to run.\n            ' #NEWLINE# def lower_for_col( column ) : #NEWLINE# #INDENT# pred = ( e [ TS_FIELD_NAME ] <= lower_dt ) #NEWLINE# colname = column . name #NEWLINE# schema = e [ colname ] . schema . measure #NEWLINE# if isinstance ( schema , Option ) : #NEWLINE# #INDENT# pred &= e [ colname ] . notnull ( ) #NEWLINE# schema = schema . ty #NEWLINE# #UNINDENT# if ( schema in floating ) : #NEWLINE# #INDENT# pred &= ( ~ e [ colname ] . isnan ( ) ) #NEWLINE# #UNINDENT# filtered = e [ pred ] #NEWLINE# lower = filtered [ TS_FIELD_NAME ] . max ( ) #NEWLINE# if have_sids : #NEWLINE# #INDENT# lower = bz . by ( filtered [ SID_FIELD_NAME ] , timestamp = lower ) . timestamp . min ( ) #NEWLINE# #UNINDENT# return lower #NEWLINE# #UNINDENT# lower = odo ( reduce ( bz . least , map ( lower_for_col , columns ) ) , pd . Timestamp , ** odo_kwargs ) #NEWLINE# if ( lower is pd . NaT ) : #NEWLINE# #INDENT# lower = lower_dt #NEWLINE# #UNINDENT# return e [ ( ( e [ TS_FIELD_NAME ] >= lower ) & ( e [ TS_FIELD_NAME ] <= upper_dt ) ) ] [ ( added_query_fields + list ( map ( getname , columns ) ) ) ] #NEWLINE# #UNINDENT# def collect_expr( e ) : #NEWLINE# #INDENT# 'Execute and merge all of the per-column subqueries.\n\n            Parameters\n            ----------\n            e : Expr\n                The baseline or deltas expression.\n\n            Returns\n            -------\n            result : pd.DataFrame\n                The resulting dataframe.\n\n            Notes\n            -----\n            This can return more data than needed. The in memory reindex will\n            handle this.\n            ' #NEWLINE# df = odo ( where ( e ) , pd . DataFrame , ** odo_kwargs ) #NEWLINE# df . sort ( TS_FIELD_NAME , inplace = True ) #NEWLINE# return df #NEWLINE# #UNINDENT# materialized_expr = collect_expr ( expr ) #NEWLINE# materialized_deltas = ( collect_expr ( deltas ) if ( deltas is not None ) else pd . DataFrame ( columns = ( added_query_fields + list ( map ( getname , columns ) ) ) ) ) #NEWLINE# if ( ( not materialized_deltas . empty ) and have_sids ) : #NEWLINE# #INDENT# materialized_deltas = materialized_deltas [ materialized_deltas [ SID_FIELD_NAME ] . isin ( assets ) ] #NEWLINE# #UNINDENT# if ( data_query_time is not None ) : #NEWLINE# #INDENT# for m in ( materialized_expr , materialized_deltas ) : #NEWLINE# #INDENT# m . loc [ : , TS_FIELD_NAME ] = m . loc [ : , TS_FIELD_NAME ] . astype ( 'datetime64[ns]' ) #NEWLINE# normalize_timestamp_to_query_time ( m , data_query_time , data_query_tz , inplace = True , ts_field = TS_FIELD_NAME ) #NEWLINE# #UNINDENT# #UNINDENT# ( sparse_output , non_novel_deltas ) = overwrite_novel_deltas ( materialized_expr , materialized_deltas , dates ) #NEWLINE# sparse_output . drop ( AD_FIELD_NAME , axis = 1 , inplace = True ) #NEWLINE# def last_in_date_group( df , reindex , have_sids = have_sids ) : #NEWLINE# #INDENT# idx = dates [ dates . searchsorted ( df [ TS_FIELD_NAME ] . values . astype ( 'datetime64[D]' ) ) ] #NEWLINE# if have_sids : #NEWLINE# #INDENT# idx = [ idx , SID_FIELD_NAME ] #NEWLINE# #UNINDENT# last_in_group = df . drop ( TS_FIELD_NAME , axis = 1 ) . groupby ( idx , sort = False ) . last ( ) #NEWLINE# if have_sids : #NEWLINE# #INDENT# last_in_group = last_in_group . unstack ( ) #NEWLINE# #UNINDENT# if reindex : #NEWLINE# #INDENT# if have_sids : #NEWLINE# #INDENT# cols = last_in_group . columns #NEWLINE# last_in_group = last_in_group . reindex ( index = dates , columns = pd . MultiIndex . from_product ( ( cols . levels [ 0 ] , assets ) , names = cols . names ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# last_in_group = last_in_group . reindex ( dates ) #NEWLINE# #UNINDENT# #UNINDENT# return last_in_group #NEWLINE# #UNINDENT# sparse_deltas = last_in_date_group ( adjustments_from_deltas , reindex = False ) #NEWLINE# dense_output = last_in_date_group ( sparse_output , reindex = True ) #NEWLINE# dense_output . ffill ( inplace = True ) #NEWLINE# if have_sids : #NEWLINE# #INDENT# adjustments_from_deltas = adjustments_from_deltas_with_sids #NEWLINE# column_view = identity #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# column_view = compose ( copy , partial ( repeat_last_axis , count = len ( assets ) ) ) #NEWLINE# adjustments_from_deltas = adjustments_from_deltas_no_sids #NEWLINE# #UNINDENT# for ( column_idx , column ) in enumerate ( columns ) : #NEWLINE# #INDENT# column_name = column . name #NEWLINE# ( yield ( column , AdjustedArray ( column_view ( dense_output [ column_name ] . values . astype ( column . dtype ) ) , mask , adjustments_from_deltas ( dates , sparse_output [ TS_FIELD_NAME ] . values , column_idx , column_name , asset_idx , sparse_deltas ) , column . missing_value ) ) )