Backtest

Overall Statistics
Total Orders 827 Average Win 0.56% Average Loss -0.50% Compounding Annual Return 4.023% Drawdown 14.200% Expectancy 0.422 Start Equity 100000 End Equity 249741.91 Net Profit 149.742% Sharpe Ratio 0.093 Sortino Ratio 0.097 Probabilistic Sharpe Ratio 0.256% Loss Rate 33% Win Rate 67% Profit-Loss Ratio 1.11 Alpha 0.008 Beta -0.046 Annual Standard Deviation 0.056 Annual Variance 0.003 Information Ratio -0.351 Tracking Error 0.169 Treynor Ratio -0.115 Total Fees $3879.42 Estimated Strategy Capacity $200000.00 Lowest Capacity Asset TBX UVHSVFDDBO85 Portfolio Turnover 2.71% Drawdown Recovery 842

# region imports
from AlgorithmImports import *
from datetime import timedelta
import math
# endregion


class DeterminedLightBrownAlpaca(QCAlgorithm):

    def initialize(self):
        self.set_cash(100_000)
        self.set_start_date(2003, 1, 1)
        self.settings.automatic_indicator_warm_up = True
        self.settings.seed_initial_prices = True        
        tickers = [
            "AGZ", "BIL", "HYG", "IEF", "IEI", "IGIB", "IGLB", "JNK",
            "LQD", "MBB", "MINT", "TBF", "TBX", "TIP", "TLT",
        ]
        self._safety = self.add_equity("SHY", Resolution.DAILY)
        self._assets = []
        for ticker in tickers:
            security = self.add_equity(ticker, Resolution.DAILY)
            weighted_return_a = IndicatorExtensions.times(self.momp(security, 63), 0.33)
            weighted_return_b = IndicatorExtensions.times(self.momp(security, 21), 0.33)
            security.return_combo = IndicatorExtensions.plus(weighted_return_a, weighted_return_b)
            security.trend_sma = SimpleMovingAverage(10)
            self.register_indicator(security, security.trend_sma, TradeBarConsolidator(CalendarType.MONTHLY))
            vol_ = IndicatorExtensions.of(StandardDeviation(63), self.logr(security, 63))
            security.volatility = IndicatorExtensions.times(vol_, math.sqrt(252))
            volatility_penalty = IndicatorExtensions.times(security.volatility, 0.34)
            security.rank_score = IndicatorExtensions.minus(security.return_combo, volatility_penalty)
            self._assets.append(security)
        self.set_warm_up(timedelta(90))

    def on_warmup_finished(self):
        time_rule = self.time_rules.at(8, 0)
        self.schedule.on(self.date_rules.month_start(self._safety), time_rule, self._rebalance)
        if self.live_mode:
            self._rebalance()
        else:
            self.schedule.on(self.date_rules.today, time_rule, self._rebalance)

    def _rebalance(self):
        ready = [asset for asset in self._assets if asset.rank_score.is_ready and asset.trend_sma.is_ready]
        if not ready:
            return
        top_ranked = sorted(ready, key=lambda x: x.rank_score.current.value)[-3:]
        passing = [asset for asset in top_ranked if asset.close > asset.trend_sma.current.value]
        to_hold = passing + ([self._safety] if len(passing) < len(top_ranked) else [])
        # Clear the safety position if we have trend-qualified assets.
        if passing and self._safety.holdings.invested:
            self.liquidate(self._safety)
        # Safety is already invested and no new trend-qualified assets were selected.
        if self._safety.holdings.invested:
            return
        targets = [PortfolioTarget(security, 1.0 / len(to_hold)) for security in to_hold]
        self.set_holdings(targets, True)