diff --git a/config.yaml b/config.yaml index 929a926..1cf2c10 100644 --- a/config.yaml +++ b/config.yaml @@ -30,6 +30,7 @@ monitor: max_p_value: 0.05 monitoring_threshold: 0.9 divergence_threshold: 0.8 + monitor_inverse: true tick_cache_time: 10 autosave: true logging: diff --git a/mt5_correlation/correlation.py b/mt5_correlation/correlation.py index d246bb8..be9db04 100644 --- a/mt5_correlation/correlation.py +++ b/mt5_correlation/correlation.py @@ -10,7 +10,6 @@ from scipy.stats.stats import pearsonr import pickle import inspect import sys -import numpy as np from mt5_correlation.mt5 import MT5 @@ -82,6 +81,9 @@ class Correlation: # below this threshold will be considered as having diverged divergence_threshold = 0.8 + # Flag to determine we monitor and report on inverse correlations + monitor_inverse = False + # Toggle on whether we are monitoring or not. Set through start_monitor and stop_monitor __monitoring = False @@ -106,13 +108,14 @@ class Correlation: # Dict: {Symbol: [retrieved datetime, ticks dataframe]} __monitor_tick_data = {} - def __init__(self, monitoring_threshold=0.9, divergence_threshold=0.8): + def __init__(self, monitoring_threshold=0.9, divergence_threshold=0.8, monitor_inverse=False): """ Initialises the Correlation class. :param monitoring_threshold: Only correlations that are greater than or equal to this threshold will be monitored. :param divergence_threshold: Correlations that are being monitored and fall below this threshold are considered to have diverged. + :param monitor_inverse: Whether we will monitor and report on negative / inverse correlations. """ # Logger self.__log = logging.getLogger(__name__) @@ -126,19 +129,28 @@ class Correlation: # Create timer for continuous monitoring self.__scheduler = sched.scheduler(time.time, time.sleep) - # Set thresholds + # Set thresholds and flags self.monitoring_threshold = monitoring_threshold self.divergence_threshold = divergence_threshold + self.monitor_inverse = monitor_inverse @property def filtered_coefficient_data(self): """ :return: Coefficient data filtered so that all base coefficients >= monitoring_threshold """ + filtered_data = None if self.coefficient_data is not None: - return self.coefficient_data.loc[self.coefficient_data['Base Coefficient'] >= self.monitoring_threshold] - else: - return None + if self.monitor_inverse: + filtered_data = self.coefficient_data \ + .loc[(self.coefficient_data['Base Coefficient'] >= self.monitoring_threshold) | + (self.coefficient_data['Base Coefficient'] <= self.monitoring_threshold * -1)] + + else: + filtered_data = self.coefficient_data.loc[self.coefficient_data['Base Coefficient'] >= + self.monitoring_threshold] + + return filtered_data def load(self, filename): """ @@ -495,7 +507,7 @@ class Correlation: if self.coefficient_data is not None and len(self.coefficient_data.index) > 0: data = self.coefficient_data.copy() - # Filter by symols if specified + # Filter by symbols 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 @@ -511,6 +523,23 @@ class Correlation: return last_calc + def get_base_coefficient(self, symbol1, symbol2): + """ + Returns the base coefficient for the specified symbol pair + :param symbol1: + :param symbol2: + :return: + """ + base_coefficient = None + if self.coefficient_data is not None: + row = self.coefficient_data[(self.coefficient_data['Symbol 1'] == symbol1) & + (self.coefficient_data['Symbol 2'] == symbol2)] + + if row is not None and len(row) == 1: + base_coefficient = row.iloc[0]['Base Coefficient'] + + return base_coefficient + def __monitor(self): """ The actual monitor method. Private. This should not be called outside of this class. Use start_monitoring and @@ -664,8 +693,11 @@ class Correlation: (self.coefficient_data['Symbol 2'] == symbol2), 'Last Calculation'] = now + # Are we an inverse correlation + inverse = self.get_base_coefficient(symbol1, symbol2) <= self.monitoring_threshold * -1 + # Calculate status and update - status = self.__calculate_status(coefficients=coefficients) + status = self.__calculate_status(coefficients=coefficients, inverse=inverse) self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) & (self.coefficient_data['Symbol 2'] == symbol2), 'Status'] = status @@ -676,22 +708,32 @@ class Correlation: data=[[symbol1, symbol2, coefficients[key], key, date_to]]) self.coefficient_history = self.coefficient_history.append(row) - def __calculate_status(self, coefficients): + def __calculate_status(self, coefficients, inverse): """ Calculates the status from the supplied set of coefficients :param coefficients: Dict of timeframes and coefficients {timeframe: coefficient} to calculate status from + :param: Whether we are calculating status based on normal or inverse correlation :return: status """ - status = None + status = STATUS_NOT_CALCULATED values = coefficients.values() - if None in values: - status = STATUS_NOT_CALCULATED - elif all(i >= self.divergence_threshold for i in values): - status = STATUS_ABOVE_DIVERGENCE_THRESHOLD - elif all(i < self.divergence_threshold for i in values): - status = STATUS_BELOW_DIVERGENCE_THRESHOLD - else: - status = STATUS_INCONSISTENT + if None not in values: + if self.monitor_inverse and inverse: + # Calculation for inverse calculations + if all(i <= self.divergence_threshold * -1 for i in values): + status = STATUS_ABOVE_DIVERGENCE_THRESHOLD + elif all(i > self.divergence_threshold * -1 for i in values): + status = STATUS_BELOW_DIVERGENCE_THRESHOLD + else: + status = STATUS_INCONSISTENT + else: + # Calculation for standard correlations + if all(i >= self.divergence_threshold for i in values): + status = STATUS_ABOVE_DIVERGENCE_THRESHOLD + elif all(i < self.divergence_threshold for i in values): + status = STATUS_BELOW_DIVERGENCE_THRESHOLD + else: + status = STATUS_INCONSISTENT return status diff --git a/mt5_correlation/gui.py b/mt5_correlation/gui.py index 718c589..b617c9f 100644 --- a/mt5_correlation/gui.py +++ b/mt5_correlation/gui.py @@ -52,7 +52,8 @@ class MonitorFrame(wx.Frame): # Create correlation instance to maintain state of calculated coefficients. Set min coefficient from config self.__cor = cor.Correlation(monitoring_threshold=self.__config.get("monitor.monitoring_threshold"), - divergence_threshold=self.__config.get("monitor.divergence_threshold")) + divergence_threshold=self.__config.get("monitor.divergence_threshold"), + monitor_inverse=self.__config.get("monitor.monitor_inverse")) # Status bar. 2 fields, one for monitoring status and one for general status. On open, monitoring status is not # monitoring. SetBackgroundColour will change colour of both. Couldn't find a way to set on single field only. @@ -221,12 +222,8 @@ 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() + self.table.data = self.__cor.filtered_coefficient_data.copy() # Format self.table.data.loc[:, 'Base Coefficient'] = self.table.data['Base Coefficient'].map('{:.5f}'.format) @@ -426,7 +423,9 @@ class MonitorFrame(wx.Frame): # 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], - history=[history_data_short, history_data_med, history_data_long], symbols=[symbol1, symbol2]) + history=[history_data_short, history_data_med, history_data_long], symbols=[symbol1, symbol2], + divergence_threshold=self.__cor.divergence_threshold, + monitor_inverse=self.__cor.monitor_inverse) # Un-hide and layout if hidden if not self.__graph.IsShown(): @@ -552,13 +551,17 @@ class GraphPanel(wx.Panel): self.__axes = None self.__fig = None - def draw(self, prices=None, ticks=None, history=None, symbols=None): + def draw(self, prices=None, ticks=None, history=None, symbols=None, divergence_threshold=None, + monitor_inverse=False): """ Plot the correlations. :param prices: Price data used to calculate base coefficient. List [Symbol1 Price Data, Symbol 2 Price Data] :param ticks: Ticks used to calculate last coefficient. List [Symbol1, Symbol2] :param history: Coefficient history data. List of data for one or more timeframes. :param symbols: Symbols. List [Symbol1, Symbol2] + :param divergence_threshold: The divergence threshold. Will be plotted on the coefficients charts if specified. + :param monitor_inverse: Are we monitoring inverse correlations. If so, a line for the inverse threshold will be + plotted if the divergence threshold is specified. :return: """ @@ -638,6 +641,10 @@ class GraphPanel(wx.Panel): f"{Config().get('monitor.calculations.medium.from')} Minutes", f"{Config().get('monitor.calculations.short.from')} Minutes"]] + # lines + horiz_lines = [None, None, None, None, [divergence_threshold, divergence_threshold * -1 + if divergence_threshold is not None and monitor_inverse else None]] + # Draw 5 charts for index in range(0, len(self.__axes)): # Titles and axis labels @@ -679,6 +686,12 @@ class GraphPanel(wx.Panel): else: self.__axes[index].set_xticklabels([]) + # Lines + if horiz_lines[index] is not None and isinstance(horiz_lines[index], list): + for line_pos in horiz_lines[index]: + if line_pos is not None: + self.__axes[index].axhline(y=line_pos, color="red", label='_nolegend_', linewidth=1) + # Legends if legends[index] is not None: self.__axes[index].legend(legends[index]) diff --git a/test/test_correlation.py b/test/test_correlation.py index ac00b58..acd65d7 100644 --- a/test/test_correlation.py +++ b/test/test_correlation.py @@ -14,7 +14,8 @@ class TestCorrelation(unittest.TestCase): mock_symbols = [Symbol(name='SYMBOL1', visible=True), Symbol(name='SYMBOL2', visible=True), Symbol(name='SYMBOL3', visible=False), - Symbol(name='SYMBOL4', visible=True)] + Symbol(name='SYMBOL4', visible=True), + Symbol(name='SYMBOL5', visible=True)] # Start and end date for price data and mock prices: base; correlated; and uncorrelated. start_date = None @@ -38,6 +39,7 @@ class TestCorrelation(unittest.TestCase): self.mock_correlated_prices = pd.DataFrame(columns=self.price_columns) self.mock_uncorrelated_prices = pd.DataFrame(columns=self.price_columns) self.mock_correlated_different_dates = pd.DataFrame(columns=self.price_columns) + self.mock_inverse_correlated_prices = pd.DataFrame(columns=self.price_columns) # Build the price data for the test. One price every 5 minutes for 500 rows. Base will use min for price, # correlated will use min + 5 and uncorrelated will use random @@ -55,6 +57,9 @@ class TestCorrelation(unittest.TestCase): self.mock_correlated_different_dates.append(pd.DataFrame(columns=self.price_columns, data=[[date + timedelta(minutes=100), date.minute + 5]])) + self.mock_inverse_correlated_prices = \ + self.mock_inverse_correlated_prices.append(pd.DataFrame(columns=self.price_columns, + data=[[date, (date.minute + 5) * -1]])) @patch('mt5_correlation.mt5.MetaTrader5') def test_calculate(self, mock): @@ -67,40 +72,54 @@ class TestCorrelation(unittest.TestCase): mock.symbols_get.return_value = self.mock_symbols # Correlation class - cor = correlation.Correlation() + cor = correlation.Correlation(monitoring_threshold=1, monitor_inverse=True) - # Calculate for price data. We should have 100% matching dates in sets. Get prices should be called 3 times. - # We dont have a SYMBOL2 as this is set as not visible. Correlations should be as follows: + # Calculate for price data. We should have 100% matching dates in sets. Get prices should be called 4 times. + # We don't have a SYMBOL3 as this is set as not visible. Correlations should be as follows: # SYMBOL1:SYMBOL2 should be fully correlated (1) # SYMBOL1:SYMBOL4 should be uncorrelated (0) - # SYMBOL2:SYMBOL4 should be uncorrelated (0) + # SYMBOL1:SYMBOL5 should be negatively correlated + # SYMBOL2:SYMBOL5 should be negatively correlated # We will not use p_value as the last set uses random numbers so p value will not be useful. mock.copy_rates_range.side_effect = [self.mock_base_prices, self.mock_correlated_prices, - self.mock_uncorrelated_prices] + self.mock_uncorrelated_prices, self.mock_inverse_correlated_prices] 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) - # Test the output. We should have 3 rows. S1:S2 c=1, S1:S4 c<1, S2:S4 c<1> - self.assertEqual(len(cor.coefficient_data.index), 3, "There should be three correlations rows calculated.") - self.assertEqual(cor.coefficient_data.iloc[0, 2], 1, "The correlation for SYMBOL1:SYMBOL2 should be 1.") - self.assertTrue(cor.coefficient_data.iloc[1, 2] < 1, "The correlation for SYMBOL1:SYMBOL4 should be <1.") - self.assertTrue(cor.coefficient_data.iloc[2, 2] < 1, "The correlation for SYMBOL2:SYMBOL4 should be <1.") + # Test the output. We should have 6 rows. S1:S2 c=1, S1:S4 c<1, S1:S5 c=-1, S2:S5 c=-1. We are not checking + # S2:S4 or S4:S5 + self.assertEqual(len(cor.coefficient_data.index), 6, "There should be six correlations rows calculated.") + self.assertEqual(cor.get_base_coefficient('SYMBOL1', 'SYMBOL2'), 1, + "The correlation for SYMBOL1:SYMBOL2 should be 1.") + self.assertTrue(cor.get_base_coefficient('SYMBOL1', 'SYMBOL4') < 1, + "The correlation for SYMBOL1:SYMBOL4 should be <1.") + self.assertEqual(cor.get_base_coefficient('SYMBOL1', 'SYMBOL5'), -1, + "The correlation for SYMBOL1:SYMBOL5 should be -1.") + self.assertEqual(cor.get_base_coefficient('SYMBOL2', 'SYMBOL5'), -1, + "The correlation for SYMBOL2:SYMBOL5 should be -1.") - # Set the monitoring threshold to 1 and get filtered correlations. There should only be 1 - cor.monitoring_threshold = 1 - self.assertEqual(len(cor.filtered_coefficient_data.index), 1, "There should only be 1 row in filtered " - "coefficient data.") + # Monitoring threshold is 1 and we are monitoring inverse. Get filtered correlations. There should be 3 (S1:S2, + # S1:S5 and S2:S5) + self.assertEqual(len(cor.filtered_coefficient_data.index), 3, + "There should be 3 rows in filtered coefficient data when we are monitoring inverse " + "correlations.") - # Now were going to recalculate, but this time: - # SYMBOL1:SYMBOL2 will have non overlapping dates and coefficient should be None. There shouldn't be a row - # SYMBOL1:SYMBOL4 will be correlated - # SYMBOL2:SYMBOL4 will have non overlapping dates and coefficient should be None. There shouldn't be a row. + # Now aren't monitoring inverse correlations. There should only be one correlation when filtered + cor.monitor_inverse = False + self.assertEqual(len(cor.filtered_coefficient_data.index), 1, + "There should be only 1 rows in filtered coefficient data when we are not monitoring inverse " + "correlations.") + + # Now were going to recalculate, but this time SYMBOL1:SYMBOL2 will have non overlapping dates and coefficient + # should be None. There shouldn't be a row. We should have correlations for S1:S4, S1:S5 and S4:S5 mock.copy_rates_range.side_effect = [self.mock_base_prices, self.mock_correlated_different_dates, - self.mock_correlated_prices] + self.mock_correlated_prices, self.mock_correlated_prices] 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) - self.assertEqual(len(cor.coefficient_data.index), 1, "There should be one correlations rows calculated.") + self.assertEqual(len(cor.coefficient_data.index), 3, "There should be three correlations rows calculated.") self.assertEqual(cor.coefficient_data.iloc[0, 2], 1, "The correlation for SYMBOL1:SYMBOL4 should be 1.") + self.assertEqual(cor.coefficient_data.iloc[1, 2], 1, "The correlation for SYMBOL1:SYMBOL5 should be 1.") + self.assertEqual(cor.coefficient_data.iloc[2, 2], 1, "The correlation for SYMBOL4:SYMBOL5 should be 1.") # Get the price data used to calculate the coefficients fro symbol 1. It should match mock_base_prices. price_data = cor.get_price_data('SYMBOL1') @@ -127,6 +146,10 @@ class TestCorrelation(unittest.TestCase): coefficient = cor.calculate_coefficient(self.mock_base_prices, self.mock_correlated_different_dates) self.assertTrue(coefficient < 1, "Coefficient should be None.") + # Test 2 inversely correlated sets + coefficient = cor.calculate_coefficient(self.mock_base_prices, self.mock_inverse_correlated_prices) + self.assertEqual(coefficient, -1, "Coefficient should be -1.") + @patch('mt5_correlation.mt5.MetaTrader5') def test_get_ticks(self, mock): """ @@ -145,7 +168,7 @@ class TestCorrelation(unittest.TestCase): # 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) + 'overlap_pct': 0, 'max_p_value': 1}, cache_time=3) cor.stop_monitor() # Get the ticks within cache time and check that they match @@ -174,13 +197,13 @@ class TestCorrelation(unittest.TestCase): mock.symbols_get.return_value = self.mock_symbols # Create correlation class. We will set a divergence threshold so that we can test status. - cor = correlation.Correlation(divergence_threshold=0.8) + cor = correlation.Correlation(divergence_threshold=0.8, monitor_inverse=True) - # Calculate for price data. We should have 100% matching dates in sets. Get prices should be called 3 times. + # Calculate for price data. We should have 100% matching dates in sets. Get prices should be called 4 times. # We dont have a SYMBOL2 as this is set as not visible. All pairs should be correlated for the purpose of this # test. mock.copy_rates_range.side_effect = [self.mock_base_prices, self.mock_correlated_prices, - self.mock_correlated_prices] + self.mock_correlated_prices, self.mock_inverse_correlated_prices] 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) @@ -190,20 +213,22 @@ class TestCorrelation(unittest.TestCase): columns = ['time', 'ask'] starttime = datetime.now() - timedelta(seconds=10) tick_data_s1 = pd.DataFrame(columns=columns) - tick_data_s3 = pd.DataFrame(columns=columns) + tick_data_s2 = pd.DataFrame(columns=columns) tick_data_s4 = pd.DataFrame(columns=columns) + tick_data_s5 = pd.DataFrame(columns=columns) now = datetime.now() price_base = 1 while starttime < now: tick_data_s1 = tick_data_s1.append(pd.DataFrame(columns=columns, data=[[starttime, price_base * 0.5]])) - tick_data_s3 = tick_data_s1.append(pd.DataFrame(columns=columns, data=[[starttime, price_base * 0.1]])) + tick_data_s2 = tick_data_s1.append(pd.DataFrame(columns=columns, data=[[starttime, price_base * 0.1]])) tick_data_s4 = tick_data_s1.append(pd.DataFrame(columns=columns, data=[[starttime, price_base * 0.25]])) + tick_data_s5 = tick_data_s1.append(pd.DataFrame(columns=columns, data=[[starttime, price_base * -0.25]])) starttime = starttime + timedelta(milliseconds=10*random.randint(0, 100)) price_base += 1 # Patch it in - mock.copy_ticks_range.side_effect = [tick_data_s1, tick_data_s3, tick_data_s4] + mock.copy_ticks_range.side_effect = [tick_data_s1, tick_data_s2, tick_data_s4, tick_data_s5] # Start the monitor. Run every second. Use ~10 and ~5 seconds of data. Were not testing the overlap and price # data quality metrics here as that is set elsewhere so these can be set to not take effect. Set cache level @@ -221,20 +246,23 @@ class TestCorrelation(unittest.TestCase): # Stop the monitor cor.stop_monitor() - # We should have 2 coefficients calculated for each symbol pair for each date_from value, so 12 in total. - self.assertEqual(len(cor.coefficient_history.index), 12) + # We should have 2 coefficients calculated for each symbol pair (6), for each date_from value (2), + # for each run (2) so 24 in total. + self.assertEqual(len(cor.coefficient_history.index), 24) # We should have 2 coefficients calculated for a single symbol pair and 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 should be ABOVE for and SYMBOL1:SYMBOL4 and - # SYMBOL2:SYMBOL4. + # The status should be BELOW for SYMBOL1:SYMBOL2 and ABOVE for SYMBOL1:SYMBOL4 and SYMBOL2:SYMBOL4. self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL2') == correlation.STATUS_BELOW_DIVERGENCE_THRESHOLD) self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL4') == correlation.STATUS_ABOVE_DIVERGENCE_THRESHOLD) self.assertTrue(cor.get_last_status('SYMBOL2', 'SYMBOL4') == correlation.STATUS_ABOVE_DIVERGENCE_THRESHOLD) + # We are monitoring inverse correlations, status for SYMBOL1:SYMBOL5 should be BELOW + self.assertTrue(cor.get_last_status('SYMBOL2', 'SYMBOL5') == correlation.STATUS_BELOW_DIVERGENCE_THRESHOLD) + @patch('mt5_correlation.mt5.MetaTrader5') def test_load_and_save(self, mock): """Calculate and run monitor for a few seconds. Store the data. Save it, load it then compare against stored @@ -246,7 +274,7 @@ class TestCorrelation(unittest.TestCase): # Patch symbol and price data, then calculate mock.symbols_get.return_value = self.mock_symbols mock.copy_rates_range.side_effect = [self.mock_base_prices, self.mock_correlated_prices, - self.mock_correlated_prices] + self.mock_correlated_prices, self.mock_inverse_correlated_prices] 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)