From 21ea7eb8094fbba9dbfcb65fc19ff16d0923ee8f Mon Sep 17 00:00:00 2001 From: Jamie Cash Date: Fri, 12 Mar 2021 15:10:07 +0000 Subject: [PATCH] Added status for monitoring --- config.yaml | 20 +-- mt5_correlation/correlation.py | 267 ++++++++++++++++++++++----------- mt5_correlation/gui.py | 61 ++++---- test/test_correlation.py | 40 +++-- 4 files changed, 246 insertions(+), 142 deletions(-) diff --git a/config.yaml b/config.yaml index 236223f..d2868ec 100644 --- a/config.yaml +++ b/config.yaml @@ -11,22 +11,22 @@ monitor: interval: 10 calculations: long: - from: 30 - min_prices: 400 + from: 60 + min_prices: 2000 max_set_size_diff_pct: 90 - overlap_pct: 80 + overlap_pct: 90 max_p_value: 0.05 medium: - from: 15 - min_prices: 300 + from: 30 + min_prices: 1000 max_set_size_diff_pct: 90 - overlap_pct: 80 + overlap_pct: 90 max_p_value: 0.05 short: - from: 5 - min_prices: 50 - max_set_size_diff_pct: 90 - overlap_pct: 80 + from: 10 + min_prices: 200 + max_set_size_diff_pct: 50 + overlap_pct: 50 max_p_value: 0.05 monitoring_threshold: 0.9 divergence_threshold: 0.8 diff --git a/mt5_correlation/correlation.py b/mt5_correlation/correlation.py index 34ff9fc..c73e083 100644 --- a/mt5_correlation/correlation.py +++ b/mt5_correlation/correlation.py @@ -15,6 +15,57 @@ import numpy as np from mt5_correlation.mt5 import MT5 +class CorrelationStatus: + """ + The status of the monitoring event for a symbol pair. + """ + + val = None + text = None + long_text = None + + def __init__(self, status_val, status_text, status_long_text=None): + """ + Creates a status. + :param status_val: + :param status_text: + :param status_long_text + :return: + """ + self.val = status_val + self.text = status_text + self.long_text = status_long_text + + def __eq__(self, other): + """ + Compare the status val. We can compare against other CorrelationStatus instances or against int. + :param other: + :return: + """ + if isinstance(other, self.__class__): + return self.val == other.val + elif isinstance(other, int): + return self.val == other + else: + return False + + def __str__(self): + """ + str is the text for the status. + :return: + """ + return self.text + + +# All status's for symbol pair from monitoring. Status set from assessing coefficient for all timeframes from last run. +STATUS_NOT_CALCULATED = CorrelationStatus(-1, 'NOT CALC', 'Coefficient could not be calculated') +STATUS_ABOVE_MONITORING_THRESHOLD = CorrelationStatus(1, 'ABOVE', 'All coefficients equal to or above the monitoring ' + 'threshold') +STATUS_BELOW_MONITORING_THRESHOLD = CorrelationStatus(2, 'BELOW', 'All coefficients below the monitoring threshold') +STATUS_INCONSISTENT = CorrelationStatus(3, 'INCONSISTENT', 'Coefficients not consistently above or below monitoring ' + 'threshold') + + class Correlation: """ A class to maintain the state of the calculated correlation coefficients. @@ -43,11 +94,6 @@ class Correlation: # The price data used to calculate the correlations __price_data = None - # The shortest timeframe (which is the largest value) for calculate_from. This will be used when we update - # the coefficient_data dataframe. All calculations for all values specified in calculate_from will be stored in - # coefficient_history, however only the shortest timeframe will be updated in coefficient_data. - __shortest_timeframe = None - # Coefficient data and history. Will be created in init call to __reset_coefficient_data coefficient_data = None coefficient_history = None @@ -180,8 +226,9 @@ class Correlation: self.coefficient_data = \ self.coefficient_data.append({'Symbol 1': symbol1, 'Symbol 2': symbol2, 'Base Coefficient': coefficient, 'UTC Date From': date_from, - 'UTC Date To': date_to, 'Timeframe': timeframe}, + 'UTC Date To': date_to, 'Timeframe': timeframe, 'Status': ''}, ignore_index=True) + self.__log.debug(f"Pair {index} of {num_pair_combinations}: {symbol1}:{symbol2} has a " f"coefficient of {coefficient}.") else: @@ -256,19 +303,9 @@ class Correlation: self.__autosave = autosave self.__filename = filename - # Store the shortest timeframe (which is the largest value) for calculate_from. This will be used when we update - # the coefficient_data dataframe. All calculations for all values specified in calculate_from will be stored in - # coefficient_history, however only the shortest timeframe will be updated in coefficient_data. - for params in self.__monitoring_params: - if self.__shortest_timeframe is None: - self.__shortest_timeframe = params['from'] - else: - self.__shortest_timeframe = min(self.__shortest_timeframe, params['from']) - # Create thread to run monitoring This will call private __monitor method that will run the calculation and # keep scheduling itself while self.monitoring is True. - params = {'interval': interval, "cache_time": cache_time, 'autosave': autosave, 'filename': filename} - thread = threading.Thread(target=self.__monitor, kwargs=params) + thread = threading.Thread(target=self.__monitor) thread.start() def stop_monitor(self): @@ -339,22 +376,31 @@ class Correlation: return coefficient - def get_coefficient_history(self, symbol1, symbol2, timeframe=None): + def get_coefficient_history(self, filters=None): """ - Returns the coefficient history for the specified symbol pair calculated during this instance. - Coefficient history does not persist between instances. - :param symbol1: - :param symbol2: - :param timeframe: Only return history for the specified timeframe. If None, this is ignored and all history - records are returned for the specified symbols + Returns the coefficient history that matches the supplied filter. + :param filters: Dict of all filters to apply. Possible values in dict are: + Symbol 1 + Symbol 2 + Coefficient + Timeframe + Date From + Date To + + If filter is not supplied, then all history is returned. + :return: dataframe containing history of coefficient data. """ - history = self.coefficient_history[(self.coefficient_history['Symbol 1'] == symbol1) & - (self.coefficient_history['Symbol 2'] == symbol2)] + history = self.coefficient_history - # If calculate from was specified, filter on it. - if timeframe is not None: - history = history[(history['Timeframe'] == timeframe)] + # Apply filters + if filters is not None: + for key in filters: + if key in history.columns: + history = history[history[key] == filters[key]] + else: + self.__log.warning(f"Invalid column name provided for filter. Filter column: {key} " + f"Valid columns: {history.columns}") return history @@ -364,25 +410,22 @@ class Correlation: :return: """ # Create dataframes for coefficient history. - coefficient_history_columns = ['Symbol 1', 'Symbol 2', 'Coefficient', 'Timeframe', 'Date From', 'Date To'] + coefficient_history_columns = ['Symbol 1', 'Symbol 2', 'Coefficient', 'Timeframe', 'Date To'] self.coefficient_history = pd.DataFrame(columns=coefficient_history_columns) # Clear tick data self.__monitor_tick_data = {} - # Clear columns from coefficient data - self.coefficient_data['Last Check'] = np.NaN - self.coefficient_data['Last Coefficient'] = np.NaN + # Clear status from coefficient data + self.coefficient_data['Status'] = '' - def get_ticks(self, symbol, date_from=None, date_to=None, cache_time=0, cache_only=False): + def get_ticks(self, symbol, date_from=None, date_to=None, cache_only=False): """ Returns the ticks for the specified symbol. Get's from cache if available and not older than cache_timeframe. :param symbol: Name of symbol to get ticks for. :param date_from: Date to get ticks from. Can only be None if getting from cache (cache_only=True) :param date_to:Date to get ticks to. Can only be None if getting from cache (cache_only=True) - :param cache_time: Number of seconds before cached data is stale. If > than this number of seconds has elapsed, - get data from source and refresh cache. :param cache_only: Only retrieve from cache. cache_time is ignored. Returns None if symbol is not available in cache. @@ -398,9 +441,9 @@ class Correlation: if cache_only: if symbol in self.__monitor_tick_data: ticks = self.__monitor_tick_data[symbol][1] - # Check if we already have it and it is not stale - elif symbol in self.__monitor_tick_data and utc_now < \ - self.__monitor_tick_data[symbol][0] + timedelta(seconds=cache_time): + # Check if we have a cache time defined, if we already have the tick data and it is not stale + elif self.__cache_time is not None and symbol in self.__monitor_tick_data and utc_now < \ + self.__monitor_tick_data[symbol][0] + timedelta(seconds=self.__cache_time): # Cached ticks are not stale. Get them ticks = self.__monitor_tick_data[symbol][1] self.__log.debug(f"Ticks for {symbol} retrieved from cache.") @@ -411,18 +454,53 @@ class Correlation: self.__log.debug(f"Ticks for {symbol} retrieved from source and cached.") return ticks - def __monitor(self, interval, cache_time=10, autosave=False, filename='autosave.cpd'): + def get_last_status(self, symbol1, symbol2): + """ + Get the last status for the specified symbol pair. + :param symbol1 + :param symbol2 + + :return: CorrelationStatus instance for symbol pair + """ + status_col = self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & + (self.coefficient_data['Symbol 2'] == symbol2), 'Status'] + status = status_col.values[0] + return status + + def get_last_calculation(self, symbol1=None, symbol2=None): + """ + Get the last calculation time the specified symbol pair. If no symbols are specified, then gets the last + calculation time across all pairs + :param symbol1 + :param symbol2 + + :return: last calculation time + """ + last_calc = None + if self.coefficient_data is not None and len(self.coefficient_data.index) > 0: + data = self.coefficient_data.copy() + + # Filter by symols if specified + data = data.loc[data['Symbol 1'] == symbol1] if symbol1 is not None else data + data = data.loc[data['Symbol 2'] == symbol2] if symbol2 is not None else data + + # Filter to remove blank dates + data = data.dropna(subset=['Last Calculation']) + + # Get the column + col = data['Last Calculation'] + + # Get max date from column + if col is not None and len(col) > 0: + last_calc = max(col.values) + + return last_calc + + def __monitor(self): """ The actual monitor method. Private. This should not be called outside of this class. Use start_monitoring and stop_monitoring. - :param interval: How often to check in seconds - :param cache_time: Tick data is cached so that we can check coefficients for multiple symbol pairs and reuse - the tick data. Number of seconds to cache tick data for before it becomes stale. - :param autosave: Whether to autosave after every monitor run. If there is no filename specified then will - create one named autosave.cpd - :param filename: Filename for autosave. Default is autosave.cpd. - :return: correlation coefficient, or None if coefficient could not be calculated. """ self.__log.debug(f"In monitor event. Monitoring: {self.__monitoring}.") @@ -430,15 +508,14 @@ class Correlation: # Only run if monitor is not stopped if self.__monitoring: # Update all coefficients - self.__update_all_coefficients(cache_time=cache_time) + self.__update_all_coefficients() # Autosave - if autosave: - self.save(filename=filename) + if self.__autosave: + self.save(filename=self.__filename) # Schedule the timer to run again - params = {'interval': interval, "cache_time": cache_time, 'autosave': autosave, 'filename': filename} - self.__scheduler.enter(delay=interval, priority=1, action=self.__monitor, kwargs=params) + self.__scheduler.enter(delay=self.__interval, priority=1, action=self.__monitor) # Log the stack. Debug stack overflow self.__log.debug(f"Current stack size: {len(inspect.stack())} Recursion limit: {sys.getrecursionlimit()}") @@ -448,13 +525,11 @@ class Correlation: self.__first_run = False self.__scheduler.run() - def __update_coefficients(self, symbol1, symbol2, cache_time=10): + def __update_coefficients(self, symbol1, symbol2): """ - Updates the long and short coefficients for the specified symbol pair + Updates the coefficients for the specified symbol pair :param symbol1: Name of symbol to calculate coefficient for. :param symbol2: Name of symbol to calculate coefficient for. - :param cache_time: Tick data is cached so that we can check coefficients for multiple symbol pairs and reuse - the tick data. Number of seconds to cache tick data for before it becomes stale. :return: correlation coefficient, or None if coefficient could not be calculated. """ @@ -473,8 +548,8 @@ class Correlation: date_from = date_to - timedelta(minutes=max_from) # Get the tick data for the longest timeframe calculation. - symbol1ticks = self.get_ticks(symbol=symbol1, date_from=date_from, date_to=date_to, cache_time=cache_time) - symbol2ticks = self.get_ticks(symbol=symbol2, date_from=date_from, date_to=date_to, cache_time=cache_time) + symbol1ticks = self.get_ticks(symbol=symbol1, date_from=date_from, date_to=date_to) + symbol2ticks = self.get_ticks(symbol=symbol2, date_from=date_from, date_to=date_to) # Resample to 1 sec OHLC, this will help with coefficient calculation ensuring that we dont have more than # one tick per second and ensuring that times can match. We will need to set the index to time for the @@ -500,6 +575,7 @@ class Correlation: # Calculate for all sets of monitoring_params if s1_prices is not None and s2_prices is not None: + coefficients = {} for params in self.__monitoring_params: # Get the from date as a datetime64 date_from_subset = pd.Timestamp(date_to - timedelta(minutes=params['from'])).to_datetime64() @@ -518,25 +594,24 @@ class Correlation: self.__log.debug(f"Symbol pair {symbol1}:{symbol2} has a coefficient of {coefficient} for last " f"{params['from']} minutes.") - # Update the coefficient data - if coefficient is not None: - self.__update_coefficient_data(symbol1=symbol1, symbol2=symbol2, coefficient=coefficient, - timeframe=params['from'], date_from=date_from_subset, - date_to=date_to) + # Add the coefficient to a dict {timeframe: coefficient}. We will update together for all for + # symbol pair and time + coefficients[params['from']] = coefficient - def __update_all_coefficients(self, cache_time=10): + # Update coefficient data for all coefficients for all timeframes for this run and symbol pair. + self.__update_coefficient_data(symbol1=symbol1, symbol2=symbol2, coefficients=coefficients, + date_to=date_to) + + def __update_all_coefficients(self): """ Updates the coefficient for all symbol pairs in that meet the min_coefficient threshold. Symbol pairs that meet the threshold can be accessed through the filtered_coefficient_data property. - - :param cache_time: Tick data is cached so that we can check coefficients for multiple symbol pairs and reuse - the tick data. Number of seconds to cache tick data for before it becomes stale. """ # Update latest coefficient for every pair for index, row in self.filtered_coefficient_data.iterrows(): symbol1 = row['Symbol 1'] symbol2 = row['Symbol 2'] - self.__update_coefficients(symbol1=symbol1, symbol2=symbol2, cache_time=cache_time) + self.__update_coefficients(symbol1=symbol1, symbol2=symbol2) def __reset_coefficient_data(self): """ @@ -545,7 +620,7 @@ class Correlation: """ # Create dataframes for coefficient data. coefficient_data_columns = ['Symbol 1', 'Symbol 2', 'Base Coefficient', 'UTC Date From', 'UTC Date To', - 'Timeframe', 'Last Check', 'Last Coefficient'] + 'Timeframe', 'Last Calculation', 'Status'] self.coefficient_data = pd.DataFrame(columns=coefficient_data_columns) # Clear coefficient history @@ -554,14 +629,12 @@ class Correlation: # Clear price data self.__price_data = None - def __update_coefficient_data(self, symbol1, symbol2, coefficient, timeframe, date_from, date_to): + def __update_coefficient_data(self, symbol1, symbol2, coefficients, date_to): """ Updates the coefficient data with the latest coefficient and adds to coefficient history. :param symbol1: :param symbol2: - :param coefficient: The coefficient calculated - :param timeframe: The number of minutes of price data used to calculate the coefficient - :param date_from: The date from for which the coefficient was calculated + :param coefficients: Dict of all coefficients calculated for this run and symbol pair. {timeframe: coefficient} :param date_to: The date from for which the coefficient was calculated :return: """ @@ -570,18 +643,40 @@ class Correlation: now = datetime.now(tz=timezone) # Update data if we have a coefficient and add to history - if coefficient is not None: - # The coefficient data table is only updated for the shortest calculation timeframe. - if timeframe == self.__shortest_timeframe: - self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & - (self.coefficient_data['Symbol 2'] == symbol2), - 'Last Check'] = now + if coefficients is not None: + # Update the coefficient data table with the Last Calculation time. + self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & + (self.coefficient_data['Symbol 2'] == symbol2), + 'Last Calculation'] = now - self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & - (self.coefficient_data['Symbol 2'] == symbol2), - 'Last Coefficient'] = coefficient + # Calculate status and update + status = self.__calculate_status(coefficients=coefficients) + self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & + (self.coefficient_data['Symbol 2'] == symbol2), + 'Status'] = status - # However the history data is always updated - row = pd.DataFrame(columns=self.coefficient_history.columns, - data=[[symbol1, symbol2, coefficient, timeframe, date_from, date_to]]) - self.coefficient_history = self.coefficient_history.append(row) + # Update history data + for key in coefficients: + row = pd.DataFrame(columns=self.coefficient_history.columns, + data=[[symbol1, symbol2, coefficients[key], key, date_to]]) + self.coefficient_history = self.coefficient_history.append(row) + + def __calculate_status(self, coefficients): + """ + Calculates the status from the supplied set of coefficients + :param coefficients: Dict of timeframes and coefficients {timeframe: coefficient} to calculate status from + :return: status + """ + status = None + values = coefficients.values() + + if None in values: + status = STATUS_NOT_CALCULATED + elif all(i >= self.monitoring_threshold for i in values): + status = STATUS_ABOVE_MONITORING_THRESHOLD + elif all(i < self.monitoring_threshold for i in values): + status = STATUS_BELOW_MONITORING_THRESHOLD + else: + status = STATUS_INCONSISTENT + + return status diff --git a/mt5_correlation/gui.py b/mt5_correlation/gui.py index 9d6d4c5..ff692d0 100644 --- a/mt5_correlation/gui.py +++ b/mt5_correlation/gui.py @@ -6,6 +6,7 @@ import matplotlib.dates import matplotlib import matplotlib.ticker as mticker +from mt5_correlation import correlation from mt5_correlation.correlation import Correlation from mt5_correlation.config import Config, SettingsDialog from datetime import datetime, timedelta @@ -34,8 +35,8 @@ class MonitorFrame(wx.Frame): COLUMN_DATE_FROM = 4 COLUMN_DATE_TO = 5 COLUMN_TIMEFRAME = 6 - COLUMN_LAST_CHECK = 7 - COLUMN_LAST_COEFFICIENT = 8 + COLUMN_LAST_CALCULATION = 7 + COLUMN_STATUS = 8 def __init__(self): # Super @@ -110,10 +111,10 @@ class MonitorFrame(wx.Frame): self.grid_correlations.SetColSize(self.COLUMN_DATE_FROM, 0) # UTC Date From. Hide self.grid_correlations.SetColSize(self.COLUMN_DATE_TO, 0) # UTC Date To. Hide self.grid_correlations.SetColSize(self.COLUMN_TIMEFRAME, 0) # Timeframe. Hide. - self.grid_correlations.SetColSize(self.COLUMN_LAST_CHECK, 100) # Last Check - self.grid_correlations.SetColSize(self.COLUMN_LAST_COEFFICIENT, 100) # Last Coefficient - self.grid_correlations.SetMinSize((520, 500)) - self.grid_correlations.SetMaxSize((520, -1)) + self.grid_correlations.SetColSize(self.COLUMN_LAST_CALCULATION, 0) # Last Calculation. Hide + self.grid_correlations.SetColSize(self.COLUMN_STATUS, 100) # Status + self.grid_correlations.SetMinSize((420, 500)) + self.grid_correlations.SetMaxSize((420, -1)) correlations_sizer.Add(self.grid_correlations, 1, wx.ALL | wx.EXPAND, 1) # Create the charts and hide as we have no data to display yet @@ -221,18 +222,18 @@ class MonitorFrame(wx.Frame): """ self.__log.debug(f"Refreshing grid. Timer running: {self.timer.IsRunning()}") + # Get coefficient data and join to history data + coef_data = self.__cor.coefficient_data.copy() + hist_data = self.__cor.get_coefficient_history() + # Update data self.table.data = self.__cor.coefficient_data.copy() # Format self.table.data.loc[:, 'Base Coefficient'] = self.table.data['Base Coefficient'].map('{:.5f}'.format) - self.table.data.loc[:, 'Last Check'] = pd.to_datetime(self.table.data['Last Check'], utc=True) - self.table.data.loc[:, 'Last Check'] = self.table.data['Last Check'].dt.strftime('%d-%m-%y %H:%M:%S') - self.table.data.loc[:, 'Last Coefficient'] = self.table.data['Last Coefficient'].map('{:.5f}'.format) - - # Remove nans. The ones from the float column will be str nan as they have been formatted - self.table.data = self.table.data.fillna('') - self.table.data.loc[:, 'Last Coefficient'] = self.table.data['Last Coefficient'].replace('nan', '') + self.table.data.loc[:, 'Last Calculation'] = pd.to_datetime(self.table.data['Last Calculation'], utc=True) + self.table.data.loc[:, 'Last Calculation'] = \ + self.table.data['Last Calculation'].dt.strftime('%d-%m-%y %H:%M:%S') # Start refresh self.grid_correlations.BeginBatch() @@ -319,7 +320,7 @@ class MonitorFrame(wx.Frame): reload_correlations = True if setting.startswith('logging.'): reload_logger = True - if setting.startswith('monitor.calculate_from'): + if setting.startswith('monitor.calculations'): reload_graph = True # Now perform the actions @@ -415,14 +416,14 @@ class MonitorFrame(wx.Frame): symbol_1_ticks = self.__cor.get_ticks(symbol1, cache_only=True) symbol_2_ticks = self.__cor.get_ticks(symbol2, cache_only=True) history_data_short = \ - self.__cor.get_coefficient_history(symbol1, symbol2, - self.__config.get('monitor.calculations.short.from')) + self.__cor.get_coefficient_history({'Symbol 1': symbol1, 'Symbol 2': symbol2, + 'Timeframe': self.__config.get('monitor.calculations.short.from')}) history_data_med = \ - self.__cor.get_coefficient_history(symbol1, symbol2, - self.__config.get('monitor.calculations.medium.from')) + self.__cor.get_coefficient_history({'Symbol 1': symbol1, 'Symbol 2': symbol2, + 'Timeframe': self.__config.get('monitor.calculations.medium.from')}) history_data_long = \ - self.__cor.get_coefficient_history(symbol1, symbol2, - self.__config.get('monitor.calculations.long.from')) + self.__cor.get_coefficient_history({'Symbol 1': symbol1, 'Symbol 2': symbol2, + 'Timeframe': self.__config.get('monitor.calculations.long.from')}) # Display if we have any data self.__log.debug(f"Refreshing history graph {symbol1}:{symbol2}.") self.__graph.draw(prices=[symbol_1_price_data, symbol_2_price_data], ticks=[symbol_1_ticks, symbol_2_ticks], @@ -442,6 +443,9 @@ class MonitorFrame(wx.Frame): if len(self.__selected_correlation) == 2: self.show_graph(symbol1=self.__selected_correlation[0], symbol2=self.__selected_correlation[1]) + # Set status message + self.SetStatusText(f"Status updated at {self.__cor.get_last_calculation():%d-%b %H:%M:%S}.", 1) + def __clear_history(self, event): """ Clears the calculated coefficient history and associated price data @@ -504,22 +508,11 @@ class DataTable(wx.grid.GridTableBase): # If column is last coefficient, get value and check against threshold. Highlight if diverged. threshold = Config().get('monitor.divergence_threshold') - if col in [MonitorFrame.COLUMN_LAST_COEFFICIENT]: - # Is coefficient <= threshold + if col in [MonitorFrame.COLUMN_STATUS]: + # Is status one of interest value = self.GetValue(row, col) if value != "": - value = float(value) - if value <= threshold: - attr.SetBackgroundColour(wx.YELLOW) - else: - attr.SetBackgroundColour(wx.WHITE) - elif col in [MonitorFrame.COLUMN_LAST_CHECK]: - # Was the last check within the last 2 monitoring intervals - value = self.GetValue(row, col) - if value != "": - value = datetime.strptime(value, '%d-%m-%y %H:%M:%S') - - if value >= datetime.now() - timedelta(minutes=2*Config().get('monitor.interval')): + if value in [correlation.STATUS_BELOW_MONITORING_THRESHOLD]: attr.SetBackgroundColour(wx.YELLOW) else: attr.SetBackgroundColour(wx.WHITE) diff --git a/test/test_correlation.py b/test/test_correlation.py index 5776c1b..d85f9fb 100644 --- a/test/test_correlation.py +++ b/test/test_correlation.py @@ -142,8 +142,15 @@ class TestCorrelation(unittest.TestCase): # Mock the tick data to contain 2 different sets. Then get twice. They should match as the data was cached. mock.copy_ticks_range.side_effect = [self.mock_base_prices, self.mock_correlated_prices] - base_ticks = cor.get_ticks('SYMBOL1', None, None, cache_time=3) - cached_ticks = cor.get_ticks('SYMBOL1', None, None, cache_time=3) + + # We need to start and stop the monitor as this will set the cache time + cor.start_monitor(interval=10, calculation_params={'from': 10, 'min_prices': 0, 'max_set_size_diff_pct': 0, + 'overlap_pct':0, 'max_p_value':1,}, cache_time=3) + cor.stop_monitor() + + # Get the ticks within cache time and check that they match + base_ticks = cor.get_ticks('SYMBOL1', None, None) + cached_ticks = cor.get_ticks('SYMBOL1', None, None) self.assertTrue(base_ticks.equals(cached_ticks), "Both sets of tick data should match as set 2 came from cache.") @@ -151,14 +158,14 @@ class TestCorrelation(unittest.TestCase): time.sleep(3) # Retrieve again. This one should be different as the cache has expired. - non_cached_ticks = cor.get_ticks('SYMBOL1', None, None, cache_time=3) + non_cached_ticks = cor.get_ticks('SYMBOL1', None, None) self.assertTrue(not base_ticks.equals(non_cached_ticks), "Both sets of tick data should differ as cached data had expired.") @patch('mt5_correlation.mt5.MetaTrader5') def test_start_monitor(self, mock): """ - Test that starting the monitor and running for 2 seconds produces two sets of coefficint history when using an + Test that starting the monitor and running for 2 seconds produces two sets of coefficient history when using an interval of 1 second. :param mock: :return: @@ -166,7 +173,7 @@ class TestCorrelation(unittest.TestCase): # Mock symbol return values mock.symbols_get.return_value = self.mock_symbols - # Correlation class + # Create correlation class cor = correlation.Correlation() # Calculate for price data. We should have 100% matching dates in sets. Get prices should be called 3 times. @@ -178,6 +185,9 @@ class TestCorrelation(unittest.TestCase): cor.calculate(date_from=self.start_date, date_to=self.end_date, timeframe=5, min_prices=100, max_set_size_diff_pct=100, overlap_pct=100, max_p_value=1) + # Set the monitoring threshold + cor.monitoring_threshold = 0.9 + # We will build some tick data for each symbol and patch it in. Tick data will be from 10 seconds ago to now. # We only need to patch in one set of tick data for each symbol as it will be cached. columns = ['time', 'ask'] @@ -202,11 +212,11 @@ class TestCorrelation(unittest.TestCase): # data quality metrics here as that is set elsewhere so these can be set to not take effect. Set cache level # high and don't use autosave. Timer runs in a separate thread so test can continue after it has started. cor.start_monitor(interval=1, calculation_params=[{'from': 0.66, 'min_prices': 0, - 'max_set_size_diff_pct': 0, 'overlap_pct':0, - 'max_p_value':1,}, + 'max_set_size_diff_pct': 0, 'overlap_pct': 0, + 'max_p_value': 1}, {'from': 0.33, 'min_prices': 0, - 'max_set_size_diff_pct': 0, 'overlap_pct':0, - 'max_p_value':1,}], cache_time=100, autosave=False) + 'max_set_size_diff_pct': 0, 'overlap_pct': 0, + 'max_p_value': 1}], cache_time=100, autosave=False) # Wait 2 seconds so timer runs twice time.sleep(2) @@ -218,8 +228,14 @@ class TestCorrelation(unittest.TestCase): self.assertEqual(len(cor.coefficient_history.index), 12) # We should have 2 coefficients calculated for a single symbol pair and timeframe - self.assertEqual(len(cor.get_coefficient_history('SYMBOL1', 'SYMBOL2', 0.66)), 2, - "We should have 2 history records for SYMBOL1:SYMBOL2 using the 0.66 min timeframe.") + self.assertEqual(len(cor.get_coefficient_history({'Symbol 1': 'SYMBOL1', 'Symbol 2': 'SYMBOL2', + 'Timeframe': 0.66})), + 2, "We should have 2 history records for SYMBOL1:SYMBOL2 using the 0.66 min timeframe.") + + # The status should be BELOW for SYMBOL1:SYMBOL2 and SYMBOL1:SYMBOL4. It should be ABOVE for SYMBOL2:SYMBOL4. + self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL2') == correlation.STATUS_BELOW_MONITORING_THRESHOLD) + self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL4') == correlation.STATUS_BELOW_MONITORING_THRESHOLD) + self.assertTrue(cor.get_last_status('SYMBOL2', 'SYMBOL4') == correlation.STATUS_ABOVE_MONITORING_THRESHOLD) @patch('mt5_correlation.mt5.MetaTrader5') def test_load_and_save(self, mock): @@ -255,7 +271,7 @@ class TestCorrelation(unittest.TestCase): # Start monitor and run for a seconds with a 1 second interval to produce some coefficient history. Then stop # the monitor cor.start_monitor(interval=1, calculation_params={'from': 0.66, 'min_prices': 0, 'max_set_size_diff_pct': 0, - 'overlap_pct': 0, 'max_p_value':1}, + 'overlap_pct': 0, 'max_p_value': 1}, cache_time=100, autosave=False) time.sleep(2) cor.stop_monitor()