<s> """<STR_LIT>""" <EOL> import pandas as pd <EOL> from zipline . pipeline . common import ( <EOL> DAYS_SINCE_PREV_DISCLOSURE , <EOL> DISCLOSURE_DATE , <EOL> NUM_SHARES , <EOL> PERCENT_SHARES , <EOL> PREVIOUS_NUM_SHARES , <EOL> PREVIOUS_PERCENT_SHARES , <EOL> PREVIOUS_DISCLOSURE_DATE , <EOL> TS_FIELD_NAME , <EOL> ) <EOL> from zipline . pipeline . data import _13DFilings <EOL> from zipline . pipeline . factors . events import BusinessDaysSince13DFilingsDate <EOL> from zipline . pipeline . loaders . _13d_filings import _13DFilingsLoader <EOL> from zipline . pipeline . loaders . utils import ( <EOL> get_values_for_date_ranges , <EOL> zip_with_floats , <EOL> zip_with_dates <EOL> ) <EOL> from zipline . testing . fixtures import WithPipelineEventDataLoader <EOL> from zipline . testing . fixtures import ZiplineTestCase <EOL> date_intervals = [ [ None , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , None ] ] <EOL> empty_df = pd . DataFrame ( <EOL> columns = [ NUM_SHARES , <EOL> PERCENT_SHARES , <EOL> DISCLOSURE_DATE , <EOL> TS_FIELD_NAME ] , <EOL> ) <EOL> empty_df [ NUM_SHARES ] = empty_df [ NUM_SHARES ] . astype ( '<STR_LIT:float>' ) <EOL> empty_df [ PERCENT_SHARES ] = empty_df [ PERCENT_SHARES ] . astype ( '<STR_LIT:float>' ) <EOL> empty_df [ TS_FIELD_NAME ] = empty_df [ TS_FIELD_NAME ] . astype ( '<STR_LIT>' ) <EOL> empty_df [ DISCLOSURE_DATE ] = empty_df [ DISCLOSURE_DATE ] . astype ( '<STR_LIT>' ) <EOL> _13d_filngs_cases = [ <EOL> pd . DataFrame ( { <EOL> NUM_SHARES : [ <NUM_LIT:1> , <NUM_LIT:15> ] , <EOL> PERCENT_SHARES : [ <NUM_LIT:10> , <NUM_LIT:20> ] , <EOL> TS_FIELD_NAME : pd . to_datetime ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> DISCLOSURE_DATE : pd . to_datetime ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> } ) , <EOL> empty_df <EOL> ] <EOL> def get_expected_previous_values ( zip_date_index_with_vals , <EOL> vals , <EOL> date_intervals , <EOL> dates ) : <EOL> return pd . DataFrame ( { <EOL> <NUM_LIT:0> : get_values_for_date_ranges ( zip_date_index_with_vals , <EOL> vals , <EOL> date_intervals , <EOL> dates ) , <EOL> <NUM_LIT:1> : zip_date_index_with_vals ( dates , [ '<STR_LIT>' ] * len ( dates ) ) , <EOL> } , index = dates ) <EOL> class _13DFilingsLoaderTestCase ( WithPipelineEventDataLoader , <EOL> ZiplineTestCase ) : <EOL> """<STR_LIT>""" <EOL> pipeline_columns = { <EOL> PREVIOUS_NUM_SHARES : <EOL> _13DFilings . number_shares . latest , <EOL> PREVIOUS_PERCENT_SHARES : <EOL> _13DFilings . percent_shares . latest , <EOL> PREVIOUS_DISCLOSURE_DATE : <EOL> _13DFilings . disclosure_date . latest , <EOL> DAYS_SINCE_PREV_DISCLOSURE : <EOL> BusinessDaysSince13DFilingsDate ( ) , <EOL> } <EOL> @ classmethod <EOL> def get_sids ( cls ) : <EOL> return range ( <NUM_LIT:2> ) <EOL> @ classmethod <EOL> def get_dataset ( cls ) : <EOL> return { sid : frame <EOL> for sid , frame <EOL> in enumerate ( _13d_filngs_cases ) } <EOL> loader_type = _13DFilingsLoader <EOL> def setup ( self , dates ) : <EOL> cols = { } <EOL> cols [ <EOL> PREVIOUS_DISCLOSURE_DATE <EOL> ] = get_expected_previous_values ( zip_with_dates , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> date_intervals , dates ) <EOL> cols [ PREVIOUS_NUM_SHARES ] = get_expected_previous_values ( <EOL> zip_with_floats , [ '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:15> ] , date_intervals , dates <EOL> ) <EOL> cols [ PREVIOUS_PERCENT_SHARES ] = get_expected_previous_values ( <EOL> zip_with_floats , [ '<STR_LIT>' , <NUM_LIT:10> , <NUM_LIT:20> ] , date_intervals , dates <EOL> ) <EOL> cols [ DAYS_SINCE_PREV_DISCLOSURE ] = self . _compute_busday_offsets ( <EOL> cols [ PREVIOUS_DISCLOSURE_DATE ] <EOL> ) <EOL> return cols </s>
