<s> """<STR_LIT>""" <EOL> import logbook <EOL> import math <EOL> import numpy as np <EOL> import zipline . utils . math_utils as zp_math <EOL> log = logbook . Logger ( '<STR_LIT>' ) <EOL> TREASURY_DURATIONS = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> def check_entry ( key , value ) : <EOL> if key != '<STR_LIT>' : <EOL> return np . isnan ( value ) or np . isinf ( value ) <EOL> else : <EOL> return False <EOL> def sharpe_ratio ( algorithm_volatility , algorithm_return , treasury_return ) : <EOL> """<STR_LIT>""" <EOL> if zp_math . tolerant_equals ( algorithm_volatility , <NUM_LIT:0> ) : <EOL> return np . nan <EOL> return ( algorithm_return - treasury_return ) / algorithm_volatility <EOL> def downside_risk ( algorithm_returns , mean_returns , normalization_factor ) : <EOL> rets = algorithm_returns . round ( <NUM_LIT:8> ) <EOL> mar = mean_returns . round ( <NUM_LIT:8> ) <EOL> mask = rets < mar <EOL> downside_diff = rets [ mask ] - mar [ mask ] <EOL> if len ( downside_diff ) <= <NUM_LIT:1> : <EOL> return <NUM_LIT:0.0> <EOL> return np . std ( downside_diff , ddof = <NUM_LIT:1> ) * math . sqrt ( normalization_factor ) <EOL> def sortino_ratio ( algorithm_period_return , treasury_period_return , mar ) : <EOL> """<STR_LIT>""" <EOL> if zp_math . tolerant_equals ( mar , <NUM_LIT:0> ) : <EOL> return <NUM_LIT:0.0> <EOL> return ( algorithm_period_return - treasury_period_return ) / mar <EOL> def information_ratio ( algorithm_returns , benchmark_returns ) : <EOL> """<STR_LIT>""" <EOL> relative_returns = algorithm_returns - benchmark_returns <EOL> relative_deviation = relative_returns . std ( ddof = <NUM_LIT:1> ) <EOL> if zp_math . tolerant_equals ( relative_deviation , <NUM_LIT:0> ) or np . isnan ( relative_deviation ) : <EOL> return <NUM_LIT:0.0> <EOL> return np . mean ( relative_returns ) / relative_deviation <EOL> def alpha ( algorithm_period_return , treasury_period_return , <EOL> benchmark_period_returns , beta ) : <EOL> """<STR_LIT>""" <EOL> return algorithm_period_return - ( treasury_period_return + beta * <EOL> ( benchmark_period_returns - treasury_period_return ) ) <EOL> def get_treasury_rate ( treasury_curves , treasury_duration , day ) : <EOL> rate = None <EOL> curve = treasury_curves . ix [ day ] <EOL> idx = TREASURY_DURATIONS . index ( treasury_duration ) <EOL> for duration in TREASURY_DURATIONS [ idx : ] : <EOL> rate = curve [ duration ] <EOL> if rate is not None : <EOL> break <EOL> return rate <EOL> def search_day_distance ( end_date , dt , env ) : <EOL> tdd = env . trading_day_distance ( dt , end_date ) <EOL> if tdd is None : <EOL> return None <EOL> assert tdd >= <NUM_LIT:0> <EOL> return tdd <EOL> def select_treasury_duration ( start_date , end_date ) : <EOL> td = end_date - start_date <EOL> if td . days <= <NUM_LIT> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> * <NUM_LIT:2> + <NUM_LIT:1> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> * <NUM_LIT:3> + <NUM_LIT:1> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> * <NUM_LIT:5> + <NUM_LIT:2> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> * <NUM_LIT:7> + <NUM_LIT:2> : <EOL> treasury_duration = '<STR_LIT>' <EOL> elif td . days <= <NUM_LIT> * <NUM_LIT:10> + <NUM_LIT:2> : <EOL> treasury_duration = '<STR_LIT>' <EOL> else : <EOL> treasury_duration = '<STR_LIT>' <EOL> return treasury_duration <EOL> def choose_treasury ( select_treasury , treasury_curves , start_date , end_date , <EOL> env , compound = True ) : <EOL> """<STR_LIT>""" <EOL> treasury_duration = select_treasury ( start_date , end_date ) <EOL> end_day = end_date . replace ( hour = <NUM_LIT:0> , minute = <NUM_LIT:0> , second = <NUM_LIT:0> , microsecond = <NUM_LIT:0> ) <EOL> search_day = None <EOL> if end_day in treasury_curves . index : <EOL> rate = get_treasury_rate ( treasury_curves , <EOL> treasury_duration , <EOL> end_day ) <EOL> if rate is not None : <EOL> search_day = end_day <EOL> if not search_day : <EOL> search_days = treasury_curves . index <EOL> i = search_days . searchsorted ( end_day ) <EOL> for prev_day in search_days [ i - <NUM_LIT:1> : : - <NUM_LIT:1> ] : <EOL> rate = get_treasury_rate ( treasury_curves , <EOL> treasury_duration , <EOL> prev_day ) <EOL> if rate is not None : <EOL> search_day = prev_day <EOL> search_dist = search_day_distance ( end_date , prev_day , env ) <EOL> break <EOL> if search_day : <EOL> if ( search_dist is None or search_dist > <NUM_LIT:1> ) and search_days [ <NUM_LIT:0> ] <= end_day <= search_days [ - <NUM_LIT:1> ] : <EOL> message = "<STR_LIT>" <EOL> message = message . format ( dt = end_date , <EOL> term = treasury_duration , <EOL> search_day = search_day ) <EOL> log . warn ( message ) <EOL> if search_day : <EOL> td = end_date - start_date <EOL> if compound : <EOL> return rate * ( td . days + <NUM_LIT:1> ) / <NUM_LIT> <EOL> else : <EOL> return rate <EOL> message = "<STR_LIT>" <EOL> message = message . format ( <EOL> dt = end_date , <EOL> term = treasury_duration <EOL> ) <EOL> raise Exception ( message ) </s>
