223 lines
10 KiB
Python
223 lines
10 KiB
Python
import datetime as dt
|
|
import logging as log
|
|
|
|
import pytz
|
|
|
|
from shared_utilities import timeframe_to_minutes, ts_of_n_minutes_ago
|
|
|
|
|
|
# log.basicConfig(level=log.ERROR)
|
|
|
|
class Candles:
|
|
def __init__(self, exchanges, config_obj, data_source):
|
|
|
|
# A reference to the app configuration
|
|
self.config = config_obj
|
|
|
|
# The maximum amount of candles to load at one time.
|
|
self.max_records = self.config.app_data.get('max_data_loaded')
|
|
|
|
# This object maintains all the cached data.
|
|
self.data = data_source
|
|
|
|
# print('Setting the candle cache.')
|
|
# # Populate the cache:
|
|
# self.set_cache(symbol=self.config.users.get_chart_view(user_name='guest', specific_property='market'),
|
|
# interval=self.config.users.get_chart_view(user_name='guest', specific_property='timeframe'),
|
|
# exchange_name=self.config.users.get_chart_view(user_name='guest', specific_property='exchange_name'))
|
|
# print('DONE Setting cache')
|
|
|
|
def get_last_n_candles(self, num_candles: int, asset: str, timeframe: str, exchange: str, user_name: str):
|
|
"""
|
|
Return the last num_candles candles of a specified timeframe, symbol, and exchange_name.
|
|
|
|
:param user_name: The name of the user that owns the exchange.
|
|
:param num_candles: int - The number of records to return.
|
|
:param asset: str - The symbol of the symbol trading pair.
|
|
:param timeframe: str - The timespan each candle represents.
|
|
:param exchange: str - The name of the exchange_name.
|
|
:return: list: list[[atr0, atr1,...], [atr0, atr1,...],...]
|
|
"""
|
|
|
|
print('\n[GET LAST N CANDLES CALLED]:', asset, exchange, timeframe, num_candles)
|
|
# Convert the timeframe to candle_length.
|
|
minutes_per_candle = timeframe_to_minutes(timeframe)
|
|
|
|
# Calculate the approximate start_datetime the first of n records will have.
|
|
start_datetime = ts_of_n_minutes_ago(n=num_candles, candle_length=minutes_per_candle)
|
|
|
|
# Ensure the start_datetime is timezone aware
|
|
if start_datetime.tzinfo is None:
|
|
raise ValueError("start_datetime is timezone naive. Please ensure it is timezone-aware.")
|
|
|
|
# Fetch records older than start_datetime.
|
|
candles = self.data.get_records_since(start_datetime=start_datetime,
|
|
ex_details=[asset, timeframe, exchange, user_name])
|
|
if len(candles.index) < num_candles:
|
|
timesince = dt.datetime.now(pytz.UTC) - start_datetime
|
|
minutes_since = int(timesince.total_seconds() / 60)
|
|
print(f"""candles[103]: Received {len(candles.index)} candles but requested {num_candles}.
|
|
At {minutes_per_candle} minutes_per_candle since {start_datetime}. There should
|
|
be {minutes_since / minutes_per_candle} candles.""")
|
|
# Return the results. The start_datetime was approximate, so we may have retrieved an extra result.
|
|
return candles[-num_candles:]
|
|
|
|
def set_new_candle(self, cdata):
|
|
"""
|
|
Sets the last candle and updates the latest values list.
|
|
|
|
:param cdata: A list of data for a single candle.[ot,o,h,l,c,ct,...]
|
|
:return: True. todo: This method probably wont be needed.
|
|
"""
|
|
raise ValueError('set_new_candle')
|
|
# this will probably rename to update_cache()
|
|
|
|
#
|
|
# if cdata[4] < self.cashed_candlesticks[-2][4]:
|
|
# color = 'rgba(255,82,82, 0.8)' # Red
|
|
# else:
|
|
# color = 'rgba(0, 150, 136, 0.8)' # Green
|
|
# self.cashed_vol.append({'time': cdata['time'], 'value': cdata['vol'], "color": color})
|
|
# self.cashed_vol.pop(0)
|
|
# return True
|
|
#
|
|
def set_cache(self, symbol=None, interval=None, exchange_name=None, user_name=None):
|
|
"""
|
|
This method requests a chart from memory to ensure the cache is initialized.
|
|
|
|
:param user_name:
|
|
:param symbol: str - The symbol of the market.
|
|
:param interval: str - timeframe of the candles.
|
|
:param exchange_name: str - The name of the exchange_name to fetch from.
|
|
:return: None
|
|
"""
|
|
# By default, initialise cache with the last viewed chart.
|
|
if not symbol:
|
|
assert user_name is not None
|
|
symbol = self.config.users.get_chart_view(user_name=user_name, prop='market')
|
|
log.info(f'set_candle_history(): No symbol provided. Using{symbol}')
|
|
if not interval:
|
|
assert user_name is not None
|
|
interval = self.config.users.get_chart_view(user_name=user_name, prop='timeframe')
|
|
log.info(f'set_candle_history(): No timeframe provided. Using{interval}')
|
|
if not exchange_name:
|
|
assert user_name is not None
|
|
exchange_name = self.config.users.get_chart_view(user_name=user_name, prop='exchange_name')
|
|
|
|
# Log the completion to the console.
|
|
log.info('set_candle_history(): Loading candle data...')
|
|
|
|
# Load candles from database
|
|
_cdata = self.get_last_n_candles(num_candles=self.max_records,
|
|
asset=symbol, timeframe=interval, exchange=exchange_name, user_name=user_name)
|
|
|
|
# Log the completion to the console.
|
|
log.info('set_candle_history(): Candle data Loaded.')
|
|
return
|
|
|
|
@staticmethod
|
|
def get_colour_coded_volume(candles):
|
|
"""
|
|
Extracts a list of volume values from the candlesticks received.
|
|
:return: The values and a color in a start_datetime keyed dictionary.
|
|
"""
|
|
red = 'rgba(255,82,82, 0.8)'
|
|
green = 'rgba(0, 150, 136, 0.8)'
|
|
|
|
def get_color(r):
|
|
row_index = int(r.name)
|
|
if (row_index - 1) not in volumes.index:
|
|
return green
|
|
if candles.close.iloc[row_index - 1] < candles.close.iloc[row_index]:
|
|
return green
|
|
else:
|
|
return red
|
|
|
|
# Make sure the index is looking nice
|
|
candles.reset_index(inplace=True)
|
|
# Extract the open_time and volume columns.
|
|
volumes = candles.loc[:, ['open_time', 'volume']]
|
|
# Add the color field calling get_color() to supply the values.
|
|
volumes["color"] = volumes.apply(get_color, axis=1)
|
|
# Rename volume column to value
|
|
volumes = volumes.rename({'volume': 'value'}, axis=1)
|
|
return volumes
|
|
|
|
def get_latest_values(self, value_name: str, symbol: str, timeframe: str,
|
|
exchange: str, user_name: str, num_record: int = 1):
|
|
"""
|
|
Returns a timestamped dictionary of any value for the last num_record candles
|
|
of any market specified. Dictionary format {time:<start_datetime>, high:<data>}
|
|
|
|
:param user_name: str - The name of the user who owns this exchange.
|
|
:param exchange: str - The name of the exchange_name the market is traded.
|
|
:param timeframe: str - The timespan each candle represents.
|
|
:param symbol: str - The symbol of the market.
|
|
:param value_name: str - 'high'|'low'|'close'
|
|
:param num_record: The number of records requested.
|
|
:return:
|
|
"""
|
|
candles = self.get_last_n_candles(asset=symbol, exchange=exchange,
|
|
timeframe=timeframe, num_candles=num_record, user_name=user_name)
|
|
if value_name == 'volume':
|
|
values = self.get_colour_coded_volume(candles)
|
|
else:
|
|
values = candles[['open_time', value_name]]
|
|
|
|
values = values.rename({'open_time': 'time'}, axis=1)
|
|
# The timestamps are in milliseconds but lightweight charts needs it divided by 1000.
|
|
values['time'] = values['time'].div(1000)
|
|
return values
|
|
|
|
@staticmethod
|
|
def convert_candles(candles):
|
|
"""
|
|
Converts a dataframe of candlesticks into the format lightweight charts expects.
|
|
|
|
:param candles: dt.dataframe
|
|
:return: List - [{'open_time': value, 'open': value,...},...]
|
|
"""
|
|
|
|
new_candles = candles.loc[:, ['open_time', 'open', 'high', 'low', 'close']]
|
|
|
|
new_candles.rename(columns={'open_time': 'time'}, inplace=True)
|
|
|
|
# The timestamps are in milliseconds but lightweight charts needs it divided by 1000.
|
|
new_candles.loc[:, ['time']] = new_candles.loc[:, ['time']].div(1000)
|
|
|
|
return new_candles
|
|
|
|
def get_candle_history(self, num_records: int, symbol: str = None, interval: str = None,
|
|
exchange_name: str = None, user_name: str = None) -> list:
|
|
"""
|
|
Returns a specified number of candle records from cached memory in the lightweight charts format.
|
|
|
|
:param num_records: int - The number of candle records to retrieve.
|
|
:param symbol: str - The symbol of the market. If None, it will be fetched from user configuration.
|
|
:param interval: str - The timeframe of the candles. If None, it will be fetched from user configuration.
|
|
:param exchange_name: str - The name of the exchange. If None, it will be fetched from user configuration.
|
|
:param user_name: str - The name of the user who owns the exchange.
|
|
:return: list - Candle records in the lightweight charts format.
|
|
"""
|
|
# By default, initialise cache with the last viewed chart.
|
|
if not symbol:
|
|
assert user_name is not None
|
|
symbol = self.config.users.get_chart_view(user_name=user_name, prop='market''market')
|
|
log.info(f'set_candle_history(): No symbol provided. Using{symbol}')
|
|
|
|
if not interval:
|
|
assert user_name is not None
|
|
interval = self.config.users.get_chart_view(user_name=user_name, prop='market''timeframe')
|
|
log.info(f'set_candle_history(): No timeframe provided. Using{interval}')
|
|
|
|
if not exchange_name:
|
|
assert user_name is not None
|
|
exchange_name = self.config.users.get_chart_view(user_name=user_name, prop='market''exchange_name')
|
|
log.info(f'get_candle_history(): No exchange name provided. Using {exchange_name}')
|
|
|
|
candlesticks = self.get_last_n_candles(num_candles=num_records, asset=symbol, timeframe=interval,
|
|
exchange=exchange_name, user_name=user_name)
|
|
|
|
# Reformat relevant candlestick data into a list of python dictionary objects.
|
|
return self.convert_candles(candlesticks)
|