<s> from mdf import ( <EOL> MDFContext , <EOL> evalnode , <EOL> queuenode , <EOL> nansumnode , <EOL> cumprodnode , <EOL> delaynode , <EOL> ffillnode , <EOL> vargroup , <EOL> datanode , <EOL> run , <EOL> DataFrameBuilder <EOL> ) <EOL> from datetime import datetime <EOL> from numpy . testing . utils import assert_almost_equal <EOL> import pandas as pd <EOL> import numpy as np <EOL> import unittest <EOL> import logging <EOL> import operator <EOL> __package__ = None <EOL> _logger = logging . getLogger ( __name__ ) <EOL> params = vargroup ( C = None , D = None ) <EOL> @ queuenode <EOL> def queue_output ( ) : <EOL> return A ( ) + B ( ) <EOL> @ queuenode <EOL> def queue_yield ( ) : <EOL> while True : <EOL> yield A ( ) + B ( ) <EOL> @ evalnode <EOL> def queue_filter ( ) : <EOL> yield False <EOL> while True : <EOL> yield True <EOL> @ queuenode ( filter = queue_filter ) <EOL> def queue_filter_test ( ) : <EOL> yield "<STR_LIT>" <EOL> while True : <EOL> yield <NUM_LIT:0> <EOL> @ nansumnode ( ) <EOL> def nansum_output ( ) : <EOL> return A ( ) + sometimes_nan_B ( ) <EOL> @ cumprodnode ( ) <EOL> def cumprod_output ( ) : <EOL> return A ( ) + B ( ) <EOL> @ evalnode <EOL> def sometimes_nan_B ( ) : <EOL> b = B ( ) <EOL> return np . nan if b % <NUM_LIT:2> else b <EOL> @ evalnode <EOL> def A ( ) : <EOL> return params . C ( ) * params . D ( ) <EOL> @ evalnode <EOL> def B ( ) : <EOL> accum = <NUM_LIT:0> <EOL> while True : <EOL> yield accum <EOL> accum += <NUM_LIT:1> <EOL> @ evalnode <EOL> def A_plus_B ( ) : <EOL> return A ( ) + B ( ) <EOL> @ evalnode <EOL> def Counter ( ) : <EOL> accum = - <NUM_LIT> <EOL> while True : <EOL> if accum != <NUM_LIT:0.0> : <EOL> yield accum <EOL> accum += <NUM_LIT:0.5> <EOL> class DelayNodeTest ( object ) : <EOL> @ evalnode <EOL> def initial_value ( cls ) : <EOL> return <NUM_LIT:0> <EOL> @ delaynode ( periods = <NUM_LIT:1> , initial_value = initial_value ) <EOL> def delayed_node ( cls ) : <EOL> i = <NUM_LIT:1> <EOL> while True : <EOL> yield i <EOL> i += <NUM_LIT:1> <EOL> @ delaynode ( periods = <NUM_LIT:1> , initial_value = initial_value , lazy = True ) <EOL> def delayed_node_lazy ( cls ) : <EOL> return cls . delay_test ( ) [ - <NUM_LIT:1> ] <EOL> @ queuenode <EOL> def delay_test ( cls ) : <EOL> return <NUM_LIT:1> + cls . delayed_node ( ) <EOL> @ queuenode <EOL> def delay_test_lazy ( cls ) : <EOL> return <NUM_LIT:1> + cls . delayed_node_lazy ( ) <EOL> @ queuenode <EOL> def ffill_queue ( ) : <EOL> return ffill_test ( ) <EOL> @ ffillnode <EOL> def ffill_test ( ) : <EOL> i = <NUM_LIT:0> <EOL> while True : <EOL> yield np . nan if i % <NUM_LIT:2> else float ( i ) <EOL> i += <NUM_LIT:1> <EOL> @ ffillnode <EOL> def ffill_array_test ( ) : <EOL> return ffill_array_test_not_filled ( ) <EOL> @ evalnode <EOL> def ffill_array_test_not_filled ( ) : <EOL> i = <NUM_LIT:0> <EOL> array = np . ndarray ( ( <NUM_LIT:5> , ) , dtype = float ) <EOL> array . fill ( <NUM_LIT> ) <EOL> yield array <EOL> array . fill ( np . nan ) <EOL> while True : <EOL> yield array <EOL> class NodeTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . daterange = pd . bdate_range ( datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) , datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:10> ) ) <EOL> self . ctx = MDFContext ( ) <EOL> self . ctx [ params . C ] = <NUM_LIT:10> <EOL> self . ctx [ params . D ] = <NUM_LIT:20> <EOL> def test_queuenode ( self ) : <EOL> self . _run ( queue_output ) <EOL> queue = self . ctx [ queue_output ] <EOL> self . assertEqual ( len ( queue ) , len ( self . daterange ) ) <EOL> def test_queueyield ( self ) : <EOL> self . _run ( queue_yield ) <EOL> queue = self . ctx [ queue_yield ] <EOL> self . assertEqual ( len ( queue ) , len ( self . daterange ) ) <EOL> def test_queue_filter ( self ) : <EOL> self . _run ( queue_filter_test ) <EOL> queue = self . ctx [ queue_filter_test ] <EOL> self . assertEqual ( list ( queue ) , [ <NUM_LIT:0> ] * ( len ( self . daterange ) - <NUM_LIT:1> ) ) <EOL> @ staticmethod <EOL> def diff_dfs ( lhs_df , rhs_df , tolerance ) : <EOL> diffs = np . abs ( lhs_df - rhs_df ) <EOL> mask = ( diffs > tolerance ) . values <EOL> mask &= ~ ( np . isnan ( lhs_df ) and np . isnan ( rhs_df ) ) . values <EOL> mask |= np . isnan ( lhs_df ) . values & ~ np . isnan ( rhs_df ) . values <EOL> mask |= np . isnan ( rhs_df ) . values & ~ np . isnan ( lhs_df ) . values <EOL> return mask . any ( ) <EOL> def test_nansumnode ( self ) : <EOL> self . _run ( nansum_output ) <EOL> nansum = self . ctx [ nansum_output ] <EOL> self . assertEqual ( nansum , <NUM_LIT> ) <EOL> def test_cumprodnode ( self ) : <EOL> self . _run ( cumprod_output ) <EOL> cumprod = self . ctx [ cumprod_output ] <EOL> self . assertEqual ( cumprod , <NUM_LIT> ) <EOL> def test_delaynode ( self ) : <EOL> self . _run ( DelayNodeTest . delay_test , DelayNodeTest . delay_test_lazy ) <EOL> value = self . ctx [ DelayNodeTest . delay_test ] <EOL> value_lazy = self . ctx [ DelayNodeTest . delay_test_lazy ] <EOL> self . assertEqual ( list ( value ) , list ( range ( <NUM_LIT:1> , len ( self . daterange ) + <NUM_LIT:1> ) ) ) <EOL> self . assertEqual ( list ( value_lazy ) , list ( range ( <NUM_LIT:1> , len ( self . daterange ) + <NUM_LIT:1> ) ) ) <EOL> def test_ffillnode ( self ) : <EOL> self . _run ( ffill_queue ) <EOL> value = self . ctx [ ffill_queue ] <EOL> self . assertEqual ( tuple ( value ) , ( <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_ffill_array ( self ) : <EOL> self . _run ( ffill_array_test ) <EOL> value = self . ctx [ ffill_array_test ] <EOL> unfilled_value = self . ctx [ ffill_array_test_not_filled ] <EOL> self . assertTrue ( np . isnan ( unfilled_value ) . all ( ) ) <EOL> self . assertEquals ( value . tolist ( ) , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> def test_datanode_ffill ( self ) : <EOL> data = pd . Series ( range ( len ( self . daterange ) ) , self . daterange , dtype = float ) <EOL> data = data [ [ bool ( i % <NUM_LIT:2> ) for i in range ( len ( data . index ) ) ] ] <EOL> expected = data . reindex ( self . daterange , method = "<STR_LIT>" ) <EOL> expected [ np . isnan ( expected ) ] = np . inf <EOL> node = datanode ( "<STR_LIT>" , data , ffill = True , missing_value = np . inf ) <EOL> qnode = node . queuenode ( ) <EOL> self . _run ( qnode ) <EOL> value = self . ctx [ qnode ] <EOL> self . assertEquals ( list ( value ) , expected . values . tolist ( ) ) <EOL> def test_lookahead_node ( self ) : <EOL> B_queue = B . queuenode ( ) <EOL> B_lookahead = B . lookaheadnode ( periods = len ( self . daterange ) ) <EOL> self . ctx . set_date ( self . daterange [ <NUM_LIT:0> ] ) <EOL> actual = self . ctx [ B_lookahead ] <EOL> self . _run ( B_queue ) <EOL> expected = self . ctx [ B_queue ] <EOL> self . assertEquals ( actual . values . tolist ( ) , list ( expected ) ) <EOL> self . assertEquals ( actual . index . tolist ( ) , list ( self . daterange ) ) <EOL> def test_apply_node ( self ) : <EOL> actual_node = A . applynode ( func = operator . add , args = ( B , ) ) . queuenode ( ) <EOL> expected_node = A_plus_B . queuenode ( ) <EOL> self . _run ( actual_node , expected_node ) <EOL> actual = self . ctx [ actual_node ] <EOL> expected = self . ctx [ expected_node ] <EOL> self . assertEquals ( actual , expected ) <EOL> def test_binary_operators_with_constant ( self ) : <EOL> self . _test ( Counter , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT:1.0> , - <NUM_LIT:0.5> , <NUM_LIT:0.5> , <NUM_LIT:1.0> , <NUM_LIT> ] ) <EOL> self . _test ( Counter + <NUM_LIT> , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( Counter - <NUM_LIT> , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( Counter * <NUM_LIT> , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( Counter / <NUM_LIT:0.5> , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( <NUM_LIT> + Counter , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( <NUM_LIT:1.0> - Counter , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0.5> , <NUM_LIT:0.0> , - <NUM_LIT:0.5> ] ) <EOL> self . _test ( <NUM_LIT> * Counter , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( <NUM_LIT:12> / ( Counter + <NUM_LIT> ) , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , <EOL> - <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> def test_binary_operators_with_node ( self ) : <EOL> self . _test ( Counter + Counter , [ - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . _test ( Counter - Counter , [ <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> ] ) <EOL> self . _test ( Counter * Counter , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1.0> , <NUM_LIT> ] ) <EOL> self . _test ( Counter / Counter , [ <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT:1.0> , <NUM_LIT:1.0> ] ) <EOL> def _test ( self , node , expected_values ) : <EOL> values = node . queuenode ( ) <EOL> self . _run ( values ) <EOL> actual = self . ctx [ values ] <EOL> self . assertEquals ( list ( actual ) , expected_values ) <EOL> def _run_for_daterange ( self , date_range , * nodes ) : <EOL> for t in date_range : <EOL> self . ctx . set_date ( t ) <EOL> for node in nodes : <EOL> self . ctx [ node ] <EOL> def _run ( self , * nodes ) : <EOL> self . _run_for_daterange ( self . daterange , * nodes ) </s>
