| Overall Statistics |
|
Total Trades 872 Average Win 1.42% Average Loss -1.29% Compounding Annual Return 55.444% Drawdown 25.700% Expectancy 0.113 Net Profit 75.174% Sharpe Ratio 1.551 Probabilistic Sharpe Ratio 62.376% Loss Rate 47% Win Rate 53% Profit-Loss Ratio 1.10 Alpha 0.462 Beta 0.201 Annual Standard Deviation 0.327 Annual Variance 0.107 Information Ratio 0.723 Tracking Error 0.39 Treynor Ratio 2.52 Total Fees $0.00 Estimated Strategy Capacity $4000000.00 |
# custom inpots
from FxTridentBase import FxTridentBase
# ********************************
class FxMultiTrident(FxTridentBase):
# ********************************
# -----------------------------------------------------------
# Initializer (Cosntructor)
# ------------------------------------------------------------
def Initializie(self):
# ------------------------------------------------------------
# base initializer
super(FxMultiTrident, self).Initializie()
# main initializer
self.SetBrokerageModel(AlphaStreamsBrokerageModel())
# setup forex brokerages
if self.fx_broker == "FXCM":
self.SetBrokerageModel(BrokerageName.FxcmBrokerage)
else:
self.SetBrokerageModel(BrokerageName.OandaBrokerage)'''
To use this library place this at the top:
from AlgoToolbox import AlgoToolbox
Then instantiate the class:
toolbox = AlgoToolbox(self, True, True)
toolbox.show_log(...)
'''
# ********************************
class AlgoToolbox:
# ********************************
# -------------------------------------------------------------------------------------------------------------
def __init__(self, qcAlgo, showDebug = False, showLog = False):
# -------------------------------------------------------------------------------------------------------------
self.qcAlgo = qcAlgo
self.showDebug = showDebug
self.showLog = showLog
# -------------------------------------------------------------------------------------------------------------
def Log(self, msg, prefix = "", suffix = ""):
# -------------------------------------------------------------------------------------------------------------
self.qcAlgo.Log("LOG: " + prefix + " : " + str(self.qcAlgo.Time) + " - " + msg + suffix)
# -------------------------------------------------------------------------------------------------------------
def show_log(self, msg, prefix = "", suffix = ""):
# -------------------------------------------------------------------------------------------------------------
if self.showLog:
self.Log(msg, prefix, suffix)
# -------------------------------------------------------------------------------------------------------------
def show_csv_log(self, msg, sep_char = ";", hide_prefix = True):
# -------------------------------------------------------------------------------------------------------------
if self.showLog:
if hide_prefix:
self.qcAlgo.Log(str(msg))
else:
self.qcAlgo.Log("LOG" + sep_char + str(msg))
# -------------------------------------------------------------------------------------------------------------
def Debug(self, msg, prefix = "", suffix = ""):
# -------------------------------------------------------------------------------------------------------------
self.qcAlgo.Debug("DEBUG: " + prefix + " : " + str(self.qcAlgo.Time) + " - " + msg + suffix)
# -------------------------------------------------------------------------------------------------------------
def show_debug(self, msg, prefix = "", suffix = ""):
# -------------------------------------------------------------------------------------------------------------
if self.showLog:
self.Debug(msg, prefix, suffix)
# -------------------------------------------------------------------------------------------------------------
def pip_diff(self, open, close):
# -------------------------------------------------------------------------------------------------------------
multiplier = self.pip_multiplier(open)
pips = round((close - open) * multiplier)
return int(pips)
# -------------------------------------------------------------------------------------------------------------
def pip_equal(self, varA, varB, pip_tolerance):
# -------------------------------------------------------------------------------------------------------------
diff = self.pip_diff(varA, varB)
return abs(diff) <= abs(pip_tolerance)
# -------------------------------------------------------------------------------------------------------------
def add_pip(self, price, pip):
# -------------------------------------------------------------------------------------------------------------
divider = self.pip_divider(price)
pip_price = pip * divider
result_price = price + pip_price
return result_price
# -------------------------------------------------------------------------------------------------------------
def pip_multiplier(self, price):
# -------------------------------------------------------------------------------------------------------------
pip_sign = 5
dec_pos = str(price).index('.')
return 10 ** (pip_sign - dec_pos)
# -------------------------------------------------------------------------------------------------------------
def pip_divider(self, price):
# -------------------------------------------------------------------------------------------------------------
multiplier = self.pip_multiplier(price)
return 1 / multiplier
# -------------------------------------------------------------------------------------------------------------
def pip_round(self, price):
# -------------------------------------------------------------------------------------------------------------
pip_size = 5
if price > 10:
pip_size = 3
result = round(price, pip_size)
return result
# -----------------------------------------------------------------------------------------------------------
def get_date(self, strDate, splitChar):
# -----------------------------------------------------------------------------------------------------------
if strDate and (not strDate.isspace()):
str_date_parts = strDate.split(splitChar)
year = int(str_date_parts[0])
month = int(str_date_parts[1])
day = int(str_date_parts[2])
return year, month, day
else:
return 0, 0, 0
# -----------------------------
def cut_by_zero_one(self, level):
# -----------------------------
if level > 1:
return 1
else:
if level < 0:
return 0
else:
return level
# -----------------------------
def cut_by_one_minusone(self, level):
# -----------------------------
if level > 1:
return 1
else:
if level < -1:
return -1
else:
return level
# -----------------------------
def normalize_percent(self, value):
# -----------------------------
return self.cut_by_one_minusone(value/100)
# -----------------------------
def normalize_percent_abs(self, value):
# -----------------------------
return abs(self.cut_by_one_minusone(value/100))from AlgoToolbox import AlgoToolbox
from QuantConnect import *
from numpy import diff
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
# ===================================
class FxEmaStar:
# ===================================
# ---------------------
def __init__(self, qcAlgo, ticker, weight, period, resolution, insight_min, showDebug = True, showLog = True):
# ---------------------
# *******************************************
# REGION Brick INIT (BEGIN)
# *******************************************
# -------------------------------------
# init brick params
# -------------------------------------
self.qcAlgo = qcAlgo
self.toolbox = qcAlgo.algo_toolbox
self.ticker = ticker
self.weight = weight
self.period = period
self.resolution = resolution
self.showDebug = showDebug
self.showLog = showLog
self.insight_min = insight_min
self.insight_model = "WEIGHTED_PREDICTION"
self.long_active = False
self.short_active = False
# derived params
self.ptt_pips_touch_tolerance = self.qcAlgo.ptt_pips_touch_tolerance
self.pet_pips_equal_tolerance = self.qcAlgo.pet_pips_equal_tolerance
self.take_profit_pips = self.qcAlgo.take_profit_pips
self.stop_loss_pips = self.qcAlgo.stop_loss_pips
# -------------------------------------
# init brick trade vars
# -------------------------------------
self.is_green_bar = False
self.is_red_bar = False
self.bar_start_time = None
self.bar_end_time = None
self.insight_default_confidence = 1
self.close_reason = "default"
# init state / trade vars
self.trade_state = 0 # 0 = wait; 1 = long; -1 = short; 1/2 = long break even; -1/2 = short break even
self.entry_price = 0
self.exit_price = 0
self.trade_profit = 0
self.lifetime_profit = 0
self.holding_long = 0
self.holding_short = 0
self.bar_open = 0
self.bar_close = 0
# -------------------------------------
# add fx-pair
# -------------------------------------
self.qcAlgo.AddForex(self.ticker, self.resolution)
# -------------------------------------
# init indicators
# -------------------------------------
self.ema_red = self.qcAlgo.EMA(self.ticker, self.qcAlgo.ema_red, self.resolution)
self.ema_blue = self.qcAlgo.EMA(self.ticker, self.qcAlgo.ema_blue, self.resolution)
self.ema_green = self.qcAlgo.EMA(self.ticker, self.qcAlgo.ema_green, self.resolution)
# -------------------------------------
# create scalp consolidator
self.scalpConsolidator = QuoteBarConsolidator(timedelta(minutes=self.period))
self.scalpConsolidator.DataConsolidated += self.Indicators_Update
self.qcAlgo.SubscriptionManager.AddConsolidator(self.ticker, self.scalpConsolidator);
# *******************************************
# REGION Brick INIT (END)
# *******************************************
# ---------------------
def emit_insight(self, time_delta_min, direction, magnitude=None, confidence=None, model=None, weight=None):
# ---------------------
# init vars
insight_timedelta = timedelta(minutes=time_delta_min)
# Creates an insight for the current symbol, predicting that it will move up/down within given timedelta
if (direction == InsightDirection.Flat):
self.qcAlgo.EmitInsights(
Insight.Price(self.ticker, insight_timedelta, InsightDirection.Flat))
else:
insight = Insight( \
self.ticker, insight_timedelta, InsightType.Price, \
direction, magnitude, confidence, model, weight)
self.qcAlgo.EmitInsights(insight)
# *******************************************
# REGION Brick METHODS (BEGIN)
# *******************************************
#-------------------------------------------------------------------
def init_holdings_long(self):
#-------------------------------------------------------------------
self.holding_long = self.qcAlgo.leverage * self.weight
#-------------------------------------------------------------------
def init_holdings_short(self):
#-------------------------------------------------------------------
self.holding_short = -1 * self.qcAlgo.leverage * self.weight
#-------------------------------------------------------------------
def Indicators_Update(self, sender, quoteBar):
#-------------------------------------------------------------------
self.bar_open = quoteBar.Open
self.bar_close = quoteBar.Close
if self.bar_open == self.bar_close:
self.is_green_bar = False
self.is_red_bar = False
else:
self.is_green_bar = self.bar_close > self.bar_open
self.is_red_bar = not self.is_green_bar
self.bar_start_time = quoteBar.Time
self.bar_end_time = quoteBar.EndTime
self.ema_red.Update(self.bar_end_time, self.bar_close);
self.ema_blue.Update(self.bar_end_time, self.bar_close);
self.ema_green.Update(self.bar_end_time, self.bar_close);
#-------------------------------------------------------------------
def trade_ondata(self, data):
#-------------------------------------------------------------------
# init
price = self.qcAlgo.Securities[self.ticker].Price
ema_red = self.ema_red.Current.Value
ema_blue = self.ema_blue.Current.Value
ema_green = self.ema_green.Current.Value
close_boundary_long = self.toolbox.add_pip(ema_green, self.ptt_pips_touch_tolerance)
close_boundary_short = self.toolbox.add_pip(ema_green, -self.ptt_pips_touch_tolerance)
# calc trate states (long)
is_long_open = self.trade_state == 0 \
and not ( \
ema_green > ema_red \
and (self.toolbox.pip_equal(ema_blue, ema_red, self.pet_pips_equal_tolerance) or ema_blue > ema_red) \
and self.is_green_bar \
and self.bar_close > ema_green and self.bar_open > ema_blue
)
is_long_close = self.trade_state > 0 \
and self.is_red_bar \
and self.bar_close < close_boundary_long
if is_long_close: self.close_reason = "is_long_ema_close"
is_long_take_profit = self.trade_state > 0 \
and self.toolbox.pip_diff(self.entry_price, price) >= self.take_profit_pips
if is_long_take_profit: self.close_reason = "is_long_take_profit"
is_long_stop_loss = self.trade_state > 0 \
and self.toolbox.pip_diff(self.entry_price, price) <= self.stop_loss_pips
if is_long_stop_loss: self.close_reason = "is_long_stop_loss"
# calc trate states (short)
is_short_open = self.trade_state == 0 \
and not ( \
ema_green < ema_red \
and (self.toolbox.pip_equal(ema_blue, ema_green, self.pet_pips_equal_tolerance) or ema_blue > ema_green) \
and self.is_red_bar \
and self.bar_close < ema_green and self.bar_open < ema_blue
)
is_short_close = self.trade_state < 0 \
and self.is_green_bar \
and self.bar_close > close_boundary_short
if is_short_close: self.close_reason = "is_short_ema_close"
is_short_take_profit = self.trade_state < 0 \
and self.toolbox.pip_diff(self.entry_price, price) <= -self.take_profit_pips
if is_short_take_profit: self.close_reason = "is_short_take_profit"
is_short_stop_loss = self.trade_state < 0 \
and self.toolbox.pip_diff(self.entry_price, price) >= -self.stop_loss_pips
if is_short_stop_loss: self.close_reason = "is_short_stop_loss"
# self.log_metrics("Price-Setup", price, ema_red, ema_blue, ema_green)
#-------------------------------------------------------------------
# WAIT-State
if is_long_open:
self.trade_state = 1 # GO Long!
self.init_holdings_long()
self.entry_price = price
self.exit_price = 0
self.emit_insight(self.period, InsightDirection.Up)
self.qcAlgo.SetHoldings(self.ticker, self.holding_long)
self.log_metrics("GO LONG! Open-Price=", price, ema_red, ema_blue, ema_green)
# quit state
return
if is_short_open:
self.trade_state = -1 # GO Short!
self.init_holdings_short()
self.entry_price = price
self.exit_price = 0
self.emit_insight(self.period, InsightDirection.Down)
self.qcAlgo.SetHoldings(self.ticker, self.holding_short)
self.log_metrics("GO SHORT! Open-Price=", price, ema_red, ema_blue, ema_green)
# quit state
return
#-------------------------------------------------------------------
# LONG-State
if is_long_close or is_long_take_profit or is_long_stop_loss:
self.exit_price = price
#self.trade_state = 0
profit_pips = self.toolbox.pip_diff(self.entry_price, self.exit_price)
self.calc_profit(profit_pips)
self.emit_insight(self.period, InsightDirection.Down)
self.log_metrics("LONG CLOSE! Reason: " + self.close_reason + ". Close-Price=", price, ema_red, ema_blue, ema_green, profit_pips)
if self.trade_state > 0:
self.close_and_go_wait(self.trade_state)
# quit state
return
#-------------------------------------------------------------------
# SHORT-State
if is_short_close or is_short_take_profit or is_short_stop_loss:
self.exit_price = price
#self.trade_state = 0
profit_pips = self.toolbox.pip_diff(self.exit_price, self.entry_price)
self.calc_profit(profit_pips)
self.emit_insight(self.period, InsightDirection.Up)
self.log_metrics("SHORT CLOSE! Reason: " + self.close_reason + ". Close-Price=", price, ema_red, ema_blue, ema_green, profit_pips)
if self.trade_state < 0:
self.close_and_go_wait(self.trade_state)
# quit state
return
#-------------------------------------------------------------------
def close_and_go_wait(self, close_state):
#-------------------------------------------------------------------
if close_state > 0:
self.holding_long = 0
self.qcAlgo.SetHoldings(self.ticker, self.holding_long)
if close_state < 0:
self.holding_short = 0
self.qcAlgo.SetHoldings(self.ticker, self.holding_short)
self.trade_state = 0
self.entry_price = 0
self.exit_price = 0
#-------------------------------------------------------------------
def calc_profit(self, pips_profit):
#-------------------------------------------------------------------
self.trade_profit = pips_profit
self.lifetime_profit = self.lifetime_profit + pips_profit
self.qcAlgo.global_lifetime_profit = self.qcAlgo.global_lifetime_profit + pips_profit
#-------------------------------------------------------------------
def log_metrics(self, msg, price, ema_red, ema_blue, ema_green, pips_profit=None):
#-------------------------------------------------------------------
self.toolbox.show_debug("-------------------------------------------")
self.toolbox.show_debug("Ticker: " + str(self.ticker))
self.toolbox.show_debug("-------------------------------------------")
self.toolbox.show_debug(msg + " " + str(price))
self.toolbox.show_debug("EMA_RED=" + str(ema_red))
self.toolbox.show_debug("EMA_BLUE=" + str(ema_blue))
self.toolbox.show_debug("EMA_GREEN=" + str(ema_green))
if pips_profit != None:
self.toolbox.show_debug("Trade PIPs Profit=" + str(pips_profit))
self.toolbox.show_debug("Lifetime Profit=" + str(self.lifetime_profit))
self.toolbox.show_debug("Global Lifetime Profit=" + str(self.qcAlgo.global_lifetime_profit))
self.toolbox.show_debug(" ")
# *******************************************
# REGION Brick METHODS (END)
# *******************************************# *************************************
class ForexGalaxyBay:
# *************************************
'''
To use this library place this at the top:
from ForexGalaxyBay import ForexGalaxyBay
Then instantiate the function:
fx = ForexGalaxyBay()
for fx_ticker in fx.selectGalaxy("MajorTen"):
self.AddForex(fx_ticker, resolution, market)
...
'''
# ----------------------------------------------------------------------------
def __init__(self):
# ----------------------------------------------------------------------------
# Currency-Pair Symbol: Weight
self.fxPairs_MajorForexBooster_Orig = { \
"EURUSD": .27
,"USDJPY": .13
,"GBPUSD": .11
,"AUDUSD": .09
,"USDCAD": .06
,"USDCHF": .06
,"NZDUSD": .04
,"EURJPY": .04
,"GBPJPY": .04
,"EURGBP": .03
,"AUDJPY": .03
,"EURAUD": .02
,"EURCAD": .02
,"AUDCAD": .02
,"USDHKD": .01
,"EURNZD": .01
,"AUDNZD": .01
,"USDMXN": .01
}
self.fxPairs_MajorForexBooster = { \
"EURUSD": .27
,"USDJPY": .13
,"GBPUSD": .11
,"AUDUSD": .09
,"USDCAD": .06
,"USDCHF": .06
,"NZDUSD": .04
,"EURJPY": .04
,"GBPJPY": .04
,"EURGBP": .03
,"AUDJPY": .03
,"EURAUD": .02
,"EURCAD": .02
,"AUDCAD": .02
,"USDHKD": .01
,"EURNZD": .01
,"AUDNZD": .01
,"USDMXN": .01
}
self.fxPairs_MajorForexBooster_Equal = { \
"EURUSD": .05
,"USDJPY": .05
,"GBPUSD": .05
,"AUDUSD": .05
,"USDCAD": .05
,"USDCHF": .05
,"NZDUSD": .05
,"EURJPY": .05
,"GBPJPY": .05
,"EURGBP": .05
,"AUDJPY": .05
,"EURAUD": .05
,"EURCAD": .05
,"AUDCAD": .05
,"USDHKD": .05
,"EURNZD": .05
,"AUDNZD": .05
,"USDMXN": .05
,"CADJPY": .05
,"CADHKD": .05
}
self.fxPairs_MajorForexBooster_Cool = { \
"EURUSD": .27
,"USDJPY": .13
,"GBPUSD": .11
,"AUDUSD": .09
,"USDCAD": .06
,"USDCHF": .06
,"NZDUSD": .04
,"EURJPY": .04
,"GBPJPY": .04
,"EURGBP": .03
,"AUDJPY": .03
,"EURAUD": .02
,"EURCAD": .02
,"AUDCAD": .02
,"USDHKD": .01
,"EURNZD": .01
,"AUDNZD": .01
,"EURMXN": .01
}
self.fxPairs_MajorForexBooster_Cool_Variant = { \
"EURUSD": .27
,"USDJPY": .12
,"GBPUSD": .12
,"AUDUSD": .11
,"USDCAD": .06
,"USDCHF": .06
,"NZDUSD": .04
,"EURJPY": .04
,"GBPJPY": .04
,"EURGBP": .03
,"AUDJPY": .03
,"EURAUD": .02
,"EURCAD": .02
,"AUDCAD": .02
,"USDHKD": .01
,"EURNZD": .01
,"AUDNZD": .01
#"USDMXN": .01
}
self.fxPairs_MajorForexBooster_Dude = { \
"EURUSD": .27
,"USDJPY": .13
,"GBPUSD": .11
,"AUDUSD": .10
,"USDCAD": .07
,"USDCHF": .07
,"NZDUSD": .05
,"EURGBP": .04
,"EURAUD": .03
,"EURCAD": .03
,"AUDCAD": .03
,"EURNZD": .02
,"AUDNZD": .02
,"USDMXN": .02
,"USDHKD": .01
}
self.fxPairs_MajorForexBooster_BigDude = { \
"EURUSD": .23
,"USDJPY": .11
,"GBPUSD": .10
,"AUDUSD": .10
,"USDMXN": .10
,"USDCAD": .07
,"USDCHF": .07
,"EURCAD": .04
,"AUDCAD": .04
,"NZDUSD": .04
,"EURGBP": .03
,"EURAUD": .02
,"EURNZD": .02
,"AUDNZD": .02
,"USDHKD": .01
}
self.fxPairs_MagicTrinity = { \
"EURUSD": .5
,"GBPUSD": .3
,"USDJPY": .2
}
self.fxPairs_MajorFive = { \
"EURUSD": .4
,"USDJPY": .2
,"GBPUSD": .2
,"AUDUSD": .1
,"USDCAD": .1
}
self.fxPairs_MajorTen_Once = { \
"EURUSD": .1
,"USDJPY": .1
,"GBPUSD": .1
,"AUDUSD": .1
,"USDCAD": .1
,"USDCHF": .1
,"NZDUSD": .1
,"EURJPY": .1
,"GBPJPY": .1
,"EURGBP": .1
}
self.fxPairs_MajorTen_Double = { \
"EURUSD": .2
,"USDJPY": .2
,"GBPUSD": .2
,"AUDUSD": .2
,"USDCAD": .2
,"USDCHF": .2
,"NZDUSD": .2
,"EURJPY": .2
,"GBPJPY": .2
,"EURGBP": .2
}
self.fxPairs_MajorTen_Orig = { \
"EURUSD": .33
,"USDJPY": .16
,"GBPUSD": .13
,"AUDUSD": .07
,"USDCAD": .06
,"USDCHF": .06
,"NZDUSD": .05
,"EURJPY": .05
,"GBPJPY": .05
,"EURGBP": .04
}
self.fxPairs_MajorTen_Ext1 = { \
"EURUSD": .3
,"USDJPY": .1
,"AUDNZD": .05
,"AUDUSD": .05
,"USDCAD": .1
,"USDCHF": .1
,"NZDUSD": .05
,"EURJPY": .1
,"GBPJPY": .1
,"EURGBP": .05
}
self.fxPairs_EurUsd = { \
"EURUSD": 1
}
self.fxPairs_EurGbp = { \
"EURGBP": 1
}
self.fxPairs_EURUSD_USDCHF = { \
"EURUSD": .5
,"USDCHF": .5
}
self.fxPairs_EURUSD_GBPUSD = { \
"EURUSD": .7
,"GBPUSD": .3
}
self.fxPairs_EURUSD_GBPUSD_EQUAL = { \
"EURUSD": .5
,"GBPUSD": .5
}
self.fxPairs_EurUsd_Orig = { \
"EURUSD": 1
}
self.fxPairs_MajorForexBooster_JPY = { \
"EURUSD": .05
,"USDJPY": .05
,"GBPUSD": .05
,"AUDUSD": .05
,"USDCAD": .05
,"USDCHF": .05
,"NZDUSD": .05
,"EURJPY": .05
,"GBPJPY": .05
,"EURGBP": .05
,"AUDCAD": .05
,"AUDJPY": .05
,"AUDNZD": .05
,"CADJPY": .05
,"EURNZD": .05
,"GBPNZD": .05
,"SGDJPY": .05
,"NZDJPY": .05
,"USDJPY": .05
}
self.fxPairs_MajorForexBooster_Asia_Ext = { \
"EURUSD": .03125
,"USDJPY": .03125
,"GBPUSD": .03125
,"AUDUSD": .03125
,"USDCAD": .03125
,"USDCHF": .03125
,"NZDUSD": .03125
,"EURJPY": .03125
,"GBPJPY": .03125
,"EURGBP": .03125
,"AUDCAD": .03125
,"AUDCHF": .03125
,"AUDHKD": .03125
,"AUDJPY": .03125
,"AUDNZD": .03125
,"CADHKD": .03125
,"CADJPY": .03125
,"CHFHKD": .03125
,"CHFJPY": .03125
,"EURHKD": .03125
,"EURNZD": .03125
,"GBPHKD": .03125
,"GBPNZD": .03125
,"HKDJPY": .03125
,"NZDHKD": .03125
,"NZDJPY": .03125
,"SGDHKD": .03125
,"SGDJPY": .03125
,"TRYJPY": .03125
,"USDHKD": .03125
,"USDJPY": .03125
,"ZARJPY": .03125
}
self.fxPairs_MajorForexBooster_Asia = { \
"USDJPY": .037
,"AUDUSD": .037
,"NZDUSD": .037
,"EURJPY": .037
,"GBPJPY": .037
,"AUDCAD": .037
,"AUDCHF": .037
,"AUDHKD": .037
,"AUDJPY": .037
,"AUDNZD": .037
,"CADHKD": .037
,"CADJPY": .037
,"CHFHKD": .037
,"CHFJPY": .037
,"EURHKD": .037
,"EURNZD": .037
,"GBPHKD": .037
,"GBPNZD": .037
,"HKDJPY": .037
,"NZDHKD": .037
,"NZDJPY": .037
,"SGDHKD": .037
,"SGDJPY": .037
,"TRYJPY": .037
,"USDHKD": .037
,"USDJPY": .037
,"ZARJPY": .037
}
self.fxPairs_MajorTen = { \
"EURUSD": .1
,"USDJPY": .1
,"GBPUSD": .1
,"AUDUSD": .1
,"USDCAD": .1
,"USDCHF": .1
,"NZDUSD": .1
,"EURJPY": .1
,"GBPJPY": .1
,"EURGBP": .1
}
self.fxPairs_MajorTen_Main = { \
"EURUSD": .1
,"USDJPY": .1
,"GBPUSD": .1
,"AUDUSD": .1
,"USDCAD": .1
,"USDCHF": .1
,"NZDUSD": .1
,"EURJPY": .1
,"GBPJPY": .1
,"EURGBP": .1
}
self.fxPairs_MajorTen_OldStyle = { \
"EURUSD": .1
,"EURCAD": .1
,"CHFUSD": .1
,"EURAUD": .1
,"GBPNZD": .1
,"NZDEUR": .1
,"GBPEUR": .1
,"USDCAD": .1
,"AUDCAD": .1
,"AUDNZD": .1
}
self.fxPairs_MajorForexBooster_All = { \
"AUDCAD": .01429
,"AUDCHF": .01429
,"AUDHKD": .01429
,"AUDJPY": .01429
,"AUDNZD": .01429
,"AUDUSD": .01429
,"CADCHF": .01429
,"CADHKD": .01429
,"CADJPY": .01429
,"CADSGD": .01429
,"CHFHKD": .01429
,"CHFJPY": .01429
,"CHFZAR": .01429
,"EURAUD": .01429
,"EURCAD": .01429
,"EURCHF": .01429
,"EURCZK": .01429
,"EURDKK": .01429
,"EURGBP": .01429
,"EURHKD": .01429
,"EURHUF": .01429
,"EURJPY": .01429
,"EURNOK": .01429
,"EURNZD": .01429
,"EURPLN": .01429
,"EURSEK": .01429
,"EURSGD": .01429
,"EURTRY": .01429
,"EURUSD": .04289
,"EURZAR": .01429
,"GBPAUD": .01429
,"GBPCAD": .01429
,"GBPCHF": .01429
,"GBPHKD": .01429
,"GBPJPY": .01429
,"GBPNZD": .01429
,"GBPPLN": .01429
,"GBPSGD": .01429
,"GBPUSD": .01429
,"GBPZAR": .01429
,"HKDJPY": .01429
,"NZDCAD": .01429
,"NZDCHF": .01429
,"NZDHKD": .01429
,"NZDJPY": .01429
,"NZDSGD": .01429
,"NZDUSD": .01429
,"SGDCHF": .01429
,"SGDHKD": .01429
,"SGDJPY": .01429
,"TRYJPY": .01429
,"USDCAD": .01429
,"USDCHF": .01429
,"USDCNH": .01429
,"USDCZK": .01429
,"USDDKK": .01429
,"USDHKD": .01429
,"USDHUF": .01429
,"USDJPY": .01429
,"USDMXN": .01429
,"USDNOK": .01429
,"USDPLN": .01429
,"USDSEK": .01429
,"USDSGD": .01429
,"USDTHB": .01429
,"USDTRY": .01429
,"USDZAR": .01429
,"ZARJPY": .01429
}
self.fxPairs_Fx_Agile_Six = { \
# selected (green)
"EURUSD": .5
,"GBPUSD": .2
,"AUDUSD": .1
,"EURGBP": .1
,"HKDJPY": .05
,"USDJPY": .05
}
self.fxPairs_Fx_Agile_Six_Even = { \
# selected (green)
"EURUSD": .4
,"GBPUSD": .2
,"AUDUSD": .1
,"EURGBP": .1
,"HKDJPY": .1
,"USDJPY": .1
}
self.fxPairs_Fx_Agile_Six_Equal = { \
# selected (green)
"EURUSD": .5
,"GBPUSD": .1
,"AUDUSD": .1
,"EURGBP": .1
,"HKDJPY": .1
,"USDJPY": .1
}
self.fxPairs_Fx_Agile_Five_Equal = { \
"EURUSD": .5
,"GBPUSD": .2
,"AUDUSD": .1
,"HKDJPY": .1
,"USDJPY": .1
}
self.fxPairs_Fx_Test = { \
"EURUSD": 1
#"GBPUSD": 1
#"AUDUSD": 1
#"EURGBP": 1
#"HKDJPY": 1
#"USDJPY": 1
#"USDCAD": 1
#"USDCHF": 1
#"EURNZD": 1
#"EURAUD": 1
#"AUDCAD": 1
#"AUDCHF": 1
#"AUDHKD": 1
#"AUDJPY": 1
#"AUDNZD": 1
#"CADCHF": 1
#"CADHKD": 1
#"CADJPY": 1
#"CADSGD": 1
#"CHFHKD": 1
#"CHFJPY": 1
#"CHFZAR": 1
#"EURCAD": 1
#"EURCHF": 1
#"EURCZK": 1
#"EURDKK": 1
#"EURHKD": 1
#"EURHUF": 1
#"EURJPY": 1
#"EURNOK": 1
#"EURPLN": 1
#"EURSEK": 1
#"EURSGD": 1
#"EURTRY": 1
#"EURZAR": 1
#"GBPAUD": 1
#"GBPCAD": 1
#"GBPCHF": 1
#"GBPHKD": 1
#"GBPJPY": 1
#"GBPNZD": 1
#"GBPPLN": 1
#"GBPSGD": 1
#"GBPZAR": 1
#"NZDCAD": 1
#"NZDCHF": 1
#"NZDHKD": 1
#"NZDJPY": 1
#"NZDSGD": 1
#"NZDUSD": 1
#"SGDCHF": 1
#"SGDHKD": 1
#"SGDJPY": 1
#"TRYJPY": 1
#"USDCNH": 1
#"USDCZK": 1
#"USDDKK": 1
#"USDHKD": 1
#"USDHUF": 1
#"USDMXN": 1
#"USDNOK": 1
#"USDPLN": 1
#"USDSEK": 1
#"USDSGD": 1
#"USDTHB": 1
#"USDTRY": 1
#"USDZAR": 1
#"ZARJPY": 1
}
# ----------------------------------------------------------------------------
def selectGalaxy(self, galaxy_type = None):
# ----------------------------------------------------------------------------
if galaxy_type.strip().upper() == "Test".strip().upper():
return self.fxPairs_Fx_Test
if galaxy_type.strip().upper() == "EurUsd".strip().upper():
return self.fxPairs_EurUsd
if galaxy_type.strip().upper() == "EurUsd_GbpUsd".strip().upper():
return self.fxPairs_EURUSD_GBPUSD
if galaxy_type.strip().upper() == "EurUsd_GbpUsd_E".strip().upper():
return self.fxPairs_EURUSD_GBPUSD_EQUAL
if galaxy_type.strip().upper() == "MajorFive".strip().upper():
return self.fxPairs_MajorFive
if galaxy_type.strip().upper() == "AgileSix".strip().upper():
return self.fxPairs_Fx_Agile_Six
if galaxy_type.strip().upper() == "AgileSixEven".strip().upper():
return self.fxPairs_Fx_Agile_Six_Even
if galaxy_type.strip().upper() == "AgileSixEqual".strip().upper():
return self.fxPairs_Fx_Agile_Six_Equal
if galaxy_type.strip().upper() == "AgileFiveEqual".strip().upper():
return self.fxPairs_Fx_Agile_Five_Equal
if galaxy_type.strip().upper() == "MagicTrinity".strip().upper():
return self.fxPairs_MagicTrinity
if galaxy_type.strip().upper() == "EurGbp".strip().upper():
return self.fxPairs_EurGbp
if galaxy_type.strip().upper() == "EurUsd_Orig".strip().upper():
return self.fxPairs_EurUsd_Orig
if galaxy_type.strip().upper() == "EURUSD_USDCHF".strip().upper():
return self.fxPairs_EURUSD_USDCHF
if galaxy_type.strip().upper() == "MajorForexBooster".strip().upper():
return self.fxPairs_MajorForexBooster
if galaxy_type.strip().upper() == "MajorForexBooster_Equal".strip().upper():
return self.fxPairs_MajorForexBooster_Equal
if galaxy_type.strip().upper() == "MFB_Equal".strip().upper():
return self.fxPairs_MajorForexBooster_Equal
if galaxy_type.strip().upper() == "MajorForexBooster_All".strip().upper():
return self.fxPairs_MajorForexBooster_All
if galaxy_type.strip().upper() == "MajorForexBooster_Asia".strip().upper():
return self.fxPairs_MajorForexBooster_Asia
if galaxy_type.strip().upper() == "MajorForexBooster_Asia_Ext".strip().upper():
return self.fxPairs_MajorForexBooster_Asia_Ext
if galaxy_type.strip().upper() == "MajorForexBooster_Cool".strip().upper():
return self.fxPairs_MajorForexBooster_Cool
if galaxy_type.strip().upper() == "MFB_Cool".strip().upper():
return self.fxPairs_MajorForexBooster_Cool
if galaxy_type.strip().upper() == "MFB_Cool_Variant".strip().upper():
return self.fxPairs_MajorForexBooster_Cool_Variant
if galaxy_type.strip().upper() == "MajorForexBooster_Dude".strip().upper():
return self.fxPairs_MajorForexBooster_Dude
if galaxy_type.strip().upper() == "MFB_Dude".strip().upper():
return self.fxPairs_MajorForexBooster_Dude
if galaxy_type.strip().upper() == "MajorForexBooster_BigDude".strip().upper():
return self.fxPairs_MajorForexBooster_BigDude
if galaxy_type.strip().upper() == "MFB_BigDude".strip().upper():
return self.fxPairs_MajorForexBooster_BigDude
if galaxy_type.strip().upper() == "MajorForexBooster_JPY".strip().upper():
return self.fxPairs_MajorForexBooster_JPY
if galaxy_type.strip().upper() == "MajorForexBooster_Orig".strip().upper():
return self.fxPairs_MajorForexBooster_Orig
if galaxy_type.strip().upper() == "MajorTen".strip().upper():
return self.fxPairs_MajorTen
if galaxy_type.strip().upper() == "MajorTen_Double".strip().upper():
return self.fxPairs_MajorTen_Double
if galaxy_type.strip().upper() == "MajorTen_Ext1".strip().upper():
return self.fxPairs_MajorTen_Ext1
if galaxy_type.strip().upper() == "MajorTen_Main".strip().upper():
return self.fxPairs_MajorTen_Main
if galaxy_type.strip().upper() == "MajorTen_OldStyle".strip().upper():
return self.fxPairs_MajorTen_OldStyle
if galaxy_type.strip().upper() == "MajorTen_Once".strip().upper():
return self.fxPairs_MajorTen_Once
if galaxy_type.strip().upper() == "MajorTen_Orig".strip().upper():
return self.fxPairs_MajorTen_Orig
# default galaxy
return self.fxPairs_EurUsd# general imports
from math import floor
from datetime import timedelta
from datetime import datetime
import math
# specific imports
from AlgoToolbox import AlgoToolbox
from ForexGalaxyBay import ForexGalaxyBay
from FxEmaStar import FxEmaStar
# **************************************************
class FxTridentBase(QCAlgorithm):
# **************************************************
#-------------------------------------------------------------------
def Initialize(self):
#-------------------------------------------------------------------
# init tools
#-------------------------------------
self.show_log = self.get_bool_param("show_log",False,False)
self.show_debug = self.get_bool_param("show_debug",False,False)
self.algo_toolbox = AlgoToolbox(self, self.show_debug, self.show_log)
self.toolbox = self.algo_toolbox
self.toolbox.show_log("------------------------------------")
self.toolbox.show_log("ENVIRONMENT:")
self.toolbox.show_log("------------------------------------")
self.toolbox.show_log("Algorithm:" + self.get_parameter("algo_name", "?"))
# init params
#-------------------------------------
self.setDateMargins()
self.fx_broker = self.get_str_param("fx_broker")
self.fx_galaxy = self.get_str_param("fx_galaxy")
self.SetCash(self.get_int_param("cash"))
self.ema_red = self.get_int_param("ema_red")
self.ema_blue = self.get_int_param("ema_blue")
self.ema_green = self.get_int_param("ema_green")
self.leverage = self.get_int_param("leverage")
# trade control
self.period_main = self.get_int_param("period_min") * 4
self.anchor_left_shift_min = 0
self.scalp_anchor_relation = 1/4
self.anchor_max_slots = 10
self.scalp_max_slots = 10
self.inverse_prediction = False
self.is_trailing = True
self.resolution = Resolution.Minute
self.anchorResolution = Resolution.Minute
self.ticker_resolution = Resolution.Minute
self.take_profit_pips = self.get_int_param("take_profit_pips")
self.stop_loss_pips = self.get_int_param("stop_loss_pips")
self.ptt_pips_touch_tolerance = self.get_float_param("touch_pip_tolerance")
self.pet_pips_equal_tolerance = self.get_float_param("equal_pip_tolerance")
# derived params
#-------------------------------------
self.lookback = self.period_main
self.anchor_multiplier = 1 / self.scalp_anchor_relation
self.history_range = self.lookback * self.anchor_multiplier
self.insight_min = self.lookback
self.anchor_period = self.period_main * self.anchor_multiplier
# states vars
#-------------------------------------
self.period_main_signal = 0
self.anchor_period_signal = 0
self.anchor_warmup = False
self.last_data = None
# trade vars
self.global_lifetime_profit = 0
# call initializing methods
#-------------------------------------
# create fx galaxy of major forex booster pairs
self.forex_galaxy = self.selectGalaxyStars()
# warming up
self.SetWarmUp(timedelta(15))
# *******************************************
# REGION Main METHODS (BEGIN)
# *******************************************
#-------------------------------------------------------------------
def OnData(self, data):
#-------------------------------------------------------------------
# init
self.last_data = data
#validation
if self.IsWarmingUp: return
# check anchor period
self.anchor_period_signal = 0
self.OnData_AnchorJob(data)
# check scalp period
if self.period_main_signal < self.period_main - 1:
self.period_main_signal = self.period_main_signal + 1
else:
self.period_main_signal = 0
self.OnData_ScalpJob(data)
#-------------------------------------------------------------------
def OnData_ScalpJob(self, data):
#-------------------------------------------------------------------
for fx_star in self.forex_galaxy:
self.forex_galaxy[fx_star].trade_ondata(data)
#-------------------------------------------------------------------
def OnData_AnchorJob(self, data):
#-------------------------------------------------------------------
pass
# -----------------------------------------------------------------------------------------------------------
def selectGalaxyStars(self):
# -----------------------------------------------------------------------------------------------------------
selected_galaxy = self.selectGalaxy(self.fx_galaxy)
result = {}
# create galaxy of fx symbol stars
for fxTickerStar in selected_galaxy:
fxStarWeight = selected_galaxy[fxTickerStar]
result[fxTickerStar] = FxEmaStar(self, fxTickerStar, fxStarWeight,
self.lookback, self.resolution, self.insight_min,
self.show_debug, self.show_log)
# finally
return result
# -----------------------------------------------------------------------------------------------------------
def selectGalaxy(self, galaxy_type):
# -----------------------------------------------------------------------------------------------------------
galaxy = ForexGalaxyBay()
return galaxy.selectGalaxy(galaxy_type)
# *******************************************
# REGION Main METHODS (END)
# *******************************************
# -----------------------------------------------------------------------------------------------------------
def get_bool_param(self, param, default=False, show_log=True):
# -----------------------------------------------------------------------------------------------------------
result = bool(self.get_parameter(param, default))
if show_log:
self.log_param(param, result)
return result
# -----------------------------------------------------------------------------------------------------------
def get_int_param(self, param, default=0, show_log=True):
# -----------------------------------------------------------------------------------------------------------
result = int(self.get_parameter(param, default))
if show_log:
self.log_param(param, result)
return result
# -----------------------------------------------------------------------------------------------------------
def get_float_param(self, param, default=0.0, show_log=True):
# -----------------------------------------------------------------------------------------------------------
result = float(self.get_parameter(param, default))
if show_log:
self.log_param(param, result)
return result
# -----------------------------------------------------------------------------------------------------------
def get_str_param(self, param, default="", show_log=True):
# -----------------------------------------------------------------------------------------------------------
result = self.get_parameter(param, default)
if show_log:
self.log_param(param, result)
return result
# -----------------------------------------------------------------------------------------------------------
def get_parameter(self, param, default = None):
# -----------------------------------------------------------------------------------------------------------
result = str(default)
try:
result = self.GetParameter(param)
except:
err_msg = param + " not found!"
self.toolbox.show_log(err_msg)
self.Debug(err_msg)
return result
# -----------------------------------------------------------------------------------------------------------
def log_param(self, param, value):
# -----------------------------------------------------------------------------------------------------------
if self.toolbox is not None:
self.toolbox.show_log(param + "=" + str(value))
# -----------------------------------------------------------------------------------------------------------
def setDateMargins(self):
# -----------------------------------------------------------------------------------------------------------
# set start date
start_date = self.toolbox.get_date(self.get_str_param("start_date"),"-")
if start_date[0] > 0:
self.SetStartDate(start_date[0], start_date[1], start_date[2])
# set end date of not empty
end_date = self.toolbox.get_date(self.get_str_param("end_date"),"-")
if end_date[0] > 0:
self.SetEndDate(end_date[0], end_date[1], end_date[2])