| Overall Statistics |
|
Total Trades 90 Average Win 1.37% Average Loss -0.23% Compounding Annual Return 1006.353% Drawdown 12.600% Expectancy 5.358 Net Profit 107.824% Sharpe Ratio 13.016 Probabilistic Sharpe Ratio 96.266% Loss Rate 8% Win Rate 92% Profit-Loss Ratio 5.94 Alpha 6.73 Beta -1.754 Annual Standard Deviation 0.522 Annual Variance 0.272 Information Ratio 12.799 Tracking Error 0.534 Treynor Ratio -3.875 Total Fees $0.00 Estimated Strategy Capacity $9700000.00 |
from AlgoToolbox import AlgoToolbox
'''
To use this library place this at the top:
from StrategyFxScalper import StrategyFxScalper
Then instantiate the function:
StrategyFxScalpCustom(StrategyFxScalper):
...
startegy = StrategyFxScalpCustom()
strategy.Run(data)
'''
# ********************************
class StrategyFxScalper:
# ********************************
# -------------------------------------------------------------------------------------------------------------
def __init__(self, qcAlgo, fxGalaxy, resolution, anchor_resolution,
scalp_period, anchor_period, show_debug, show_log, anchor_slots): #, trade_units):
# -------------------------------------------------------------------------------------------------------------
self.buy_flag = 0
self.sell_flag = 0
self.limitOrderTicket = None
self.profitTargetOrderTicket = None
self.stopLossOrderTicket = None
self.scalp_ema_fast = 0
self.scalp_ema_slow = 0
self.anchor_ema_fast = 0
self.anchor_ema_slow = 0
self.qcAlgo = qcAlgo
self.fxGalaxy = fxGalaxy
self.BrickData = {}
self.fxHoldingCoeff = 1 / len(fxGalaxy)
self.resolution = resolution
self.anchorResolution = anchor_resolution
self.scalp_period = scalp_period
self.anchor_period = anchor_period
self.anchor_max_slots = anchor_slots
#self.trade_units_amount = trade_units
self.holding_value = 0
self.showDebug = show_debug
self.showLog = show_log
self.toolbox = AlgoToolbox(self.qcAlgo, self.showDebug, self.showLog)
# -------------------------------------------------------------------------------------------------------------
def Run(self, data, scalp_call = False, anchor_call = False):
# -------------------------------------------------------------------------------------------------------------
# validation
# exit if warming up
if self.qcAlgo.IsWarmingUp:
return
# init
ctxData = data
if ctxData is None:
ctxData = self.BrickData
# loop through all currencies in the galaxy
for fxStarBrick in self.fxGalaxy.values():
# assign current strategy
fxStarBrick.strategy = self
# debug if needed
if self.showDebug:
if scalp_call or anchor_call: # debug for scalp/anchor live calls only!
if fxStarBrick.IsChartHour(ctxData.Time): # debug for the chart time only
self.debugRunFx(ctxData, fxStarBrick)
# save brick data
msg_no_brick_data = "No data for " + fxStarBrick.symbol.Value + "!!! Stop Run-Workflow ...";
err_no_brick_data = "ERROR: No data for " + fxStarBrick.symbol.Value + "!!! Stop Run-Workflow. Error-Message: ";
data_exists = ctxData.ContainsKey(fxStarBrick.symbol)
if data_exists:
#try:
self.BrickData = ctxData
# live-run strategy for actual currency (live)
if not scalp_call and not anchor_call:
self.RunFx(ctxData, fxStarBrick)
# live-run strategy for actual currency (scalp period) if allowed
if scalp_call:
self.RunFxScalp(ctxData, fxStarBrick)
# live-run strategy for actual currency (anchor period) if allowed
if anchor_call:
self.RunFxAnchor(ctxData, fxStarBrick)
#except:
# # EXIT IF NO SYMBOL-DATA!!!
# self.apriori_debug(err_no_brick_data + sys.exc_info()[0])
# self.apriori_log(err_no_brick_data + sys.exc_info()[0])
# return
else:
# EXIT IF NO SYMBOL-DATA!!!
self.apriori_log(msg_no_brick_data)
self.apriori_debug(msg_no_brick_data)
return
# -------------------------------------------------------------------------------------------------------------
def RunScalp(self, fxBrick):
# -------------------------------------------------------------------------------------------------------------
# validation
# exit if warming up
if self.qcAlgo.IsWarmingUp:
return
# go!
self.RunFxScalp(self.BrickData, fxBrick)
# -------------------------------------------------------------------------------------------------------------
def RunAnchor(self, fxBrick):
# -------------------------------------------------------------------------------------------------------------
# validation
# exit if warming up
if self.qcAlgo.IsWarmingUp:
return
# go!
self.RunFxAnchor(self.BrickData, fxBrick)
# base (override in custom)
# -------------------------------------------------------------------------------------------------------------
def RunFx(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
# init
fxBrick.UpdateScalpFluentData(data)
fxBrick.UpdateAnchorFluentData(data)
# go!
pass
# base (override in custom)
# -------------------------------------------------------------------------------------------------------------
def RunFxScalp(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
# debug if needed
if self.showDebug:
self.debugRunFx(data, fxBrick)
# init
fxBrick.ResetScalpFluentData()
# go!
# check trades
self.CheckTradeRules(data, fxBrick)
# -------------------------------------------------------------------------------------------------------------
def CheckTradeRules(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
fxBrick.Check_Trade_Rules(data)
# base (override in custom)
# -------------------------------------------------------------------------------------------------------------
def RunFxAnchor(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
# debug if needed
if self.showDebug:
self.debugRunFx(data, fxBrick)
# init
fxBrick.ResetAnchorFluentData()
fxBrick.AnchorIsWarmedUp()
# go!
pass
# ****************************************************
# * TRADE RULES REGION (BASE / BEGIN) *
# ****************************************************
# -------------------------------------------------------------------------------------------------------------
def OnTrigger_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTrigger_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTriggerDrop_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTriggerDrop_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnEntry_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnEntry_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnEntry_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnExit_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnExit_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnStopLossAndTakeProfit_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnStopLossAndTakeProfit_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnFinishTrade_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnFinishTrade_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnStopLoss_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnStopLoss_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeContinue_Long(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeContinue_Short(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
pass
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# ****************************************************
# * TRADE RULES REGION TRADE RULES (BASE / END) *
# ****************************************************
# ******************************************************
# * TRADE-ACTION RULES REGION TRADE RULES (BASE / BEGIN) *
# ******************************************************
# -------------------------------------------------------------------------------------------------------------
def OnTradeOpen_Long_Holdings(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
if self.qcAlgo.emit_trades:
# init
fxBrick.trade_action("OPEN_LONG")
# BUY via SetHoldings()
self.holding_value = 1 * self.qcAlgo.forex_leverage * self.qcAlgo.cash_max_ratio * fxBrick.weight
self.qcAlgo.SetHoldings(fxBrick.symbol, self.holding_value)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeOpen_Long(self, data, fxBrick, set_limits=True):
# -------------------------------------------------------------------------------------------------------------
if self.qcAlgo.emit_trades:
# reset limits
fxBrick.reset_limits()
# !!! prevent initial canceling of the limit orders !!!
#fxBrick.veto_cancel_orders()
# init
fxBrick.trade_action("OPEN_LONG")
price = fxBrick.data.Close
# set absolute open level
fxBrick.trade_abs_start_level = price
fxBrick.trade_abs_stop_level = 0
# BUY via order (LONG)
# get holgig coeffitient
self.holding_value = 1 * self.qcAlgo.forex_leverage * self.qcAlgo.cash_max_ratio * fxBrick.weight
# get weighted rounded quantity
quantity = (self.qcAlgo.Portfolio.Cash / price) * self.holding_value
rounded_quantity = fxBrick.GetRoundedOrderSize(quantity)
# place long order (buy)
self.qcAlgo.Buy(fxBrick.symbol, rounded_quantity)
if set_limits:
# calc by PIP rounde price
take_profit = self.pip_round(fxBrick.trade_take_profit)
stop_loss = self.pip_round(fxBrick.stop_loss_level)
# take profit long
fxBrick.ProfitTarget = self.qcAlgo.LimitOrder(fxBrick.symbol, -rounded_quantity, take_profit)
# stop loss long
fxBrick.StopLoss = self.qcAlgo.StopMarketOrder(fxBrick.symbol, -rounded_quantity, stop_loss)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeClose_Long(self, data, fxBrick, closeRatio = 1):
# closeRatio = 1 - close all holdings, 0.5 - close 50% of holdings, 0 - nothing to close
# -------------------------------------------------------------------------------------------------------------
# avoid part closing on 1. exit (part-close)
if closeRatio < 1 and not self.qcAlgo.exit_1_close:
return
# init
price = fxBrick.data.Close
# set absolute open level
fxBrick.trade_abs_stop_level = price
if self.qcAlgo.emit_trades:
# close trade on given ratio
fxBrick.trade_action("CLOSE_LONG_" + str(closeRatio * 100) + "_PERCENT")
self.trade_close(data, fxBrick, closeRatio)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeOpen_Short_Holdings(self, data, fxBrick):
# -------------------------------------------------------------------------------------------------------------
if self.qcAlgo.emit_trades:
fxBrick.trade_action("OPEN_SHORT")
# SELL via SetHoldings()
self.holding_value = -1 * self.qcAlgo.forex_leverage * self.qcAlgo.cash_max_ratio * fxBrick.weight
self.qcAlgo.SetHoldings(fxBrick.symbol, self.holding_value)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeOpen_Short(self, data, fxBrick, set_limits=True):
# -------------------------------------------------------------------------------------------------------------
if self.qcAlgo.emit_trades:
# reset limits
fxBrick.reset_limits()
# !!! prevent initial canceling of the limit orders !!!
#fxBrick.veto_cancel_orders()
# init
fxBrick.trade_action("OPEN_SHORT")
price = fxBrick.data.Close
# set absolute open level
fxBrick.trade_abs_start_level = price
fxBrick.trade_abs_stop_level = 0
# SELL via order (SHORT)
# get holgig coeffitient
self.holding_value = self.qcAlgo.forex_leverage * self.qcAlgo.cash_max_ratio * fxBrick.weight
# get weighted rounded quantity
quantity = (self.qcAlgo.Portfolio.Cash / price) * self.holding_value
rounded_quantity = fxBrick.GetRoundedOrderSize(quantity)
# place short order (sell)
self.qcAlgo.Sell(fxBrick.symbol, rounded_quantity)
if set_limits:
# calc by PIP rounde price
take_profit = self.pip_round(fxBrick.trade_take_profit)
stop_loss = self.pip_round(fxBrick.stop_loss_level)
# take profit short
fxBrick.ProfitTarget = self.qcAlgo.LimitOrder(fxBrick.symbol, rounded_quantity, take_profit)
# stop loss short
fxBrick.StopLoss = self.qcAlgo.StopMarketOrder(fxBrick.symbol, rounded_quantity, stop_loss)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def OnTradeClose_Short(self, data, fxBrick, closeRatio = 1):
# closeRatio = 1 - close all holdings, 0.5 - close 50% of holdings, 0 - nothing to close
# -------------------------------------------------------------------------------------------------------------
# avoid part closing on 1. exit (part-close)
if closeRatio < 1 and not self.qcAlgo.exit_1_close:
return
# init
price = fxBrick.data.Close
# set absolute open level
fxBrick.trade_abs_stop_level = price
if self.qcAlgo.emit_trades:
# adjust short coeff
if self.holding_value > 0:
self.holding_value = -1 * self.holding_value
# close trade on given ratio
fxBrick.trade_action("CLOSE_SHORT_" + str(closeRatio * 100) + "_PERCENT")
self.trade_close(data, fxBrick, closeRatio)
fxBrick.Emit_Insight()
# TODO 1: Execute Rule
# TODO 2: Plot Data if executed
# -------------------------------------------------------------------------------------------------------------
def trade_close(self, data, fxBrick, closeRatio = 1):
# closeRatio = 1 - close all holdings, 0.5 - close 50% of holdings, 0 - nothing to close
# -------------------------------------------------------------------------------------------------------------
current_holding = 1 - closeRatio
if current_holding < 0:
current_holding = 0
if current_holding > 1:
current_holding = 1
# re-balance holding coefficient
self.holding_value = current_holding * self.holding_value
# close via SetHoldings()
self.qcAlgo.SetHoldings(fxBrick.symbol, self.holding_value)
# get profit benchmarks
fxBrick.get_profit_benchmarks()
# ******************************************************
# * TRADE-ACTION RULES REGION TRADE RULES (BASE / END) *
# ******************************************************
# -------------------------------------------------------------------------------------------------------------
def debugRunFx(self, bigData, fxStarBrick):
# -------------------------------------------------------------------------------------------------------------
try:
data = bigData[fxStarBrick.symbol]
# validation
# exit if chart is disabled
if not fxStarBrick.IsChartTime(data.Time):
return
# go!
self.debug("===================================================" )
self.debug("Run:" )
self.debug("Brick-Data: " + str(data.Time) + " : " + str(fxStarBrick.symbol))
self.debug("Current OHLC: " + str(data.Open) + " / " + str(data.High) + " / " \
+ str(data.Low) + " / " + str(data.Close))
self.debug("Fluent-Anchor OHLC: " \
+ str(fxStarBrick.anchor_fluent_open) + " / " \
+ str(fxStarBrick.anchor_fluent_high) + " / " \
+ str(fxStarBrick.anchor_fluent_low) + " / " \
+ str(fxStarBrick.anchor_fluent_close))
for idx in range(fxStarBrick.scalp_max_slots):
self.debug("TS-"+str(idx)+": " + str(fxStarBrick.scalpQuoteBar_RollingWindow[idx].Time))
self.debug("S-RW-"+str(idx)+": " + str(fxStarBrick.scalpQuoteBar_RollingWindow[idx]))
self.debug("-----------------------------------------------" )
self.debug("Scalp Indicators:")
for ind_rw_key in fxStarBrick.scalpIndicators_RollingWindows:
self.debug(str("S-IND:" + ind_rw_key))
rw = fxStarBrick.scalpIndicators_RollingWindows[ind_rw_key]
# show current indicator rolling window for current currency
for i in range(fxStarBrick.scalp_max_slots + 1):
self.debug(str(rw[i]))
self.debug("-------------------------------------------------------------------------------" )
for idx in range(fxStarBrick.anchor_max_slots):
self.debug("TA-"+str(idx)+": " + str(fxStarBrick.anchorQuoteBar_RollingWindow[idx].Time))
self.debug("A-RW-"+str(idx)+": " + str(fxStarBrick.anchorQuoteBar_RollingWindow[idx]))
self.debug("-----------------------------------------------" )
self.debug("Anchor Indicators:")
for ind_rw_key in fxStarBrick.anchorIndicators_RollingWindows.keys():
self.debug(str("A-IND:" + ind_rw_key))
rw = fxStarBrick.anchorIndicators_RollingWindows[ind_rw_key]
# show current indicator rolling window for current currency
for i in range(fxStarBrick.anchor_max_slots + 1):
self.debug(str(rw[i]))
self.debug("-------------------------------------------------------------------------------" )
except:
self.debug("Error debugRunFx() with" + str(fxStarBrick.symbol))
# -------------------------------------------------------------------------------------------------------------
def debug(self, msg):
# -------------------------------------------------------------------------------------------------------------
self.toolbox.show_debug(msg)
# -------------------------------------------------------------------------------------------------------------
def log(self, msg):
# -------------------------------------------------------------------------------------------------------------
self.toolbox.show_log(msg)
# -------------------------------------------------------------------------------------------------------------
def Debug(self, msg):
# -------------------------------------------------------------------------------------------------------------
self.toolbox.show_debug(msg)
# -------------------------------------------------------------------------------------------------------------
def apriori_debug(self, msg):
# -------------------------------------------------------------------------------------------------------------
self.toolbox.Debug(msg)
# -------------------------------------------------------------------------------------------------------------
def apriori_log(self, msg):
# -------------------------------------------------------------------------------------------------------------
self.toolbox.Log(msg)
# -------------------------------------------------------------------------------------------------------------
def pip_round(self, price):
# -------------------------------------------------------------------------------------------------------------
return self.toolbox.pip_round(price)
# *********************************
# * PLOT REGION (BEGIN) *
# *********************************
# -------------------------------------------------------------------------------------------------------------
def plot_scalp_chart(self, fxBrick, bar):
# -------------------------------------------------------------------------------------------------------------
self.plot_price_chart(fxBrick.scalp_chart.Name, fxBrick, bar)
# -------------------------------------------------------------------------------------------------------------
def plot_anchor_chart(self, fxBrick, bar):
# -------------------------------------------------------------------------------------------------------------
self.plot_price_chart(fxBrick.anchor_chart.Name, fxBrick, bar)
# -------------------------------------------------------------------------------------------------------------
def plot_scalp_ema_cross(self, fxBrick, value):
# -------------------------------------------------------------------------------------------------------------
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_ema_cross_point, value)
# -------------------------------------------------------------------------------------------------------------
def plot_anchor_ema_cross(self, fxBrick, value):
# -------------------------------------------------------------------------------------------------------------
self.plot(fxBrick.anchor_chart.Name, fxBrick.price_chart_serial_ema_cross_point, value)
# -------------------------------------------------------------------------------------------------------------
def plot_price_chart(self, chart_name, fxBrick, bar):
# -------------------------------------------------------------------------------------------------------------
self.plot(chart_name, fxBrick.price_chart_serial_open, bar.Open)
self.plot(chart_name, fxBrick.price_chart_serial_close, bar.Close)
self.plot(chart_name, fxBrick.price_chart_serial_close_consolidated, bar.Close)
self.plot(chart_name, fxBrick.price_chart_serial_open_price_point, bar.Open)
self.plot(chart_name, fxBrick.price_chart_serial_high_price_point, bar.High)
self.plot(chart_name, fxBrick.price_chart_serial_low_price_point, bar.Low)
self.plot(chart_name, fxBrick.price_chart_serial_close_price_point, bar.Close)
# -------------------------------------------------------------------------------------------------------------
def plot_trigger(self, fxBrick, value, isShort):
# -------------------------------------------------------------------------------------------------------------
if isShort:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_trigger_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_trigger_point, value)
# -------------------------------------------------------------------------------------------------------------
def plot_entry(self, fxBrick, value, isShort, isPoint):
# -------------------------------------------------------------------------------------------------------------
if isShort:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_entry_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_entry, value)
else:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_entry_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_entry, value)
# -------------------------------------------------------------------------------------------------------------
def plot_stop_loss(self, fxBrick, value, isShort, isPoint):
# -------------------------------------------------------------------------------------------------------------
if isShort:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_stop_loss_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_stop_loss, value)
else:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_stop_loss_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_stop_loss, value)
# -------------------------------------------------------------------------------------------------------------
def plot_exit_1(self, fxBrick, value, isShort, isPoint):
# -------------------------------------------------------------------------------------------------------------
if isShort:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_exit_1_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_exit_1, value)
else:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_exit_1_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_exit_1, value)
# -------------------------------------------------------------------------------------------------------------
def plot_exit_2(self, fxBrick, value, isShort, isPoint):
# -------------------------------------------------------------------------------------------------------------
if isShort:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_exit_2_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_short_exit_2, value)
else:
if isPoint:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_exit_2_point, value)
else:
self.plot(fxBrick.scalp_chart.Name, fxBrick.price_chart_serial_long_exit_2, value)
# -------------------------------------------------------------------------------------------------------------
def plot(self, chart_name, serial_name, value):
# -------------------------------------------------------------------------------------------------------------
self.qcAlgo.Plot(chart_name, serial_name, value)
# *********************************
# * PLOT REGION (END) *
# *********************************# custom inpots
from ForexTrendBase import ForexTrendBase
from FxRedBirdBrick_South import FxRedBirdBrick_South
from StrategyFxRedBird_South import StrategyFxRedBird_South
# ********************************
class ForexRedBird_South(ForexTrendBase):
# ********************************
# -----------------------------------------------------------
# Initializer (Cosntructor)
# ------------------------------------------------------------
def Initializie(self):
# ------------------------------------------------------------
# base initializer
super(ForexRedBird_South, self).Initializie()
# main initializer
self.SetBrokerageModel(AlphaStreamsBrokerageModel())
# setup forex brokerages
if self.Broker == "FXCM":
self.SetBrokerageModel(BrokerageName.FxcmBrokerage)
else:
self.SetBrokerageModel(BrokerageName.OandaBrokerage)
# ------------------
# custom selector method
# -----------------------------------------------------------------------------------------------------------
def selectStrategy(self):
# -----------------------------------------------------------------------------------------------------------
return StrategyFxRedBird_South(self, self.forex_galaxy, self.resolution, self.anchorResolution, self.scalp_period,
self.anchor_period, self.show_debug, self.show_log, self.anchor_max_slots)
# ------------------
# custom selector
# -----------------------------------------------------------------------------------------------------------
def selectGalaxyStars(self):
# -----------------------------------------------------------------------------------------------------------
# init strategy galaxy
selected_galaxy = self.selectGalaxy(self.get_str_param("selected_galaxy"))
#self.toolbox.show_log
result = {}
# create galaxy of fx symbol bricks
for fxTickerStar in selected_galaxy:
fxStarWeight = selected_galaxy[fxTickerStar]
result[fxTickerStar] = FxRedBirdBrick_South(self, fxTickerStar, fxStarWeight,
self.resolution, self.anchorResolution,
self.scalp_period, self.anchor_period,
self.show_debug, self.show_log,
self.anchor_max_slots, self.scalp_max_slots)
# finally
return result
# -----------------------------------------------------------------------------------------------------------
def OnOrderEvent(self, orderEvent):
# -----------------------------------------------------------------------------------------------------------
return
'''
# init
filledOrderId = orderEvent.OrderId
overrule_veto = False
# looping all symbol bricks
for fxBrickSymbol in self.forex_galaxy:
if fxBrickSymbol == str(orderEvent.Symbol):
fxBrick = self.forex_galaxy[fxBrickSymbol]
#if fxBrick.Ignore_Cancel_Event:
# if fxBrick.trade_state == fxBrick.STAND_BY:
# return
#else:
show_action = False #not fxBrick.Ignore_Cancel_Event
state_reset = False #not fxBrick.Ignore_Cancel_Event
# If the ProfitTarget order was filled, close the StopLoss order
if fxBrick.match_order(fxBrick.ProfitTarget, filledOrderId):
if fxBrick.cancel_stop_loss(overrule_veto, show_action, state_reset):
fxBrick.trade_action_filled_tp_cancel_sl()
# If the StopLoss order was filled, close the ProfitTarget
if fxBrick.match_order(fxBrick.StopLoss, filledOrderId):
if fxBrick.cancel_take_profit(overrule_veto, show_action, state_reset):
fxBrick.trade_action_filled_sl_cancel_tp()
'''from ForexSymbolBrick import ForexSymbolBrick
from QuantConnect.Data import *
from QuantConnect import *
class FxRedBirdBrick_South(ForexSymbolBrick):
def __init__(self,qcAlgo, ticker, weight, resolution, anchorResolution, scalpPeriod, anchorPeriod,
showDebug = False, showLog = False, anchor_max_slots = 24, scalp_max_slots = 10):
super(FxRedBirdBrick_South, self).__init__(qcAlgo, ticker, weight, resolution, anchorResolution, scalpPeriod, anchorPeriod, \
showDebug, showLog, anchor_max_slots, scalp_max_slots)
# params candidates
self.delta_accelerate_max_ratio = 3 #5 #3 #2.7 # 0 deactivate
self.orange_risk_invert_weight = 5
self.red_risk_shift_percent = 10
self.exit2_level_min_steps_count = -1 #7 # deactivate = -1
self.RCT_pullback_trailing_stop_ratio = 0 # 0 = deactivate
# trailing props
self.TTG_trailing_trade_gain = 0
self.TLG_trailing_live_gain = 0
self.LLo_lowest_low_open = 0
self.LLc_lowest_low_close = 0
self.HHo_highest_high_open = 0
self.HHc_highest_high_close = 0
self.LVo_live_open = 0
self.LVc_live_close = 0
# delta props
self.minor_delta = 0
self.major_delta = 0
self.delta_acceleration = 0
# semafor props
self.IsLevel_Exit_1_Semafor = False
self.IsLevel_Exit_2_Semafor = False
self.exit2_level_steps_count = 0
# trade control props
self.trade_lowest_low = 0
self.trade_highest_high = 0
self.lowest_low_last_trailng_qbars = 0
self.highest_high_last_trailng_qbars = 0
self.touch_level = 0
self.exit_1_rate = self.qcAlgo.exit_1_rate
# machine states
self.EXIT_1_SHORT = self.EXIT_1_SHORT_BASE
self.EXIT_1_LONG = self.EXIT_1_LONG_BASE
self.EXIT_2_SHORT = self.EXIT_2_SHORT_BASE
self.EXIT_2_LONG = self.EXIT_2_LONG_BASE
# fast-ema trigger tolerance to bar touch delta (pips)
self.bar_to_fast_indicator_touch_delta_trigger_pips = self.qcAlgo.touch_trigger_pip_tolerance
# slow-ema stopper toleracnce to bar touch delta (pips)
self.bar_to_slow_inidcator_touch_delta_stopper_pips = self.qcAlgo.touch_trigger_pip_tolerance
# * TRADE RULES/EVENTS REGION (CUSTOM / BEGIN) *
def calc_trade_props(self, rolling_window_qoute_bar, lookback_steps, bar_touch_margins, trigger_pips_up, trigger_pips_down):
# long trade props
if self.is_long_trade:
self.stop_loss_level = self.touch_level
self.trade_highest_high = self.Get_Trade_Scalp_Highest_High(rolling_window_qoute_bar, lookback_steps, bar_touch_margins)
self.entry_level = self.plus_pip(self.trade_highest_high, trigger_pips_up)
self.trade_risk = abs(self.entry_level - self.stop_loss_level)
self.trade_break_even = self.entry_level + self.trade_risk
self.trade_exit = self.trade_break_even + self.trade_risk
self.trade_take_profit = self.trade_exit
# short trade props
if self.is_short_trade:
self.stop_loss_level = self.touch_level
self.trade_lowest_low = self.Get_Trade_Scalp_Lowest_Low(rolling_window_qoute_bar, lookback_steps, bar_touch_margins)
self.entry_level = self.plus_pip(self.trade_lowest_low, trigger_pips_down)
self.trade_risk = abs(self.entry_level - self.stop_loss_level)
self.trade_break_even = self.entry_level - self.trade_risk
self.trade_exit = self.trade_break_even - self.trade_risk
self.trade_take_profit = self.trade_exit
def invert_break_even(self):
# long trade props
if self.is_long_trade:
self.trade_break_even = self.entry_level - self.orange_risk_invert_weight * self.trade_risk
# short trade props
if self.is_short_trade:
self.trade_break_even = self.entry_level + self.orange_risk_invert_weight * self.trade_risk
def calc_on_break_even(self):
# common trade props (short/long)
self.stop_loss_level = self.entry_level
# self.trade_break_even = 0
def calc_diff_range(self, rw_qb, d_idx, u_idx):
o = 0
c = 0
for i in range(d_idx, u_idx):
if self.is_long_trade:
if o == 0 or o > rw_qb[i].Open:
o = rw_qb[i].Open
if c == 0 or c < rw_qb[i].Close:
c = rw_qb[i].Close
if self.is_short_trade:
if o == 0 or o < rw_qb[i].Open:
o = rw_qb[i].Open
if c == 0 or c > rw_qb[i].Close:
c = rw_qb[i].Close
return o, c
def calc_diff_delta(self, rw_qb):
# init
a_len_major = rw_qb.Size
a_len_minor = int(a_len_major / 2)
# go!
minor_o, minor_c = self.calc_diff_range(rw_qb, 0, a_len_minor)
major_o, major_c = self.calc_diff_range(rw_qb, a_len_minor, a_len_major)
self.minor_delta = minor_c - minor_o
self.major_delta = major_c - major_o
if self.minor_delta != 0:
self.delta_acceleration = abs(self.major_delta / self.minor_delta)
else:
self.delta_acceleration = 0
def Is_Support_Level(self, rw_qb):
if self.delta_accelerate_max_ratio == 0:
return False
else:
self.calc_diff_delta(rw_qb)
return self.major_delta < 0 and self.delta_acceleration > self.delta_accelerate_max_ratio
#return self.delta_acceleration > self.delta_accelerate_max_ratio
def Is_Resistance_Level(self, rw_qb):
if self.delta_accelerate_max_ratio == 0:
return False
else:
self.calc_diff_delta(rw_qb)
return self.major_delta > 0 and self.delta_acceleration > self.delta_accelerate_max_ratio
#return self.delta_acceleration > self.delta_accelerate_max_ratio
def calc_rolling_trade_props(self, rolling_window_qoute_bar, lookback_steps, bar_touch_margins, trigger_pips_up, trigger_pips_down):
force_open_close_touch_margins = "open_close"
# init
self.oc_lowest_low_last_trailng_qbars = self.Get_Trade_Scalp_Lowest_Low(rolling_window_qoute_bar, lookback_steps, force_open_close_touch_margins)
self.oc_highest_high_last_trailng_qbars = self.Get_Trade_Scalp_Highest_High(rolling_window_qoute_bar, lookback_steps, force_open_close_touch_margins)
self.TTG_trailing_trade_gain = abs(self.oc_highest_high_last_trailng_qbars - self.oc_lowest_low_last_trailng_qbars)
self.TLG_trailing_live_gain = self.scalp_fluent_close
# LONG trailing stop loss
if self.trade_state == self.EXIT_1_LONG:
self.lowest_low_last_trailng_qbars = self.Get_Trade_Scalp_Lowest_Low(rolling_window_qoute_bar, lookback_steps, bar_touch_margins)
self.stop_loss_level = self.plus_pip(self.lowest_low_last_trailng_qbars, trigger_pips_down)
self.trade_exit = 0 # exit via stop_loss only, while in trailing_exit-mode
self.trade_take_profit = self.trade_exit
# SHORT trailing stop loss
if self.trade_state == self.EXIT_1_SHORT:
self.highest_high_last_trailng_qbars = self.Get_Trade_Scalp_Highest_High(rolling_window_qoute_bar, lookback_steps, bar_touch_margins)
self.stop_loss_level = self.plus_pip(self.highest_high_last_trailng_qbars, trigger_pips_up)
self.trade_exit = 0 # exit via stop_loss only, while in trailing_exit-mode
self.trade_take_profit = self.trade_exit
# calc lookback profit
def update_hh_ll_limits(self):
#init
live_s_o = self.scalp_fluent_open
live_s_c = self.scalp_fluent_close
LLo = self.LLo_lowest_low_open
LLc = self.LLc_lowest_low_close
HHo = self.HHo_highest_high_open
HHc = self.HHc_highest_high_close
#go!
self.LVo_live_open = live_s_o
self.LVc_live_close = live_s_c
if self.is_long_trade:
if LLo == 0 or LLo > live_s_o:
self.LLo_lowest_low_open = live_s_o
if HHc == 0 or HHc < live_s_c:
self.HHc_highest_high_close = live_s_c
if self.is_short_trade:
if LLc == 0 or LLc > live_s_c:
self.LLc_lowest_low_close = live_s_c
if HHo == 0 or HHo < live_s_o:
self.HHo_highest_high_open = live_s_o
def Is_Pullback_Trailing_Exit(self):
# init
result = False
# compare
# long trade props
if self.is_long_trade:
if self.LVc_live_close < self.LVo_live_open:
self.TLG_trailing_live_gain = self.HHc_highest_high_close - self.LVc_live_close
self.TTG_trailing_trade_gain = self.HHc_highest_high_close - self.LLo_lowest_low_open
result = self.LLo_lowest_low_open > 0 and self.HHc_highest_high_close > 0 \
and self.LVo_live_open > self.LLo_lowest_low_open \
and abs(self.TLG_trailing_live_gain / self.TTG_trailing_trade_gain) > self.RCT_pullback_trailing_stop_ratio
# short trade props
if self.is_short_trade:
if self.LVc_live_close > self.LVo_live_open:
self.TLG_trailing_live_gain = self.LVc_live_close - self.LLc_lowest_low_close
self.TTG_trailing_trade_gain = self.HHo_highest_high_open - self.LLc_lowest_low_close
result = self.HHo_highest_high_open > 0 and self.LLc_lowest_low_close > 0 \
and self.LVo_live_open < self.HHo_highest_high_open \
and abs(self.TLG_trailing_live_gain / self.TTG_trailing_trade_gain) > self.RCT_pullback_trailing_stop_ratio
# finally
return result
def Check_Trade_Rules(self, data):
# base
super(FxRedBirdBrick_South, self).Check_Trade_Rules(data)
# init environment vars
rw_ai_fast = self.anchorIndicators_RollingWindows["ema_8_anchor"]
rw_ai_slow = self.anchorIndicators_RollingWindows["ema_21_anchor"]
rw_si_fast = self.scalpIndicators_RollingWindows["ema_8_scalp"]
rw_si_slow = self.scalpIndicators_RollingWindows["ema_21_scalp"]
rw_a_qb = self.anchorQuoteBar_RollingWindow
rw_s_qb = self.scalpQuoteBar_RollingWindow
trailing_exit_stop_bars_count = self.qcAlgo.trailing_exit_stop_loss_bars
live_s_time = self.scalp_fluent_time
live_s_o = self.scalp_fluent_open
live_s_h = self.scalp_fluent_high
live_s_l = self.scalp_fluent_low
live_s_c = self.scalp_fluent_close
live_s_value = live_s_c
post_s_o = rw_s_qb[0].Open
post_s_h = rw_s_qb[0].High
post_s_l = rw_s_qb[0].Low
post_s_c = rw_s_qb[0].Close
prev_s_o = rw_s_qb[1].Open
prev_s_h = rw_s_qb[1].High
prev_s_l = rw_s_qb[1].Low
prev_s_c = rw_s_qb[1].Close
live_a_time = self.anchor_fluent_time
live_a_o = self.anchor_fluent_open
live_a_h = self.anchor_fluent_high
live_a_l = self.anchor_fluent_low
live_a_c = self.anchor_fluent_close
live_a_value = live_a_c
live_ai_fast = self.anchor_ema_current_value(8)
live_ai_slow = self.anchor_ema_current_value(21)
live_si_fast = self.scalp_ema_current_value(8)
live_si_mid = self.scalp_ema_current_value(13)
live_si_slow = self.scalp_ema_current_value(21)
s_narrow = self.is_narrow_ema_scalp
s_delta_pips = self.scalp_emas_pips_delta
s_delta_min_pips = self.qcAlgo.scalp_delta_min_pips
a_delta = self.anchor_emas_pips_delta
a_delta_min_pips = self.qcAlgo.anchor_delta_min_pips
bar_touch_margins = self.qcAlgo.bar_touch_margins
tolerance = self.qcAlgo.touch_trigger_pip_tolerance
trigger_touch_tolerance = self.bar_to_fast_indicator_touch_delta_trigger_pips
stopper_touch_tolerance = self.bar_to_slow_inidcator_touch_delta_stopper_pips
trigger_pips_up = self.qcAlgo.pullback_trigger_pips_up
trigger_pips_down = self.qcAlgo.pullback_trigger_pips_down
exit_trailing = self.qcAlgo.exit_strategy_trailing_exit
exit_break_even = self.qcAlgo.exit_strategy_break_even_exit
take_profit_limit = self.qcAlgo.take_profit
stop_loss_limit = self.qcAlgo.take_profit
# exit_trailing startegy has bigger prioriry
if exit_trailing and exit_break_even:
exit_break_even = False
exit_trailing = True
else:
if not exit_trailing and not exit_break_even:
exit_trailing = True
exit_break_even = False
# Go!
if self.check_trade_allowed:
# Init trade section vars
self.trade_confidence = 1
live_idx = 0
prior_idx = 1
level = live_s_value
# * *
# * HANDLE TRADE STATE-MACHINE *
# * *
if self.trade_state == self.STAND_BY:
# --- handle STAND-BY (begin)
self.IsLevel_TradeEnabled_Semafors[live_idx] = \
self.IsLevel_TradeEnabled(live_ai_fast, live_ai_slow, live_a_o, live_a_h, live_a_l, live_a_c, tolerance, bar_touch_margins)
live_trade_enabled = self.IsLevel_TradeEnabled_Semafors[live_idx]
self.Indicator_Deltas[live_idx] = abs(live_ai_fast - live_ai_slow)
live_indicator_delta = self.Indicator_Deltas[live_idx]
self.Direction_Semafors[live_idx] = \
self.Check_Trade_Direction(live_ai_fast, live_ai_slow, live_a_o, live_a_h, live_a_l, live_a_c, tolerance, bar_touch_margins)
live_trade_direction = self.Direction_Semafors[live_idx]
# check if trade is enabled
if live_trade_enabled:
# check anchor history rules
hist_trade_enabled = True
hist_trade_direction_match = True
hist_ind_expanded = True
hist_ind_expanded_sum = 0
hist_ind_expanded_count = 0
hist_ind_expanded_avg = 0
for ha_idx in range(1, self.anchor_max_slots + 1):
# init
self.Indicator_Deltas[ha_idx] = 0
qb_idx = ha_idx - 1
hist_ai_fast = self.value(rw_ai_fast[ha_idx])
hist_ai_slow = self.value(rw_ai_slow[ha_idx])
hist_indicator_delta = abs(hist_ai_fast - hist_ai_slow)
# calc history enablers
self.IsLevel_TradeEnabled_Semafors[ha_idx] = \
self.IsLevel_TradeEnabled(hist_ai_fast , hist_ai_slow, \
rw_a_qb[qb_idx].Open, rw_a_qb[qb_idx].High, rw_a_qb[qb_idx].Low, rw_a_qb[qb_idx].Close, \
tolerance, bar_touch_margins)
# calc history direction