diff --git a/mt5_correlation/correlation.py b/mt5_correlation/correlation.py index 8f28396..1c4a45d 100644 --- a/mt5_correlation/correlation.py +++ b/mt5_correlation/correlation.py @@ -255,8 +255,8 @@ class Correlation: """ Calculates the correlation coefficient between two sets of price data. Uses close price. - :param symbol1_prices: Pandas dataframe containing prices or ticks for symbol 1 - :param symbol2_prices: Pandas dataframe containing prices or ticks for symbol 2 + :param symbol1_prices: Pandas dataframe containing prices for symbol 1 + :param symbol2_prices: Pandas dataframe containing prices for symbol 2 :param min_prices: The minimum number of prices that should be used to calculate coefficient. If this threshold is not met then returned coefficient will be None :param max_set_size_diff_pct: Correlations will only be calculated if the sizes of the two price data sets are @@ -511,6 +511,10 @@ class Correlation: coefficient_history_columns = ['Symbol 1', 'Symbol 2', 'Coefficient', 'UTC Date From', 'UTC Date To'] self.coefficient_history = pd.DataFrame(columns=coefficient_history_columns) + # Clear price data and tick data + self.__price_data = None + self.__monitor_tick_data = {} + def __update_coefficient_data(self, symbol1, symbol2, coefficient, date_from, date_to): """ Updates the coefficient data with the latest coefficient and adds to coefficient history. diff --git a/test/test_correlation.py b/test/test_correlation.py new file mode 100644 index 0000000..cab52dc --- /dev/null +++ b/test/test_correlation.py @@ -0,0 +1,279 @@ +import unittest +from unittest.mock import patch +import time +import mt5_correlation.correlation as correlation +import pandas as pd +from datetime import datetime, timedelta +from test_mt5 import Symbol +import random +import os + + +class TestCorrelation(unittest.TestCase): + # Mock symbols. 4 Symbols, 3 visible. + mock_symbols = [Symbol(name='SYMBOL1', visible=True), + Symbol(name='SYMBOL2', visible=True), + Symbol(name='SYMBOL3', visible=False), + Symbol(name='SYMBOL4', visible=True)] + + # Start and end date for price data and mock prices: base; correlated; and uncorrelated. + start_date = None + end_date = None + price_columns = None + mock_base_prices = None + mock_correlated_prices = None + mock_uncorrelated_prices = None + + def setUp(self): + """ + Creates some price data fro use in tests + :return: + """ + # Start and end date for price data and mock price dataframes. One for: base; correlated; uncorrelated and + # different dates. + self.start_date = datetime(2021, 1, 1, 1, 5, 0) + self.end_date = datetime(2021, 1, 1, 11, 30, 0) + self.price_columns = ['time', 'close'] + self.mock_base_prices = pd.DataFrame(columns=self.price_columns) + 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) + + # 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 + for date in (self.start_date + timedelta(minutes=m) for m in range(0, 500*5, 5)): + self.mock_base_prices = self.mock_base_prices.append(pd.DataFrame(columns=self.price_columns, + data=[[date, date.minute]])) + self.mock_correlated_prices = \ + self.mock_correlated_prices.append(pd.DataFrame(columns=self.price_columns, + data=[[date, date.minute + 5]])) + self.mock_uncorrelated_prices = \ + self.mock_uncorrelated_prices.append(pd.DataFrame(columns=self.price_columns, + data=[[date, random.randint(0, 1000000)]])) + + self.mock_correlated_different_dates = \ + self.mock_correlated_different_dates.append(pd.DataFrame(columns=self.price_columns, + data=[[date + timedelta(minutes=100), + date.minute + 5]])) + + @patch('mt5_correlation.mt5.MetaTrader5') + def test_calculate(self, mock): + """ + Test the calculate method. Uses mock for MT5 symbols and prices. + :param mock: + :return: + """ + # Mock symbol return values + mock.symbols_get.return_value = self.mock_symbols + + # Correlation class + cor = correlation.Correlation() + + # 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: + # SYMBOL1:SYMBOL2 should be fully correlated (1) + # SYMBOL1:SYMBOL4 should be uncorrelated (0) + # SYMBOL2:SYMBOL4 should be uncorrelated (0) + # 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] + 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.") + + # 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.") + + # 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. + mock.copy_rates_range.side_effect = [self.mock_base_prices, self.mock_correlated_different_dates, + 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(cor.coefficient_data.iloc[0, 2], 1, "The correlation for SYMBOL1:SYMBOL4 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') + self.assertTrue(price_data.equals(self.mock_base_prices), "Price data returned post calculation should match " + "mock price data.") + + def test_calculate_coefficient(self): + """ + Tests the coefficient calculation. + :return: + """ + # Correlation class + cor = correlation.Correlation() + + # Test 2 correlated sets + coefficient = cor.calculate_coefficient(self.mock_base_prices, self.mock_correlated_prices) + self.assertEqual(coefficient, 1, "Coefficient should be 1.") + + # Test 2 uncorrelated sets. Set p value to 1 to force correlation to be returned. + coefficient = cor.calculate_coefficient(self.mock_base_prices, self.mock_uncorrelated_prices, max_p_value=1) + self.assertTrue(coefficient < 1, "Coefficient should be < 1.") + + # Test 2 sets where prices dont overlap + coefficient = cor.calculate_coefficient(self.mock_base_prices, self.mock_correlated_different_dates) + self.assertTrue(coefficient < 1, "Coefficient should be None.") + + @patch('mt5_correlation.mt5.MetaTrader5') + def test_get_ticks(self, mock): + """ + Test that caching works. For the purpose of this test, we can use price data rather than tick data. + Mock 2 different sets of prices. Get three times. Base, One within cache threshold and one outside. Set 1 + should match set 2 but differ from set 3. + :param mock: + :return: + """ + + # Correlation class to test + cor = correlation.Correlation() + + # 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) + self.assertTrue(base_ticks.equals(cached_ticks), + "Both sets of tick data should match as set 2 came from cache.") + + # Wait 3 seconds + 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) + 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 + interval of 1 second. + :param mock: + :return: + """ + # Mock symbol return values + mock.symbols_get.return_value = self.mock_symbols + + # Correlation class + cor = correlation.Correlation() + + # 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. 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] + + 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) + + # 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'] + starttime = datetime.now() - timedelta(seconds=10) + tick_data_s1 = pd.DataFrame(columns=columns) + tick_data_s3 = pd.DataFrame(columns=columns) + tick_data_s4 = 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_s4 = 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] + + # Start the monitor. Run every second. Use ~10 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 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, from_mins=0.66, min_prices=0, 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) + + # Stop the monitor + cor.stop_monitor() + + # We should have 2 coefficients calculated for each symbol pair + self.assertEqual(len(cor.coefficient_history.index), 6) + + @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 + data.""" + + # Correlation class + cor = correlation.Correlation() + + # 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] + 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) + + # Patch the tick data + 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_s4 = 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_s4 = 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 + mock.copy_ticks_range.side_effect = [tick_data_s1, tick_data_s3, tick_data_s4] + + # 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, from_mins=0.66, min_prices=0, max_set_size_diff_pct=0, overlap_pct=0, + max_p_value=1, cache_time=100, autosave=False) + time.sleep(2) + cor.stop_monitor() + + # Get copies of data that will be saved. + cd_copy = cor.coefficient_data + pd_copy = cor.get_price_data('SYMBOL1') + mtd_copy = cor.get_ticks('SYMBOL1', cache_only=True) + ch_copy = cor.coefficient_history + + # Save, reset data, then reload + cor.save("unittest.cpd") + cor.load("unittest.cpd") + + # Test that the reloaded data matches the original + self.assertTrue(cd_copy.equals(cor.coefficient_data), + "Saved and reloaded coefficient data should match original.") + self.assertTrue(pd_copy.equals(cor.get_price_data('SYMBOL1')), + "Saved and reloaded price data should match original.") + self.assertTrue(mtd_copy.equals(cor.get_ticks('SYMBOL1', cache_only=True)), + "Saved and reloaded tick data should match original.") + self.assertTrue(ch_copy.equals(cor.coefficient_history), + "Saved and reloaded coefficient history should match original.") + + # Cleanup. delete the file + os.remove("unittest.cpd") + + +if __name__ == '__main__': + unittest.main()