| Overall Statistics |
|
Total Trades 324 Average Win 0.78% Average Loss -0.23% Compounding Annual Return 28.565% Drawdown 2.500% Expectancy 1.008 Net Profit 42.840% Sharpe Ratio 1.236 Probabilistic Sharpe Ratio 66.560% Loss Rate 54% Win Rate 46% Profit-Loss Ratio 3.34 Alpha 0.202 Beta 0.011 Annual Standard Deviation 0.165 Annual Variance 0.027 Information Ratio 0.068 Tracking Error 0.291 Treynor Ratio 18.48 Total Fees $0.00 Estimated Strategy Capacity $3800000.00 Lowest Capacity Asset BTCUSD XJ |
from typing import Dict, List
from AlgorithmImports import *
from datetime import timedelta
class ExampleAlpha(QCAlgorithm):
bar_period = timedelta(hours=4)
window_size = 52
resolution = Resolution.Hour
market = Market.GDAX
def Initialize(self):
if not self.LiveMode:
self.DebugMode = True
self.SetCash(10000.00)
self.SetStartDate(2020, 1, 1)
self.SetEndDate(2021, 6, 1)
crypto_tickers = ["BTCUSD", "ETHUSD"]
symbols=[Symbol.Create(x, SecurityType.Crypto, self.market) for x in crypto_tickers]
self.UniverseSettings.Resolution = self.resolution
if self.market == Market.GDAX:
self.UniverseSettings.Leverage = 1
elif self.market == Market.Kraken:
self.UniverseSettings.Leverage = 3
self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))
self.SetAlpha(ExampleAlphaModel(self, self.bar_period, self.window_size, self.resolution))
self.SetPortfolioConstruction(InsightWeightingPortfolioConstructionModel())
self.SetExecution(ImmediateExecutionModel())
self.SetRiskManagement(NullRiskManagementModel())
class IndicatorTwo(PythonIndicator):
@property
def ready(self) -> bool:
return self.tradebar_window.Count > self.WarmUpPeriod
def __init__(self):
self.name = "IndicatorTwo"
self.Period = 52
self.Value = 0.0
PythonIndicator.__init__(self.name, self.Period)
self.WarmUpPeriod = 52
self._obv = OnBalanceVolume()
self._slow_length = 13
self._fast_length = 7
self._ema_slow = ExponentialMovingAverage("ema_slow", self._slow_length)
self._ema_fast = ExponentialMovingAverage("ema_fast", self._fast_length)
self.tradebar_window = RollingWindow[TradeBar](52)
self.fast_ema_window = RollingWindow[float](52)
self.slow_ema_window = RollingWindow[float](52)
self.obv_window = RollingWindow[float](52)
def Update(self, input: TradeBar) -> bool:
self.tradebar_window.Add(input)
self._obv.Update(input)
self._ema_slow.Update(input.EndTime, self._obv.Current.Value)
self._ema_fast.Update(input.EndTime, self._obv.Current.Value)
self.fast_ema_window.Add(self._ema_fast.Current.Value)
self.slow_ema_window.Add(self._ema_slow.Current.Value)
self.obv_window.Add(self._obv.Current.Value)
self.Value = self._obv.Current.Value
return self.ready
class IndicatorOne(PythonIndicator):
def ready(self) -> bool:
self.tradebar_window.Count > self.WarmUpPeriod
def __init__(self):
PythonIndicator.__init__("IndicatorOne", 20)
self.WarmUpPeriod = 20
self._obv = OnBalanceVolume()
self._slow_length = 13
self._fast_length = 7
self._ema_slow = ExponentialMovingAverage("ema_slow", self._slow_length)
self._ema_fast = ExponentialMovingAverage("ema_fast", self._fast_length)
self.tradebar_window = RollingWindow[TradeBar](20)
self.fast_ema_window = RollingWindow[float](20) #TODO
self.slow_ema_window = RollingWindow[float](20) #TODO
self.obv_window = RollingWindow[float](20) #TODO
self.Value = 0.0
def Update(self, input: TradeBar) -> bool:
self.tradebar_window.Add(input)
self._obv.Update(input)
self._ema_slow.Update(input.EndTime, self._obv.Current.Value)
self._ema_fast.Update(input.EndTime, self._obv.Current.Value)
self.fast_ema_window.Add(self._ema_fast.Current.Value)
self.slow_ema_window.Add(self._ema_slow.Current.Value)
self.obv_window.Add(self._obv.Current.Value)
self.Value = self._obv.Current.Value
return self.ready
class SymbolData:
@property
def ready(self) -> bool:
# return self.tradebar_window.IsReady and self.indicator_one.ready and self.indicator_two.ready
return self.tradebar_window.IsReady and self.indicator_one.ready #uncomment
@property
def is_band_positive(self) -> bool:
# return self.indicator_one.fast_ema_window[0] > self.indicator_one.slow_ema_window[0] and self.indicator_two.fast_ema_window[0] > self.indicator_two.slow_ema_window[0]
return self.indicator_one.fast_ema_window[0] > self.indicator_one.slow_ema_window[0]
def __init__(self, algorithm: QCAlgorithm, security: Security, bar_period: timedelta, window_size: int, resolution: Resolution):
self.algorithm = algorithm
self.security = security
self.symbol = security.Symbol
self.indicator_one = IndicatorOne()
# self.indicator_two = IndicatorTwo()
self.bar_period = bar_period
self.window_size = window_size
self.previous_direction = InsightDirection.Down
self.resolution = resolution
self.tradebar_window = RollingWindow[TradeBar](window_size)
self.consolidator = TradeBarConsolidator(bar_period)
self.consolidator.DataConsolidated += self.consolidator_handler
self.sample_count = 0
self.warming_up = True
self.consolidator_ready = False
# Warmup #
total_time = Time.Multiply(self.bar_period, float(self.window_size))
# algorithm.Debug(total_time)
history = algorithm.History(self.symbol, total_time, self.resolution)
# algorithm.Debug(history)
if history.empty:
algorithm.Debug("HIsToRY IS EmpTy")
#TODO: handle error
symbol_history = history.loc[self.symbol]
for idx, row in symbol_history.iterrows():
# algorithm.Debug(idx)
# algorithm.Debug(row)
tradebar = TradeBar(
idx,
self.symbol,
row["open"],
row["high"],
row["low"],
row["close"],
row["volume"],
self.bar_period
)
self.consolidator.Update(tradebar)
self.warming_up = False
def hook_up_consolidator(self, algorithm: QCAlgorithm):
algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.consolidator)
self.consolidator_ready = True
algorithm.Debug("Add consolidator")
def consolidator_handler(self, sender, tradebar):
self.indicator_one.Update(tradebar)
# self.indicator_two.Update(tradebar)
self.tradebar_window.Add(tradebar)
class ExampleAlphaModel(AlphaModel):
def __init__(self, algorithm: QCAlgorithm, bar_period: timedelta, window_size: int, resolution: Resolution):
self.symbol_data_keyed_by_symbol: Dict[Symbol, SymbolData] = {}
self.bar_period = bar_period
self.window_size = window_size
self.resolution = resolution
self.sample_count = 0
def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
insights = []
for key, symbol_data in self.symbol_data_keyed_by_symbol.items():
if symbol_data.security.Price == 0:
continue
if not symbol_data.ready:
continue
if not symbol_data.warming_up and not symbol_data.consolidator_ready:
symbol_data.hook_up_consolidator(algorithm)
continue
if symbol_data.consolidator_ready:
direction = InsightDirection.Up if symbol_data.is_band_positive else InsightDirection.Down
if direction != symbol_data.previous_direction:
new_insight = Insight(symbol_data.security.Symbol, timedelta(minutes=1), InsightType.Price, direction)
new_insight.Weight = 0.5
insights.append(new_insight)
symbol_data.previous_direction = direction
return insights
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges):
for added_security in changes.AddedSecurities:
symbol_data = SymbolData(algorithm, added_security, self.bar_period, self.window_size, self.resolution)
algorithm.Debug(symbol_data)
self.symbol_data_keyed_by_symbol[added_security.Symbol] = symbol_data
for removed_security in changes.RemovedSecurities:
symbol_data = self.symbol_data_keyed_by_symbol.pop(removed_security.Symbol, None)
if symbol_data is not None:
algorithm.SubscriptionManager.RemoveConsolidator(removed_security.Symbol, symbol_data.consolidator)