This repository has been archived by the owner on Jun 8, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 152
/
MasterMoniGoManiHyperStrategy.py
1522 lines (1299 loc) · 88.7 KB
/
MasterMoniGoManiHyperStrategy.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# -*- coding: utf-8 -*-
# --- ↓ Do not remove these libs ↓ -------------------------------------------------------------------------------------
import copy
import json
import logging
import math
import os
import sys
from abc import ABC
from datetime import datetime, timedelta
from functools import reduce
from typing import Any, Dict, List, Optional, Union
import numpy as np # noqa
import pandas as pd # noqa
import talib.abstract as ta
from numpy import timedelta64
from pandas import DataFrame
from scipy.interpolate import interp1d
from yaml import full_load
from freqtrade.data.history import load_pair_history
from freqtrade.enums import RunMode
from freqtrade.exchange import timeframe_to_prev_date
from freqtrade.misc import deep_merge_dicts, round_dict
from freqtrade.optimize.space import Categorical, Dimension, Integer, SKDecimal
from freqtrade.persistence import Trade
from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, merge_informative_pair, timeframe_to_minutes
logger = logging.getLogger(__name__)
# --- ↑ Do not remove these libs ↑ -------------------------------------------------------------------------------------
class MasterMoniGoManiHyperStrategy(IStrategy, ABC):
"""
####################################################################################
#### ####
### MoniGoMani Master Framework v0.13.0 by Rikj000 ###
## ----------------------------- ##
# Isn't that what we all want? Our money to go many? #
# Well that's what this Freqtrade framework hopes to do for you! #
## By giving you/HyperOpt a lot of signals to alter the weight from ##
### ------------------------------------------------------ ###
## Big thank you to xmatthias and everyone who helped on MoniGoMani, ##
## Freqtrade Discord support was also really helpful so thank you too! ##
### ------------------------------------------------------- ###
## Disclaimer: This strategy is under development. ##
# I do not recommend running it live until further development/testing. #
## TEST IT BEFORE USING IT! ##
### ▄▄█▀▀▀▀▀█▄▄ ###
## ------------------------------------- ▄█▀ ▄ ▄ ▀█▄ ##
### If you like my work, feel free to donate or use one of █ ▀█▀▀▀▀▄ █ ###
## my referral links, that would also greatly be appreciated █ █▄▄▄▄▀ █ ##
# ICONOMI: https://www.iconomi.com/register?ref=zQQPK █ █ █ █ #
## Binance: https://www.binance.com/en/register?ref=97611461 ▀█▄ ▀▀█▀█▀ ▄█▀ ##
### BTC: 19LL2LCMZo4bHJgy15q1Z1bfe7mV4bfoWK ▀▀█▄▄▄▄▄█▀▀ ###
#### ####
####################################################################################
"""
# MGM trend names
mgm_trends = ['downwards', 'sideways', 'upwards']
# Initialize empty buy/sell/protection_params dictionaries
buy_params = {}
sell_params = {}
protection_params = {}
# Load the MoniGoMani config names from '.hurry'
mgm_config_name = mgm_config_hyperopt_name = None
hurry_config_path = f'{os.getcwd()}/.hurry'
if os.path.isfile(hurry_config_path) is True:
with open(hurry_config_path, 'r') as yml_file:
config = full_load(yml_file) or {}
if 'config' in config:
hurry_config = config['config']
mgm_config_name = hurry_config['mgm_config_names']['mgm-config']
mgm_config_hyperopt_name = hurry_config['mgm_config_names']['mgm-config-hyperopt']
if (mgm_config_name is None) or (mgm_config_hyperopt_name is None):
sys.exit('MoniGoManiHyperStrategy - ERROR - The MoniGoMani Configuration filenames could not be loaded from'
'".hurry"... Please run "python3 ./mgm-hurry setup" to create your ".hurry" file')
# Load the MoniGoMani settings
mgm_config_path = f'{os.getcwd()}/user_data/{mgm_config_name}'
if os.path.isfile(mgm_config_path) is True:
# Load the 'mgm-config.json' file as an object and parse it as a dictionary
file_object = open(mgm_config_path, )
json_data = json.load(file_object)
mgm_config = json_data['monigomani_settings']
else:
sys.exit(f'MoniGoManiHyperStrategy - ERROR - The main MoniGoMani configuration file "mgm-config" can\'t '
f'be found at: {mgm_config_path}... Please provide the correct file and/or alter "mgm_config_name" in '
f'".hurry"')
# Apply the loaded MoniGoMani Settings
try:
backtest_timeframe = mgm_config['timeframes']['backtest_timeframe']
core_trend_timeframe = mgm_config['timeframes']['core_trend_timeframe']
roi_timeframe = mgm_config['timeframes']['roi_timeframe']
timeframe = mgm_config['timeframes']['timeframe']
startup_candle_count = mgm_config['startup_candle_count']
precision = mgm_config['precision']
min_weighted_signal_value = mgm_config['weighted_signal_spaces']['min_weighted_signal_value']
max_weighted_signal_value = mgm_config['weighted_signal_spaces']['max_weighted_signal_value']
min_trend_total_signal_needed_value = mgm_config[
'weighted_signal_spaces']['min_trend_total_signal_needed_value']
min_trend_total_signal_needed_candles_lookback_window_value = mgm_config[
'weighted_signal_spaces']['min_trend_total_signal_needed_candles_lookback_window_value']
max_trend_total_signal_needed_candles_lookback_window_value = mgm_config[
'weighted_signal_spaces']['max_trend_total_signal_needed_candles_lookback_window_value']
min_trend_signal_triggers_needed_value = mgm_config[
'weighted_signal_spaces']['min_trend_signal_triggers_needed']
search_threshold_weighted_signal_values = mgm_config[
'weighted_signal_spaces']['search_threshold_weighted_signal_values']
search_threshold_trend_total_signal_needed_candles_lookback_window_value = mgm_config[
'weighted_signal_spaces']['search_threshold_trend_total_signal_needed_candles_lookback_window_value']
search_threshold_trend_signal_triggers_needed = mgm_config[
'weighted_signal_spaces']['search_threshold_trend_signal_triggers_needed']
roi_delay = mgm_config['roi_spaces']['roi_delay']
roi_table_step_size = mgm_config['roi_spaces']['roi_table_step_size']
roi_time_interval_scaling = mgm_config['roi_spaces']['roi_time_interval_scaling']
roi_value_step_scaling = mgm_config['roi_spaces']['roi_value_step_scaling']
roi_when_downwards = mgm_config['roi_spaces']['roi_when_downwards']
roi_when_sideways = mgm_config['roi_spaces']['roi_when_sideways']
roi_when_upwards = mgm_config['roi_spaces']['roi_when_upwards']
stoploss_min_value = mgm_config['stoploss_spaces']['stoploss_min_value']
stoploss_max_value = mgm_config['stoploss_spaces']['stoploss_max_value']
trailing_stop_positive_min_value = mgm_config['stoploss_spaces']['trailing_stop_positive_min_value']
trailing_stop_positive_max_value = mgm_config['stoploss_spaces']['trailing_stop_positive_max_value']
trailing_stop_positive_offset_min_value = mgm_config[
'stoploss_spaces']['trailing_stop_positive_offset_min_value']
trailing_stop_positive_offset_max_value = mgm_config[
'stoploss_spaces']['trailing_stop_positive_offset_max_value']
mgm_unclogger_params = mgm_config['unclogger_spaces']
mgm_protection_params = mgm_config['protection_spaces']
minimal_roi = mgm_config['default_stub_values']['minimal_roi']
stoploss = mgm_config['default_stub_values']['stoploss']
trailing_stop = mgm_config['default_stub_values']['trailing_stop']
trailing_stop_positive = mgm_config['default_stub_values']['trailing_stop_positive']
trailing_stop_positive_offset = mgm_config['default_stub_values']['trailing_stop_positive_offset']
trailing_only_offset_is_reached = mgm_config['default_stub_values']['trailing_only_offset_is_reached']
debuggable_weighted_signal_dataframe = mgm_config['debuggable_weighted_signal_dataframe']
use_mgm_logging = mgm_config['use_mgm_logging']
mgm_log_levels_enabled = mgm_config['mgm_log_levels_enabled']
except KeyError as missing_setting:
sys.exit(f'MoniGoManiHyperStrategy - ERROR - '
f'The main MoniGoMani configuration file "mgm-config" is missing some settings.'
f'\nPlease make sure that all MoniGoMani related settings are existing inside this file!'
f'\n{missing_setting} has been detected as missing from the file...'
f'\nCompare with the latest "mgm-config.example" to see if you are up to date with the latest settings'
f': \nhttps://github.com/Rikj000/MoniGoMani/blob/development/user_data/mgm-config.example.json')
# If results from a previous HyperOpt Run are found then continue the next HyperOpt Run upon them
mgm_config_hyperopt_path = f'{os.getcwd()}/user_data/{mgm_config_hyperopt_name}'
if os.path.isfile(mgm_config_hyperopt_path) is True:
# Try to load the previous 'mgm-config-hyperopt' file as an object and parse it as a dictionary
# if the parse fails, warn and continue as if it didn't exist.
try:
file_object = open(mgm_config_hyperopt_path, )
mgm_config_hyperopt = json.load(file_object)
except ValueError as e:
mgm_config_hyperopt = {}
logger.warning(f'MoniGoManiHyperStrategy - WARN - {mgm_config_hyperopt_path} is inaccessible or is '
f'not valid JSON, disregarding existing "mgm-config-hyperopt" file and '
f'treating as first hyperopt run!')
# Convert the loaded 'mgm-config-hyperopt' data to the needed HyperOpt Results format if it's found
# Default stub values from 'mgm-config' are used otherwise.
if mgm_config_hyperopt != {}:
for space in mgm_config_hyperopt['params']:
if space in ['buy', 'sell', 'protection']:
for param, param_value in mgm_config_hyperopt['params'][space].items():
if param.startswith('buy'):
buy_params[param] = param_value
elif param.startswith('sell'):
sell_params[param] = param_value
else:
protection_params[param] = param_value
if space == 'roi':
minimal_roi = mgm_config_hyperopt['params'][space]
if space == 'stoploss':
stoploss = mgm_config_hyperopt['params'][space][space]
if space == 'trailing':
trailing_stop = mgm_config_hyperopt['params'][space]['trailing_stop']
trailing_stop_positive = mgm_config_hyperopt['params'][space]['trailing_stop_positive']
trailing_stop_positive_offset = mgm_config_hyperopt[
'params'][space]['trailing_stop_positive_offset']
trailing_only_offset_is_reached = mgm_config_hyperopt[
'params'][space]['trailing_only_offset_is_reached']
else:
mgm_config_hyperopt = {}
# Create dictionary to store custom information MoniGoMani will be using in RAM
initial_custom_info: dict = {'open_trades': {}, 'unclogger_cooldown_pairs': {}}
custom_info: dict = copy.deepcopy(initial_custom_info)
# Initialize some parameters which will be automatically configured/used by MoniGoMani
use_custom_stoploss = True # Leave this enabled (Needed for open_trade custom_information_storage)
is_dry_live_run_detected = True # Class level runmode detection, Gets set automatically
informative_timeframe = timeframe # Gets set automatically
timeframe_multiplier = None # Gets set automatically
separator = 1.5 # Gets set automatically
separator_candle_weight_reducer = 0.03 # Gets set automatically
# Initialize comparison values to check if total signals utilized by HyperOpt are possible
total_signals_possible = {}
total_triggers_possible = {}
for trend in mgm_trends:
for space in ['buy', 'sell']:
total_signals_possible[f'{space}_{trend}'] = 0
total_triggers_possible[f'{space}_{trend}'] = 0
class HyperOpt:
@staticmethod
def generate_roi_table(params: Dict) -> Dict[int, float]:
"""
Generates a Custom Long Continuous ROI (Return of Interest) Table with fewer gaps in it.
Configurable step_size is loaded in from the Master MGM Framework.
:param params: (Dict) Base Parameters used for the ROI Table calculation
:return Dict: Generated ROI Table
"""
step = MasterMoniGoManiHyperStrategy.roi_table_step_size
minimal_roi = {0: params['roi_p1'] + params['roi_p2'] + params['roi_p3'],
params['roi_t3']: params['roi_p1'] + params['roi_p2'],
params['roi_t3'] + params['roi_t2']: params['roi_p1'],
params['roi_t3'] + params['roi_t2'] + params['roi_t1']: 0}
max_value = max(map(int, minimal_roi.keys()))
min_value = MasterMoniGoManiHyperStrategy.roi_delay
f = interp1d(list(map(int, minimal_roi.keys())), list(minimal_roi.values()))
x = list(range(min_value, max_value, step))
y = list(map(float, map(f, x)))
if y[-1] != 0:
x.append(x[-1] + step)
y.append(0)
return dict(zip(x, y))
@staticmethod
def roi_space() -> List[Dimension]:
"""
Create a ROI (Return of Interest) space. Define values to search for each ROI steps.
This method implements adaptive ROI HyperSpace with varied ranges for parameters which automatically adapts
to the un-zoomed base_weighted_signal_timeframe used by the MGM Framework during BackTesting & HyperOpting.
:return List: Generated ROI Space
"""
# Default scaling coefficients for the ROI HyperSpace. Can be changed to adjust resulting ranges of the ROI
# tables. Increase if you need wider ranges in the ROI HyperSpace, decrease if shorter ranges are needed:
# roi_t_alpha: Limits for the time intervals in the ROI Tables. Components are scaled linearly.
roi_t_alpha = MasterMoniGoManiHyperStrategy.roi_time_interval_scaling
# roi_p_alpha: Limits for the ROI value steps. Components are scaled logarithmically.
roi_p_alpha = MasterMoniGoManiHyperStrategy.roi_value_step_scaling
# Load in the ROI timeframe size from the Master MGM Framework
timeframe_min = timeframe_to_minutes(MasterMoniGoManiHyperStrategy.roi_timeframe)
# The scaling is designed so that it maps exactly to the legacy Freqtrade roi_space()
# method for the 5m timeframe.
roi_t_scale = timeframe_min / 5
roi_p_scale = math.log1p(timeframe_min) / math.log1p(5)
roi_limits = {'roi_t1_min': int(10 * roi_t_scale * roi_t_alpha),
'roi_t1_max': int(120 * roi_t_scale * roi_t_alpha),
'roi_t2_min': int(10 * roi_t_scale * roi_t_alpha),
'roi_t2_max': int(60 * roi_t_scale * roi_t_alpha),
'roi_t3_min': int(10 * roi_t_scale * roi_t_alpha),
'roi_t3_max': int(40 * roi_t_scale * roi_t_alpha),
'roi_p1_min': 0.01 * roi_p_scale * roi_p_alpha,
'roi_p1_max': 0.04 * roi_p_scale * roi_p_alpha,
'roi_p2_min': 0.01 * roi_p_scale * roi_p_alpha,
'roi_p2_max': 0.07 * roi_p_scale * roi_p_alpha,
'roi_p3_min': 0.01 * roi_p_scale * roi_p_alpha,
'roi_p3_max': 0.20 * roi_p_scale * roi_p_alpha}
# Generate MGM's custom long continuous ROI table
logger.debug(f'Using ROI space limits: {roi_limits}')
p = {'roi_t1': roi_limits['roi_t1_min'], 'roi_t2': roi_limits['roi_t2_min'],
'roi_t3': roi_limits['roi_t3_min'], 'roi_p1': roi_limits['roi_p1_min'],
'roi_p2': roi_limits['roi_p2_min'], 'roi_p3': roi_limits['roi_p3_min']}
logger.info(f'Min ROI table: {round_dict(MasterMoniGoManiHyperStrategy.HyperOpt.generate_roi_table(p), 3)}')
p = {'roi_t1': roi_limits['roi_t1_max'], 'roi_t2': roi_limits['roi_t2_max'],
'roi_t3': roi_limits['roi_t3_max'], 'roi_p1': roi_limits['roi_p1_max'],
'roi_p2': roi_limits['roi_p2_max'], 'roi_p3': roi_limits['roi_p3_max']}
logger.info(f'Max ROI table: {round_dict(MasterMoniGoManiHyperStrategy.HyperOpt.generate_roi_table(p), 3)}')
return [Integer(roi_limits['roi_t1_min'], roi_limits['roi_t1_max'], name='roi_t1'),
Integer(roi_limits['roi_t2_min'], roi_limits['roi_t2_max'], name='roi_t2'),
Integer(roi_limits['roi_t3_min'], roi_limits['roi_t3_max'], name='roi_t3'),
SKDecimal(roi_limits['roi_p1_min'], roi_limits['roi_p1_max'], decimals=3, name='roi_p1'),
SKDecimal(roi_limits['roi_p2_min'], roi_limits['roi_p2_max'], decimals=3, name='roi_p2'),
SKDecimal(roi_limits['roi_p3_min'], roi_limits['roi_p3_max'], decimals=3, name='roi_p3')]
@staticmethod
def stoploss_space() -> List[Dimension]:
"""
Define custom stoploss search space with configurable parameters for the Stoploss Value to search.
Override it if you need some different range for the parameter in the 'stoploss' optimization hyperspace.
:return List: Generated Stoploss Space
"""
return [SKDecimal(MasterMoniGoManiHyperStrategy.stoploss_max_value,
MasterMoniGoManiHyperStrategy.stoploss_min_value, decimals=3, name='stoploss')]
@staticmethod
def trailing_space() -> List[Dimension]:
"""
Define custom trailing search space with parameters configurable in 'mgm-config'
:return List: Generated Trailing Space
"""
return [
# It was decided to always set trailing_stop is to True if the 'trailing' hyperspace is used.
# Otherwise, hyperopt will vary other parameters that won't have effect if
# trailing_stop is set False.
# This parameter is included into the hyperspace dimensions rather than assigning
# it explicitly in the code in order to have it printed in the results along with
# other 'trailing' hyperspace parameters.
Categorical([True], name='trailing_stop'),
SKDecimal(MasterMoniGoManiHyperStrategy.trailing_stop_positive_min_value,
MasterMoniGoManiHyperStrategy.trailing_stop_positive_max_value,
decimals=3, name='trailing_stop_positive'),
# 'trailing_stop_positive_offset' should be greater than 'trailing_stop_positive',
# so this intermediate parameter is used as the value of the difference between
# them. The value of the 'trailing_stop_positive_offset' is constructed in the
# generate_trailing_params() method.
# This is similar to the hyperspace dimensions used for constructing the ROI tables.
SKDecimal(MasterMoniGoManiHyperStrategy.trailing_stop_positive_offset_min_value,
MasterMoniGoManiHyperStrategy.trailing_stop_positive_offset_max_value,
decimals=3, name='trailing_stop_positive_offset_p1'),
Categorical([True, False], name='trailing_only_offset_is_reached')]
# Initialize the hyperoptable freqtrade protections
@property
def protections(self):
protections = []
# Generates the hyperoptable protections with the correct spaces
for protection_config in self.mgm_protection_params:
protection = {}
if isinstance(protection_config, dict) is True:
for protection_parameter_key in protection_config:
protection_parameter_value = protection_config[protection_parameter_key]
# If the protection parameter value is a dict, then fetch the hyperoptable parameter value from MGM
if isinstance(protection_parameter_value, dict) is True:
protection_parameter_name = (
f'{protection_config["method"]}_{protection_parameter_key}' if 'id' not in protection_config
else f'{protection_config["method"]}_{protection_config["id"]}_{protection_parameter_key}')
protection_parameter = getattr(self, f'protection_{protection_parameter_name}')
protection[protection_parameter_key] = protection_parameter.value
# Else just append the protection parameter value
elif protection_parameter_key != 'id':
protection[protection_parameter_key] = protection_parameter_value
protections.append(protection)
return protections
def __init__(self, config: dict):
"""
First method to be called once during the MoniGoMani class initialization process
:param config: (dict)
"""
i = 'Initialization'
if RunMode(config.get('runmode', RunMode.OTHER)) in (RunMode.BACKTEST, RunMode.HYPEROPT):
self.timeframe = self.backtest_timeframe
self.mgm_logger('info', 'TimeFrame-Zoom', f'Auto updating to zoomed "backtest_timeframe": {self.timeframe}')
self.is_dry_live_run_detected = False
self.mgm_logger('info', i, 'Current run mode detected as: HyperOpting/BackTesting. '
'Auto updated is_dry_live_run_detected to: False')
self.mgm_logger('info', i, 'Calculating and storing "timeframe_multiplier"')
self.timeframe_multiplier = int(timeframe_to_minutes(self.informative_timeframe)
/ timeframe_to_minutes(self.timeframe))
if self.timeframe_multiplier < 1:
raise SystemExit('MoniGoManiHyperStrategy - ERROR - TimeFrame-Zoom - '
'"timeframe" must be bigger than "backtest_timeframe"')
else:
if os.path.isfile(self.mgm_config_hyperopt_path) is False:
sys.exit(f'MoniGoManiHyperStrategy - ERROR - The MoniGoMani HyperOpt Results configuration file '
f'({self.mgm_config_hyperopt_name}) can\'t be found at: {self.mgm_config_hyperopt_path}... '
f'Please Optimize your MoniGoMani before Dry/Live running! Once optimized provide the correct '
f'file and/or alter "mgm_config_names" in ".hurry"')
self.is_dry_live_run_detected = True
self.mgm_logger('info', i, 'Current run mode detected as: Dry/Live-Run. '
'Auto updated is_dry_live_run_detected to: True')
if self.mgm_config['unclogger_spaces']['unclogger_enabled'] is True:
self.separator = self.mgm_config['unclogger_spaces'][
'unclogger_trend_lookback_candles_window_recent_past_weight_separator']
separator_window = (self.separator / 1) - (1 / self.separator)
trend_lookback_candles_window = self.get_param_value('sell___unclogger_trend_lookback_candles_window')
self.separator_candle_weight_reducer = (separator_window / trend_lookback_candles_window)
super().__init__(config)
@staticmethod
def populate_frequi_plots(weighted_signal_plots: dict) -> dict:
"""
Merges the Weighted Signal Plots together with the Buy/Sell Signal Plots and the Trend Detection Plots for
a nice visualization in FreqUI
:param weighted_signal_plots: FreqUI plotting data used for weighted signals (and their indicators)
:return dict: Complete FreqUI plotting data containing weighted signal + other MGM Framework plotting
"""
# Plot configuration to show all signals used in MoniGoMani in FreqUI (Use load from Strategy in FreqUI)
framework_plots = {
# Main Plots - Trend Indicator (SAR)
'main_plot': {
'sar': {'color': '#2c05f6'}
},
# Sub Plots - Each dict defines one additional plot
'subplots': {
# Sub Plots - Trend Detection
'MoniGoMani Core Trend': {
'mgm_trend': {'color': '#7fba3c'}
},
'Hilbert Transform (Trend vs Cycle)': {
'ht_trendmode': {'color': '#6f1a7b'}
},
# Sub Plots - Final Buy + Sell Signals
'Buy + Sell Signals Firing': {
'buy': {'color': '#09d528'},
'sell': {'color': '#d19e28'}
},
'Total Buy + Sell Signal Strength': {
'total_buy_signal_strength': {'color': '#09d528'},
'total_sell_signal_strength': {'color': '#d19e28'}
},
'Weighted Buy + Sell Signals Firing': {
'buy_signals_triggered': {'color': '#09d528'},
'sell_signals_triggered': {'color': '#d19e28'}
}
}
}
return deep_merge_dicts(framework_plots, weighted_signal_plots)
def _populate_core_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds the core indicators used to define trends to the strategy engine.
:param dataframe: (DataFrame) DataFrame with data from the exchange
:param metadata: (dict) Additional information, like the currently traded pair
:return: a Dataframe with all core trend indicators for MoniGoMani
"""
# Momentum Indicators
# -------------------
# Hilbert Transform - Trend vs Cycle
dataframe['ht_trendmode'] = ta.HT_TRENDMODE(dataframe)
# Parabolic SAR
dataframe['sar'] = ta.SAR(dataframe)
# Core Trend Detection
# --------------------
dataframe.loc[(dataframe['ht_trendmode'] == 1) & (dataframe['sar'] > dataframe['close']), 'trend'] = 'downwards'
dataframe.loc[(dataframe['ht_trendmode'] == 0) | (dataframe['sar'] == dataframe['close']), 'trend'] = 'sideways'
dataframe.loc[(dataframe['ht_trendmode'] == 1) & (dataframe['sar'] < dataframe['close']), 'trend'] = 'upwards'
# Add DataFrame column for visualization in FreqUI when Dry/Live RunMode is detected or when not using TFZ
if (self.is_dry_live_run_detected is True) or (self.informative_timeframe == self.backtest_timeframe):
dataframe.loc[(dataframe['trend'] == 'downwards'), 'mgm_trend'] = -1
dataframe.loc[(dataframe['trend'] == 'sideways'), 'mgm_trend'] = 0
dataframe.loc[(dataframe['trend'] == 'upwards'), 'mgm_trend'] = 1
return dataframe
def minutes_to_timeframe(self, minutes: int) -> str:
"""
Calculates the corresponding timeframe for the amount of minutes provided
:param minutes: (int) Amount of minutes to parse to the closest candle size
:return: (str) The parsed timeframe / candle size
"""
if minutes < 1:
timeframe_number = 60 * minutes
timeframe_size = 's' # Return seconds
elif minutes < 60:
timeframe_number = minutes
timeframe_size = 'm' # Return minutes
elif minutes < 1440:
timeframe_number = minutes / 60
timeframe_size = 'h' # Return hours
elif minutes < 10080:
timeframe_number = minutes / 1440
timeframe_size = 'd' # Return days
elif minutes < 40320:
timeframe_number = minutes / 10080
timeframe_size = 'w' # Return weeks
else:
timeframe_number = minutes / 40320
timeframe_size = 'M' # Return months
if (timeframe_number - int(timeframe_number) == 0) is False:
sys.exit(f'MoniGoManiHyperStrategy - ERROR - MoniGoMani could not correctly parse the provided minutes '
f'({minutes}m) to a usable timeframe format ({str(timeframe_number)}{timeframe_size})! '
f'Please adjust the "timeframes" section of your "mgm-config"!')
return f'{str(int(timeframe_number))}{timeframe_size}'
def _populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds indicators based on Run-Mode & TimeFrame-Zoom:
If Dry/Live-running or BackTesting/HyperOpting without TimeFrame-Zoom it just pulls
'timeframe' (1h candles) to compute indicators.
If BackTesting/HyperOpting with TimeFrame-Zoom it pulls 'informative_pairs' (1h candles)
to compute indicators, but then tests upon 'backtest_timeframe' (5m or 1m candles)
to simulate price movement during that 'timeframe' (1h candle).
:param dataframe: (DataFrame) DataFrame with data from the exchange
:param metadata: (dict) Additional information, like the currently traded pair
:return DataFrame: DataFrame for MoniGoMani with all mandatory indicator data populated
"""
tfz = 'TimeFrame-Zoom'
# Compute indicator data during Backtesting / Hyperopting when TimeFrame-Zooming
if (self.is_dry_live_run_detected is False) and (self.informative_timeframe != self.backtest_timeframe):
self.mgm_logger('info', tfz, f'Backtesting/Hyperopting this strategy with a informative_timeframe '
f'({self.informative_timeframe} candles) and a zoomed backtest_timeframe '
f'({self.backtest_timeframe} candles)')
# Populate core trend indicators
core_trend = load_pair_history(pair=metadata['pair'],
datadir=self.config['datadir'],
timeframe=self.core_trend_timeframe,
# ToDo: calculate correct startup_candles needed for HT_TRENDMODE and SAR
startup_candles=self.startup_candle_count,
data_format=self.config.get('dataformat_ohlcv', 'json'))
core_trend = self._populate_core_trend(core_trend, metadata)
# Warning! This method gets ALL downloaded data for the given timeframe (when in BackTesting mode).
# If you have many months or years downloaded for this pair, this will take a long time!
informative = load_pair_history(pair=metadata['pair'],
datadir=self.config['datadir'],
timeframe=self.informative_timeframe,
startup_candles=self.startup_candle_count,
data_format=self.config.get('dataformat_ohlcv', 'json'))
# Throw away older data that isn't needed.
first_informative = dataframe['date'].min().floor('H')
informative = informative[informative['date'] >= first_informative]
# Merge core trend to informative data frame
informative = merge_informative_pair(
informative, core_trend[['date', 'ht_trendmode', 'sar', 'trend']].copy(),
self.informative_timeframe, self.core_trend_timeframe, ffill=True)
skip_columns = [f'{s}_{self.core_trend_timeframe}' for s in
['date', 'open', 'high', 'low', 'close', 'volume']]
informative.rename(columns=lambda s: s.replace('_{}'.format(self.core_trend_timeframe),
'') if (s not in skip_columns) else s, inplace=True)
# Populate indicators at a larger timeframe
informative = self.do_populate_indicators(informative.copy(), metadata)
# Drop unused columns to keep the dataframe lightweight
drop_columns = ['open', 'high', 'low', 'close', 'volume', f'date_{self.core_trend_timeframe}']
informative.drop(drop_columns, inplace=True, axis=1)
# Merge indicators back in with, filling in missing values.
dataframe = merge_informative_pair(dataframe, informative, self.timeframe,
self.informative_timeframe, ffill=True)
# Rename columns, since merge_informative_pair adds `_<timeframe>` to the end of each name.
# Skip over date etc..
skip_columns = [f'{s}_{self.informative_timeframe}' for s in
['date', 'open', 'high', 'low', 'close', 'volume']]
dataframe.rename(columns=lambda s: s.replace('_{}'.format(self.informative_timeframe),
'') if (s not in skip_columns) else s, inplace=True)
dataframe.drop([f'date_{self.informative_timeframe}'], inplace=True, axis=1)
# Compute indicator data normally during Dry & Live Running or when not using TimeFrame-Zoom
else:
self.mgm_logger('info', tfz,
f'Dry/Live-running MoniGoMani with normal timeframe ({self.timeframe} candles)')
# Populate core trend indicators
core_trend = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.core_trend_timeframe)
core_trend = self._populate_core_trend(core_trend, metadata)
# Merge core trend to main data frame
dataframe = merge_informative_pair(
dataframe, core_trend[['date', 'ht_trendmode', 'sar', 'trend', 'mgm_trend']].copy(),
self.timeframe, self.core_trend_timeframe, ffill=True)
skip_columns = [f'{s}_{self.core_trend_timeframe}' for s in
['date', 'open', 'high', 'low', 'close', 'volume']]
dataframe.rename(columns=lambda s: s.replace('_{}'.format(self.core_trend_timeframe),
'') if (s not in skip_columns) else s, inplace=True)
dataframe.drop([f'date_{self.core_trend_timeframe}'], inplace=True, axis=1)
# Just populate indicators.
dataframe = self.do_populate_indicators(dataframe, metadata)
return dataframe
def get_param_value(self, parameter_name: str):
"""
Fetches a parameter value from the initialized MoniGoMani class by name
:param parameter_name: (str) Name of the parameter
:return: (double) Parameter value divided by precision and rounded
"""
parameter = getattr(self, parameter_name)
return round(parameter.value / self.precision)
def get_all_current_open_trades(self, trade: 'Trade') -> List:
"""
Fetches all the trades currently open depending on the current RunMode of Freqtrade
:param trade: (trade) Current open trade object.
:return List: List containing all current open trades
"""
cis = 'custom_stoploss - Custom Information Storage'
if self.is_dry_live_run_detected is True:
self.mgm_logger('debug', cis, 'Fetching all currently open trades during Dry/Live Run')
all_open_trades = Trade.get_trades([Trade.is_open.is_(True)]).order_by(Trade.open_date).all()
# Fetch all open trade data during Back Testing & Hyper Opting
else:
self.mgm_logger('debug', cis, 'Fetching all currently open trades during BackTesting/HyperOpting')
all_open_trades = trade.trades_open
self.mgm_logger('debug', cis, f'Up-to-date open trades ({str(len(all_open_trades))}) fetched!')
self.mgm_logger('debug', cis, f'all_open_trades contents: {repr(all_open_trades)}')
return all_open_trades
def get_unclogger_trade_trend_data(self, pair: str, current_time: datetime) -> dict:
"""
Fetches the trade trend data for a pair over a length of the sell___unclogger_trend_lookback_candles_window
:param pair: (str) Pair of which the trend data needs to be fetched
:param current_time: (datetime) Current time, represents the start of unclogger_trend_lookback_candles_window
:return: (dict) Dictionary containing the trend data for the pairs unclogger_trend_lookback_candles_window
"""
# Fetch all needed 'trend' trade data
stored_trend_dataframe = {}
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
self.mgm_logger('debug', 'Open Trade Unclogger', 'Fetching all needed "trend" trade data')
trend_lookback_candles_window = self.get_param_value('sell___unclogger_trend_lookback_candles_window')
for candle in range(1, trend_lookback_candles_window + 1):
candle_time = self.convert_candle_time(current_time=current_time, current_candle=candle)
candle_trend = dataframe.loc[dataframe['date'] == candle_time].squeeze()['trend']
if isinstance(candle_trend, str):
stored_trend_dataframe[candle] = candle_trend
else:
break
return stored_trend_dataframe
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
Open Trade Custom Information Storage & Garbage Collector
---------------------------------------------------------
MoniGoMani (currently) only uses this function to store custom information from all open_trades at that given
moment during BackTesting/HyperOpting or Dry/Live-Running
Further it also does garbage collection to make sure no old closed trade data remains in custom_info
The actual normal "custom_stoploss" usage for which this function is generally used isn't used by MGM (yet)!
This custom_stoploss function should be able to work in tandem with Trailing stoploss!
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break MoniGoMani.
:return float: New stoploss value, relative to the current-rate
"""
cis = 'custom_stoploss - Custom Information Storage'
gc = cis + ' Garbage Collector'
# Open Trade Custom Information Storage
# -------------------------------------
# Fetch all open trade data depending on RunMode
all_open_trades = self.get_all_current_open_trades(trade)
# Store current pair's open_trade + it's current profit in custom_info
for open_trade in all_open_trades:
if str(open_trade.pair) == pair:
if str(open_trade.pair) not in self.custom_info['open_trades']:
self.custom_info['open_trades'][str(open_trade.pair)] = {}
self.custom_info['open_trades'][str(open_trade.pair)]['trade'] = str(open_trade)
self.custom_info['open_trades'][str(open_trade.pair)]['current_profit'] = current_profit
self.mgm_logger('info', cis, f'Storing trade + current profit/loss for pair ({pair}) in custom_info')
break
# Custom Information Storage Garbage Collector
# --------------------------------------------
# Check if any old open_trade garbage needs to be removed
if len(all_open_trades) < len(self.custom_info['open_trades']):
garbage_trade_amount = len(self.custom_info['open_trades']) - len(all_open_trades)
self.mgm_logger('info', gc,
f'Old open trade garbage detected for {str(garbage_trade_amount)} trades, starting cleanup')
for garbage_trade in range(garbage_trade_amount):
for stored_trade in self.custom_info['open_trades']:
pair_still_open = False
for open_trade in all_open_trades:
if str(stored_trade) == str(open_trade.pair):
self.mgm_logger('debug', gc, f'Open trade found, '
f'no action needed for pair ({stored_trade}) in custom_info')
pair_still_open = True
break
# Remove old open_trade garbage
if not pair_still_open:
self.mgm_logger('info', gc,
f'No open trade found for pair ({stored_trade}), removing from custom_info')
self.custom_info['open_trades'].pop(stored_trade)
self.mgm_logger('debug', gc,
f'Successfully removed garbage_trade {str(garbage_trade)} from custom_info!')
break
# Print all stored open trade info in custom_storage
self.mgm_logger('debug', cis, f'Open trades ({str(len(self.custom_info["open_trades"]))}) '
f'in custom_info updated successfully!')
self.mgm_logger('debug', cis, f'custom_info["open_trades"] contents: {repr(self.custom_info["open_trades"])}')
# Always return a value bigger than the initial stoploss to keep using the initial stoploss.
# Since we (currently) only want to use this function for custom information storage!
return -1
def custom_sell(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""
Open Trade Unclogger:
---------------------
Override Sell Signal: When enabled attempts to unclog the bot when it's stuck with losing trades & unable to
trade more new trades.
It will only unclog a losing trade when all following checks have been full-filled:
=> Check if everything in custom_storage is up-to-date with all_open_trades
=> Check if there are enough losing trades open for unclogging to occur
=> Check if there is a losing trade open for the pair currently being run through the MoniGoMani loop
=> Check if trade has been open for X minutes (long enough to give it a recovery chance)
=> Check if total open trades losing % is met
=> Check if open_trade's trend changed negatively during past X candles
Please configurable/hyperoptable in the sell_params dictionary under the hyperopt results copy/paste section.
Only used when 'unclogger_enabled' is set to 'true'.
:param pair: Pair that's currently analyzed
:param trade: trade object.
:param current_time: datetime object, containing the current datetime
:param current_rate: Rate, calculated based on pricing settings in ask_strategy.
:param current_profit: Current profit (as ratio), calculated based on current_rate.
:param **kwargs: Ensure to keep this here so updates to this won't break MoniGoMani.
:return: True or string if a custom sell should occur, otherwise None
"""
if ((self.mgm_config['unclogger_spaces']['unclogger_enabled'] is
True) and (pair in self.custom_info['open_trades']) and (self.custom_info['open_trades'][pair] != {})):
# Initialize some re-occurring logging strings
abort = 'No unclogging needed! '
proceed = ' Proceeding to the next check!'
otu = 'Open Trade Unclogger'
cis = 'custom_sell - Custom Information Storage'
su = 'sell___unclogger_'
try:
# Open Trade Custom Information Storage
# -------------------------------------
# Fetch all open trade data depending on RunMode
all_open_trades = self.get_all_current_open_trades(trade)
# Check if everything in custom_storage is up-to-date with all_open_trades
if len(all_open_trades) > len(self.custom_info['open_trades']):
self.mgm_logger('warning', cis, f'Open trades ({str(len(self.custom_info["open_trades"]))}) in '
f'custom_storage do not match yet with trades in live open trades '
f'({str(len(all_open_trades))}) aborting unclogger for now!')
return None # By default, we don't want a force sell to occur
# Open Trade Unclogger
# --------------------
self.mgm_logger('debug', otu, 'Running trough all checks to see if unclogging is needed')
# Check if there are enough losing trades open for unclogging to occur
self.mgm_logger('debug', otu,
'Fetching all currently losing_open_trades from custom information storage')
losing_open_trades = {}
for stored_trade in self.custom_info['open_trades']:
stored_current_profit = self.custom_info['open_trades'][stored_trade]['current_profit']
if stored_current_profit < 0:
if pair not in losing_open_trades:
losing_open_trades[str(stored_trade)] = {}
losing_open_trades[str(stored_trade)] = stored_current_profit
self.mgm_logger('debug', otu, f'Fetched losing_open_trades ({str(len(losing_open_trades))}) '
f'from custom information storage!')
minimal_losing_trades_open = self.get_param_value(f'{su}minimal_losing_trades_open')
if len(losing_open_trades) < minimal_losing_trades_open:
self.mgm_logger('debug', otu, f'{abort}Not enough losing trades currently open!')
return None # By default, we don't want a force sell to occur
self.mgm_logger('debug', otu, f'Enough losing trades detected!{proceed}')
# Check if there is a losing trade open for the pair currently being run through the MoniGoMani
if pair not in losing_open_trades:
self.mgm_logger('debug', otu, f'{abort}Currently checked pair ({pair}) is not '
f'making a loss at this point in time!')
return None # By default, we don't want a force sell to occur
self.mgm_logger('debug', otu, f'Currently checked pair ({pair}) is losing!{proceed}')
trade_open_time = trade.open_date_utc.replace(tzinfo=None)
self.mgm_logger('debug', otu, f'Trade open time: {str(trade_open_time)}')
min_losing_trade_duration = self.get_param_value(f'{su}minimal_losing_trade_duration_minutes')
minimal_open_time = (current_time.replace(tzinfo=None) - timedelta(minutes=min_losing_trade_duration))
self.mgm_logger('debug', otu, f'Minimal open time: {str(minimal_open_time)}')
if trade_open_time > minimal_open_time:
self.mgm_logger('debug', otu, f'{abort}Currently checked pair ({pair}) has not '
f'been open been open for long enough!')
return None # By default, we don't want a force sell to occur
self.mgm_logger('debug', otu, f'Trade has been open for long enough!{proceed}')
# Check if total open trades losing % is met
percentage_open_trades_losing = int((len(losing_open_trades) / len(all_open_trades)) * 100)
self.mgm_logger('debug', otu, f'percentage_open_trades_losing: {str(percentage_open_trades_losing)}%')
trades_losing_percentage_needed = self.get_param_value(f'{su}open_trades_losing_percentage_needed')
if percentage_open_trades_losing < trades_losing_percentage_needed:
self.mgm_logger('debug', otu,
f'{abort}Percentage of open trades losing needed has not been satisfied!')
return None # By default, we don't want a force sell to occur
self.mgm_logger('debug', otu, f'Percentage of open trades losing needed has been satisfied!{proceed}')
# Fetch current dataframe for the pair currently being run through MoniGoMani
trend_lookback_candles_window = self.get_param_value(f'{su}trend_lookback_candles_window')
self.mgm_logger('debug', otu, f'Fetching currently needed "trend" dataframe data to check how pair '
f'({pair}) has been doing in during the last '
f'{str(trend_lookback_candles_window)} candles')
# Fetch all needed 'trend' trade data
stored_trend_dataframe = self.get_unclogger_trade_trend_data(pair=pair, current_time=current_time)
# Check if enough trend data has been stored to do the next check
if len(stored_trend_dataframe) < trend_lookback_candles_window:
self.mgm_logger('debug', otu, f'{abort}Not enough trend data stored yet!')
return None # By default, we don't want a force sell to occur
# Print all fetched 'trend' trade data
self.mgm_logger('debug', otu, f'All needed "trend" trade data '
f'({str(len(stored_trend_dataframe))}) fetched!')
self.mgm_logger('debug', otu, f'stored_trend_dataframe contents: {repr(stored_trend_dataframe)}')
# Check if the currently detected trend is positive
negative_trend = True
for trend in self.mgm_trends:
if ((self.mgm_config['unclogger_spaces'][f'unclogger_trend_lookback_window_uses_{trend}_candles']
is False) & (stored_trend_dataframe[1] == trend)):
negative_trend = False
break
if negative_trend is False:
self.mgm_logger('debug', otu, f'{abort}Positive trend currently detected!')
return None # By default, we don't want a force sell to occur
# Check if open_trade's trend changed negatively during past X candles
self.mgm_logger('debug', otu, f'Calculating amount of unclogger_trend_lookback_candles_window '
f'"satisfied" for pair: {pair}')
unclogger_weighted_candles_satisfied = 0
for lookback_candle in range(1, trend_lookback_candles_window + 1):
for trend in self.mgm_trends:
if (self.mgm_config['unclogger_spaces'][f'unclogger_trend_lookback_window_uses_{trend}_candles'
] & (stored_trend_dataframe[lookback_candle] == trend)):
unclogger_weighted_candles_satisfied += (
self.separator - (lookback_candle * self.separator_candle_weight_reducer))
self.mgm_logger('debug', otu, f'Amount of unclogger_trend_lookback_candles_window "satisfied": '
f'{str(unclogger_weighted_candles_satisfied)} for pair: {pair}')
# Calculate the percentage of the lookback window currently satisfied
unclogger_candles_percentage_satisfied = (unclogger_weighted_candles_satisfied /
trend_lookback_candles_window) * 100
# Override Sell Signal: Unclog trade by forcing a sell & attempt to continue
# the profit climb with the "freed up trading slot"
trend_lookback_candles_window_percentage_needed = self.get_param_value(
f'{su}trend_lookback_candles_window_percentage_needed')
if unclogger_candles_percentage_satisfied >= trend_lookback_candles_window_percentage_needed:
# Buy Cooldown Window Custom Information Storage
if pair not in self.custom_info['unclogger_cooldown_pairs']:
self.custom_info['unclogger_cooldown_pairs'][pair] = []
buy_cooldown_minutes_window = self.get_param_value(f'{su}buy_cooldown_minutes_window')
self.custom_info['unclogger_cooldown_pairs'][pair].append({
'start_time': current_time,
'end_time': current_time + timedelta(minutes=buy_cooldown_minutes_window)
})
self.mgm_logger('info', otu, 'Unclogging losing trade...')
return 'MGM_unclogging_losing_trade'
else:
self.mgm_logger('info', otu, 'No need to unclog open trade...')
except Exception as e:
self.mgm_logger('error', otu, f'Following error has occurred: {str(e)}')
return None # By default, we don't want a force sell to occur
def convert_candle_time(self, current_time: datetime, current_candle: int = 1) -> datetime:
"""
Converts the current_time to a candle_time (of an informative_timeframe candle size),
which can be used to query the dataframe
:param current_time: (datetime) Current time object
:param current_candle: (int) Amount of candles to offset (look back) from the current_time
:return: (datetime) Converted candle time object
"""
# Convert the candle time to the one being used by the 'informative_timeframe'
candle_multiplier = int(self.informative_timeframe.rstrip('mhdwM'))
candle_time = (timeframe_to_prev_date(self.informative_timeframe, current_time) -
timedelta(minutes=int(current_candle * candle_multiplier)))
if self.informative_timeframe.find('h') != -1:
candle_time = (timeframe_to_prev_date(self.informative_timeframe, current_time) -
timedelta(hours=int(current_candle * candle_multiplier)))
elif self.informative_timeframe.find('d') != -1:
candle_time = (timeframe_to_prev_date(self.informative_timeframe, current_time) -
timedelta(days=int(current_candle * candle_multiplier)))
elif self.informative_timeframe.find('w') != -1:
candle_time = (timeframe_to_prev_date(self.informative_timeframe, current_time) -
timedelta(weeks=int(current_candle * candle_multiplier)))
elif self.informative_timeframe.find('M') != -1:
candle_time = (timeframe_to_prev_date(self.informative_timeframe, current_time) -
timedelta64(int(current_candle * candle_multiplier), 'M'))
return candle_time
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, **kwargs) -> bool:
"""
Open Trade Unclogger Buy Cooldown Window
----------------------------------------
Override Buy Signal - Cancels out a buy order when all the following are fulfilled:
- The Open Trade Unclogger is enabled
- The Buy Cooldown Window set into place after unclogging said losing pair has not expired yet
Timing for this function is critical, it's needed to avoid doing heavy tasks or network requests in this method.
:param pair: Pair that's about to be bought.
:param order_type: Order type (as configured in order_types). usually limit or market.
:param amount: Amount in target (quote) currency that's going to be traded.
:param rate: Rate that's going to be used when using limit orders
:param time_in_force: Time in force. Defaults to GTC (Good-til-cancelled).
:param current_time: datetime object, containing the current datetime
:param **kwargs: Ensure to keep this here so updates to this won't break MoniGoMani.
:return bool: When True is returned, then the buy-order is placed on the exchange. False aborts the process
"""
uc = 'Unclogger Cooldown'
n = '\n '
if ((self.mgm_config['unclogger_spaces']['unclogger_enabled'] is
True) and (pair in self.custom_info['unclogger_cooldown_pairs'])):
for cooldown_period in self.custom_info['unclogger_cooldown_pairs'][pair][:]:
self.mgm_logger('debug', uc, f'{pair} CoolDown Period:'
f'{n}Cooldown Start Time: {cooldown_period["start_time"]}'
f'{n}Cooldown End Time: {cooldown_period["end_time"]}'
f'{n}Current Time: {current_time}')
if cooldown_period['end_time'] < current_time:
self.mgm_logger('debug', uc, f'Cooldown period for unclogged pair ({pair}) has expired, '
f'removing from custom_info! (CurrentTime: {current_time})')
self.custom_info['unclogger_cooldown_pairs'][pair].remove(cooldown_period)
elif cooldown_period['start_time'] < current_time < cooldown_period['end_time']:
self.mgm_logger('debug', uc, 'Blocking buy signal since pair is cooling down...')
return False # Block the buy signal from going through when the pair is still under cooldown
return True # Allow the buy signal to go through if the pair is not under cooldown
def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float,
rate: float, time_in_force: str, sell_reason: str,
current_time: datetime, **kwargs) -> bool:
"""
Weighted Signals Sell Profit Only
---------------------------------
Override Sell Signal - Configurable setting, if enabled
weighted sell signals require to be profitable to go through.