20 Commits

Author SHA1 Message Date
Jamie Cash 7d97b78f8c Improved divergence graph. 2021-06-04 17:40:44 +01:00
Jamie Cash 06c980a45a Graphs use colormap defined in settings 2021-06-02 17:49:54 +01:00
Jamie Cash 1656c6a4d2 Visual improvements to diverged symbols chart and added log file and help file windows. 2021-06-02 16:03:30 +01:00
Jamie Cash 66cdf47406 Added diverged symbols chart 2021-06-01 17:45:11 +01:00
Jamie Cash b947b928e7 Added diverged symbols view 2021-06-01 14:20:57 +01:00
Jamie Cash f6d05873d2 Made graphs less squashed 2021-05-28 17:41:53 +01:00
Jamie Cash 029e86d358 Converted to MDI 2021-05-28 17:34:28 +01:00
Jamie Cash c977e3ac1c Split out config to seperate project (wxconfig). wxconfig now a dependency. Previous config code removed. 2021-05-26 16:37:15 +01:00
Jamie Cash 225f869276 Fixed unit tests for new status's 2021-05-25 11:53:32 +01:00
Jamie Cash 1c59945745 Included diverging and converging status calculation 2021-05-25 11:16:41 +01:00
Jamie Cash 443ad822ad Merged charts to create more space. 2021-03-16 17:56:54 +00:00
Jamie Cash d0049546fe Added support for inverse correlations 2021-03-16 15:02:19 +00:00
Jamie Cash 4fdf4bedc7 Used monitoring_threshold to calculate status. 2021-03-15 11:30:08 +00:00
Jamie Cash 21ea7eb809 Added status for monitoring 2021-03-12 15:10:07 +00:00
Jamie Cash ab6c2ae338 Clear last check and last coefficient column on clear history. 2021-03-11 09:50:52 +00:00
Jamie Cash 8c2cbf5df8 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. 2021-03-10 15:22:17 +00:00
Jamie Cash 0c17c7e5a9 UI improvements 2021-03-09 16:16:30 +00:00
Jamie Cash d08fef0cbb Added support for multiple timeframes for coefficient calculation. 2021-03-08 17:50:24 +00:00
Jamie Cash 15f1ca2288 Added unit tests for correlation 2021-03-05 17:39:48 +00:00
Jamie Cash a77559cc4e Added unit tests for MT5 2021-03-04 18:18:47 +00:00
21 changed files with 2198 additions and 1375 deletions
+10 -51
View File
@@ -6,14 +6,14 @@ Calculates correlation coefficient between all symbols in MetaTrader5 Market Wat
2) Set up your python environment; and
3) Install the required libraries.
```
```shell
pip install -r mt5-correlation/requirements.txt
```
# Usage
If you set up a virtual environment in the Setup step, ensure this is activated. Then run the script.
```
```shell
python -m mt5_correlations/mt5_correlations.py
```
@@ -21,56 +21,15 @@ This will open a GUI.
## Calculating Baseline Coefficients
On the first time that you run, you will want to calculate the initial set of correlations.
1) Open settings and review the settings under the 'Calculate' tab. These settings are:
- from.days: The number of days of data to be used to calculate the coefficient.
- timeframe: The timeframe for price candles to use for the calculation. Possible values are:
* 1: 1 Minute candles
* 2: 2 Minute candles
* 3: 3 Minute candles
* 4: 4 Minute candles
* 5: 5 Minute candles
* 6: 6 Minute candles
* 10: 10 Minute candles
* 15: 15 Minute candles
* 20: 20 Minute candles
* 30: 30 Minute candles
* 16385: 1 Hour candles
* 16386: 2 Hour candles
* 16387: 3 Hour candles
* 16388: 4 Hour candles
* 16390: 6 Hour candles
* 16392: 8 Hour candles
* 16396: 12 Hour candles
* 16408: 1 Day candles
* 32769: 1 Week candles
* 49153: 1 Month candles
- min_prices: The minimum number of candles required to calculate a coefficient from. If any of the symbols do not have at least this number of candles then the coefficient won't be calculated.
- max_set_size_diff_pct: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of candle data must be at least this values % of the largest set.
- 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 coefficient to be considered valid. A full explanation on the correlation coefficient P value is available here: https://docs.scipy.org/doc/scipy-0.14.0/reference/generated/scipy.stats.pearsonr.html.
2) Set the threshold for the correlations to monitor. This can be set under the Settings 'Monitoring' tab and is named monitoring.threshold. Only settings with a coefficient over this threshold will be displayed and monitored.
3) Calculate the coefficients by selecting File/Calculate. All symbol pairs that have a correlation coefficient greater than the monitoring threshold will be displayed. A graph showing the price candle data used to calculate the coefficient for the pair will be displayed when you select the row.
1) Open settings and review the settings under the 'Calculate' tab. Hover over the individual settings for help.
2) Set the threshold for the correlations to monitor. This can be set under the Settings 'Monitoring' tab and is named 'Monitoring Threshold'. Only settings with a coefficient over this threshold will be displayed and monitored.
3) Calculate the coefficients by selecting File/Calculate. All symbol pairs that have a correlation coefficient greater than the monitoring threshold will be displayed. A graph showing the price candle data used to calculate the coefficient for the pair will be displayed when you select the row.
4) You may want to save. Select File/Save and choose a file name. This file will contain all the calculated coefficients, and the price data used to calculate them. This file can be loaded to avoid having to recalculate the baseline coefficients every time you use the application.
## Monitoring for Divergence
Once you have calculated or loaded the baseline coefficients, they can be monitored for divergence.
1) Open settings and review the settings under the 'Monitor' tab. These settings are:
- from.minutes: The number of minutes of data to be used to calculate the coefficient.
- interval: The number of seconds between monitoring events. The application will monitor for divergence every {interval} seconds.
- min_prices: Tick data will be converted to 1 second candles prior to calculation. This will enable data to be matched between symbols. This setting specifies the minimum number of price candles required to calculate a coefficient from. If any of the symbols do not have at least number of candles then the coefficient won't be calculated.
- max_set_size_diff_pct: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of price candles must be at least this values % of the largest set.
- overlap_pct: The dates and times in the two sets of data must match. The ticks will be converted to 1 second price candles before calculation. The coefficient will only be calculated against the times from the candles that overlap. Any non overlapping times 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 coefficient to be considered valid. A full explanation on the correlation coefficient P value is available here: https://docs.scipy.org/doc/scipy-0.14.0/reference/generated/scipy.stats.pearsonr.html.
- monitoring_threshold: The application will only display correlated pairs that have a correlation coefficient greater than or equal to this value.
- divergence_threshold: The application will consider a pair to have diverged if the correlation coefficient falls below this threshold. These will be highlighted in yellow.
- tick_cache_time: Every calculation requires tick data for both symbols. Tick data will be cached for this number of seconds before being retrieved from MetaTrader. Some caching is recommended as a single monitoring run will request the same data for symbols that form multiple correlated pairs.
- autosave: Whether to auto save after every monitoring event. If a file was opened or has been saved, then the data will be saved to this file, otherwise the data will be saved to a file named autosave.cpd.
2) Switch the monitoring toggle to on. The application will continuously monitor for divergence. The data frame will be updated with the last time that the correlation was checked, and the last coefficient. The chart frame will contain 5 charts which will be updated after every monitoring event:
- 2 charts, one for each symbol in the correlated pair, showing the price data used to calculate the baseline coefficient.
- 2 charts, one for each symbol in the correlated pair, showing the tick data used to calculate the latest coefficient.
- 1 chart showing every correlation coefficient calculated for the symbol pair.
1) Open settings and review the settings under the 'Monitor' tab. Hover over the individual settings for help.
2) Switch the monitoring toggle to on. The application will continuously monitor for divergence. The data frame will be updated with the last time that the correlation was checked, and the last coefficient. The chart frame will contain 3 charts which will be updated after every monitoring event:
- One showing the price history data used to calculate the baseline coefficient for both symbols in the correlated pair;
- One showing the tick data used to calculate the latest coefficient for both symbols in the correlated pair.
- One showing every correlation coefficient calculated for the symbol pair.
+29 -13
View File
@@ -8,15 +8,29 @@ calculate:
overlap_pct: 90
max_p_value: 0.05
monitor:
from:
minutes: 30
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: 300
max_set_size_diff_pct: 50
overlap_pct: 50
max_p_value: 0.05
medium:
from: 10
min_prices: 100
max_set_size_diff_pct: 50
overlap_pct: 50
max_p_value: 0.05
short:
from: 2
min_prices: 30
max_set_size_diff_pct: 50
overlap_pct: 50
max_p_value: 0.05
monitoring_threshold: 0.9
divergence_threshold: 0.8
monitor_inverse: true
tick_cache_time: 10
autosave: true
logging:
@@ -55,18 +69,20 @@ logging:
- console
- file
propagate: 0
charts:
colormap: Dark2
developer:
inspection: false
inspection: true
window:
x: 42
y: 51
width: 1512
height: 975
x: -4
y: 1
width: 1626
height: 1043
style: 541072960
settings_window:
x: 354
y: 299
width: 664
height: 317
width: 624
height: 328
style: 524352
...
+104
View File
@@ -0,0 +1,104 @@
---
calculate:
from:
days:
__label: Calculate from (days)
__helptext: The number of days of data to be used to calculate the coefficient.
timeframe:
__label: Timeframe
__helptext: The timeframe for price candles to use for the calculation. Possible values are 1=1 Minute candles; 2=2 Minute candles; 3=3 Minute candles; 4=4 Minute candles; 5=5 Minute candles; 6=6 Minute candles; 10=10 Minute candles; 15=15 Minute candles; 20=20 Minute candles; 30=30 Minute candles; 16385=1 Hour candles; 16386=2 Hour candles; 16387=3 Hour candles; 16388=4 Hour candles; 16390=6 Hour candles; 16392=8 Hour candles; 16396=12 Hour candles; 16408=1 Day candles; 32769=1 Week candles; or 49153=1 Month candles.
min_prices:
__label: Min Prices
__helptext: The minimum number of candles required to calculate a coefficient from. If any of the symbols do not have at least this number of candles then the coefficient won't be calculated.
max_set_size_diff_pct:
__label: Min Set Size Difference %
__helptext: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of candle data must be at least this values % of the largest set.
overlap_pct:
__label: Overlap %
__helptext: 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:
__label: Max Pearsonr P Value
__helptext: The maximum P value for the coefficient to be considered valid. A full explanation on the correlation coefficient P value is available in the scipy pearsonr documentation.
monitor:
interval:
__label: Monitoring Interval
__helptext: The number of seconds between monitoring events.
calculations:
long:
__label: Long
__helptext: The settings for the correlation calculation using the longest timeframe.
from:
__label: From (Minutes)
__helptext: The number of minutes of data to be used to calculate the long coefficient.
min_prices:
__label: Min Prices
__helptext: Tick data will be converted to 1 second candles prior to calculation. This will enable data to be matched between symbols. This setting specifies the minimum number of price candles required to calculate a coefficient from. If any of the symbols do not have at least number of candles then the coefficient won't be calculated.
max_set_size_diff_pct:
__label: Max Set Size Difference %
__helptext: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of price candles must be at least this values % of the largest set.
overlap_pct:
__label: Overlap %
__helptext: The dates and times in the two sets of data must match. The ticks will be converted to 1 second price candles before calculation. The coefficient will only be calculated against the times from the candles that overlap. Any non overlapping times 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:
__label: Max Pearsonr P Value
__helptext: The maximum P value for the coefficient to be considered valid. A full explanation on the correlation coefficient P value is available in the scipy pearsonr documentation.
medium:
__label: Medium
__helptext: The settings for the correlation calculation using a medium timeframe.
from:
__label: From (Minutes)
__helptext: The number of minutes of data to be used to calculate the medium coefficient.
min_prices:
__label: Min Prices
__helptext: Tick data will be converted to 1 second candles prior to calculation. This will enable data to be matched between symbols. This setting specifies the minimum number of price candles required to calculate a coefficient from. If any of the symbols do not have at least number of candles then the coefficient won't be calculated.
max_set_size_diff_pct:
__label: Max Set Size Difference %
__helptext: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of price candles must be at least this values % of the largest set.
overlap_pct:
__label: Overlap %
__helptext: The dates and times in the two sets of data must match. The ticks will be converted to 1 second price candles before calculation. The coefficient will only be calculated against the times from the candles that overlap. Any non overlapping times 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:
__label: Max Pearsonr P Value
__helptext: The maximum P value for the coefficient to be considered valid. A full explanation on the correlation coefficient P value is available in the scipy pearsonr documentation.
short:
__label: Short
__helptext: The settings for the correlation calculation using the shortest timeframe.
from:
__label: From (Minutes)
__helptext: The number of minutes of data to be used to calculate the short coefficient.
min_prices:
__label: Min Prices
__helptext: Tick data will be converted to 1 second candles prior to calculation. This will enable data to be matched between symbols. This setting specifies the minimum number of price candles required to calculate a coefficient from. If any of the symbols do not have at least number of candles then the coefficient won't be calculated.
max_set_size_diff_pct:
__label: Max Set Size Difference %
__helptext: For a meaningful coefficient calculation, the two sets of data should be of a similar size. This setting specifies the % difference allowed for a correlation to be calculated. The smallest set of price candles must be at least this values % of the largest set.
overlap_pct:
__label: Overlap %
__helptext: The dates and times in the two sets of data must match. The ticks will be converted to 1 second price candles before calculation. The coefficient will only be calculated against the times from the candles that overlap. Any non overlapping times 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:
__label: Max Pearsonr P Value
__helptext: The maximum P value for the coefficient to be considered valid. A full explanation on the correlation coefficient P value is available in the scipy pearsonr documentation.
monitoring_threshold:
__label: Monitoring Threshold
__helptext: Only pairs with a coefficient over this threshold will be displayed and monitored.
divergence_threshold:
__label: Divergence Threshold
__helptext: The application will consider a pair to have diverged if the correlation coefficient for all timeframes (long, medium and short) falls below this threshold.
monitor_inverse:
__label: Monitor Inverse
__helptext: Monitor Inverse Correlations (uses negative scale with -1 being fully inversly correlated)
tick_cache_time:
__label: Tick Cache Time
__helptext: Every calculation requires tick data for both symbols. Tick data will be cached for this number of seconds before being retrieved from MetaTrader. Some caching is recommended as a single monitoring run will request the same data for symbols that form multiple correlated pairs.
autosave:
__label: Auto Save
__helptext: Whether to auto save after every monitoring event. If a file was opened or has been saved, then the data will be saved to this file, otherwise the data will be saved to a file named autosave.cpd.
charts:
colormap:
__label: Color Map
__helptext: The matplotlib color pallet to use for plotting graphs. A list of pallets is available at https://matplotlib.org/stable/tutorials/colors/colormaps.html
developer:
inspection:
__label: Inspection
__helptext: Provide GUI Inspection guidelines for developers modifying the GUI.
...
+3 -1
View File
@@ -1,3 +1,5 @@
import os
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
HELP_FILE = fr"{ROOT_DIR}\README.md"
LOG_FILE = fr"{ROOT_DIR}\debug.log"
+6 -6
View File
@@ -3,8 +3,8 @@ Application to monitor previously correlated symbol pairs for correlation diverg
"""
import definitions
import logging.config
from mt5_correlation.gui import MonitorFrame
from mt5_correlation.config import Config
from mt5_correlation.gui import CorrelationMDIFrame
import wxconfig as cfg
import wx
import wx.lib.mixins.inspection as wit
@@ -18,20 +18,20 @@ class InspectionApp(wx.App, wit.InspectionMixin):
if __name__ == "__main__":
# Load the config
Config().load(fr"{definitions.ROOT_DIR}\config.yaml")
cfg.Config().load(fr"{definitions.ROOT_DIR}\config.yaml", meta=fr"{definitions.ROOT_DIR}\configmeta.yaml")
# Get logging config and configure the logger
log_config = Config().get('logging')
log_config = cfg.Config().get('logging')
logging.config.dictConfig(log_config)
# Do we have inspection turned on. Create correct version of app
inspection = Config().get('developer.inspection')
inspection = cfg.Config().get('developer.inspection')
if inspection:
app = InspectionApp()
else:
app = wx.App(False)
# Start the app
frame = MonitorFrame()
frame = CorrelationMDIFrame()
frame.Show()
app.MainLoop()
-474
View File
@@ -1,474 +0,0 @@
import yaml
import wx
import logging
class Config(object):
"""
Provides access to application configuration parameters stored in config.yaml.
"""
config_filepath = None
__config = None
__instance = None
def __new__(cls):
"""
Singleton. Get instance of this class. Create if not already created.
:return:
"""
if cls.__instance is None:
cls.__instance = super(Config, cls).__new__(cls)
return cls.__instance
def load(self, path):
"""
Loads the applications config file
:param path: Path to config file
:return:
"""
with open(path, 'r') as yamlfile:
self.__config = yaml.safe_load(yamlfile)
# Store path so that we can save later
self.config_filepath = path
def save(self):
"""
Saves config file
:return:
"""
with open(self.config_filepath, 'w') as file:
file.write("---\n")
yaml.dump(self.__config, file, sort_keys=False)
file.write("...")
def get(self, path):
"""
Gets a config property value.
:param path: path to property. Path separated by .
:return: property value
"""
elements = path.split('.')
last = None
for element in elements:
if last is None:
last = self.__config[element]
else:
last = last[element]
return last
def get_root_nodes(self):
"""
Returns all root notes as a list
:return: dict of root notes of YAML config file
"""
nodes = []
for key in self.__config:
nodes.append(key)
return nodes
def set(self, path, value):
"""
Sets a config property value
:param path: path to property. Path separated by .
:param value: Value to set property to
:return:
"""
obj = self.__config
key_list = path.split(".")
for k in key_list[:-1]:
obj = obj[k]
obj[key_list[-1]] = value
class SettingsDialog(wx.Dialog):
"""
A dialog box for changing settings. A tab for each root node, with a tree view on left for every branch and a text
box for every value.
"""
# Settings
__settings = None # Will set in init.
# Store any settings that have changed
changed_settings = {}
def __init__(self, parent, exclude=None):
"""
Open the settings dialog
:param parent: The parent frame for this dialog
:param exclude: List of settings root nodes to exclude from this dialog
"""
# Super Constructor
wx.Dialog.__init__(self, parent=parent, id=wx.ID_ANY, title="Settings",
pos=wx.Point(x=Config().get('settings_window.x'),
y=Config().get('settings_window.y')),
size=wx.Size(width=Config().get('settings_window.width'),
height=Config().get('settings_window.height')),
style=Config().get('settings_window.style'))
self.SetTitle("Settings")
# Create logger and get config
self.__log = logging.getLogger(__name__)
self.__settings = Config()
# Dict of changes. Will commit only on ok
self.__changes = {}
# Dialog should be resizable
self.SetWindowStyle(wx.RESIZE_BORDER)
# Settings to exclude. Just settings_window if None. Add settings_window if not specified.
exclude = ['settings_window'] if exclude is None else exclude
if 'settings_window' not in exclude:
exclude.append('settings_window')
# We want 2 vertical sections, the tabbed notebook and the buttons. The buttons sizer will have 2 horizontal
# sections, one for each button.
main_sizer = wx.BoxSizer(wx.VERTICAL) # Notebook panel
button_sizer = wx.BoxSizer(wx.HORIZONTAL) # Button sizer
# Notebook
self.__notebook = wx.Notebook(self, wx.ID_ANY) # The notebook
# A tab for each root node in config. We will store the tabs components in lists which can be accessed by the
# index returned from notebook.GetSelectedItem()
root_nodes = self.__settings.get_root_nodes()
self.__tabs = []
for node in root_nodes:
# Exclude?
if node not in exclude:
# Create new tab
self.__tabs.append(SettingsTab(self, self.__notebook, node))
# Add tab to notebook
self.__notebook.AddPage(self.__tabs[-1], node)
# Buttons
button_ok = wx.Button(self, label="Update")
button_cancel = wx.Button(self, label="Cancel")
button_sizer.Add(button_ok, 0, wx.ALL, 1)
button_sizer.Add(button_cancel, 0, wx.ALL, 1)
# Add notebook and button sizer to main sizer and set main sizer for window
main_sizer.Add(self.__notebook, 1, wx.ALL | wx.EXPAND, 5)
main_sizer.Add(button_sizer)
self.SetSizer(main_sizer)
# Bind buttons & notebook page select.
button_ok.Bind(wx.EVT_BUTTON, self.__on_ok)
button_cancel.Bind(wx.EVT_BUTTON, self.__on_cancel)
self.Bind(wx.EVT_NOTEBOOK_PAGE_CHANGED, self.__on_page_select)
# Bind window close event
self.Bind(wx.EVT_CLOSE, self.__on_close, self)
# Call on_page_select to select the first page
self.__on_page_select(event=None)
def __on_page_select(self, event):
# Call the tabs select method to populate
index = self.__notebook.GetSelection()
self.__tabs[index].select()
def __on_cancel(self, event):
# Clear changed settings and close
self.changed_settings = {}
self.EndModal(wx.ID_CANCEL)
self.Close()
def __on_ok(self, event):
# Update settings and save
delkeys = []
for setting in self.changed_settings:
# Get the current and new setting
orig_value = self.__settings.get(setting)
new_value = self.changed_settings[setting]
# We need to retain data type. New values will all be string as they were retrieved from textctl.
# Get the data type of the original and cast new to it. Note boolean needs to be handled differently as it
# doesn't cast directly.
if isinstance(orig_value, bool):
new_value = new_value.lower() in ['true', '1', 'yes', 't']
else:
new_value = type(orig_value)(new_value)
# If they are the same, discard from changes. We will use a list of items to delete (delkeys) as we cant
# delete whilst iterating. If they are different, update settings.
if orig_value == new_value:
delkeys.append(setting)
else:
self.__settings.set(setting, new_value)
# Now delete the items that were the same from changed_settings. changed_settings may be used by settings
# dialog caller.
for key in delkeys:
del(self.changed_settings[key])
# Save the settings and close dialog
self.__settings.save()
self.EndModal(wx.ID_OK)
self.Destroy()
def __on_close(self, event):
# Save pos and size
x, y = self.GetPosition()
width, height = self.GetSize()
self.__settings.set('settings_window.x', x)
self.__settings.set('settings_window.y', y)
self.__settings.set('settings_window.width', width)
self.__settings.set('settings_window.height', height)
# Style
style = self.GetWindowStyle()
self.__settings.set('settings_window.style', style)
class SettingsTab(wx.Panel):
"""
A notebook tab containing the settings tree and values for a settings root node.
"""
# Root node for this tab
__root_node_name = None
# Parent frame. Set during constructor
__parent_frame = None
# Each tab has: a tree view; a values panel; and a list of value text boxes bound to a change
# event.
__tree = None
__tab_sizer = None
# Currently displayed value panel. Will be switched when tree menu items are selected.
__current_value_panel = None
def __init__(self, parent_frame, notebook, root_node):
"""
Creates a tab for the settings notebook.
:param parent_frame: The frame containing the notebook.
:param notebook. The notebook that this tab should be part of.
:param root_node. The root node name for the settings
"""
# Super Constructor
wx.Panel.__init__(self, parent=notebook)
# Store the parent frame and get the settings for this tab.
self.__parent_frame = parent_frame
# Store the root node for this tab
self.__root_node_name = root_node
# Create logger
self.__log = logging.getLogger(__name__)
# Build the tab and set it's sizer.
self.__tab_sizer = wx.BoxSizer(wx.HORIZONTAL)
self.SetSizer(self.__tab_sizer)
# Create tree control and add it to sizer
self.__tree = SettingsTree(self, self.__root_node_name)
self.__tab_sizer.Add(self.__tree, 1, wx.ALL | wx.EXPAND, 1)
# Bind tree selection changed
self.__tree.Bind(wx.EVT_TREE_SEL_CHANGED, self.__on_tree_select)
def select(self):
"""
To be called when this tab is selected. Populate value sizer for the selected item, If no item is selected,
populate for root.
:return:
"""
selected_item = self.__tree.GetSelection()
if selected_item.ID is None:
root_node = self.__tree.GetRootItem()
setting_path = self.__tree.GetItemData(root_node)
else:
setting_path = self.__tree.GetItemData(selected_item)
# Set the panel
self.__switch_value_panel(setting_path)
def __on_tree_select(self, event):
"""
Called when an item in the tree is selected. Displays the correct settings panel
:param event:
:return:
"""
# Get Selected item and check that it is a tree item
tree_item = event.GetItem()
if not tree_item.IsOk():
return
# Get the setting path from item data
setting_path = self.__tree.GetItemData(tree_item)
# Switch the panel
self.__switch_value_panel(setting_path)
def __switch_value_panel(self, setting_path):
"""
Switched the value panel to the correct one for the settings path
:param setting_path:
:return:
"""
# Get current panel and delete.
if self.__current_value_panel is not None:
self.__current_value_panel.Destroy()
# Create the new value panel and add to sizer.
self.__current_value_panel = SettingsValuePanel(self.__parent_frame, self, setting_path)
self.__tab_sizer.Add(self.__current_value_panel, 1, wx.ALL | wx.EXPAND, 1)
# Redraw
self.__tab_sizer.Layout()
class SettingsTree(wx.TreeCtrl):
"""
A Tree control containing the settings nodes settings node
"""
__root_node_name = None
def __init__(self, settings_tab, settings_node):
"""
Creates a tree control for specified settings node.
:param settings_tab. The settings_tab on which this tree control should be displayed.
:param settings_node. The node name for the settings who's values will be presented
"""
# Super Constructor
wx.TreeCtrl.__init__(self, parent=settings_tab)
# Set root node
self.__root_node_name = settings_node
# Build the tree
self.__build_tree(None, self.__root_node_name)
# Expand it
# self.ExpandAll()
# Set max size. Width should be best size width, height should be auto (-1)
best_width = self.GetBestSize()[0] * 2 # Hack, best size not working
self.SetMaxSize((best_width, -1))
def __build_tree(self, node, node_name):
"""
Recursive function to build the tree and value panels from the node using the settings
:param node: The tree view node to build from. If none, builds from root.
:param node_name. The name of the node to build from.
"""
# Build root node if node is None
if node is None:
node = self.AddRoot(self.__root_node_name)
self.SetItemData(node, self.__root_node_name)
# Get settings
node_path = self.GetItemData(node)
settings = Config().get(node_path)
# Iterate settings, adding branches. Recurse to add sub branches.
for setting in settings:
# Get settings path
settings_path = f"{self.GetItemData(node)}.{setting}"
# Get value. If dict, add the node and recursively call this function again.
value = settings[setting]
if type(value) is dict:
# Add the node and set its settings path
node_id = self.AppendItem(node, setting)
self.SetItemData(node_id, settings_path)
# Recurse
self.__build_tree(node_id, value)
class SettingsValuePanel(wx.ScrolledWindow):
"""
A panel containing text boxes for editing values for a settings node
"""
__value_sizer = None
__value_boxes = []
def __init__(self, parent_frame, settings_tab, node):
"""
Creates a panel containing values for a settings node.
:param parent_frame: The frame containing the notebook.
:param settings_tab. The settings_tab on which this panel should be displayed.
:param node. The node name for the settings who's values will be presented
"""
# Super Constructor
wx.ScrolledWindow.__init__(self, parent=settings_tab)
# Create logger
self.__log = logging.getLogger(__name__)
# Store the parent frame and get the settings for this node.
self.__parent_frame = parent_frame
self.__settings = Config().get(node)
leaf_settings = {}
for setting in self.__settings:
if type(self.__settings[setting]) is not dict:
leaf_settings[setting] = self.__settings[setting]
# Add the value sizer for settings values.
self.__value_sizer = wx.FlexGridSizer(rows=len(leaf_settings), cols=2, vgap=2, hgap=2)
# Add the sizer to the panel
self.SetSizer(self.__value_sizer)
# Display every value
for setting in leaf_settings:
# Setting path
setting_path = f"{node}.{setting}"
# Value. Make sure that we display changed value if already changed
if setting_path in self.__parent_frame.changed_settings:
value = self.__parent_frame.changed_settings[setting_path]
else:
value = leaf_settings[setting]
# Add a label and value text box
label = wx.StaticText(self, wx.ID_ANY, setting, style=wx.ALIGN_LEFT)
self.__value_boxes.append(wx.TextCtrl(self, wx.ID_ANY, f"{value}", style=wx.ALIGN_LEFT))
self.__value_sizer.AddMany([(label, 0, wx.EXPAND), (self.__value_boxes[-1], 0, wx.EXPAND)])
# Bind to text change. We need to generate a handler as this will have a parameter.
self.__value_boxes[-1].Bind(wx.EVT_TEXT, self.__get_on_change_evt_handler(setting_path=setting_path))
# Layout the value sizer
self.__value_sizer.Layout()
# Setup scrollbars
self.SetScrollbars(1, 1, 1000, 1000)
def __get_on_change_evt_handler(self, setting_path):
"""
Returns a new event handler with a parameter of the settings path
:param setting_path:
:return:
"""
def on_value_changed(event):
old_val = self.__settings.get(setting_path)
self.__parent_frame.changed_settings[setting_path] = event.String
self.__log.debug(f"Value changed from {old_val} to {self.__parent_frame.changed_settings[setting_path]} "
f"for {setting_path}.")
return on_value_changed
+403 -154
View File
@@ -14,21 +14,94 @@ import sys
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_CORRELATED = CorrelationStatus(1, 'CORRELATED', 'Coefficients for all timeframes are equal to or above the '
'divergence threshold')
STATUS_DIVERGED = CorrelationStatus(2, 'DIVERGED', 'Coefficients for all timeframes are below the divergence threshold')
STATUS_INCONSISTENT = CorrelationStatus(3, 'INCONSISTENT', 'Coefficients not consistently above or below divergence '
'threshold and are neither trending towards divergence or '
'convergence')
STATUS_DIVERGING = CorrelationStatus(4, 'DIVERGING', 'Coefficients, when ordered by timeframe, are trending '
'towards convergence. The shortest timeframe is below the '
'divergence threshold and the longest timeframe is above the '
'divergence threshold')
STATUS_CONVERGING = CorrelationStatus(5, 'CONVERGING', 'Coefficients, when ordered by timeframe, are trending '
'towards divergence. The shortest timeframe is above the '
'divergence threshold and the longest timeframe is below the '
'divergence threshold')
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
# coefficient than ths won't be monitored.
monitoring_threshold = 0.9
# Threshold for divergence. Correlation coefficients that were previously above the monitoring_threshold and fall
# 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
__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
@@ -36,7 +109,7 @@ class Correlation:
# The price data used to calculate the correlations
__price_data = None
# Coefficient data and history. Will be created as dataframes in Init
# Coefficient data and history. Will be created in init call to __reset_coefficient_data
coefficient_data = None
coefficient_history = None
@@ -44,11 +117,19 @@ class Correlation:
# Dict: {Symbol: [retrieved datetime, ticks dataframe]}
__monitor_tick_data = {}
def __init__(self):
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__)
# Connection to metatrader
# Connection to MetaTrader5
self.__mt5 = MT5()
# Create dataframe for coefficient data
@@ -57,15 +138,56 @@ class Correlation:
# Create timer for continuous monitoring
self.__scheduler = sched.scheduler(time.time, time.sleep)
# 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
@property
def diverged_symbols(self):
"""
:return: dataframe containing all diverged, diverging or converging symbols and count of number of
divergences for those symbols.
"""
filtered_data = None
if self.coefficient_data is not None:
# Only rows where we have a divergence
filtered_data = self.coefficient_data \
.loc[(self.coefficient_data['Status'] == STATUS_DIVERGED) |
(self.coefficient_data['Status'] == STATUS_DIVERGING) |
(self.coefficient_data['Status'] == STATUS_CONVERGING)]
# We only need the symbols
all_symbols = pd.DataFrame(columns=['Symbol', 'Count'],
data={'Symbol': filtered_data['Symbol 1'].append(filtered_data['Symbol 2']),
'Count': 1})
# Group and count. Reset index so that we have named SYMBOL column.
filtered_data = all_symbols.groupby(by='Symbol').count().reset_index()
# Sort
filtered_data = filtered_data.sort_values('Count', ascending=False)
return filtered_data
def load(self, filename):
"""
@@ -108,7 +230,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:
@@ -135,7 +260,7 @@ class Correlation:
# Loop through all symbol pair combinations and calculate coefficient. Make sure you don't double count pairs
# eg. (USD/GBP AUD/USD vs AUD/USD USD/GBP). Use grid of all symbols with i and j axis. j starts at i + 1 to
# avoid duplicating. We will store all coefficients in a dataframe for export as CSV.
# avoid duplicating. We will store all coefficients in a dataframe.
index = 0
# There will be (x^2 - x) / 2 pairs where x is number of symbols
num_pair_combinations = int((len(symbols) ** 2 - len(symbols)) / 2)
@@ -165,8 +290,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:
@@ -178,12 +304,8 @@ class Correlation:
# If we were monitoring, we stopped, so start again.
if was_monitoring:
self.start_monitor(interval=self.__monitoring_params['interval'],
from_mins=self.__monitoring_params['from_mins'],
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):
"""
@@ -197,20 +319,26 @@ class Correlation:
return price_data
def start_monitor(self, interval, from_mins, 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 from_mins: The number of minutes of tick data to use for calculations
: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
@@ -228,14 +356,20 @@ 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
# 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, 'from_mins': from_mins,
'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.
thread = threading.Thread(target=self.__monitor)
thread.start()
def stop_monitor(self):
@@ -255,8 +389,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
@@ -290,8 +424,8 @@ class Correlation:
symbol1_prices_filtered = symbol1_prices[symbol1_prices['time'].isin(intersect_dates)]
symbol2_prices_filtered = symbol2_prices[symbol2_prices['time'].isin(intersect_dates)]
# Calculate coefficient. Only use if p value is < 0.01 (highly likely that coefficient is valid and null
# hypothesis is false).
# Calculate coefficient. Only use if p value is < max_p_value (highly likely that coefficient is valid
# and null hypothesis is false).
coefficient_with_p_value = pearsonr(symbol1_prices_filtered['close'], symbol2_prices_filtered['close'])
coefficient = None if coefficient_with_p_value[1] > max_p_value else coefficient_with_p_value[0]
@@ -306,27 +440,56 @@ class Correlation:
return coefficient
def get_coefficient_history(self, symbol1, symbol2):
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:
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
# 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
def get_ticks(self, symbol, date_from=None, date_to=None, cache_time=0, cache_only=False):
def clear_coefficient_history(self):
"""
Clears the coefficient history for all symbol pairs
:return:
"""
# Create dataframes for coefficient history.
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 status from coefficient data
self.coefficient_data['Status'] = ''
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.
@@ -342,9 +505,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.")
@@ -355,26 +518,70 @@ class Correlation:
self.__log.debug(f"Ticks for {symbol} retrieved from source and cached.")
return ticks
def __monitor(self, interval, from_mins, 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 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 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
# 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 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
stop_monitoring.
:param interval: How often to check in seconds
:param from_mins: The number of minutes of tick data to use for calculations
: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
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}.")
@@ -382,20 +589,14 @@ class Correlation:
# Only run if monitor is not stopped
if self.__monitoring:
# Update all coefficients
self.__update_all_coefficients(from_mins=from_mins, 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()
# Autosave
if autosave:
self.save(filename=filename)
if self.__autosave:
self.save(filename=self.__filename)
# Schedule the timer to run again
params = {'interval': interval, 'from_mins': from_mins, '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}
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()}")
@@ -405,120 +606,117 @@ class Correlation:
self.__first_run = False
self.__scheduler.run()
def __update_coefficient(self, symbol1, symbol2, from_mins, 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):
"""
Updates the coefficient 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 from_mins: The number of minutes of tick data to use for calculations
: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.
"""
coefficient = None
# 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.
# Date range for data
timezone = pytz.timezone("Etc/UTC")
date_to = datetime.now(tz=timezone)
date_from = date_to - timedelta(minutes=from_mins)
date_from = date_to - timedelta(minutes=max_from)
# Get the tick data
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)
# Get the tick data for the longest timeframe calculation.
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
# 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:
# 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
if symbol1ticks is not None and symbol2ticks is not None and \
len(symbol1ticks.index) > 0 and len(symbol2ticks.index) > 0:
symbol1ticks = symbol1ticks.set_index('time')
symbol2ticks = symbol2ticks.set_index('time')
try:
symbol1prices = symbol1ticks['ask'].resample('1S').ohlc()
symbol2prices = symbol2ticks['ask'].resample('1S').ohlc()
symbol1ticks = symbol1ticks.set_index('time')
symbol2ticks = symbol2ticks.set_index('time')
s1_prices = symbol1ticks['ask'].resample('1S').ohlc()
s2_prices = symbol2ticks['ask'].resample('1S').ohlc()
except RecursionError:
self.__log.warning(f"Coefficient could not be calculated for {symbol1}:{symbol2} as prices could not "
self.__log.warning(f"Coefficient could not be calculated for {symbol1}:{symbol2}. prices could not "
f"be resampled.")
else:
symbol1prices.reset_index(inplace=True)
symbol2prices.reset_index(inplace=True)
symbol1prices = symbol1prices[symbol1prices['close'].notna()]
symbol2prices = symbol2prices[symbol2prices['close'].notna()]
s1_prices.reset_index(inplace=True)
s2_prices.reset_index(inplace=True)
s1_prices = s1_prices[s1_prices['close'].notna()]
s2_prices = s2_prices[s2_prices['close'].notna()]
# Calculate the coefficient
coefficient = self.calculate_coefficient(symbol1_prices=symbol1prices, symbol2_prices=symbol2prices,
min_prices=min_prices,
max_set_size_diff_pct=max_set_size_diff_pct,
overlap_pct=overlap_pct, max_p_value=max_p_value)
# 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()
self.__log.debug(f"Symbol pair {symbol1}:{symbol2} has a coefficient of {coefficient}.")
else:
coefficient = None
# Get subset of the price data
s1_prices_subset = s1_prices[(s1_prices['time'] >= date_from_subset)]
s2_prices_subset = s2_prices[(s2_prices['time'] >= date_from_subset)]
# Update the coefficient data
if coefficient is not None:
self.__update_coefficient_data(symbol1=symbol1, symbol2=symbol2, coefficient=coefficient,
date_from=date_from, date_to=date_to)
# Calculate the coefficient
coefficient = \
self.calculate_coefficient(symbol1_prices=s1_prices_subset, symbol2_prices=s2_prices_subset,
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'])
return coefficient
self.__log.debug(f"Symbol pair {symbol1}:{symbol2} has a coefficient of {coefficient} for last "
f"{params['from']} minutes.")
def __update_all_coefficients(self, from_mins, min_prices=100, max_set_size_diff_pct=90, overlap_pct=90,
max_p_value=0.05, cache_time=10):
# Add the coefficient to a dict {timeframe: coefficient}. We will update together for all for
# symbol pair and time
coefficients[params['from']] = coefficient
# 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 from_mins: The number of minutes of tick data to use for calculations
: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_coefficient(symbol1=symbol1, symbol2=symbol2, from_mins=from_mins,
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)
def __reset_coefficient_data(self):
"""
Clears coefficient data and history.
:return:
"""
# Create dataframe for coefficient data
# 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)
# Create dataframe for coefficient history
coefficient_history_columns = ['Symbol 1', 'Symbol 2', 'Coefficient', 'UTC Date From', 'UTC Date To']
self.coefficient_history = pd.DataFrame(columns=coefficient_history_columns)
# Clear coefficient history
self.clear_coefficient_history()
def __update_coefficient_data(self, symbol1, symbol2, coefficient, date_from, date_to):
# Clear price data
self.__price_data = None
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:
:param date_from:
:param date_to:
: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:
"""
@@ -526,15 +724,66 @@ class Correlation:
now = datetime.now(tz=timezone)
# Update data if we have a coefficient and add to history
if coefficient is not None:
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 Check'] = now
'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, inverse=inverse)
self.coefficient_data.loc[(self.coefficient_data['Symbol 1'] == symbol1) &
(self.coefficient_data['Symbol 2'] == symbol2),
'Last Coefficient'] = coefficient
'Status'] = status
row = pd.DataFrame(columns=self.coefficient_history.columns,
data=[[symbol1, symbol2, coefficient, 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, 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 = STATUS_NOT_CALCULATED
# Only continue if we have calculated all coefficients, otherwise we will return STATUS_NOT_CALCULATED
if None not in coefficients.values():
# Get the values ordered by timeframe descending
ordered_values = []
for key in sorted(coefficients, reverse=True):
ordered_values.append(coefficients[key])
if self.monitor_inverse and inverse:
# Calculation for inverse calculations
if all(i <= self.divergence_threshold * -1 for i in ordered_values):
status = STATUS_CORRELATED
elif all(i > self.divergence_threshold * -1 for i in ordered_values):
status = STATUS_DIVERGED
elif all(ordered_values[i] <= ordered_values[i+1] for i in range(0, len(ordered_values)-1, 1)):
status = STATUS_CONVERGING
elif all(ordered_values[i] > ordered_values[i+1] for i in range(0, len(ordered_values)-1, 1)):
status = STATUS_DIVERGING
else:
status = STATUS_INCONSISTENT
else:
# Calculation for standard correlations
if all(i >= self.divergence_threshold for i in ordered_values):
status = STATUS_CORRELATED
elif all(i < self.divergence_threshold for i in ordered_values):
status = STATUS_DIVERGED
elif all(ordered_values[i] <= ordered_values[i+1] for i in range(0, len(ordered_values)-1, 1)):
status = STATUS_DIVERGING
elif all(ordered_values[i] > ordered_values[i+1] for i in range(0, len(ordered_values)-1, 1)):
status = STATUS_CONVERGING
else:
status = STATUS_INCONSISTENT
return status
-604
View File
@@ -1,604 +0,0 @@
import wx
import wx.grid
import matplotlib.pyplot as plt
from matplotlib.backends.backend_wxagg import FigureCanvasWxAgg as FigureCanvas
import matplotlib.dates
import matplotlib
from mt5_correlation.correlation import Correlation
from mt5_correlation.config import Config, SettingsDialog
from datetime import datetime, timedelta
import pytz
import pandas as pd
import logging
import logging.config
matplotlib.use('WXAgg')
class MonitorFrame(wx.Frame):
__cor = None
__rows = 0 # Need to track as we need to notify grid if row count changes.
__opened_filename = None # So we can save to same file as we opened
__config = None # The applications config
__selected_correlation = [] # List of Symbol 1 & Symbol 2
# Columns for coefficient table
COLUMN_INDEX = 0
COLUMN_SYMBOL1 = 1
COLUMN_SYMBOL2 = 2
COLUMN_BASE_COEFFICIENT = 3
COLUMN_DATE_FROM = 4
COLUMN_DATE_TO = 5
COLUMN_TIMEFRAME = 6
COLUMN_LAST_CHECK = 7
COLUMN_LAST_COEFFICIENT = 8
def __init__(self):
# Super
wx.Frame.__init__(self, parent=None, id=wx.ID_ANY, title="Divergence Monitor",
pos=wx.Point(x=Config().get('window.x'),
y=Config().get('window.y')),
size=wx.Size(width=Config().get('window.width'),
height=Config().get('window.height')),
style=Config().get('window.style'))
# Create logger and get config
self.__log = logging.getLogger(__name__)
self.__config = Config()
# Create correlation instance to maintain state of calculated coefficients. Set min coefficient from config
self.__cor = Correlation()
self.__cor.monitoring_threshold = self.__config.get("monitor.monitoring_threshold")
# Status bar
self.statusbar = self.CreateStatusBar(1)
# Menu Bar and file menu
self.menubar = wx.MenuBar()
file_menu = wx.Menu()
# File menu items
menu_item_open = file_menu.Append(wx.ID_ANY, "Open", "Open correlations file.")
menu_item_save = file_menu.Append(wx.ID_ANY, "Save", "Save correlations file.")
menu_item_saveas = file_menu.Append(wx.ID_ANY, "Save As", "Save correlations file.")
file_menu.AppendSeparator()
menu_item_calculate = file_menu.Append(wx.ID_ANY, "Calculate", "Calculate base coefficients.")
file_menu.AppendSeparator()
menu_item_settings = file_menu.Append(wx.ID_ANY, "Settings", "Change application settings.")
file_menu.AppendSeparator()
menu_item_exit = file_menu.Append(wx.ID_ANY, "Exit", "Close the application")
# Add file menu and set menu bar
self.menubar.Append(file_menu, "File")
self.SetMenuBar(self.menubar)
# Main window. We want 2 horizontal sections, the grid showing correlations and a graph. In the correlations
# section, we want 2 vertical sections, the monitor toggle and the correlations grid. For the toggle we want 2
# sections, a label and a toggle.
# ---------------------------------------------------------------
# |label | toggle | |
# |-----------------------| |
# | | |
# |Correlations Grid | Graphs |
# | | |
# | | |
# | | |
# | | |
# ----------------------------------------------------------------
panel = wx.Panel(self, wx.ID_ANY)
toggle_sizer = wx.BoxSizer(wx.HORIZONTAL) # Label and toggle
correlations_sizer = wx.BoxSizer(wx.VERTICAL) # Toggle sizer and correlations grid
self.__main_sizer = wx.BoxSizer(wx.HORIZONTAL) # Correlations sizer and graphs panel
panel.SetSizer(self.__main_sizer)
# Create the label and toggle, populate the toggle sizer and add the toggle sizer to the correlations sizer
monitor_toggle_label = wx.StaticText(panel, id=wx.ID_ANY, label="Monitoring")
toggle_sizer.Add(monitor_toggle_label, 0, wx.ALL, 1)
self.monitor_toggle = wx.ToggleButton(panel, wx.ID_ANY, label="Off")
self.monitor_toggle.SetBackgroundColour(wx.RED)
toggle_sizer.Add(self.monitor_toggle, 0, wx.ALL, 1)
correlations_sizer.Add(toggle_sizer, 0, wx.ALL, 1)
# Create the correlations grid. This is a data table using pandas dataframe for underlying data. Add the
# correlations_grid to the correlations sizer.
self.table = DataTable(self.__cor.filtered_coefficient_data)
self.grid_correlations = wx.grid.Grid(panel, wx.ID_ANY)
self.grid_correlations.SetTable(self.table, takeOwnership=True)
self.grid_correlations.EnableEditing(False)
self.grid_correlations.EnableDragRowSize(False)
self.grid_correlations.EnableDragColSize(False)
self.grid_correlations.EnableDragGridSize(False)
self.grid_correlations.SetSelectionMode(wx.grid.Grid.SelectRows)
self.grid_correlations.SetRowLabelSize(0)
self.grid_correlations.SetColSize(self.COLUMN_INDEX, 0) # Index. Hide
self.grid_correlations.SetColSize(self.COLUMN_SYMBOL1, 100) # Symbol 1
self.grid_correlations.SetColSize(self.COLUMN_SYMBOL2, 100) # Symbol 2
self.grid_correlations.SetColSize(self.COLUMN_BASE_COEFFICIENT, 100) # Base Coefficient
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))
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
self.__graph = GraphPanel(panel)
self.__graph.Hide()
# Add the correlations sizer and the charts to the main sizer.
self.__main_sizer.Add(correlations_sizer, 0, wx.ALL | wx.EXPAND, 1)
self.__main_sizer.Add(self.__graph, 1, wx.ALL | wx.EXPAND, 1)
# Layout the window.
self.Layout()
# Set up timer to refresh grid
self.timer = wx.Timer(self)
# Bind monitor button
self.monitor_toggle.Bind(wx.EVT_TOGGLEBUTTON, self.monitor)
# Bind timer
self.Bind(wx.EVT_TIMER, self.__timer_event, self.timer)
# Bind menu items
self.Bind(wx.EVT_MENU, self.open_file, menu_item_open)
self.Bind(wx.EVT_MENU, self.save_file, menu_item_save)
self.Bind(wx.EVT_MENU, self.save_file_as, menu_item_saveas)
self.Bind(wx.EVT_MENU, self.calculate_coefficients, menu_item_calculate)
self.Bind(wx.EVT_MENU, self.open_settings, menu_item_settings)
self.Bind(wx.EVT_MENU, self.quit, menu_item_exit)
# Bind row select
self.Bind(wx.grid.EVT_GRID_SELECT_CELL, self.select_cell, self.grid_correlations)
# Bind window close event
self.Bind(wx.EVT_CLOSE, self.on_close, self)
def open_file(self, event):
with wx.FileDialog(self, "Open Coefficients file", wildcard="cpd (*.cpd)|*.cpd",
style=wx.FD_OPEN | wx.FD_FILE_MUST_EXIST) as fileDialog:
if fileDialog.ShowModal() == wx.ID_CANCEL:
return # the user changed their mind
# Load the file chosen by the user.
self.__opened_filename = fileDialog.GetPath()
self.SetStatusText(f"Loading file {self.__opened_filename}.")
self.__cor.load(self.__opened_filename)
# Refresh data in grid
self.refresh_grid()
self.SetStatusText(f"File {self.__opened_filename} loaded.")
def save_file(self, event):
self.SetStatusText(f"Saving file as {self.__opened_filename}")
if self.__opened_filename is None:
self.save_file_as(event)
else:
self.__cor.save(self.__opened_filename)
self.SetStatusText(f"File saved as {self.__opened_filename}")
def save_file_as(self, event):
with wx.FileDialog(self, "Save Coefficients file", wildcard="cpd (*.cpd)|*.cpd",
style=wx.FD_SAVE) as fileDialog:
if fileDialog.ShowModal() == wx.ID_CANCEL:
return # the user changed their mind
# Save the file and price data file, changing opened filename so next save writes to new file
self.SetStatusText(f"Saving file as {self.__opened_filename}")
self.__opened_filename = fileDialog.GetPath()
self.__cor.save(self.__opened_filename)
self.SetStatusText(f"File saved as {self.__opened_filename}")
def calculate_coefficients(self, event):
# set time zone to UTC to avoid local offset issues, and get from and to dates (a week ago to today)
timezone = pytz.timezone("Etc/UTC")
utc_to = datetime.now(tz=timezone)
utc_from = utc_to - timedelta(days=self.__config.get('calculate.from.days'))
# Calculate
self.SetStatusText("Calculating coefficients.")
self.__cor.calculate(date_from=utc_from, date_to=utc_to,
timeframe=self.__config.get('calculate.timeframe'),
min_prices=self.__config.get('calculate.min_prices'),
max_set_size_diff_pct=self.__config.get('calculate.max_set_size_diff_pct'),
overlap_pct=self.__config.get('calculate.overlap_pct'),
max_p_value=self.__config.get('calculate.max_p_value'))
self.SetStatusText("")
# Show calculated data
self.refresh_grid()
def quit(self, event):
# Close
self.Close()
def refresh_grid(self):
"""
Refreshes grid. Notifies if rows have been added or deleted.
:return:
"""
self.__log.debug(f"Refreshing grid. Timer running: {self.timer.IsRunning()}")
# 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', '')
# Start refresh
self.grid_correlations.BeginBatch()
# Check if num rows in dataframe has changed, and send appropriate APPEND or DELETE messages
cur_rows = len(self.__cor.filtered_coefficient_data.index)
if cur_rows < self.__rows:
# Data has been deleted. Send message
msg = wx.grid.GridTableMessage(self.table, wx.grid.GRIDTABLE_NOTIFY_ROWS_DELETED,
self.__rows - cur_rows, self.__rows - cur_rows)
self.grid_correlations.ProcessTableMessage(msg)
elif cur_rows > self.__rows:
# Data has been added. Send message
msg = wx.grid.GridTableMessage(self.table, wx.grid.GRIDTABLE_NOTIFY_ROWS_APPENDED,
cur_rows - self.__rows) # how many
self.grid_correlations.ProcessTableMessage(msg)
self.grid_correlations.EndBatch()
# Send updated message
msg = wx.grid.GridTableMessage(self.table, wx.grid.GRIDTABLE_REQUEST_VIEW_GET_VALUES)
self.grid_correlations.ProcessTableMessage(msg)
# Update row count
self.__rows = cur_rows
def monitor(self, event):
# Check state of toggle button. If on, then start monitoring, else stop
if self.monitor_toggle.GetValue():
self.__log.info("Starting monitoring.")
self.monitor_toggle.SetBackgroundColour(wx.GREEN)
self.monitor_toggle.SetLabelText("On")
self.SetStatusText("Monitoring for changes to coefficients.")
self.timer.Start(self.__config.get('monitor.interval')*1000)
# Autosave filename
filename = self.__opened_filename if self.__opened_filename is not None else 'autosave.cpd'
self.__cor.start_monitor(interval=self.__config.get('monitor.interval'),
from_mins=self.__config.get('monitor.from.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'),
cache_time=self.__config.get('monitor.tick_cache_time'),
autosave=self.__config.get('monitor.autosave'),
filename=filename)
else:
self.__log.info("Stopping monitoring.")
self.monitor_toggle.SetBackgroundColour(wx.RED)
self.monitor_toggle.SetLabelText("Off")
self.SetStatusText("Monitoring stopped.")
self.timer.Stop()
self.__cor.stop_monitor()
def open_settings(self, event):
"""
Opens the settings dialog
:return:
"""
settings_dialog = SettingsDialog(parent=self, exclude=['window'])
res = settings_dialog.ShowModal()
if res == wx.ID_OK:
# Reload relevant parts of app
restart_monitor_timer = False
restart_gui_timer = False
reload_correlations = False
reload_logger = False
reload_graph = False
for setting in settings_dialog.changed_settings:
# If any 'monitor.' settings except 'monitor.divergence_threshold have changed then restart
# monitoring timer with new settings.
# If 'monitor.interval has changed then restart gui timer.
# If 'monitor.monitoring_threshold' has changed, then refresh correlation data.
# If any 'logging.' settings have changed, then reload logger config.
if setting.startswith('monitor.') and setting != 'monitor.divergence_threshold':
restart_monitor_timer = True
if setting == 'monitor.interval':
restart_gui_timer = True
if setting == 'monitor.monitoring_threshold':
reload_correlations = True
if setting.startswith('logging.'):
reload_logger = True
if setting.startswith('monitor.from.'):
reload_graph = True
# Now perform the actions
if restart_monitor_timer:
self.__log.info("Settings updated. Reloading monitoring timer.")
self.__cor.stop_monitor()
self.__cor.start_monitor(interval=self.__config.get('monitor.interval'),
from_mins=self.__config.get('monitor.from.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'))
if restart_gui_timer:
self.__log.info("Settings updated. Restarting gui timer.")
self.timer.Stop()
self.timer.Start(self.__config.get('monitor.interval') * 1000)
if reload_correlations:
self.__log.info("Settings updated. Updating monitoring threshold and reloading grid.")
self.__cor.monitoring_threshold = self.__config.get("monitor.monitoring_threshold")
self.refresh_grid()
if reload_logger:
self.__log.info("Settings updated. Reloading logger.")
log_config = Config().get('logging')
logging.config.dictConfig(log_config)
if reload_graph:
self.__log.info("Settings updated. Reloading graph.")
if len(self.__selected_correlation) == 2:
self.show_graph(symbol1=self.__selected_correlation[0], symbol2=self.__selected_correlation[1])
def on_close(self, event):
"""
Window closing. Save coefficients and stop monitoring.
:param event:
:return:
"""
# Save pos and size
x, y = self.GetPosition()
width, height = self.GetSize()
self.__config.set('window.x', x)
self.__config.set('window.y', y)
self.__config.set('window.width', width)
self.__config.set('window.height', height)
# Style
style = self.GetWindowStyle()
self.__config.set('window.style', style)
self.__config.save()
# Stop monitoring
self.__cor.stop_monitor()
# Kill graph as it seems to be stopping script from ending
self.__graph = None
# End
event.Skip()
def select_cell(self, event):
"""
A cell was selected. Show the graph for the correlation.
:param event:
:return:
"""
# Get row and symbols.
row = event.GetRow()
symbol1 = self.grid_correlations.GetCellValue(row, self.COLUMN_SYMBOL1)
symbol2 = self.grid_correlations.GetCellValue(row, self.COLUMN_SYMBOL2)
self.__selected_correlation = [symbol1, symbol2]
self.show_graph(symbol1, symbol2)
def show_graph(self, symbol1, symbol2):
"""
Displays the graph for the specified symbols correlation history
:param symbol1:
:param symbol2:
:return:
"""
# Get the price data for the base coefficient calculation, tick data to calculate last coefficient and and the
# coefficient history data
symbol_1_price_data = self.__cor.get_price_data(symbol1)
symbol_2_price_data = self.__cor.get_price_data(symbol2)
symbol_1_ticks = self.__cor.get_ticks(symbol1, cache_only=True)
symbol_2_ticks = self.__cor.get_ticks(symbol2, cache_only=True)
history_data = self.__cor.get_coefficient_history(symbol1, symbol2)
times = history_data['UTC Date To']
coefficients = history_data['Coefficient']
# Display if we have any data
self.__log.debug(f"Refreshing history graph {symbol1}:{symbol2}.")
self.__graph.draw(times=times, coefficients=coefficients, prices=[symbol_1_price_data, symbol_2_price_data],
ticks=[symbol_1_ticks, symbol_2_ticks], symbols=[symbol1, symbol2])
# Un-hide and layout if hidden
if not self.__graph.IsShown():
self.__graph.Show()
self.__main_sizer.Layout()
def __timer_event(self, event):
"""
Called on timer event. Refreshes grid and updatates selected graph.
:return:
"""
self.refresh_grid()
if len(self.__selected_correlation) == 2:
self.show_graph(symbol1=self.__selected_correlation[0], symbol2=self.__selected_correlation[1])
class DataTable(wx.grid.GridTableBase):
"""
A data table that holds data in a pandas dataframe
"""
def __init__(self, data=None):
wx.grid.GridTableBase.__init__(self)
self.headerRows = 1
if data is None:
data = pd.DataFrame()
self.data = data
# Get divergence threshold from app config
self.divergence_threshold = Config().get('monitor.divergence_threshold')
def GetNumberRows(self):
return len(self.data)
def GetNumberCols(self):
return len(self.data.columns) + 1
def GetValue(self, row, col):
if col == 0:
return self.data.index[row]
return self.data.iloc[row, col - 1]
def SetValue(self, row, col, value):
self.data.iloc[row, col - 1] = value
def GetColLabelValue(self, col):
if col == 0:
if self.data.index.name is None:
return 'Index'
else:
return self.data.index.name
return str(self.data.columns[col - 1])
def GetTypeName(self, row, col):
return wx.grid.GRID_VALUE_STRING
def GetAttr(self, row, col, prop):
attr = wx.grid.GridCellAttr()
# If column is last coefficient, get value and check against threshold. Highlight if diverged.
threshold = Config().get('monitor.divergence_threshold')
if col == MonitorFrame.COLUMN_LAST_COEFFICIENT:
value = self.GetValue(row, col)
if value != "":
value = float(value)
if value <= threshold:
attr.SetBackgroundColour(wx.YELLOW)
else:
attr.SetBackgroundColour(wx.WHITE)
return attr
class GraphPanel(wx.Panel):
def __init__(self, parent):
"""
A panel to show the graphs
:param parent: The parent panel
"""
# Super
wx.Panel.__init__(self, parent)
# 3 axis, 2 price data for calculate, 2 price data for last coefficient and coefficient history.
# All will have axis labels and top and right boarders
# removed
self.__fig, self.__axes = plt.subplots(nrows=5, ncols=1)
# Create the canvas
self.__canvas = FigureCanvas(self, -1, self.__fig)
# Date format for x axes
self.__tick_fmt_date = matplotlib.dates.DateFormatter('%d-%b')
self.__tick_fmt_time = matplotlib.dates.DateFormatter('%H:%M:%S')
# Sizer etc.
self.__sizer = wx.BoxSizer(wx.VERTICAL)
self.__sizer.Add(self.__canvas, 1, wx.LEFT | wx.TOP | wx.GROW)
self.SetSizer(self.__sizer)
self.Fit()
def __del__(self):
# Close all plots
plt.close('all')
self.__axes = None
self.__fig = None
def draw(self, times, coefficients, prices=None, symbols=None, ticks=None):
"""
Plot the correlations.
:param times: Series of time values for x axis for coefficient history chart
:param coefficients: Series of coefficients values for y axis of coefficient history chart
:param prices: Price data used to calculate base coefficient. List [Symbol1 Price Data, Symbol 2 Price Data]
:param symbols: Symbols. List [Symbol1, Symbol2]
:param ticks: Ticks used to calculate last coefficient. List [Symbol1, Symbol2]
:return:
"""
# Clear. We will need to redraw
for ax in self.__axes:
ax.clear()
if symbols is not None and len(symbols) == 2:
# Axis ranges
price_chart_date_range = [min(min(prices[0]['time']), min(prices[1]['time'])),
max(max(prices[0]['time']), max(prices[1]['time']))]
tick_chart_date_range = [min(min(ticks[0]['time']), min(ticks[1]['time'])),
max(max(ticks[0]['time']), max(ticks[1]['time']))]
# Chart config
titles = [f"Base Coefficient Price Data for {symbols[0]}", f"Base Coefficient Price Data for {symbols[1]}",
f"Coefficient Tick Data for {symbols[0]}", f"Coefficient Tick Data for {symbols[1]}",
f"Coefficient History for {symbols[0]}:{symbols[1]}"]
xlims = [price_chart_date_range, price_chart_date_range, tick_chart_date_range, tick_chart_date_range, None]
ylims = [None, None, None, None, [-1, 1]]
xlabels = [None, None, None, None, None]
ylabels = ['Price', 'Price', 'Price', 'Price', 'Coefficient']
tick_labels = [[], prices[1]['time'], [], ticks[1]['time'], times]
mtick_fmts = [None, self.__tick_fmt_date, None, self.__tick_fmt_time, self.__tick_fmt_time]
mtick_rot = [0, 45, 0, 45, 45]
xdata = [prices[0]['time'], prices[1]['time'], ticks[0]['time'], ticks[1]['time'], times]
ydata = [prices[0]['close'], prices[1]['close'], ticks[0]['ask'], ticks[1]['ask'], coefficients]
types = ['plot', 'plot', 'plot', 'plot', 'scatter']
# Draw 5 charts
for index in range(0, len(self.__axes)):
# Titles and axis labels
self.__axes[index].set_title(titles[index])
self.__axes[index].set_xlabel(xlabels[index])
self.__axes[index].set_ylabel(ylabels[index])
# Limits
if xlims[index] is not None:
self.__axes[index].set_xlim(xlims[index])
if ylims[index] is not None:
self.__axes[index].set_ylim(ylims[index])
# Tick labels and formats
self.__axes[index].xaxis.set_ticklabels(tick_labels[index])
if mtick_fmts[index] is not None:
self.__axes[index].xaxis.set_major_formatter(mtick_fmts[index])
plt.setp(self.__axes[index].xaxis.get_majorticklabels(), rotation=mtick_rot[index])
# Remove typ and right boarders
self.__axes[index].spines["top"].set_visible(False)
self.__axes[index].spines["right"].set_visible(False)
# Plot
if types[index] == 'plot':
self.__axes[index].plot(xdata[index], ydata[index])
elif types[index] == 'scatter':
self.__axes[index].scatter(xdata[index], ydata[index], s=1)
# Layout with padding between charts
self.__fig.tight_layout(pad=0.5)
# Redraw canvas
self.__canvas.draw()
+1
View File
@@ -0,0 +1 @@
from mt5_correlation.gui.mdi import CorrelationMDIFrame
+359
View File
@@ -0,0 +1,359 @@
import abc
import importlib
import logging
import pytz
import wx
import wx.lib.inspection as ins
import wxconfig
import wxconfig as cfg
from datetime import datetime, timedelta
from mt5_correlation import correlation as cor
class CorrelationMDIFrame(wx.MDIParentFrame):
"""
The MDI Frame window for the correlation monitoring application
"""
# The correlation instance that calculates coefficients and monitors for divergence. Needs to be accessible to
# child frames.
cor = None
__opened_filename = None # So we can save to same file as we opened
__log = None # The logger
__menu_item_monitor = None # We need to store this menu item so that we can check if it is checked or not.
def __init__(self):
# Super
wx.MDIParentFrame.__init__(self, parent=None, id=wx.ID_ANY, title="Divergence Monitor",
pos=wx.Point(x=cfg.Config().get('window.x'), y=cfg.Config().get('window.y')),
size=wx.Size(width=cfg.Config().get('window.width'),
height=cfg.Config().get('window.height')),
style=cfg.Config().get('window.style'))
# Create logger
self.__log = logging.getLogger(__name__)
# Create correlation instance to maintain state of calculated coefficients. Set params from config
self.cor = cor.Correlation(monitoring_threshold=cfg.Config().get("monitor.monitoring_threshold"),
divergence_threshold=cfg.Config().get("monitor.divergence_threshold"),
monitor_inverse=cfg.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.
self.__statusbar = self.CreateStatusBar(2)
self.__statusbar.SetStatusWidths([100, -1])
self.SetStatusText("Not Monitoring", 0)
# Create menu bar and bind menu items to methods
self.menubar = wx.MenuBar()
# File menu and items
menu_file = wx.Menu()
self.Bind(wx.EVT_MENU, self.__on_open_file, menu_file.Append(wx.ID_ANY, "&Open", "Open correlations file."))
self.Bind(wx.EVT_MENU, self.__on_save_file, menu_file.Append(wx.ID_ANY, "Save", "Save correlations file."))
self.Bind(wx.EVT_MENU, self.__on_save_file_as,
menu_file.Append(wx.ID_ANY, "Save As", "Save correlations file."))
menu_file.AppendSeparator()
self.Bind(wx.EVT_MENU, self.__on_open_settings,
menu_file.Append(wx.ID_ANY, "Settings", "Change application settings."))
menu_file.AppendSeparator()
self.Bind(wx.EVT_MENU, self.__on_exit, menu_file.Append(wx.ID_ANY, "Exit", "Close the application"))
self.menubar.Append(menu_file, "&File")
# Coefficient menu and items
menu_coef = wx.Menu()
self.Bind(wx.EVT_MENU, self.__on_calculate,
menu_coef.Append(wx.ID_ANY, "Calculate", "Calculate base coefficients."))
self.__menu_item_monitor = menu_coef.Append(wx.ID_ANY, "Monitor",
"Monitor correlated pairs for changes to coefficient.",
kind=wx.ITEM_CHECK)
self.Bind(wx.EVT_MENU, self.__on_monitor, self.__menu_item_monitor)
menu_coef.AppendSeparator()
self.Bind(wx.EVT_MENU, self.__on_clear,
menu_coef.Append(wx.ID_ANY, "Clear", "Clear coefficient and price history."))
self.menubar.Append(menu_coef, "Coefficient")
# View menu and items
menu_view = wx.Menu()
self.Bind(wx.EVT_MENU, self.__on_view_status, menu_view.Append(wx.ID_ANY, "Status",
"View status of correlations."))
self.Bind(wx.EVT_MENU, self.__on_view_diverged, menu_view.Append(wx.ID_ANY, "Diverged Symbols",
"View diverged symbols."))
self.menubar.Append(menu_view, "&View")
# Help menu and items
help_menu = wx.Menu()
self.Bind(wx.EVT_MENU, self.__on_view_log, help_menu.Append(wx.ID_ANY, "View Log", "Show application log."))
self.Bind(wx.EVT_MENU, self.__on_view_help, help_menu.Append(wx.ID_ANY, "Help",
"Show application usage instructions."))
self.menubar.Append(help_menu, "&Help")
# Set menu bar
self.SetMenuBar(self.menubar)
# Set up timer to refresh
self.timer = wx.Timer(self)
self.Bind(wx.EVT_TIMER, self.__on_timer, self.timer)
# Bind window close event
self.Bind(wx.EVT_CLOSE, self.__on_close, self)
def __on_close(self, event):
"""
Window closing. Save coefficients and stop monitoring.
:param event:
:return:
"""
# Save pos and size
x, y = self.GetPosition()
width, height = self.GetSize()
cfg.Config().set('window.x', x)
cfg.Config().set('window.y', y)
cfg.Config().set('window.width', width)
cfg.Config().set('window.height', height)
# Style
style = self.GetWindowStyle()
cfg.Config().set('window.style', style)
cfg.Config().save()
# Stop monitoring
self.cor.stop_monitor()
# End
event.Skip()
def __on_open_file(self, evt):
with wx.FileDialog(self, "Open Coefficients file", wildcard="cpd (*.cpd)|*.cpd",
style=wx.FD_OPEN | wx.FD_FILE_MUST_EXIST) as fileDialog:
if fileDialog.ShowModal() == wx.ID_CANCEL:
return # the user changed their mind
# Load the file chosen by the user.
self.__opened_filename = fileDialog.GetPath()
self.SetStatusText(f"Loading file {self.__opened_filename}.", 1)
self.cor.load(self.__opened_filename)
# Show calculated data and refresh all opened frames
self.__on_view_status(evt)
self.__refresh()
self.SetStatusText(f"File {self.__opened_filename} loaded.", 1)
def __on_save_file(self, evt):
self.SetStatusText(f"Saving file as {self.__opened_filename}", 1)
if self.__opened_filename is None:
self.__on_save_file_as(evt)
else:
self.cor.save(self.__opened_filename)
self.SetStatusText(f"File saved as {self.__opened_filename}", 1)
def __on_save_file_as(self, evt):
with wx.FileDialog(self, "Save Coefficients file", wildcard="cpd (*.cpd)|*.cpd",
style=wx.FD_SAVE) as fileDialog:
if fileDialog.ShowModal() == wx.ID_CANCEL:
return # the user changed their mind
# Save the file and price data file, changing opened filename so next save writes to new file
self.SetStatusText(f"Saving file as {self.__opened_filename}", 1)
self.__opened_filename = fileDialog.GetPath()
self.cor.save(self.__opened_filename)
self.SetStatusText(f"File saved as {self.__opened_filename}", 1)
def __on_open_settings(self, evt):
settings_dialog = cfg.SettingsDialog(parent=self, exclude=['window'])
res = settings_dialog.ShowModal()
if res == wx.ID_OK:
# Stop the monitor
self.cor.stop_monitor()
# Build calculation params and restart the monitor
calculation_params = [cfg.Config().get('monitor.calculations.long'),
cfg.Config().get('monitor.calculations.medium'),
cfg.Config().get('monitor.calculations.short')]
self.cor.start_monitor(interval=cfg.Config().get('monitor.interval'),
calculation_params=calculation_params,
cache_time=cfg.Config().get('monitor.tick_cache_time'),
autosave=cfg.Config().get('monitor.autosave'),
filename=self.__opened_filename)
# Refresh all open child frames
self.__refresh()
def __on_exit(self, evt):
# Close
self.Close()
def __on_calculate(self, evt):
# set time zone to UTC to avoid local offset issues, and get from and to dates (a week ago to today)
timezone = pytz.timezone("Etc/UTC")
utc_to = datetime.now(tz=timezone)
utc_from = utc_to - timedelta(days=cfg.Config().get('calculate.from.days'))
# Calculate
self.SetStatusText("Calculating coefficients.", 1)
self.cor.calculate(date_from=utc_from, date_to=utc_to,
timeframe=cfg.Config().get('calculate.timeframe'),
min_prices=cfg.Config().get('calculate.min_prices'),
max_set_size_diff_pct=cfg.Config().get('calculate.max_set_size_diff_pct'),
overlap_pct=cfg.Config().get('calculate.overlap_pct'),
max_p_value=cfg.Config().get('calculate.max_p_value'))
self.SetStatusText("", 1)
# Show calculated data and refresh frames
self.__on_view_status(evt)
self.__refresh()
def __on_monitor(self, evt):
# Check state of toggle menu. If on, then start monitoring, else stop
if self.__menu_item_monitor.IsChecked():
self.__log.info("Starting monitoring for changes to coefficients.")
self.SetStatusText("Monitoring", 0)
self.__statusbar.SetBackgroundColour('green')
self.__statusbar.Refresh()
self.timer.Start(cfg.Config().get('monitor.interval') * 1000)
# Autosave filename
filename = self.__opened_filename if self.__opened_filename is not None else 'autosave.cpd'
# Build calculation params and start monitor
calculation_params = [cfg.Config().get('monitor.calculations.long'),
cfg.Config().get('monitor.calculations.medium'),
cfg.Config().get('monitor.calculations.short')]
self.cor.start_monitor(interval=cfg.Config().get('monitor.interval'),
calculation_params=calculation_params,
cache_time=cfg.Config().get('monitor.tick_cache_time'),
autosave=cfg.Config().get('monitor.autosave'),
filename=filename)
else:
self.__log.info("Stopping monitoring.")
self.SetStatusText("Not Monitoring", 0)
self.__statusbar.SetBackgroundColour('lightgray')
self.__statusbar.Refresh()
self.timer.Stop()
self.cor.stop_monitor()
def __on_clear(self, evt):
# Clear the history
self.cor.clear_coefficient_history()
# Refresh opened child frames
self.__refresh()
def __on_timer(self, evt):
# Refresh opened child frames
self.__refresh()
# Set status message
self.SetStatusText(f"Status updated at {self.cor.get_last_calculation():%d-%b %H:%M:%S}.", 1)
def __on_view_status(self, evt):
FrameManager.open_frame(parent=self, frame_module='mt5_correlation.gui.mdi_child_status',
frame_class='MDIChildStatus',
raise_if_open=True)
def __on_view_diverged(self, evt):
FrameManager.open_frame(parent=self, frame_module='mt5_correlation.gui.mdi_child_diverged_symbols',
frame_class='MDIChildDivergedSymbols',
raise_if_open=True)
def __on_view_log(self, evt):
FrameManager.open_frame(parent=self, frame_module='mt5_correlation.gui.mdi_child_log',
frame_class='MDIChildLog',
raise_if_open=True)
def __on_view_help(self, evt):
FrameManager.open_frame(parent=self, frame_module='mt5_correlation.gui.mdi_child_help',
frame_class='MDIChildHelp',
raise_if_open=True)
def __refresh(self):
"""
Refresh all open child frames
:return:
"""
children = self.GetChildren()
for child in children:
if isinstance(child, CorrelationMDIChild):
child.refresh()
elif isinstance(child, wx.StatusBar) or isinstance(child, ins.InspectionFrame) or \
isinstance(child, wxconfig.SettingsDialog):
# Ignore
pass
else:
raise Exception(f"MDI Child for application must implement CorrelationMDIChild. MDI Child is "
f"{type(child)}.")
class CorrelationMDIChild(wx.MDIChildFrame):
"""
Interface for all MDI Children supported by the MDIParent
"""
@abc.abstractmethod
def refresh(self):
"""
Must be implemented. Refreshes the content. Called by MDIParents __refresh method
:return:
"""
raise NotImplementedError
class FrameManager:
"""
Manages the opening and raising of MDIChild frames
"""
@staticmethod
def open_frame(parent, frame_module, frame_class, raise_if_open=True, **kwargs):
"""
Opens the frame specified by the frame class
:param parent: The MDIParentFrame to open the child frame into
:param frame_module: A string specifying the module containing the frame class to open or raise
:param frame_class: A string specifying the frame class to open or raise
:param raise_if_open: Whether the frame should raise rather than open if an instance is already open.
:param kwargs: A dict of parameters to pass to frame constructor. These will also be checked in raise_if_open
to determine uniqueness (i.e. If a frame of the same class is already open but its params are different,
then the frame will be opened again with the new params instead of being raised.)
:return:
"""
# Load the module and class
module = importlib.import_module(frame_module)
clazz = getattr(module, frame_class)
# Do we have an opened instance
opened_instance = None
for child in parent.GetChildren():
if isinstance(child, clazz):
# do the args match
match = True
for key in kwargs:
if kwargs[key] != getattr(child, key):
match = False
# Only open existing instance if args matched
if match:
opened_instance = child
# If we dont have an opened instance or raise_on_open is False then open new frame, otherwise raise it
if opened_instance is None or raise_if_open is False:
if len(kwargs) == 0:
clazz(parent=parent).Show(True)
else:
clazz(parent=parent, **kwargs).Show(True)
else:
opened_instance.Raise()
@@ -0,0 +1,215 @@
import logging
import matplotlib.dates
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
import wx
import wxconfig as cfg
import wx.lib.scrolledpanel as scrolled
from matplotlib.backends.backend_wxagg import FigureCanvasWxAgg as FigureCanvas
import mt5_correlation.gui.mdi as mdi
class MDIChildCorrelationGraph(mdi.CorrelationMDIChild):
"""
Shows the graphs for the specified correlation
"""
symbols = None # Symbols for correlation. Public as we use to check if window for the symbol pair is already open.
# Date formats for graphs
__tick_fmt_date = matplotlib.dates.DateFormatter('%d-%b')
__tick_fmt_time = matplotlib.dates.DateFormatter('%H:%M:%S')
# Colors for graph lines for symbol1 and symbol2. Will use first 2 colours in colormap
__colours = matplotlib.cm.get_cmap(cfg.Config().get("charts.colormap")).colors
# Fig, axes and canvas
__fig = None
__axs = None
__canvas = None
def __init__(self, parent, **kwargs):
# Super
wx.MDIChildFrame.__init__(self, parent=parent, id=wx.ID_ANY,
title=f"Correlation Status for {kwargs['symbols'][0]}:{kwargs['symbols'][1]}")
# Create logger
self.__log = logging.getLogger(__name__)
# Store the symbols
self.symbols = kwargs['symbols']
# We will freeze this frame and thaw once constructed to avoid flicker.
self.Freeze()
# Draw the empty graphs. We will populate with data in refresh. We will have 3 charts:
# 1) Data used to calculate base coefficient for both symbols (2 lines on chart);
# 2) Data used to calculate latest coefficient for both symbols (2 lines on chart); and
# 3) Coefficient history and the divergence threshold lines
# Create fig and 3 axes.
self.__fig, self.__axs = plt.subplots(3)
# Create additional axis for second line on charts 1 & 2
self.__s2axs = [self.__axs[0].twinx(), self.__axs[1].twinx()]
# Set titles
self.__axs[0].set_title(f"Base Coefficient Price Data for {self.symbols[0]}:{self.symbols[1]}")
self.__axs[1].set_title(f"Coefficient Tick Data for {self.symbols[0]}:{self.symbols[1]}")
self.__axs[2].set_title(f"Coefficient History for {self.symbols[0]}:{self.symbols[1]}")
# Set Y Labels and tick colours for charts 1 & 2. Left for symbol1, right for symbol2
for i in range(0, 2):
self.__axs[i].set_ylabel(f"{self.symbols[0]}", color=self.__colours[0], labelpad=10)
self.__axs[i].tick_params(axis='y', labelcolor=self.__colours[0])
self.__s2axs[i].set_ylabel(f"{self.symbols[1]}", color=self.__colours[1], labelpad=10)
self.__s2axs[i].tick_params(axis='y', labelcolor=self.__colours[1])
# Set Y label and limits for 3rd chart. Limits will be coefficients range from -1 to 1
self.__axs[2].set_ylabel('Coefficient', labelpad=10)
self.__axs[2].set_ylim([-1, 1])
# Set X labels to ''. Workaround as matplotlib is not leaving space for ticks
for ax in self.__axs:
ax.set_xlabel(" ", labelpad=10)
# Layout with padding between charts
self.__fig.tight_layout(pad=0.5)
# Create panel and sizer. This will provide scrollbar
panel = scrolled.ScrolledPanel(self, wx.ID_ANY)
sizer = wx.BoxSizer()
panel.SetSizer(sizer)
# Add fig to canvas and canvas to sizer. Thaw window to update
self.__canvas = FigureCanvas(panel, wx.ID_ANY, self.__fig)
sizer.Add(self.__canvas, 1, wx.ALL | wx.EXPAND)
self.Thaw()
# Setup scrolling
panel.SetupScrolling()
# Refresh to show content
self.refresh()
def refresh(self):
"""
Refresh the graph
:return:
"""
# Get the price data for the base coefficient calculation, tick data that was used to calculate last
# coefficient and and the coefficient history data
price_data = [self.GetMDIParent().cor.get_price_data(self.symbols[0]),
self.GetMDIParent().cor.get_price_data(self.symbols[1])]
tick_data = [self.GetMDIParent().cor.get_ticks(self.symbols[0], cache_only=True),
self.GetMDIParent().cor.get_ticks(self.symbols[1], cache_only=True)]
history_data = []
for timeframe in cfg.Config().get('monitor.calculations'):
frm = cfg.Config().get(f'monitor.calculations.{timeframe}.from')
history_data.append(self.GetMDIParent().cor.get_coefficient_history(
{'Symbol 1': self.symbols[0], 'Symbol 2': self.symbols[1], 'Timeframe': frm}))
# Check what data we have available
price_data_available = price_data is not None and len(price_data) == 2 and price_data[0] is not None and \
price_data[1] is not None and len(price_data[0]) > 0 and len(price_data[1]) > 0
tick_data_available = tick_data is not None and len(tick_data) == 2 and tick_data[0] is not None and \
tick_data[1] is not None and len(tick_data[0]) > 0 and len(tick_data[1]) > 0
history_data_available = history_data is not None and len(history_data) > 0
# Get all plots for coefficient history. History can contain multiple plots for different timeframes. They
# will all be plotted on the same chart.
times = []
coefficients = []
if history_data_available:
for hist in history_data:
times.append(hist['Date To'])
coefficients.append(hist['Coefficient'])
# Update graphs where we have data available
if price_data_available:
# Update range and ticks
xrange = [min(min(price_data[0]['time']), min(price_data[1]['time'])),
max(max(price_data[0]['time']), max(price_data[1]['time']))]
self.__axs[0].set_xlim(xrange)
# Plot both lines
self.__axs[0].plot(price_data[0]['time'], price_data[0]['close'],
color=self.__colours[0])
self.__s2axs[0].plot(price_data[1]['time'], price_data[1]['close'],
color=self.__colours[1])
# Ticks, labels and formats. Fixing xticks with FixedLocator but also using MaxNLocator to avoid
# cramped x-labels
if len(price_data[0]['time']) > 0:
self.__axs[0].xaxis.set_major_locator(mticker.MaxNLocator(10))
ticks_loc = self.__axs[0].get_xticks().tolist()
self.__axs[0].xaxis.set_major_locator(mticker.FixedLocator(ticks_loc))
self.__axs[0].set_xticklabels(ticks_loc)
self.__axs[0].xaxis.set_major_formatter(self.__tick_fmt_date)
plt.setp(self.__axs[0].xaxis.get_majorticklabels(), rotation=45)
if tick_data_available:
# Update range and ticks
xrange = [min(min(tick_data[0]['time']), min(tick_data[1]['time'])),
max(max(tick_data[0]['time']), max(tick_data[1]['time']))]
self.__axs[1].set_xlim(xrange)
# Plot both lines
self.__axs[1].plot(tick_data[0]['time'], tick_data[0]['ask'],
color=self.__colours[0])
self.__s2axs[1].plot(tick_data[1]['time'], tick_data[1]['ask'],
color=self.__colours[1])
if len(tick_data[0]['time']) > 0:
self.__axs[1].xaxis.set_major_locator(mticker.MaxNLocator(10))
ticks_loc = self.__axs[1].get_xticks().tolist()
self.__axs[1].xaxis.set_major_locator(mticker.FixedLocator(ticks_loc))
self.__axs[1].set_xticklabels(ticks_loc)
self.__axs[1].xaxis.set_major_formatter(self.__tick_fmt_time)
plt.setp(self.__axs[1].xaxis.get_majorticklabels(), rotation=45)
if history_data_available:
# Plot. There may be more than one set of data for chart. One for each coefficient date range. Convert
# single data to list, then loop to plot
xdata = times if isinstance(times, list) else [times, ]
ydata = coefficients if isinstance(coefficients, list) else [coefficients, ]
for i in range(0, len(xdata)):
self.__axs[2].scatter(xdata[i], ydata[i], s=1)
# Ticks, labels and formats. Fixing xticks with FixedLocator but also using MaxNLocator to avoid
# cramped x-labels
if len(times[0].array) > 0:
self.__axs[2].xaxis.set_major_locator(mticker.MaxNLocator(10))
ticks_loc = self.__axs[2].get_xticks().tolist()
self.__axs[2].xaxis.set_major_locator(mticker.FixedLocator(ticks_loc))
self.__axs[2].set_xticklabels(ticks_loc)
self.__axs[2].xaxis.set_major_formatter(self.__tick_fmt_time)
plt.setp(self.__axs[2].xaxis.get_majorticklabels(), rotation=45)
# Legend
self.__axs[2].legend([f"{cfg.Config().get('monitor.calculations.long.from')} Minutes",
f"{cfg.Config().get('monitor.calculations.medium.from')} Minutes",
f"{cfg.Config().get('monitor.calculations.short.from')} Minutes"])
# Lines showing divergence threshold. 2 if we are monitoring inverse correlations.
divergence_threshold = self.GetMDIParent().cor.divergence_threshold
monitor_inverse = self.GetMDIParent().cor.monitor_inverse
if divergence_threshold is not None:
self.__axs[2].axhline(y=divergence_threshold, color="red", label='_nolegend_', linewidth=1)
if monitor_inverse:
self.__axs[2].axhline(y=divergence_threshold * -1, color="red", label='_nolegend_', linewidth=1)
# Redraw canvas
self.__canvas.draw()
def __del__(self):
# Close all plots
plt.close('all')
@@ -0,0 +1,156 @@
import logging
import pandas as pd
import wx
import wx.grid
import mt5_correlation.gui.mdi as mdi
# Columns for diverged symbols table
COLUMN_INDEX = 0
COLUMN_SYMBOL = 1
COLUMN_NUM_DIVERGENCES = 2
class MDIChildDivergedSymbols(mdi.CorrelationMDIChild):
"""
Shows the status of all correlations that are within the monitoring threshold
"""
# The table and grid containing the symbols that form part of the diverged correlations. Defined at instance level
# to enable refresh.
__table = None
__grid = None
# Number of rows. Required for and updated by refresh method
__rows = 0
__log = None # The logger
def __init__(self, parent):
# Super
wx.MDIChildFrame.__init__(self, parent=parent, id=wx.ID_ANY, title="Diverged Symbols",
size=wx.Size(width=240, height=-1), style=wx.DEFAULT_FRAME_STYLE)
# Create logger
self.__log = logging.getLogger(__name__)
# Panel and sizer for table
panel = wx.Panel(self, wx.ID_ANY)
sizer = wx.BoxSizer(wx.VERTICAL)
panel.SetSizer(sizer)
# Create the grid. This is a data table using pandas dataframe for underlying data. Add the
# grid to the sizer.
self.__table = _DataTable(columns=self.GetMDIParent().cor.diverged_symbols.columns)
self.__grid = wx.grid.Grid(panel, wx.ID_ANY)
self.__grid.SetTable(self.__table, takeOwnership=True)
self.__grid.EnableEditing(False)
self.__grid.EnableDragRowSize(False)
self.__grid.EnableDragColSize(True)
self.__grid.EnableDragGridSize(True)
self.__grid.SetSelectionMode(wx.grid.Grid.SelectRows)
self.__grid.SetRowLabelSize(0)
self.__grid.SetColSize(COLUMN_INDEX, 0) # Index. Hide
self.__grid.SetColSize(COLUMN_SYMBOL, 100) # Symbol
self.__grid.SetColSize(COLUMN_NUM_DIVERGENCES, 100) # Num divergences
self.__grid.SetMinSize((220, 100))
sizer.Add(self.__grid, 1, wx.ALL | wx.EXPAND)
# Bind row doubleclick
self.Bind(wx.grid.EVT_GRID_CELL_LEFT_DCLICK, self.__on_doubleckick_row, self.__grid)
# Refresh to populate
self.refresh()
def refresh(self):
"""
Refreshes grid. Notifies if rows have been added or deleted.
:return:
"""
self.__log.debug(f"Refreshing grid.")
# Update data
self.__table.data = self.GetMDIParent().cor.diverged_symbols.copy()
# Start refresh
self.__grid.BeginBatch()
# Check if num rows in dataframe has changed, and send appropriate APPEND or DELETE messages
cur_rows = len(self.GetMDIParent().cor.diverged_symbols.index)
if cur_rows < self.__rows:
# Data has been deleted. Send message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_NOTIFY_ROWS_DELETED,
self.__rows - cur_rows, self.__rows - cur_rows)
self.__grid.ProcessTableMessage(msg)
elif cur_rows > self.__rows:
# Data has been added. Send message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_NOTIFY_ROWS_APPENDED,
cur_rows - self.__rows) # how many
self.__grid.ProcessTableMessage(msg)
self.__grid.EndBatch()
# Send updated message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_REQUEST_VIEW_GET_VALUES)
self.__grid.ProcessTableMessage(msg)
# Update row count
self.__rows = cur_rows
def __on_doubleckick_row(self, evt):
"""
Open the graphs when a row is doubleclicked.
:param evt:
:return:
"""
row = evt.GetRow()
symbol = self.__grid.GetCellValue(row, COLUMN_SYMBOL)
mdi.FrameManager.open_frame(parent=self.GetMDIParent(),
frame_module='mt5_correlation.gui.mdi_child_divergedgraph',
frame_class='MDIChildDivergedGraph',
raise_if_open=True,
symbol=symbol)
class _DataTable(wx.grid.GridTableBase):
"""
A data table that holds data in a pandas dataframe.
"""
data = None # The data for this table. A Pandas DataFrame
def __init__(self, columns):
wx.grid.GridTableBase.__init__(self)
self.headerRows = 1
self.data = pd.DataFrame(columns=columns)
def GetNumberRows(self):
return len(self.data)
def GetNumberCols(self):
return len(self.data.columns) + 1
def GetValue(self, row, col):
if row < self.RowsCount and col < self.ColsCount:
return self.data.index[row] if col == 0 else self.data.iloc[row, col - 1]
else:
raise Exception(f"Trying to access row {row} and col {col} which does not exist.")
def SetValue(self, row, col, value):
self.data.iloc[row, col - 1] = value
def GetColLabelValue(self, col):
if col == 0:
if self.data.index.name is None:
return 'Index'
else:
return self.data.index.name
return str(self.data.columns[col - 1])
def GetTypeName(self, row, col):
return wx.grid.GRID_VALUE_STRING
def GetAttr(self, row, col, prop):
attr = wx.grid.GridCellAttr()
return attr
@@ -0,0 +1,194 @@
import logging
import matplotlib.dates
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
import wx
import wx.lib.scrolledpanel as scrolled
import wxconfig as cfg
from matplotlib.backends.backend_wxagg import FigureCanvasWxAgg as FigureCanvas
from mpl_toolkits.axes_grid1 import host_subplot
from mpl_toolkits import axisartist
import mt5_correlation.gui.mdi as mdi
from mt5_correlation import correlation as cor
class MDIChildDivergedGraph(mdi.CorrelationMDIChild):
"""
Shows the graphs for the specified correlation
"""
symbol = None # Symbol to chart divergence for. Public as we use to check if window for the symbol is already open.
# Logger
__log = None
# Date formats for graphs
__tick_fmt_date = matplotlib.dates.DateFormatter('%d-%b')
__tick_fmt_time = matplotlib.dates.DateFormatter('%H:%M:%S')
# Graph Canvas
__canvas = None
# Colors for graph lines
__colours = matplotlib.cm.get_cmap(cfg.Config().get("charts.colormap")).colors
# We will store the other symbols last plotted. This will save us rebuilding teh figure if teh symbols haven't
# changed.
__other_symbols = None
def __init__(self, parent, **kwargs):
# Super
wx.MDIChildFrame.__init__(self, parent=parent, id=wx.ID_ANY,
title=f"Divergence Graph for {kwargs['symbol']}")
# Create logger
self.__log = logging.getLogger(__name__)
# Store the symbol
self.symbol = kwargs['symbol']
# Create panel and sizer
panel = scrolled.ScrolledPanel(self, wx.ID_ANY)
sizer = wx.BoxSizer()
panel.SetSizer(sizer)
# Create figure and canvas. Add canvas to sizer
self.__canvas = FigureCanvas(panel, wx.ID_ANY, plt.figure())
panel.GetSizer().Add(self.__canvas, 1, wx.ALL | wx.EXPAND)
panel.SetupScrolling()
# Refresh to show content
self.refresh()
def refresh(self):
"""
Refresh the graph
:return:
"""
# Get tick data for base symbol
symbol_tick_data = self.GetMDIParent().cor.get_ticks(self.symbol, cache_only=True)
# Get the other symbols and their tick data
other_symbols_data = self.__get_other_symbols_data()
# Delete all axes from the figure. They will need to be recreated as the symbols may be different
for axes in self.__canvas.figure.axes:
axes.remove()
# Plot for all other symbols
num_subplots = 1 if len(other_symbols_data.keys()) == 0 else len(other_symbols_data.keys())
plotnum = 1
axs = [] # Store the axes for sharing x axis
for other_symbol in other_symbols_data:
axs.append(self.__canvas.figure.add_subplot(num_subplots, 1, plotnum))
self.__plot(axes=axs[-1], base_symbol=self.symbol, other_symbol=other_symbol,
base_symbol_data=symbol_tick_data, other_symbol_data=other_symbols_data[other_symbol])
# Next plot
plotnum += 1
# Share x axis of the last axes with all the others
self.__share_xaxis(axs)
# Redraw canvas
self.__canvas.figure.tight_layout(pad=0.5)
self.__canvas.draw()
def __get_other_symbols_data(self):
"""
Gets the data required for the graphs
:param self:
:return: dict of other symbols their tick data
"""
# Get the symbols that this one has diverged against.
data = self.GetMDIParent().cor.filtered_coefficient_data
filtered_data = data.loc[(
(data['Status'] == cor.STATUS_DIVERGED) |
(data['Status'] == cor.STATUS_DIVERGING) |
(data['Status'] == cor.STATUS_CONVERGING)
) &
(
(data['Symbol 1'] == self.symbol) |
(data['Symbol 2'] == self.symbol)
)]
# Get all symbols and remove the base one. We will need to ensure that this is first in the list
other_symbols = list(filtered_data['Symbol 1'].append(filtered_data['Symbol 2']).drop_duplicates())
if self.symbol in other_symbols:
other_symbols.remove(self.symbol)
# Get the tick data other symbols and add to dict
other_tick_data = {}
for symbol in other_symbols:
tick_data = self.GetMDIParent().cor.get_ticks(symbol, cache_only=True)
other_tick_data[symbol] = tick_data
return other_tick_data
def __plot(self, axes, base_symbol, other_symbol, base_symbol_data, other_symbol_data):
"""
Plots the data on the axes
:param axes: The subplot to plot onto
:param base_symbol
:param other_symbol:
"""
# Create the other axes. Will need an axes for the base symbol data and another for the other symbol data
other_axes = axes.twinx()
# Set plot title and axis labels
axes.set_title(f"Tick Data for {base_symbol}:{other_symbol}")
axes.set_ylabel(base_symbol, color=self.__colours[0], labelpad=10)
other_axes.set_ylabel(other_symbol, color=self.__colours[1], labelpad=10)
# Set the tick and axis colors
self.__set_axes_color(axes, self.__colours[0], 'left')
self.__set_axes_color(other_axes, self.__colours[1], 'right')
# Plot both lines
axes.plot(base_symbol_data['time'], base_symbol_data['ask'], color=self.__colours[0])
other_axes.plot(other_symbol_data['time'], other_symbol_data['ask'], color=self.__colours[1])
@staticmethod
def __set_axes_color(axes, color, axis_loc='right'):
"""
Set the color for the axes, including axis line, ticks and tick labels
:param axes: The axes to set color for.
:param color: The color to set to
:param axis_loc: The location of the axis, label and ticks. Either left for base symbol or right for others
:return:
"""
# Change the color for the axis line, ticks and tick labels
axes.spines[axis_loc].set_color(color)
axes.tick_params(axis='y', colors=color)
def __share_xaxis(self, axs):
"""
Share the xaxis of the last axes with all other axes. Remove axis tick labels for all but the last. Format axis
tick labels for the last.
:param axs:
:return:
"""
if len(axs) > 0:
last_ax = axs[-1]
for ax in axs:
# If we are not on the last one, share and hide tick labels. If we are on the last one, format tick
# labels.
if ax != last_ax:
ax.sharex(last_ax)
plt.setp(ax.xaxis.get_majorticklabels(), visible=False)
else:
# Ticks, labels and formats. Fixing xticks with FixedLocator but also using MaxNLocator to avoid
# cramped x-labels
ax.xaxis.set_major_locator(mticker.MaxNLocator(10))
ticks_loc = ax.get_xticks().tolist()
ax.xaxis.set_major_locator(mticker.FixedLocator(ticks_loc))
ax.set_xticklabels(ticks_loc)
ax.xaxis.set_major_formatter(self.__tick_fmt_time)
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
def __del__(self):
# Close all plots
plt.close('all')
+41
View File
@@ -0,0 +1,41 @@
import definitions
import markdown
import wx
import wx.html
import mt5_correlation.gui.mdi as mdi
class MDIChildHelp(mdi.CorrelationMDIChild):
"""
Shows the README.md file
"""
def __init__(self, parent):
# Super
mdi.CorrelationMDIChild.__init__(self, parent=parent, id=wx.ID_ANY, pos=wx.DefaultPosition, title="Help",
size=wx.Size(width=800, height=-1),
style=wx.DEFAULT_FRAME_STYLE)
# Panel and sizer for help file
panel = wx.Panel(self, wx.ID_ANY)
sizer = wx.BoxSizer()
panel.SetSizer(sizer)
# HtmlWindow
html_widget = wx.html.HtmlWindow(parent=panel, id=wx.ID_ANY, style=wx.html.HW_SCROLLBAR_AUTO | wx.html.HW_NO_SELECTION)
sizer.Add(html_widget, 1, wx.ALL | wx.EXPAND)
# Load the help file, convert markdown to HTML and display. The markdown library doesnt understand the shell
# format so we will remove.
markdown_text = open(definitions.HELP_FILE).read()
markdown_text = markdown_text.replace("```shell", "```")
html = markdown.markdown(markdown_text)
html_widget.SetPage(html)
def refresh(self):
"""
Nothing to do on refresh. Help file doesnt change during outside of development.
:return:
"""
pass
+43
View File
@@ -0,0 +1,43 @@
import definitions
import wx
import wx.html
import mt5_correlation.gui.mdi as mdi
class MDIChildLog(mdi.CorrelationMDIChild):
"""
Shows the debug.log file
"""
__log_window = None # Widget to display log file in
def __init__(self, parent):
# Super
mdi.CorrelationMDIChild.__init__(self, parent=parent, id=wx.ID_ANY, pos=wx.DefaultPosition, title="Log",
size=wx.Size(width=800, height=200),
style=wx.DEFAULT_FRAME_STYLE)
# Panel and sizer for help file
panel = wx.Panel(self, wx.ID_ANY)
sizer = wx.BoxSizer()
panel.SetSizer(sizer)
# Log file window
self.__log_window = wx.TextCtrl(parent=panel, id=wx.ID_ANY, style=wx.HSCROLL | wx.TE_MULTILINE | wx.TE_READONLY)
sizer.Add(self.__log_window, 1, wx.ALL | wx.EXPAND)
# Refresh to populate
self.refresh()
def refresh(self):
"""
Refresh the log file
:return:
"""
# Load the help file
self.__log_window.LoadFile(definitions.LOG_FILE)
# Scroll to bottom
self.__log_window.SetInsertionPoint(-1)
+189
View File
@@ -0,0 +1,189 @@
import logging
import pandas as pd
import wx
import wx.grid
from mt5_correlation import correlation as cor
import mt5_correlation.gui.mdi as mdi
# Columns for coefficient table
COLUMN_INDEX = 0
COLUMN_SYMBOL1 = 1
COLUMN_SYMBOL2 = 2
COLUMN_BASE_COEFFICIENT = 3
COLUMN_DATE_FROM = 4
COLUMN_DATE_TO = 5
COLUMN_TIMEFRAME = 6
COLUMN_LAST_CALCULATION = 7
COLUMN_STATUS = 8
class MDIChildStatus(mdi.CorrelationMDIChild):
"""
Shows the status of all correlations that are within the monitoring threshold
"""
# The table and grid containing the status of correlations. Defined at instance level to enable refresh.
__table = None
__grid = None
# Number of rows. Required for and updated by refresh method
__rows = 0
__log = None # The logger
def __init__(self, parent):
# Super
wx.MDIChildFrame.__init__(self, parent=parent, id=wx.ID_ANY, title="Correlation Status",
size=wx.Size(width=440, height=-1), style=wx.DEFAULT_FRAME_STYLE)
# Create logger
self.__log = logging.getLogger(__name__)
# Panel and sizer for table
panel = wx.Panel(self, wx.ID_ANY)
sizer = wx.BoxSizer(wx.VERTICAL)
panel.SetSizer(sizer)
# Create the correlations grid. This is a data table using pandas dataframe for underlying data. Add the
# correlations_grid to the correlations sizer.
self.__table = _DataTable(columns=self.GetMDIParent().cor.filtered_coefficient_data.columns)
self.__grid = wx.grid.Grid(panel, wx.ID_ANY)
self.__grid.SetTable(self.__table, takeOwnership=True)
self.__grid.EnableEditing(False)
self.__grid.EnableDragRowSize(False)
self.__grid.EnableDragColSize(True)
self.__grid.EnableDragGridSize(True)
self.__grid.SetSelectionMode(wx.grid.Grid.SelectRows)
self.__grid.SetRowLabelSize(0)
self.__grid.SetColSize(COLUMN_INDEX, 0) # Index. Hide
self.__grid.SetColSize(COLUMN_SYMBOL1, 100) # Symbol 1
self.__grid.SetColSize(COLUMN_SYMBOL2, 100) # Symbol 2
self.__grid.SetColSize(COLUMN_BASE_COEFFICIENT, 100) # Base Coefficient
self.__grid.SetColSize(COLUMN_DATE_FROM, 0) # UTC Date From. Hide
self.__grid.SetColSize(COLUMN_DATE_TO, 0) # UTC Date To. Hide
self.__grid.SetColSize(COLUMN_TIMEFRAME, 0) # Timeframe. Hide.
self.__grid.SetColSize(COLUMN_LAST_CALCULATION, 0) # Last Calculation. Hide
self.__grid.SetColSize(COLUMN_STATUS, 100) # Status
self.__grid.SetMinSize((420, 500))
sizer.Add(self.__grid, 1, wx.ALL | wx.EXPAND)
# Bind row doubleclick
self.Bind(wx.grid.EVT_GRID_CELL_LEFT_DCLICK, self.__on_doubleckick_row, self.__grid)
# Refresh to populate
self.refresh()
def refresh(self):
"""
Refreshes grid. Notifies if rows have been added or deleted.
:return:
"""
self.__log.debug(f"Refreshing grid.")
# Update data
self.__table.data = self.GetMDIParent().cor.filtered_coefficient_data.copy()
# Format
self.__table.data.loc[:, 'Base Coefficient'] = self.__table.data['Base Coefficient'].map('{:.5f}'.format)
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.BeginBatch()
# Check if num rows in dataframe has changed, and send appropriate APPEND or DELETE messages
cur_rows = len(self.GetMDIParent().cor.filtered_coefficient_data.index)
if cur_rows < self.__rows:
# Data has been deleted. Send message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_NOTIFY_ROWS_DELETED,
self.__rows - cur_rows, self.__rows - cur_rows)
self.__grid.ProcessTableMessage(msg)
elif cur_rows > self.__rows:
# Data has been added. Send message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_NOTIFY_ROWS_APPENDED,
cur_rows - self.__rows) # how many
self.__grid.ProcessTableMessage(msg)
self.__grid.EndBatch()
# Send updated message
msg = wx.grid.GridTableMessage(self.__table, wx.grid.GRIDTABLE_REQUEST_VIEW_GET_VALUES)
self.__grid.ProcessTableMessage(msg)
# Update row count
self.__rows = cur_rows
def __on_doubleckick_row(self, evt):
"""
Open the graphs when a row is doubleclicked.
:param evt:
:return:
"""
row = evt.GetRow()
symbol1 = self.__grid.GetCellValue(row, COLUMN_SYMBOL1)
symbol2 = self.__grid.GetCellValue(row, COLUMN_SYMBOL2)
mdi.FrameManager.open_frame(parent=self.GetMDIParent(),
frame_module='mt5_correlation.gui.mdi_child_correlationgraph',
frame_class='MDIChildCorrelationGraph',
raise_if_open=True,
symbols=[symbol1, symbol2])
class _DataTable(wx.grid.GridTableBase):
"""
A data table that holds data in a pandas dataframe. Contains highlighting rules for status.
"""
data = None # The data for this table. A Pandas DataFrame
def __init__(self, columns):
wx.grid.GridTableBase.__init__(self)
self.headerRows = 1
self.data = pd.DataFrame(columns=columns)
def GetNumberRows(self):
return len(self.data)
def GetNumberCols(self):
return len(self.data.columns) + 1
def GetValue(self, row, col):
if row < self.RowsCount and col < self.ColsCount:
return self.data.index[row] if col == 0 else self.data.iloc[row, col - 1]
else:
raise Exception(f"Trying to access row {row} and col {col} which does not exist.")
def SetValue(self, row, col, value):
self.data.iloc[row, col - 1] = value
def GetColLabelValue(self, col):
if col == 0:
if self.data.index.name is None:
return 'Index'
else:
return self.data.index.name
return str(self.data.columns[col - 1])
def GetTypeName(self, row, col):
return wx.grid.GRID_VALUE_STRING
def GetAttr(self, row, col, prop):
attr = wx.grid.GridCellAttr()
# Check that we are not out of bounds
if row < self.RowsCount:
# If column is status, check and highlight if diverging or converging.
if col in [COLUMN_STATUS]:
# Is status one of interest
value = self.GetValue(row, col)
if value != "":
if value in [cor.STATUS_DIVERGING]:
attr.SetBackgroundColour(wx.RED)
elif value in [cor.STATUS_CONVERGING]:
attr.SetBackgroundColour(wx.GREEN)
else:
attr.SetBackgroundColour(wx.WHITE)
return attr
+6 -3
View File
@@ -1,8 +1,11 @@
pandas==1.2.1
matplotlib==3.3.4
matplotlib==3.4.2
MetaTrader5==5.0.34
pytz==2021.1
scipy==1.6.0
logging==0.4.9.6
pyyaml==5.4.1
wxpython==4.1.1
markdown==3.3.4
wxpython==4.1.1
mock==4.0.3
numpy==1.20.0
wxconfig==1.2
-49
View File
@@ -1,49 +0,0 @@
import unittest
from mt5_correlation.config import Config
class TestConfig(unittest.TestCase):
def test_load_and_get(self):
config = Config()
config.load("testconfig.yaml")
val121 = config.get('test1.test1_2.val1_2_1')
self.assertEqual(val121, 'val1_2_1', "Get returned incorrect value.")
def test_set_and_save(self):
config = Config()
config.load("testconfig.yaml")
path = 'test1.test1_2.val1_2_1'
# Save new value, storing orig so we can restore later
orig_value = config.get(path)
config.set(path, "newval")
config.save()
# Reopen config and get value to see if it is previously saved value
config = Config()
config.load("testconfig.yaml")
saved_value = config.get(path)
self.assertEqual(saved_value, 'newval', "New value was not saved and returned.")
# Restore and save file
config.set(path, orig_value)
config.save()
def test_get_root_nodes(self):
config = Config()
config.load("testconfig.yaml")
# Get root nodes
root_nodes = config.get_root_nodes()
# There should be 2
self.assertTrue(len(root_nodes) == 2, "There should be 2 root nodes.")
# The first should be test1 and the second should be test2
self.assertEqual(root_nodes[0], 'test1', "First root node should be 'test1'.")
self.assertEqual(root_nodes[1], 'test2', "Second root node should be 'test2'.")
if __name__ == '__main__':
unittest.main()
+362
View File
@@ -0,0 +1,362 @@
import unittest
from unittest.mock import patch, PropertyMock
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),
Symbol(name='SYMBOL5', 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)
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
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]]))
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):
"""
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(monitoring_threshold=1, monitor_inverse=True)
# 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)
# 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_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 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.")
# 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 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]
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), 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 for 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.")
# 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):
"""
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]
# 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.")
# 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)
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 coefficient history when using an
interval of 1 second.
:param mock:
:return:
"""
# Mock symbol return values
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, monitor_inverse=True)
# 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_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)
# 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_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_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_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
# 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},
{'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)
# Stop the monitor
cor.stop_monitor()
# 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 DIVERGED for SYMBOL1:SYMBOL2 and CORRELATED for SYMBOL1:SYMBOL4 and SYMBOL2:SYMBOL4.
self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL2') == correlation.STATUS_DIVERGED)
self.assertTrue(cor.get_last_status('SYMBOL1', 'SYMBOL4') == correlation.STATUS_CORRELATED)
self.assertTrue(cor.get_last_status('SYMBOL2', 'SYMBOL4') == correlation.STATUS_CORRELATED)
# We are monitoring inverse correlations, status for SYMBOL1:SYMBOL5 should be DIVERGED
self.assertTrue(cor.get_last_status('SYMBOL2', 'SYMBOL5') == correlation.STATUS_DIVERGED)
@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, 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)
# 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, 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()
# 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")
@patch('mt5_correlation.correlation.Correlation.coefficient_data', new_callable=PropertyMock)
def test_diverged_symbols(self, mock):
"""
Test that diverged_symbols property correctly groups symbols and counts.
:param mock:
:return:
"""
# Correlation class
cor = correlation.Correlation()
# Mock the correlation data. Symbol 1 has diverged 3 times; symbols 2 has diverged twice; symbol 3 has
# diverged once; symbols 4 has diverged twice and symbol 5 has not diverged at all. Use all diverged status'
# (diverged, diverging & converging). Also add a row for a non diverged pair.
mock.return_value = pd.DataFrame(columns=['Symbol 1', 'Symbol 2', 'Status'], data=[
['SYMBOL1', 'SYMBOL2', correlation.STATUS_DIVERGED],
['SYMBOL1', 'SYMBOL3', correlation.STATUS_DIVERGING],
['SYMBOL1', 'SYMBOL4', correlation.STATUS_CONVERGING],
['SYMBOL2', 'SYMBOL4', correlation.STATUS_DIVERGED],
['SYMBOL2', 'SYMBOL3', correlation.STATUS_CORRELATED]])
# Get the diverged_symbols data and check the counts
diverged_symbols = cor.diverged_symbols
self.assertEqual(diverged_symbols.loc[(diverged_symbols['Symbol'] == 'SYMBOL1')]['Count'].iloc[0], 3,
"Symbol 1 has diverged three times.")
self.assertEqual(diverged_symbols.loc[(diverged_symbols['Symbol'] == 'SYMBOL2')]['Count'].iloc[0], 2,
"Symbol 2 has diverged twice.")
self.assertEqual(diverged_symbols.loc[(diverged_symbols['Symbol'] == 'SYMBOL3')]['Count'].iloc[0], 1,
"Symbol 3 has diverged once.")
self.assertEqual(diverged_symbols.loc[(diverged_symbols['Symbol'] == 'SYMBOL4')]['Count'].iloc[0], 2,
"Symbol 4 has diverged twice.")
self.assertFalse('SYMBOL5' in diverged_symbols['Symbol'])
if __name__ == '__main__':
unittest.main()
+77
View File
@@ -0,0 +1,77 @@
import unittest
from unittest.mock import patch
import mt5_correlation.mt5 as mt5
import pandas as pd
from datetime import datetime
class Symbol:
""" A Mock symbol class"""
name = None
visible = None
def __init__(self, name, visible):
self.name = name
self.visible = visible
class TestMT5(unittest.TestCase):
"""
Unit test for MT5. Uses mock to mock MetaTrader5 connection.
"""
# Mock symbols. 5 Symbols, 4 visible.
mock_symbols = [Symbol(name='SYMBOL1', visible=True),
Symbol(name='SYMBOL2', visible=True),
Symbol(name='SYMBOL3', visible=False),
Symbol(name='SYMBOL4', visible=True),
Symbol(name='SYMBOL5', visible=True)]
# Mock prices for symbol 1
mock_prices = pd.DataFrame(columns=['time', 'close'],
data=[[datetime(2021, 1, 1, 1, 5, 0), 123.123],
[datetime(2021, 1, 1, 1, 10, 0), 123.124],
[datetime(2021, 1, 1, 1, 15, 0), 123.125],
[datetime(2021, 1, 1, 1, 20, 0), 125.126],
[datetime(2021, 1, 1, 1, 25, 0), 123.127],
[datetime(2021, 1, 1, 1, 30, 0), 123.128]])
@patch('mt5_correlation.mt5.MetaTrader5')
def test_get_symbols(self, mock):
# Mock return value
mock.symbols_get.return_value = self.mock_symbols
# Call get_symbols
symbols = mt5.MT5().get_symbols()
# There should be four, as one is set as not visible
self.assertTrue(len(symbols) == 4, "There should be 5 symbols returned from MT5.")
@patch('mt5_correlation.mt5.MetaTrader5')
def test_get_prices(self, mock):
# Mock return value
mock.copy_rates_range.return_value = self.mock_prices
# Call get prices
prices = mt5.MT5().get_prices(symbol='SYMBOL1', from_date='01-JAN-2021 01:00:00',
to_date='01-JAN-2021 01:10:25', timeframe=mt5.TIMEFRAME_M5)
# There should be 6
self.assertTrue(len(prices.index) == 6, "There should be 6 prices.")
@patch('mt5_correlation.mt5.MetaTrader5')
def test_get_ticks(self, mock):
# Mock return value
mock.copy_ticks_range.return_value = self.mock_prices
# Call get ticks
ticks = mt5.MT5().get_ticks(symbol='SYMBOL1', from_date='01-JAN-2021 01:00:00', to_date='01-JAN-2021 01:10:25')
# There should be 6
self.assertTrue(len(ticks.index) == 6, "There should be 6 prices.")
if __name__ == '__main__':
unittest.main()
-20
View File
@@ -1,20 +0,0 @@
---
test1:
test1_1:
val1_1_1: val1_1_1
val1_1_2: val1_1_2
val1_1_3: val1_1_3
test1_2:
val1_2_1: val1_2_1
val1_2_2: val1_2_2
val1_2_3: val1_2_3
test2:
test2_1:
val2_1_1: val1_1_1
val2_1_2: val1_1_2
val2_1_3: val1_1_3
test2_2:
val2_2_1: val1_2_1
val2_2_2: val1_2_2
val2_2_3: val1_2_3
...