| Overall Statistics |
|
Total Orders 1324 Average Win 0.15% Average Loss -0.29% Compounding Annual Return -57.022% Drawdown 20.600% Expectancy -0.066 Start Equity 100000 End Equity 88323.33 Net Profit -11.677% Sharpe Ratio -2.051 Sortino Ratio -2.866 Probabilistic Sharpe Ratio 6.431% Loss Rate 38% Win Rate 62% Profit-Loss Ratio 0.52 Alpha -0.457 Beta -1.647 Annual Standard Deviation 0.238 Annual Variance 0.056 Information Ratio -1.583 Tracking Error 0.32 Treynor Ratio 0.296 Total Fees $4926.31 Estimated Strategy Capacity $630000.00 Lowest Capacity Asset UVXY V0H08FY38ZFP Portfolio Turnover 374.85% |
#region imports
from AlgorithmImports import *
#endregion
from AlgorithmImports import *
class SymbolData:
def __init__(self, symbol, k1, k2, range_period, consolidator_resolution):
self.symbol = symbol
self.range_window = RollingWindow[TradeBar](range_period)
self.consolidator = TradeBarConsolidator(consolidator_resolution)
def on_data_consolidated(sender, consolidated):
\
self.range_window.add(consolidated)
if self.range_window.is_ready:
hh = max([x.high for x in self.range_window])
hc = max([x.close for x in self.range_window])
lc = min([x.close for x in self.range_window])
ll = min([x.low for x in self.range_window])
range = max([hh - lc, hc - ll])
self.upper_line = consolidated.close + k1 * range
self.lower_line = consolidated.close - k2 * range
# event fired at new consolidated trade bar
self.consolidator.data_consolidated += on_data_consolidated
# Returns the interior consolidator
def get_consolidator(self):
return self.consolidator
@property
def is_ready(self):
return self.range_window.is_ready#region imports
from AlgorithmImports import *
#endregion
from AlgorithmImports import *
from SymbolData import *
# Your New Python File
class DualThrustAlphaModel(AlphaModel):
def __init__(self,
k1,
k2,
range_period,
resolution = Resolution.HOUR,
bars_to_consolidate = 1):
'''Initializes a new instance of the class
Args:
k1: Coefficient for upper band
k2: Coefficient for lower band
range_period: Amount of last bars to calculate the range
resolution: The resolution of data sent into the EMA indicators
bars_to_consolidate: If we want alpha to work on trade bars whose length is different
from the standard resolution - 1m 1h etc. - we need to pass this parameters along
with proper data resolution'''
# coefficient that used to determine upper and lower borders of a breakout channel
self.k1 = k1
self.k2 = k2
# period the range is calculated over
self.range_period = range_period
# initialize with empty dict.
self._symbol_data_by_symbol = dict()
# time for bars we make the calculations on
resolution_in_time_span = Extensions.to_time_span(resolution)
self.consolidator_time_span = Time.multiply(resolution_in_time_span, bars_to_consolidate)
# in 5 days after emission an insight is to be considered expired
self.period = timedelta(5)
def update(self, algorithm, data):
insights = []
for symbol, symbol_data in self._symbol_data_by_symbol.items():
if not symbol_data.is_ready:
continue
holding = algorithm.portfolio[symbol]
price = algorithm.securities[symbol].price
# buying condition
# - (1) price is above upper line
# - (2) and we are not long. this is a first time we crossed the line lately
if price > symbol_data.upper_line and not holding.is_long:
insight_close_time_utc = algorithm.utc_time + self.period
insights.append(Insight.price(symbol, insight_close_time_utc, InsightDirection.UP))
# selling condition
# - (1) price is lower that lower line
# - (2) and we are not short. this is a first time we crossed the line lately
if price < symbol_data.lower_line and not holding.is_short:
insight_close_time_utc = algorithm.utc_time + self.period
insights.append(Insight.price(symbol, insight_close_time_utc, InsightDirection.DOWN))
return insights
def on_securities_changed(self, algorithm, changes):
# added
for symbol in [x.symbol for x in changes.added_securities]:
if symbol not in self._symbol_data_by_symbol:
# add symbol/symbol_data pair to collection
symbol_data = self.SymbolData(symbol, self.k1, self.k2, self.range_period, self.consolidator_time_span)
self._symbol_data_by_symbol[symbol] = symbol_data
# register consolidator
algorithm.subscription_manager.add_consolidator(symbol, symbol_data.get_consolidator())
# removed
for symbol in [x.symbol for x in changes.removed_securities]:
symbol_data = self._symbol_data_by_symbol.pop(symbol, None)
if symbol_data is None:
algorithm.error("Unable to remove data from collection: DualThrustAlphaModel")
else:
# unsubscribe consolidator from data updates
algorithm.subscription_manager.remove_consolidator(symbol, symbol_data.get_consolidator())from AlgorithmImports import *
from VIXAlpha import *
from SymbolData import *
class SELLPUTFRAMEWORK(QCAlgorithm):
def initialize(self):
# -- STRATEGY INPUT PARAMETERS --
self.range_period = 22
self.bbl = 20
self.lb = 50
self.ph = 0.85
self.mult = 2.0
self.consolidator_bars = 20
# set leverage
self.leverage = 3
self.yoy_return = 0.6
# Settings
self.universe_settings.resolution = Resolution.MINUTE
self.set_start_date(2024,4,1)
self.set_cash(100000)
# Warming up
resolution_in_time_span = Extensions.to_time_span(self.universe_settings.resolution)
warm_up_time_span = Time.multiply(resolution_in_time_span, self.consolidator_bars)
self.set_warm_up(warm_up_time_span)
# Universe Selection
tickers = ['TQQQ', 'SQQQ', 'UVXY']
symbols = [ Symbol.create(ticker, SecurityType.EQUITY, Market.USA) for ticker in tickers]
self.set_universe_selection(ManualUniverseSelectionModel(symbols))
# Alpha Model
self.set_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(minutes = 20), 0.025, None))
## Portfolio Construction
self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(Resolution.DAILY))
## Execution
self.set_execution(ImmediateExecutionModel())
## Risk Management
self.set_risk_management(MaximumDrawdownPercentPerSecurity(0.01))
def on_order_event(self, order_event):
if order_event.status == OrderStatus.FILLED:
self.debug("Purchased Stock: {0}".format(order_event.symbol))