gavindiaz/mt5-correlation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Simplified coefficient class. Monitoring stores params so that they dont need to be passed around. Added support for different params for each timeframe. Updated GUI to show three timeframes for each calculation.

Browse Source
This commit is contained in:
Jamie Cash
2021-03-10 15:22:17 +00:00
parent 0c17c7e5a9
commit 8c2cbf5df8
4 changed files with 144 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.yaml
+19 -9
View File
@@ -8,16 +8,26 @@ calculate:
overlap_pct: 90
max_p_value: 0.05
monitor:
calculate_from:
long:
minutes: 30
short:
minutes: 10
interval: 10
min_prices: 400
max_set_size_diff_pct: 90
overlap_pct: 80
max_p_value: 0.05
calculations:
long:
from: 30
min_prices: 400
max_set_size_diff_pct: 90
overlap_pct: 80
max_p_value: 0.05
medium:
from: 15
min_prices: 300
max_set_size_diff_pct: 90
overlap_pct: 80
max_p_value: 0.05
short:
from: 5
min_prices: 50
max_set_size_diff_pct: 90
overlap_pct: 80
max_p_value: 0.05
monitoring_threshold: 0.9
divergence_threshold: 0.8
tick_cache_time: 10
mt5_correlation/correlation.py
+78 -86
View File
@@ -19,7 +19,7 @@ class Correlation:
A class to maintain the state of the calculated correlation coefficients.
"""
# Connection to metatrader
# Connection to MetaTrader5
__mt5 = None
# Minimum base coefficient for monitoring. Symbol pairs with a lower correlation
@@ -28,7 +28,13 @@ class Correlation:
# Toggle on whether we are monitoring or not. Set through start_monitor and stop_monitor
__monitoring = False
__monitoring_params = {}
# Monitoring calculation params, interval, cache_time, autosave and filename. Passed to start_monitor
__monitoring_params = []
__interval = None
__cache_time = None
__autosave = None
__filename = None
# First run of scheduler
__first_run = True
@@ -53,7 +59,7 @@ class Correlation:
# Logger
self.__log = logging.getLogger(__name__)
# Connection to metatrader
# Connection to MetaTrader5
self.__mt5 = MT5()
# Create dataframe for coefficient data
@@ -113,7 +119,10 @@ class Correlation:
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
within this pct of each other
:param overlap_pct:
:param overlap_pct:The dates and times in the two sets of data must match. The coefficient will only be
calculated against the dates that overlap. Any non overlapping dates will be discarded. This setting
specifies the minimum size of the overlapping data when compared to the smallest set as a %. A coefficient
will not be calculated if this threshold is not met.
:param max_p_value: The maximum p value for the correlation to be meaningful
:return:
@@ -183,12 +192,8 @@ class Correlation:
# If we were monitoring, we stopped, so start again.
if was_monitoring:
self.start_monitor(interval=self.__monitoring_params['interval'],
calculate_from=self.__monitoring_params['calculate_from'],
min_prices=self.__monitoring_params['min_prices'],
max_set_size_diff_pct=self.__monitoring_params['max_set_size_diff_pct'],
overlap_pct=self.__monitoring_params['overlap_pct'],
max_p_value=self.__monitoring_params['max_p_value'])
self.start_monitor(interval=self.__interval, calculation_params=self.__monitoring_params,
cache_time=self.__cache_time, autosave=self.__autosave, filename=self.__filename)
def get_price_data(self, symbol):
"""
@@ -202,21 +207,26 @@ class Correlation:
return price_data
def start_monitor(self, interval, calculate_from, min_prices=100, max_set_size_diff_pct=90,
overlap_pct=90, max_p_value=0.05, cache_time=10, autosave=False, filename='autosave.cpd'):
def start_monitor(self, interval, calculation_params, cache_time=10, autosave=False, filename='autosave.cpd'):
"""
Starts monitor to continuously update the coefficient for all symbol pairs in that meet the min_coefficient
threshold.
:param interval: How often to check in seconds
:param calculate_from: The number of minutes of tick data to use for calculation. This can be a single value or
a list. If a list, then calculations will be performed for every from date in list.
: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
within this pct of each other
:param overlap_pct:
:param max_p_value: The maximum p value for the correlation to be meaningful
:param calculation_params: A single dict or list of dicts containing the parameters for the coefficient
calculations. On every iteration, a coefficient will be calculated for every set of params in list. Params
contain the following values:
from: The number of minutes of tick data to use for calculation. This can be a single value or
a list. If a list, then calculations will be performed for every from date in list.
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
max_set_size_diff_pct: Correlations will only be calculated if the sizes of the two price data sets are
within this pct of each other
overlap_pct: The dates and times in the two sets of data must match. The coefficient will only be
calculated against the dates that overlap. Any non overlapping dates will be discarded. This
setting specifies the minimum size of the overlapping data when compared to the smallest set as a %.
A coefficient will not be calculated if this threshold is not met.
max_p_value: The maximum p value for the correlation to be meaningful
: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
@@ -234,20 +244,30 @@ class Correlation:
self.__log.debug(f"Starting monitor.")
self.__monitoring = True
# Store the calculation params. If it isn't a list, convert to list of one to make code simpler later on.
self.__monitoring_params = calculation_params if isinstance(calculation_params, list) \
else [calculation_params, ]
# Store the other params. We will need these later if monitor is stopped and needs to be restarted. This
# happens in calculate.
self.__interval = interval
self.__cache_time = cache_time
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.
self.__shortest_timeframe = min(calculate_from) if isinstance(calculate_from, list) else calculate_from
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. Store the params. We will need to use these if we have
# to stop and restart the monitor. Note, this happens during calculate
self.__monitoring_params = {'interval': interval, 'calculate_from': calculate_from,
'min_prices': min_prices,
'max_set_size_diff_pct': max_set_size_diff_pct, 'overlap_pct': overlap_pct,
'max_p_value': max_p_value, 'cache_time': cache_time, 'autosave': autosave,
'filename': filename}
thread = threading.Thread(target=self.__monitor, kwargs=self.__monitoring_params)
# 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.start()
def stop_monitor(self):
@@ -386,21 +406,12 @@ class Correlation:
self.__log.debug(f"Ticks for {symbol} retrieved from source and cached.")
return ticks
def __monitor(self, interval, calculate_from, min_prices=100, max_set_size_diff_pct=90,
overlap_pct=90, max_p_value=0.05, cache_time=10, autosave=False, filename='autosave.cpd'):
def __monitor(self, interval, cache_time=10, autosave=False, filename='autosave.cpd'):
"""
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 calculate_from: The number of minutes of tick data to use for calculation. This can be a single value or
a list. If a list, then calculations will be performed for every from date in list.
: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
within this pct of each other
:param overlap_pct:
:param max_p_value: The maximum p value for the correlation to be meaningful
: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
@@ -414,19 +425,14 @@ class Correlation:
# Only run if monitor is not stopped
if self.__monitoring:
# Update all coefficients
self.__update_all_coefficients(calculate_from=calculate_from,
min_prices=min_prices, max_set_size_diff_pct=max_set_size_diff_pct,
overlap_pct=overlap_pct, max_p_value=max_p_value, cache_time=cache_time)
self.__update_all_coefficients(cache_time=cache_time)
# Autosave
if autosave:
self.save(filename=filename)
# Schedule the timer to run again
params = {'interval': interval, 'calculate_from': calculate_from,
'min_prices': min_prices, 'max_set_size_diff_pct': max_set_size_diff_pct,
'overlap_pct': overlap_pct, 'max_p_value': max_p_value, "cache_time": cache_time,
'autosave': autosave, 'filename': filename}
params = {'interval': interval, "cache_time": cache_time, 'autosave': autosave, 'filename': filename}
self.__scheduler.enter(delay=interval, priority=1, action=self.__monitor, kwargs=params)
# Log the stack. Debug stack overflow
@@ -437,43 +443,39 @@ class Correlation:
self.__first_run = False
self.__scheduler.run()
def __update_coefficients(self, symbol1, symbol2, calculate_from, min_prices=100,
max_set_size_diff_pct=90, overlap_pct=90, max_p_value=0.05, cache_time=10):
def __update_coefficients(self, symbol1, symbol2, cache_time=10):
"""
Updates the long and short 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 calculate_from: The number of minutes of tick data to use for calculation. This can be a single value or
a list. If a list, then calculations will be performed for every from date in list.
: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
within this pct of each other
:param overlap_pct:
:param max_p_value: The maximum p value for the correlation to be meaningful
: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.
"""
# Convert calculate from to list of one if only one value is provided
if not isinstance(calculate_from, list):
calculate_from = [calculate_from, ]
# Get the largest value of from in monitoring_params. This will be used to retrieve the data. We will only
# retrieve once and use for every set of params by getting subset of the data.
max_from = None
for params in self.__monitoring_params:
if max_from is None:
max_from = params['from']
else:
max_from = max(max_from, params['from'])
# Get dates
# From and to dates for calculations. From should be furthest away if list is provided in calculate_from
# Date range for data
timezone = pytz.timezone("Etc/UTC")
date_to = datetime.now(tz=timezone)
date_from = date_to - timedelta(minutes=max(calculate_from))
date_from = date_to - timedelta(minutes=max_from)
# Get the tick data for the longest timeframe calculation. We will extract shorter timeframes from it if list
# was provided in calculate_from to avoid retrieving multiple times.
# 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)
# 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
# resample then revert back to a 'time' column. We will then need to remove rows with nan in 'close' price
s1_prices = None
s2_prices = None
if symbol1ticks is not None and symbol2ticks is not None and len(symbol1ticks.index) > 0 and \
len(symbol2ticks.index) > 0:
@@ -491,10 +493,11 @@ class Correlation:
s1_prices = s1_prices[s1_prices['close'].notna()]
s2_prices = s2_prices[s2_prices['close'].notna()]
# Calculate for all timeframes
for from_mins in calculate_from:
# Calculate for all sets of monitoring_params
if s1_prices is not None and s2_prices is not None:
for params in self.__monitoring_params:
# Get the from date as a datetime64
date_from_subset = pd.Timestamp(date_to - timedelta(minutes=from_mins)).to_datetime64()
date_from_subset = pd.Timestamp(date_to - timedelta(minutes=params['from'])).to_datetime64()
# Get subset of the price data
s1_prices_subset = s1_prices[(s1_prices['time'] >= date_from_subset)]
@@ -503,43 +506,32 @@ class Correlation:
# Calculate the coefficient
coefficient = \
self.calculate_coefficient(symbol1_prices=s1_prices_subset, symbol2_prices=s2_prices_subset,
min_prices=min_prices, max_set_size_diff_pct=max_set_size_diff_pct,
overlap_pct=overlap_pct, max_p_value=max_p_value)
min_prices=params['min_prices'],
max_set_size_diff_pct=params['max_set_size_diff_pct'],
overlap_pct=params['overlap_pct'], max_p_value=params['max_p_value'])
self.__log.debug(f"Symbol pair {symbol1}:{symbol2} has a coefficient of {coefficient} for last "
f"{from_mins} minutes.")
f"{params['from']} minutes.")
# Update the coefficient data
if coefficient is not None:
self.__update_coefficient_data(symbol1=symbol1, symbol2=symbol2, coefficient=coefficient,
timeframe=from_mins, date_from=date_from_subset, date_to=date_to)
timeframe=params['from'], date_from=date_from_subset,
date_to=date_to)
def __update_all_coefficients(self, calculate_from, min_prices=100, max_set_size_diff_pct=90,
overlap_pct=90, max_p_value=0.05, cache_time=10):
def __update_all_coefficients(self, cache_time=10):
"""
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 calculate_from: The number of minutes of tick data to use for calculation. This can be a single value or
a list. If a list, then calculations will be performed for every from date in list.
: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
within this pct of each other
:param overlap_pct:
:param max_p_value: The maximum p value for the correlation to be meaningful
: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.
"""
# 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, calculate_from=calculate_from,
min_prices=min_prices, max_set_size_diff_pct=max_set_size_diff_pct,
overlap_pct=overlap_pct, max_p_value=max_p_value, cache_time=cache_time)
self.__update_coefficients(symbol1=symbol1, symbol2=symbol2, cache_time=cache_time)
def __reset_coefficient_data(self):
"""
mt5_correlation/gui.py
+38 -18
View File
@@ -272,13 +272,13 @@ class MonitorFrame(wx.Frame):
# Autosave filename
filename = self.__opened_filename if self.__opened_filename is not None else 'autosave.cpd'
# Build calculation params and start monitor
calculation_params = [self.__config.get('monitor.calculations.long'),
self.__config.get('monitor.calculations.medium'),
self.__config.get('monitor.calculations.short')]
self.__cor.start_monitor(interval=self.__config.get('monitor.interval'),
calculate_from=[self.__config.get('monitor.calculate_from.long.minutes'),
self.__config.get('monitor.calculate_from.short.minutes')],
min_prices=self.__config.get('monitor.min_prices'),
max_set_size_diff_pct=self.__config.get('monitor.max_set_size_diff_pct'),
overlap_pct=self.__config.get('monitor.overlap_pct'),
max_p_value=self.__config.get('monitor.max_p_value'),
calculation_params=calculation_params,
cache_time=self.__config.get('monitor.tick_cache_time'),
autosave=self.__config.get('monitor.autosave'),
filename=filename)
@@ -326,13 +326,18 @@ class MonitorFrame(wx.Frame):
if restart_monitor_timer:
self.__log.info("Settings updated. Reloading monitoring timer.")
self.__cor.stop_monitor()
# Build calculation params and start monitor
calculation_params = [self.__config.get('monitor.calculations.long'),
self.__config.get('monitor.calculations.medium'),
self.__config.get('monitor.calculations.short')]
self.__cor.start_monitor(interval=self.__config.get('monitor.interval'),
calculate_from=[self.__config.get('monitor.calculate_from.long.minutes'),
self.__config.get('monitor.calculate_from.short.minutes')],
min_prices=self.__config.get('monitor.min_prices'),
max_set_size_diff_pct=self.__config.get('monitor.max_set_size_diff_pct'),
overlap_pct=self.__config.get('monitor.overlap_pct'),
max_p_value=self.__config.get('monitor.max_p_value'))
calculation_params=calculation_params,
cache_time=self.__config.get('monitor.tick_cache_time'),
autosave=self.__config.get('monitor.autosave'),
filename=self.__opened_filename)
if restart_gui_timer:
self.__log.info("Settings updated. Restarting gui timer.")
self.timer.Stop()
@@ -411,14 +416,17 @@ class MonitorFrame(wx.Frame):
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.calculate_from.short.minutes'))
self.__config.get('monitor.calculations.short.from'))
history_data_med = \
self.__cor.get_coefficient_history(symbol1, symbol2,
self.__config.get('monitor.calculations.medium.from'))
history_data_long = \
self.__cor.get_coefficient_history(symbol1, symbol2,
self.__config.get('monitor.calculate_from.long.minutes'))
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],
history=[history_data_short, history_data_long], symbols=[symbol1, symbol2])
history=[history_data_short, history_data_med, history_data_long], symbols=[symbol1, symbol2])
# Un-hide and layout if hidden
if not self.__graph.IsShown():
@@ -494,6 +502,7 @@ 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
value = self.GetValue(row, col)
if value != "":
value = float(value)
@@ -501,6 +510,16 @@ class DataTable(wx.grid.GridTableBase):
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')):
attr.SetBackgroundColour(wx.YELLOW)
else:
attr.SetBackgroundColour(wx.WHITE)
return attr
@@ -550,7 +569,7 @@ class GraphPanel(wx.Panel):
# Check what data we have available
price_data_available = prices is not None and len(prices) == 2 and \
prices[0] is not None and prices[1] is not None
prices[0] is not None and prices[1] is not None
tick_data_available = ticks is not None and len(ticks) == 2 and ticks[0] is not None and ticks[1] is not None
history_data_available = history is not None and len(history) > 0
symbols_selected = symbols is not None and len(symbols) == 2
@@ -619,8 +638,9 @@ class GraphPanel(wx.Panel):
types = ['plot', 'plot', 'plot', 'plot', 'scatter']
# Legends
legends = [None, None, None, None, [f"{Config().get('monitor.calculate_from.short.minutes')} Minutes",
f"{Config().get('monitor.calculate_from.long.minutes')} Minutes"]]
legends = [None, None, None, None, [f"{Config().get('monitor.calculations.long.from')} Minutes",
f"{Config().get('monitor.calculations.medium.from')} Minutes",
f"{Config().get('monitor.calculations.short.from')} Minutes"]]
# Draw 5 charts
for index in range(0, len(self.__axes)):
test/test_correlation.py
+9 -4
View File
@@ -201,8 +201,12 @@ class TestCorrelation(unittest.TestCase):
# 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
# 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, calculate_from=[0.66, 0.33], min_prices=0, max_set_size_diff_pct=0,
overlap_pct=0, max_p_value=1, cache_time=100, autosave=False)
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,},
{'from': 0.33, '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)
@@ -250,8 +254,9 @@ 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, calculate_from=0.66, min_prices=0, max_set_size_diff_pct=0,
overlap_pct=0, max_p_value=1, cache_time=100, autosave=False)
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},
cache_time=100, autosave=False)
time.sleep(2)
cor.stop_monitor()