<s> """<STR_LIT>""" <EOL> from mdf import ( <EOL> MDFContext , <EOL> varnode , <EOL> evalnode , <EOL> now <EOL> ) <EOL> import mdf <EOL> from datetime import datetime , timedelta <EOL> import unittest <EOL> import logging <EOL> import pandas as pa <EOL> import numpy as np <EOL> __package__ = None <EOL> _logger = logging . getLogger ( __name__ ) <EOL> index = pa . Index ( [ datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:3> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:6> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:7> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:8> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:9> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:10> ) , <EOL> datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT> ) ] ) <EOL> data = pa . Series ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> , <NUM_LIT:10> ] , index = index , dtype = float ) <EOL> datanode = mdf . datanode ( "<STR_LIT:data>" , data ) <EOL> df = pa . DataFrame ( { "<STR_LIT:A>" : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> , <NUM_LIT:10> ] , <EOL> "<STR_LIT:B>" : [ np . nan , <NUM_LIT:9> , <NUM_LIT:8> , np . nan , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:4> , <NUM_LIT:3> , <NUM_LIT:2> , <NUM_LIT:1> ] } , <EOL> index = index , dtype = float ) <EOL> dfnode = mdf . datanode ( "<STR_LIT>" , df ) <EOL> filter = pa . Series ( [ False , True , False , True , False , <EOL> True , False , True , False , True ] , index = index ) <EOL> filternode = mdf . datanode ( "<STR_LIT>" , filter ) <EOL> queue_expected = [ data [ data . index <= d ] . tolist ( ) for d in index ] <EOL> queue_filtered_expected = [ data [ filter ] [ data [ filter ] . index <= d ] . tolist ( ) for d in index ] <EOL> delay_expected = data . shift ( <NUM_LIT:1> ) <EOL> delay_filtered_expected = data [ filter ] . shift ( <NUM_LIT:1> ) . reindex ( index , method = "<STR_LIT>" ) <EOL> delay_df_expected = df . shift ( <NUM_LIT:1> ) <EOL> delay_df_filtered_expected = df [ filter ] . shift ( <NUM_LIT:1> ) . reindex ( index , method = "<STR_LIT>" ) <EOL> nansum_expected = data . cumsum ( skipna = True ) . ffill ( ) <EOL> nansum_filtered_expected = data [ filter ] . cumsum ( skipna = True ) . reindex ( index ) . ffill ( ) <EOL> nansum_df_expected = df . cumsum ( skipna = True ) . ffill ( ) <EOL> nansum_df_filtered_expected = df [ filter ] . cumsum ( skipna = True ) . reindex ( index ) . ffill ( ) <EOL> cumprod_expected = data . cumprod ( ) . ffill ( ) <EOL> cumprod_filtered_expected = data [ filter ] . cumprod ( ) . reindex ( index ) . ffill ( ) <EOL> cumprod_df_expected = df . cumprod ( ) . ffill ( ) <EOL> cumprod_df_filtered_expected = df [ filter ] . cumprod ( ) . reindex ( index ) . ffill ( ) <EOL> def _normalize ( x ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( x , float ) : <EOL> if np . isnan ( x ) : <EOL> return - np . inf <EOL> return x <EOL> if isinstance ( x , ( pa . DataFrame , pa . Series ) ) : <EOL> x = x . values <EOL> if x . ndim == <NUM_LIT:1> : <EOL> return tuple ( [ _normalize ( y ) for y in x ] ) <EOL> r = x . copy ( ) <EOL> r [ np . isnan ( r ) ] = - np . inf <EOL> return tuple ( r . tolist ( ) ) <EOL> class NodeFilterTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . ctx = MDFContext ( index [ <NUM_LIT:0> ] ) <EOL> def test_queue ( self ) : <EOL> actual = self . _run ( datanode . queuenode ( as_list = True ) ) <EOL> self . assertEqual ( actual . tolist ( ) , queue_expected ) <EOL> actual = self . _run ( datanode . queuenode ( filter = filternode , as_list = True ) ) <EOL> self . assertEqual ( actual . tolist ( ) , queue_filtered_expected ) <EOL> def test_delay ( self ) : <EOL> actual = self . _run ( datanode . delaynode ( periods = <NUM_LIT:1> , initial_value = np . nan ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( delay_expected ) ) <EOL> actual = self . _run ( datanode . delaynode ( periods = <NUM_LIT:1> , initial_value = np . nan , filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( delay_filtered_expected ) ) <EOL> def test_delay_df ( self ) : <EOL> actual = self . _run ( dfnode . delaynode ( periods = <NUM_LIT:1> , initial_value = np . nan ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( delay_df_expected ) ) <EOL> actual = self . _run ( dfnode . delaynode ( periods = <NUM_LIT:1> , initial_value = np . nan , filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( delay_df_filtered_expected ) ) <EOL> def test_nansum ( self ) : <EOL> actual = self . _run ( datanode . nansumnode ( ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( nansum_expected ) ) <EOL> actual = self . _run ( datanode . nansumnode ( filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( nansum_filtered_expected ) ) <EOL> def test_nansum_df ( self ) : <EOL> actual = self . _run ( dfnode . nansumnode ( ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( nansum_df_expected ) ) <EOL> actual = self . _run ( dfnode . nansumnode ( filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( nansum_df_filtered_expected ) ) <EOL> def test_cumprod ( self ) : <EOL> actual = self . _run ( datanode . cumprodnode ( ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( cumprod_expected ) ) <EOL> actual = self . _run ( datanode . cumprodnode ( filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( cumprod_filtered_expected ) ) <EOL> def test_cumprod_df ( self ) : <EOL> actual = self . _run ( dfnode . cumprodnode ( ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( cumprod_df_expected ) ) <EOL> actual = self . _run ( dfnode . cumprodnode ( filter = filternode ) ) <EOL> self . assertEqual ( _normalize ( actual ) , _normalize ( cumprod_df_filtered_expected ) ) <EOL> def _run ( self , node ) : <EOL> result = [ ] <EOL> for t in index : <EOL> self . ctx . set_date ( t ) <EOL> result . append ( self . ctx [ node ] ) <EOL> if isinstance ( result [ <NUM_LIT:0> ] , pa . Series ) : <EOL> return pa . DataFrame ( result , index = index ) <EOL> return pa . Series ( result , index , dtype = object ) </s>
