| Overall Statistics |
|
Total Orders 1807 Average Win 3.20% Average Loss -1.81% Compounding Annual Return 401.139% Drawdown 58.000% Expectancy 0.574 Start Equity 100000 End Equity 184940152.06 Net Profit 184840.152% Sharpe Ratio 4.051 Sortino Ratio 4.671 Probabilistic Sharpe Ratio 98.891% Loss Rate 43% Win Rate 57% Profit-Loss Ratio 1.76 Alpha 2.88 Beta 0.309 Annual Standard Deviation 0.718 Annual Variance 0.515 Information Ratio 3.879 Tracking Error 0.726 Treynor Ratio 9.419 Total Fees $5305761.48 Estimated Strategy Capacity $700000.00 Lowest Capacity Asset GDXU XJSPWMCOQ4BP Portfolio Turnover 37.18% Drawdown Recovery 195 |
# =============================================================================
# "Balanced-v2" — 4-sleeve deployable book, ONE daily account, fund-of-funds
# netting. Sleeves + book weights (sum=1.0, gross<=100%, never borrows):
# A4 reversal 0.35 (1-day cross-sectional low-reversal, mcap-weighted)
# gen263 momentum 0.25 (sector-neutral multi-horizon momentum + brakes)
# 531 momentum-breadth 0.25 (285-authoritative momentum + breadth + BIL sweep)
# kinfo regime 0.15 (0379+0273 Nasdaq regime rotation, *0.98 buffer)
# net_w[sym] = 0.35*a4 + 0.25*gen263 + 0.25*531 + 0.15*(0.98*kinfo) (gross<=1)
#
# Derived from the prior "BookB_gen263" book (A4/gen263/410/kinfo) by (1) keeping
# the A4, gen263, kinfo sleeves and the ENTIRE netting + T+1 MarketOnOpenOrder
# execution harness VERBATIM, (2) REMOVING the 410 Ledoit-Wolf ETF sleeve, and
# (3) ADDING a 531 momentum-breadth sleeve ported from S531_285_authoritative.py.
#
# EXECUTION CONTRACT (all sleeves): a sleeve NEVER calls Liquidate/SetHoldings/
# market_order. Each only WRITES its in-bucket target dict (sum<=1.0) + flags
# self._dirty. Risk-off / empty branches set that sleeve's dict = {} (531 sweeps
# its bucket to BIL, its own yield hedge) — never a book-wide liquidate. The
# shared _execute_net harness nets the four dicts, submits sells before buys, and
# fills everything via MarketOnOpenOrder at the T+1 open (decision on T close, no
# look-ahead). The harness scales each dict by its book weight.
#
# 531 uses its OWN breadth latch (allow_531/was_risk_off_531/max_stress_531) and
# its OWN band-ceiling history (band_hist_531) so it never corrupts gen263's
# identically-named state. It shares the read-only per-symbol EMA/ADX/stretch/
# close-window indicators (same band_len=189, hist_len=126, adx_period=14) that
# OnData already maintains — no new hot-path work.
#
# Dates: get_parameter -> default 2010-01-01 .. 2026-06-26.
# =============================================================================
from AlgorithmImports import *
import numpy as np
class MultiSleeveMomentumRotation(QCAlgorithm):
PRICE_WINDOW_SIZE = 210 # covers the largest lookback used (200) + buffer
LEVERAGE = 1.0 # target 100% of NAV in positions (margin)
CASH_BUFFER_PCT = 0.04 # always leave this fraction of NAV uninvested
# Simple flat-% intraday stop loss, checked every few minutes during
# market hours (requires Minute-resolution price data -- indicators
# below stay Resolution.DAILY, so the sleeve logic itself is unaffected).
# Basis = that day's open. Backtested at 15% as the best of {5,10,15,20}%
# tried on this strategy -- Pareto-beat the no-stop baseline on CAGR,
# drawdown, Sharpe, and Sortino all at once.
ENABLE_INTRADAY_SL = True
SL_PCT = 0.15
INTRADAY_CHECK_MINUTES = 5
def initialize(self) -> None:
self.set_start_date(2020, 1, 1)
self.set_end_date(2024, 9, 1)
#self.set_start_date(2024, 9, 1)
self.set_cash(100_000)
self.set_brokerage_model(
BrokerageName.INTERACTIVE_BROKERS_BROKERAGE,
AccountType.MARGIN,
)
self.set_benchmark("SPY")
self.settings.minimum_order_margin_portfolio_percentage = 0.0
# Static requirements compiled from the original decision tree.
all_tickers = ['AAPX', 'AGG', 'AGQ', 'AMZN', 'AMZU', 'AMZZ', 'BABA', 'BAM', 'BIL', 'BITX', 'BN', 'BND', 'BTAL', 'BX', 'COIN', 'CONL', 'FAS', 'FAZ', 'FBL', 'GDXD', 'GDXU', 'GGLL', 'HOOD', 'IBKR', 'IEF', 'IOO', 'KKR', 'KMLM', 'MA', 'META', 'NVDL', 'PLTR', 'PSQ', 'QLD', 'QQQ', 'RGTI', 'SCHW', 'SH', 'SOFI', 'SOXS', 'SPXL', 'SPY', 'SQQQ', 'TECL', 'TECS', 'TLT', 'TMF', 'TQQQ', 'TSLA', 'TSLR', 'UPRO', 'UVXY', 'V', 'VTR', 'VTV', 'WELL', 'XLF', 'XLK']
rsi_pairs = [('AGG', 20), ('AGQ', 10), ('BAM', 10), ('BN', 10), ('BND', 10), ('BX', 10), ('FAS', 10), ('GDXU', 10), ('HOOD', 10), ('IBKR', 10), ('IEF', 10), ('IOO', 10), ('KKR', 10), ('KMLM', 10), ('MA', 10), ('PSQ', 10), ('PSQ', 20), ('QQQ', 10), ('SCHW', 10), ('SH', 60), ('SOFI', 10), ('SPXL', 10), ('SPY', 10), ('SQQQ', 10), ('TLT', 10), ('TLT', 20), ('TQQQ', 10), ('V', 10), ('VTR', 10), ('VTV', 10), ('WELL', 10), ('XLF', 10), ('XLK', 10)]
# -- Subscribe & build per-ticker daily price history + RSI indicators
self._syms = {}
self._price_window = {}
for t in sorted(all_tickers):
try:
sym = self.add_equity(t, Resolution.MINUTE).symbol
except Exception as e:
self.log(f"WARNING: could not subscribe {t}: {e}")
continue
self._syms[t] = sym
self._price_window[t] = RollingWindow[float](self.PRICE_WINDOW_SIZE)
self._rsi = {}
for ticker, window in rsi_pairs:
if ticker not in self._syms:
continue
self._rsi[(ticker, window)] = self.rsi(
self._syms[ticker], window, MovingAverageType.WILDERS, Resolution.DAILY
)
self.set_warm_up(self.PRICE_WINDOW_SIZE + 5, Resolution.DAILY)
# -- Seed price windows from history so day-0 evaluation has data
for t, sym in self._syms.items():
hist = self.history(sym, self.PRICE_WINDOW_SIZE + 5, Resolution.DAILY)
if not hist.empty:
closes = hist["close"].values
for c in closes[::-1]: # oldest first push, so window[0] ends most recent
if c > 0:
self._price_window[t].add(float(c))
self._trade_count = 0
self._day_open = {} # Symbol -> today's open, for the intraday SL
# -- Feed daily closes into the rolling windows right after each
# session close, then compute the rebalance target and place
# MarketOnOpenOrders for the next session's open.
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.after_market_close("SPY", 0),
self._update_price_windows,
)
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.after_market_close("SPY", 1),
self._rebalance,
)
# -- Snapshot today's open for every held name, then poll every few
# minutes during market hours for the intraday stop loss.
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.after_market_open("SPY", 0),
self._snapshot_day_open,
)
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.every(timedelta(minutes=self.INTRADAY_CHECK_MINUTES)),
self._check_intraday_sl,
)
# ── Complete strategy decision tree ────────────────────────────
def _strategy_targets(self) -> dict:
if self._rsi_value('GDXU', 10) > 0.79:
target_allocation = self._allocate('GDXD')
elif self._rsi_value('GDXU', 10) < 0.3:
target_allocation = self._allocate('GDXU')
elif self._cumulative_return('QQQ', 90) > self._cumulative_return('QQQ', 70):
if self._cumulative_return('GDXU', 70) < self._cumulative_return('GDXU', 75):
target_allocation = self._allocate('GDXU')
elif self._close('SPY') > self._sma_close('SPY', 200):
if self._rsi_value('TQQQ', 10) > 0.79:
target_allocation = self._allocate('UVXY')
elif self._rsi_value('SPXL', 10) > 0.8:
target_allocation = self._allocate('UVXY')
elif self._rsi_value('SPY', 10) > 0.79:
if self._rsi_value('SPY', 10) > 0.81 or (self._rsi_value('IOO', 10) > 0.81 or (self._rsi_value('TQQQ', 10) > 0.81 or (self._rsi_value('VTV', 10) > 0.81 or self._rsi_value('XLF', 10) > 0.81))):
target_allocation = self._allocate('UVXY')
else:
target_allocation = self._combine_weighted([(0.75, self._allocate('UVXY')), (0.25, self._combine_equal(self._allocate('BIL'), self._allocate('BTAL')))])
elif self._rsi_value('IOO', 10) > 0.79:
if self._rsi_value('IOO', 10) > 0.81 or (self._rsi_value('TQQQ', 10) > 0.81 or (self._rsi_value('VTV', 10) > 0.81 or self._rsi_value('XLF', 10) > 0.81)):
target_allocation = self._allocate('UVXY')
else:
target_allocation = self._combine_weighted([(0.75, self._allocate('UVXY')), (0.25, self._combine_equal(self._allocate('BIL'), self._allocate('BTAL')))])
elif self._rsi_value('TQQQ', 10) > 0.79:
if self._rsi_value('TQQQ', 10) > 0.81 or (self._rsi_value('VTV', 10) > 0.81 or self._rsi_value('XLF', 10) > 0.81):
target_allocation = self._allocate('UVXY')
else:
target_allocation = self._combine_weighted([(0.75, self._allocate('UVXY')), (0.25, self._combine_equal(self._allocate('BIL'), self._allocate('BTAL')))])
elif self._rsi_value('VTV', 10) > 0.79:
if self._rsi_value('VTV', 10) > 0.81 or self._rsi_value('XLF', 10) > 0.81:
target_allocation = self._allocate('UVXY')
else:
target_allocation = self._combine_weighted([(0.75, self._allocate('UVXY')), (0.25, self._combine_equal(self._allocate('BIL'), self._allocate('BTAL')))])
elif self._rsi_value('XLF', 10) > 0.79:
if self._rsi_value('XLF', 10) > 0.81:
target_allocation = self._allocate('UVXY')
else:
target_allocation = self._combine_weighted([(0.75, self._allocate('UVXY')), (0.25, self._combine_equal(self._allocate('BIL'), self._allocate('BTAL')))])
elif self._rsi_value('TQQQ', 10) < 0.3:
target_allocation = self._allocate('TQQQ')
elif self._rsi_value('SPY', 10) < 0.3:
target_allocation = self._allocate('SPXL')
elif self._close('SPY') > self._sma_close('SPY', 200):
if self._rsi_value('XLK', 10) > self._rsi_value('KMLM', 10):
if self._sma_return('AAPX', 10) > 0.0:
aapx_candidate = self._allocate('AAPX')
else:
aapx_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('NVDL', 10) > 0.0:
nvdl_candidate = self._allocate('NVDL')
else:
nvdl_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('BITX', 10) > 0.0:
bitx_candidate = self._allocate('BITX')
else:
bitx_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('TSLA', 10) > 0.0:
tslr_candidate = self._allocate('TSLR')
else:
tslr_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('META', 10) > 0.0:
fbl_candidate = self._allocate('FBL')
else:
fbl_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('GGLL', 10) > 0.0:
ggll_candidate = self._allocate('GGLL')
else:
ggll_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('AMZN', 10) > 0.0:
amzu_candidate = self._allocate('AMZU')
else:
amzu_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('RGTI', 10) > 0.0:
rgti_candidate = self._allocate('RGTI')
else:
rgti_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('PLTR', 10) > 0.0:
pltr_candidate = self._allocate('PLTR')
else:
pltr_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('BABA', 10) > 0.0:
baba_candidate = self._allocate('BABA')
else:
baba_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
if self._sma_return('COIN', 10) > 0.0:
conl_candidate = self._allocate('CONL')
else:
conl_candidate = self._combine_equal(self._allocate('BIL'), self._allocate('TQQQ'))
ranked_candidates = [(self._std_return('AAPX', 20), aapx_candidate), (self._std_return('NVDL', 20), nvdl_candidate), (self._std_return('BITX', 20), bitx_candidate), (self._std_return('TSLA', 20), tslr_candidate), (self._std_return('META', 20), fbl_candidate), (self._std_return('GGLL', 20), ggll_candidate), (self._std_return('AMZN', 20), amzu_candidate), (self._std_return('RGTI', 20), rgti_candidate), (self._std_return('PLTR', 20), pltr_candidate), (self._std_return('BABA', 20), baba_candidate), (self._std_return('COIN', 20), conl_candidate)]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
elif self._close('KMLM') < self._sma_close('KMLM', 20):
ranked_candidates = [(self._sma_return('AAPX', 15), self._allocate('AAPX')), (self._sma_return('NVDL', 15), self._allocate('NVDL')), (self._sma_return('BITX', 15), self._allocate('BITX')), (self._sma_return('TSLR', 15), self._allocate('TSLR')), (self._sma_return('FBL', 15), self._allocate('FBL')), (self._sma_return('GGLL', 15), self._allocate('GGLL')), (self._sma_return('AMZZ', 15), self._allocate('AMZZ')), (self._sma_return('RGTI', 15), self._allocate('RGTI')), (self._sma_return('PLTR', 15), self._allocate('PLTR')), (self._sma_return('BABA', 15), self._allocate('BABA')), (self._sma_return('CONL', 15), self._allocate('CONL'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
else:
target_allocation = self._combine_equal(self._allocate('TECS'), self._allocate('SOXS'), self._allocate('SQQQ'))
else:
if self._rsi_value('TLT', 20) > self._rsi_value('PSQ', 20):
primary_hedge_allocation = self._allocate('QQQ')
elif self._close('TQQQ') > self._sma_close('TQQQ', 20):
if self._rsi_value('PSQ', 10) < 0.35:
primary_hedge_allocation = self._allocate('PSQ')
elif self._rsi_value('AGG', 20) > self._rsi_value('SH', 60):
primary_hedge_allocation = self._allocate('TQQQ')
else:
primary_hedge_allocation = self._allocate('PSQ')
elif self._rsi_value('IEF', 10) > self._rsi_value('PSQ', 20):
primary_hedge_allocation = self._allocate('PSQ')
else:
primary_hedge_allocation = self._allocate('SQQQ')
if self._cumulative_return('QQQ', 60) < -0.12:
if self._rsi_value('BND', 10) > self._rsi_value('QQQ', 10):
secondary_hedge_allocation = self._allocate('QLD')
else:
secondary_hedge_allocation = self._allocate('BTAL')
elif self._close('TQQQ') > self._sma_close('TQQQ', 20):
if self._rsi_value('PSQ', 10) < 0.35:
secondary_hedge_allocation = self._allocate('PSQ')
elif self._rsi_value('AGG', 20) > self._rsi_value('SH', 60):
secondary_hedge_allocation = self._allocate('TQQQ')
else:
secondary_hedge_allocation = self._allocate('PSQ')
elif self._rsi_value('IEF', 10) > self._rsi_value('PSQ', 20):
secondary_hedge_allocation = self._allocate('PSQ')
else:
secondary_hedge_allocation = self._allocate('SQQQ')
target_allocation = self._combine_equal(primary_hedge_allocation, secondary_hedge_allocation)
elif self._rsi_value('TQQQ', 10) < 0.31:
target_allocation = self._allocate('TECL')
elif self._rsi_value('SPY', 10) < 0.3:
target_allocation = self._allocate('UPRO')
elif self._close('TQQQ') < self._sma_close('TQQQ', 20):
ranked_candidates = [(self._rsi_value('SQQQ', 10), self._allocate('SQQQ')), (self._rsi_value('TLT', 10), self._allocate('TLT'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
elif self._rsi_value('SQQQ', 10) < 0.31:
target_allocation = self._allocate('SQQQ')
else:
target_allocation = self._allocate('TQQQ')
elif self._cumulative_return('TLT', 95) < self._cumulative_return('QQQ', 35):
target_allocation = self._allocate('GDXD')
elif self._close('SPY') > self._sma_close('SPY', 200):
if self._sma_return('FAS', 50) > self._sma_return('FAS', 200):
if self._sma_return('FAS', 10) > self._sma_return('QQQ', 20):
fas_candidate = self._allocate('FAS')
else:
ranked_candidates = [(self._sma_return('AAPX', 15), self._allocate('AAPX')), (self._sma_return('NVDL', 15), self._allocate('NVDL')), (self._sma_return('BITX', 15), self._allocate('BITX')), (self._sma_return('TSLR', 15), self._allocate('TSLR')), (self._sma_return('FBL', 15), self._allocate('FBL')), (self._sma_return('GGLL', 15), self._allocate('GGLL')), (self._sma_return('AMZZ', 15), self._allocate('AMZZ')), (self._sma_return('RGTI', 15), self._allocate('RGTI')), (self._sma_return('PLTR', 15), self._allocate('PLTR')), (self._sma_return('BABA', 15), self._allocate('BABA')), (self._sma_return('CONL', 15), self._allocate('CONL'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
fas_candidate = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
ranked_candidates = [(self._sma_return('V', 20), self._allocate('V')), (self._sma_return('SOFI', 20), self._allocate('SOFI')), (self._sma_return('MA', 20), self._allocate('MA')), (self._sma_return('BX', 20), self._allocate('BX')), (self._sma_return('SCHW', 20), self._allocate('SCHW')), (self._sma_return('KKR', 20), self._allocate('KKR')), (self._sma_return('BN', 20), self._allocate('BN')), (self._sma_return('WELL', 20), self._allocate('WELL')), (self._sma_return('VTR', 20), self._allocate('VTR')), (self._sma_return('BAM', 20), self._allocate('BAM')), (self._sma_return('HOOD', 20), self._allocate('HOOD')), (self._sma_return('IBKR', 20), self._allocate('IBKR')), (self._sma_return('FAS', 20), fas_candidate)]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:3]])
else:
ranked_candidates = [(self._rsi_value('V', 10), self._allocate('V')), (self._rsi_value('SOFI', 10), self._allocate('SOFI')), (self._rsi_value('MA', 10), self._allocate('MA')), (self._rsi_value('BX', 10), self._allocate('BX')), (self._rsi_value('SCHW', 10), self._allocate('SCHW')), (self._rsi_value('KKR', 10), self._allocate('KKR')), (self._rsi_value('BN', 10), self._allocate('BN')), (self._rsi_value('WELL', 10), self._allocate('WELL')), (self._rsi_value('VTR', 10), self._allocate('VTR')), (self._rsi_value('BAM', 10), self._allocate('BAM')), (self._rsi_value('HOOD', 10), self._allocate('HOOD')), (self._rsi_value('IBKR', 10), self._allocate('IBKR'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=False)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:3]])
elif self._close('FAS') < self._sma_close('FAS', 100):
if self._rsi_value('FAS', 10) > 0.31:
ranked_candidates = [(self._sma_return('AAPX', 15), self._allocate('AAPX')), (self._sma_return('NVDL', 15), self._allocate('NVDL')), (self._sma_return('BITX', 15), self._allocate('BITX')), (self._sma_return('TSLR', 15), self._allocate('TSLR')), (self._sma_return('FBL', 15), self._allocate('FBL')), (self._sma_return('GGLL', 15), self._allocate('GGLL')), (self._sma_return('AMZZ', 15), self._allocate('AMZZ')), (self._sma_return('RGTI', 15), self._allocate('RGTI')), (self._sma_return('PLTR', 15), self._allocate('PLTR')), (self._sma_return('BABA', 15), self._allocate('BABA')), (self._sma_return('CONL', 15), self._allocate('CONL'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
ranked_candidates = [(self._sma_return('TMF', 15), self._allocate('TMF')), (self._sma_return('FAZ', 15), self._allocate('FAZ')), (self._sma_return('AAPX', 15), self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]]))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
else:
ranked_candidates = [(self._rsi_value('AGQ', 10), self._allocate('AGQ')), (self._rsi_value('FAS', 10), self._allocate('FAS'))]
ranked_candidates.sort(key=lambda candidate: candidate[0], reverse=True)
target_allocation = self._combine_equal(*[allocation for _, allocation in ranked_candidates[:1]])
else:
target_allocation = self._allocate('FAS')
return target_allocation
# ── Allocation and ranking helpers ─────────────────────────────
def _allocate(self, ticker: str) -> dict:
symbol = self._syms.get(ticker)
return {symbol: 1.0} if symbol is not None else {}
@staticmethod
def _merge_allocations(weighted_allocations) -> dict:
combined = {}
for allocation, multiplier in weighted_allocations:
for symbol, weight in allocation.items():
combined[symbol] = combined.get(symbol, 0.0) + weight * multiplier
return combined
def _combine_equal(self, *allocations) -> dict:
if not allocations:
return {}
equal_weight = 1.0 / len(allocations)
return self._merge_allocations(
(allocation, equal_weight) for allocation in allocations
)
def _combine_weighted(self, weighted_allocations) -> dict:
return self._merge_allocations(
(allocation, weight) for weight, allocation in weighted_allocations
)
def _select_ranked(self, candidates, count: int, descending: bool) -> dict:
ranked = sorted(candidates, key=lambda candidate: candidate[0], reverse=descending)
selected_allocations = [resolver() for _, resolver in ranked[:count]]
return self._combine_equal(*selected_allocations)
# ── Indicator helpers ──────────────────────────────────────────
def _window(self, ticker: str, required: int):
window = self._price_window.get(ticker)
return window if window and window.is_ready and window.count >= required else None
def _close(self, ticker: str) -> float:
window = self._window(ticker, 1)
return float(window[0]) if window else 0.0
def _rsi_value(self, ticker: str, period: int) -> float:
indicator = self._rsi.get((ticker, period))
return float(indicator.current.value / 100.0) if indicator and indicator.is_ready else 0.5
def _sma_close(self, ticker: str, period: int) -> float:
window = self._window(ticker, period)
return float(np.mean([window[i] for i in range(period)])) if window else 0.0
def _daily_return(self, ticker: str) -> float:
window = self._window(ticker, 2)
return float(window[0] / window[1] - 1.0) if window else 0.0
def _returns(self, ticker: str, period: int):
window = self._window(ticker, period + 1)
return [window[i] / window[i + 1] - 1.0 for i in range(period)] if window else None
def _sma_return(self, ticker: str, period: int) -> float:
returns = self._returns(ticker, period)
return float(np.mean(returns)) if returns else 0.0
def _std_return(self, ticker: str, period: int) -> float:
returns = self._returns(ticker, period)
return float(np.std(returns)) if returns else 0.0
def _cumulative_return(self, ticker: str, period: int) -> float:
window = self._window(ticker, period + 1)
return float(window[0] / window[period] - 1.0) if window else 0.0
# ── Scheduled events ─────────────────────────────────────────────
def _update_price_windows(self) -> None:
for t, sym in self._syms.items():
close = self.securities[sym].close
if close > 0:
self._price_window[t].add(float(close))
def _snapshot_day_open(self) -> None:
"""Record today's open for every subscribed ticker; the intraday SL
is measured off this, not average cost."""
if not self.ENABLE_INTRADAY_SL:
return
for sym in self._syms.values():
price = self.securities[sym].price
if price > 0:
self._day_open[sym] = float(price)
def _check_intraday_sl(self) -> None:
if not self.ENABLE_INTRADAY_SL or self.is_warming_up:
return
if not self.is_market_open(self._syms.get("SPY")):
return
for h in list(self.portfolio.values()):
if not h.invested:
continue
sym = h.symbol
open_px = self._day_open.get(sym)
price = self.securities[sym].price
if not open_px or open_px <= 0 or price <= 0:
continue
chg = price / open_px - 1.0
if chg <= -self.SL_PCT:
self.liquidate(sym, tag=f"intraday SL {chg:.1%} vs open (limit {self.SL_PCT:.0%})")
self.log(f"[SL] {self.time} {sym.value} {chg:.1%} vs open -> liquidated intraday")
def _rebalance(self) -> None:
if self.is_warming_up:
return
combined = self._strategy_targets()
total_w = sum(combined.values())
if total_w <= 0:
return
targets = set(combined)
for h in list(self.portfolio.values()):
if h.invested and h.symbol not in targets:
self.market_on_open_order(h.symbol, -h.quantity)
pv = self.portfolio.total_portfolio_value * self.LEVERAGE * (1.0 - self.CASH_BUFFER_PCT)
# All orders here are MarketOnOpenOrder -- none settle until tomorrow's
# open, so the exits submitted above can't be relied on to have freed
# up any margin yet when a buy order below gets validated. Track a
# local margin budget (starting from what's actually free right now,
# with a safety haircut) and cap/skip buy orders against it instead of
# submitting the full computed size and letting the broker reject it.
margin_budget = self.portfolio.margin_remaining * 0.9
MARGIN_RATE_ESTIMATE = 0.5 # matches this account's observed Reg-T rate
for sym, wt in combined.items():
price = self.securities[sym].price
if price <= 0:
continue
target_qty = int(pv * wt / price)
delta = target_qty - int(self.portfolio[sym].quantity)
if delta == 0:
continue
if delta > 0:
est_margin_needed = delta * price * MARGIN_RATE_ESTIMATE
if est_margin_needed > margin_budget:
affordable_qty = int(margin_budget / (price * MARGIN_RATE_ESTIMATE))
if affordable_qty <= 0:
self.log(f"[MARGIN-SKIP] {self.time.date()} {sym.value} skipped, no margin budget left")
continue
self.log(f"[MARGIN-CAP] {self.time.date()} {sym.value} capped {delta} -> {affordable_qty} shares (insufficient margin)")
delta = affordable_qty
est_margin_needed = delta * price * MARGIN_RATE_ESTIMATE
margin_budget -= est_margin_needed
self.market_on_open_order(sym, delta)
self._trade_count += 1
net = "+".join(f"{round(w*100):.0f}%{s.value}"
for s, w in sorted(combined.items(), key=lambda x: -x[1]) if w > 0.005)
self.log(f"[{self._trade_count:04d}] {self.time.date()} | net={net}")
def on_end_of_algorithm(self) -> None:
self.log(f"\n Final NAV: ${self.portfolio.total_portfolio_value:>15,.2f} | Rebalances: {self._trade_count}")