Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_resampled_merge(testdata_1m_btc):
resampled = resample_to_interval(testdata_1m_btc, 5)
merged = resampled_merge(testdata_1m_btc, resampled)
assert (len(merged) == len(testdata_1m_btc))
assert "resample_5_open" in merged
assert "resample_5_close" in merged
assert "resample_5_low" in merged
assert "resample_5_high" in merged
assert "resample_5_date" not in merged
assert "resample_5_volume" not in merged
# Verify the assignment goes to the correct candle
# If resampling to 5m, then the resampled value needs to be on the 5m candle.
assert sum(merged.loc[merged['date'] == '2017-11-14 22:54:00', 'resample_5_close'].isna()) == 1
assert sum(merged.loc[merged['date'] == '2017-11-14 22:55:00', 'resample_5_close'].isna()) == 0
assert sum(merged.loc[merged['date'] == '2017-11-14 22:56:00', 'resample_5_close'].isna()) == 1
def test_resampled_merge_contains_indicator(testdata_1m_btc):
resampled = resample_to_interval(testdata_1m_btc, 5)
resampled['cmf'] = chaikin_money_flow(resampled, 5)
merged = resampled_merge(testdata_1m_btc, resampled)
print(merged)
assert "resample_5_cmf" in merged
def test_resample_to_interval(testdata_1m_btc):
result = resample_to_interval(testdata_1m_btc, 5)
# should be roughly a factor 5
assert (len(testdata_1m_btc) / len(result) > 4.5)
assert (len(testdata_1m_btc) / len(result) < 5.5)
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe['maShort'] = ta.EMA(dataframe, timeperiod=8)
dataframe['maMedium'] = ta.EMA(dataframe, timeperiod=21)
##################################################################################
# required for graphing
bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2)
dataframe['bb_lowerband'] = bollinger['lower']
dataframe['bb_upperband'] = bollinger['upper']
dataframe['bb_middleband'] = bollinger['mid']
dataframe_long = resample_to_interval(dataframe, timeframe_to_minutes(self.ticker_interval) * 12)
dataframe_long['sma'] = ta.SMA(dataframe_long, timeperiod=50, price='close')
dataframe = resampled_merge(dataframe, dataframe_long, fill_na=True)
return dataframe
def populate_indicators(self, dataframe: DataFrame) -> DataFrame:
from technical.util import resample_to_interval
from technical.util import resampled_merge
dataframe['sma5'] = ta.SMA(dataframe, timeperiod=5)
dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200)
# resample our dataframes
dataframe_short = resample_to_interval(dataframe, self.get_ticker_indicator() * 2)
dataframe_long = resample_to_interval(dataframe, self.get_ticker_indicator() * 8)
# compute our RSI's
dataframe_short['rsi'] = ta.RSI(dataframe_short, timeperiod=14)
dataframe_long['rsi'] = ta.RSI(dataframe_long, timeperiod=14)
# merge dataframe back together
dataframe = resampled_merge(dataframe, dataframe_short)
dataframe = resampled_merge(dataframe, dataframe_long)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe.fillna(method='ffill', inplace=True)
return dataframe
def populate_indicators(self, dataframe: DataFrame) -> DataFrame:
from technical.util import resample_to_interval
from technical.util import resampled_merge
dataframe['sma5'] = ta.SMA(dataframe, timeperiod=5)
dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200)
# resample our dataframes
dataframe_short = resample_to_interval(dataframe, self.get_ticker_indicator() * 2)
dataframe_long = resample_to_interval(dataframe, self.get_ticker_indicator() * 8)
# compute our RSI's
dataframe_short['rsi'] = ta.RSI(dataframe_short, timeperiod=14)
dataframe_long['rsi'] = ta.RSI(dataframe_long, timeperiod=14)
# merge dataframe back together
dataframe = resampled_merge(dataframe, dataframe_short)
dataframe = resampled_merge(dataframe, dataframe_long)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe.fillna(method='ffill', inplace=True)
return dataframe