| Overall Statistics |
|
Total Orders 149 Average Win 0.52% Average Loss -1.02% Compounding Annual Return -20.673% Drawdown 15.700% Expectancy -0.081 Start Equity 37700 End Equity 35225.42 Net Profit -6.564% Sharpe Ratio -0.742 Sortino Ratio -0.785 Probabilistic Sharpe Ratio 15.887% Loss Rate 39% Win Rate 61% Profit-Loss Ratio 0.51 Alpha -0.267 Beta 0.9 Annual Standard Deviation 0.241 Annual Variance 0.058 Information Ratio -1.558 Tracking Error 0.178 Treynor Ratio -0.199 Total Fees $165.94 Estimated Strategy Capacity $37000000.00 Lowest Capacity Asset WAG R735QTJ8XC9X Portfolio Turnover 27.59% |
# region imports
from AlgorithmImports import *
from alpha_1637 import Strategy001AlphaModel
from porfolio_1737 import Stragety001PortfolioConstructionModel
# endregion
class MeanReversionAlgorithm(QCAlgorithm):
def initialize(self):
# self.settings.daily_precise_end_time = False
self.set_start_date(2024, 5, 30)
self.set_end_date(2024, 12, 30)
self.set_cash(37700) # Set Strategy Cash
self.set_warm_up(timedelta(days=3), Resolution.DAILY)
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
self.add_universe_selection(ETFConstituentsUniverseSelectionModel("QQQ"))
self.universe_settings.resolution = Resolution.DAILY
self.set_benchmark("QQQ")
self.add_alpha(Strategy001AlphaModel())
self.set_portfolio_construction(Stragety001PortfolioConstructionModel())
self.set_execution(ImmediateExecutionModel())
#region imports
from AlgorithmImports import *
import time
import random
#endregion
class Strategy001AlphaModel(AlphaModel):
basket_capacity = 8
trade_bar_count = 0
trade_bar_none_count = 0
def __init__(self):
self.entry_rolling_window_dict = {}
self.exit_rolling_window_dict = {}
def update(self, algorithm, data):
self.entry_insights_update(algorithm, data)
self.exit_insights_update(algorithm, data)
if data.time.weekday() == 4: # close time on Friday
algorithm.debug(f"{self.trade_bar_none_count}/{self.trade_bar_count}={self.trade_bar_none_count / self.trade_bar_count} trade_bar is None {data.time.year}-{data.time.month}-{data.time.day} {data.time.hour}:{data.time.minute}:{data.time.second}")
algorithm.debug(f"len(data)={len(data)}")
if len(data) < 10:
for symbol, trade_bar in data.items():
algorithm.debug(f"{symbol.value}")
exit_insights_count = self.exit_insights_create(algorithm, data)
entry_insights_list = []
if data.time.weekday() == 0: # close time on Monday
entry_insights_list = self.entry_insights_create(algorithm, data)
securities_count = 0
for security_holding in algorithm.portfolio.values():
if security_holding.invested:
securities_count += 1
index = min(self.basket_capacity - (securities_count - exit_insights_count), self.basket_capacity)
entry_insights_list = entry_insights_list[:index]
return entry_insights_list
def entry_insights_create(self, algorithm, data):
entry_insights_list = []
for symbol, trade_bar in data.items():
if (self.entry_rolling_window_dict[symbol].count == 3) \
and (not algorithm.portfolio[symbol].invested):
if (self.entry_rolling_window_dict[symbol][0].close < self.entry_rolling_window_dict[symbol][1].close) \
and (self.entry_rolling_window_dict[symbol][1].close < self.entry_rolling_window_dict[symbol][2].close) \
and (self.entry_rolling_window_dict[symbol][0].close > 5.0):
entry_insights_list.append(
Insight(
symbol=symbol,
period=timedelta(days=365 * 10),
type=InsightType.PRICE,
direction=InsightDirection.UP,
magnitude=self.entry_rolling_window_dict[symbol][0].volume,
confidence=1.0))
sorted_active_insights = sorted(entry_insights_list, key=lambda x: x.magnitude, reverse=True)
return sorted_active_insights
def exit_insights_create(self, algorithm, data):
exit_insights_count = 0
for symbol, trade_bar in data.items():
if (self.exit_rolling_window_dict[symbol].count == 2) and algorithm.portfolio[symbol].invested:
if self.exit_rolling_window_dict[symbol][0].close > self.exit_rolling_window_dict[symbol][1].high:
assert algorithm.insights.contains_key(symbol)
assert len(algorithm.insights[symbol]) == 1
algorithm.insights[symbol][0].expire(algorithm.utc_time)
exit_insights_count += 1
return exit_insights_count
def exit_insights_update(self, algorithm, data):
for symbol, trade_bar in data.items():
if trade_bar is None:
# algorithm.error(f"{symbol.value}:trade_bar is None, reset exit_rolling_window")
self.trade_bar_none_count += 1
self.exit_rolling_window_dict[symbol].reset()
else:
self.exit_rolling_window_dict[symbol].add(trade_bar)
self.trade_bar_count += 1
def entry_insights_update(self, algorithm, data):
if data.time.weekday() in [3, 4, 0]: # close time on Thursday, Friday and Monday
for symbol, trade_bar in data.items():
if trade_bar is None:
# algorithm.error(f"{symbol.value}:trade_bar is None, reset entry_rolling_window")
self.trade_bar_none_count += 1
self.entry_rolling_window_dict[symbol].reset()
else:
self.entry_rolling_window_dict[symbol].add(trade_bar)
self.trade_bar_count += 1
def on_securities_changed(self, algorithm, changes):
for security in changes.added_securities:
self.entry_rolling_window_dict.setdefault(security.symbol, RollingWindow[TradeBar](3))
self.exit_rolling_window_dict.setdefault(security.symbol, RollingWindow[TradeBar](2))
#region imports
from AlgorithmImports import *
import time
from alpha_1637 import Strategy001AlphaModel
#endregion
# Portfolio construction scaffolding class; basic method args.
class Stragety001PortfolioConstructionModel(EqualWeightingPortfolioConstructionModel):
total_position = 1.6
# Determines the target percent for each insight
def determine_target_percent(self, activeInsights: List[Insight]) -> Dict[Insight, float]:
targets = {}
for insight in activeInsights:
if not self.algorithm.portfolio[insight.symbol].invested: # new targets
if insight.direction is InsightDirection.UP:
targets[insight] = self.total_position / Strategy001AlphaModel.basket_capacity
return targets