| Overall Statistics |
|
Total Orders 1940 Average Win 0.45% Average Loss -0.29% Compounding Annual Return 13.965% Drawdown 26.800% Expectancy 0.277 Start Equity 100000 End Equity 192386.12 Net Profit 92.386% Sharpe Ratio 0.424 Sortino Ratio 0.485 Probabilistic Sharpe Ratio 5.025% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 1.54 Alpha 0.027 Beta 0.712 Annual Standard Deviation 0.163 Annual Variance 0.027 Information Ratio 0.072 Tracking Error 0.134 Treynor Ratio 0.097 Total Fees $1800.16 Estimated Strategy Capacity $58000000.00 Lowest Capacity Asset FOXBV X2S9UGTP4UHX Portfolio Turnover 3.37% Drawdown Recovery 575 |
"""
Two-Sleeve Portfolio: S3 Momentum (40%) + Value/Quality/Trend (60%), UPRO hedge
Cash account -- no margin, no shorting.
S3: SectorTopUniverse (top ~100/sector by mcap, NYSE/NASDAQ/AMEX, price>$5,
mcap>$5B). Momentum (21/63/126/189/252d), ADX<35, price above 189d EMA band.
10 positions, capped 20% each, exposure tapers with breadth stress.
Rebalances 4th Monday/month (holiday-rolled, 30min before close), staleness
catch-up.
Value: coarse top-1000 by $ volume, price>$5 -> fine filter P/E [5,15],
D/E<1, div yield>1%, ROI>10%. 21d OLS log-price slope signal; fills to 15
from next-best trend scores if short. P/E ascending, inverse-vol weighted
(equal-weight fallback), capped 15% each. DD guard halves exposure above
10% DD. Rebalances 1st trading day/month, same catch-up as S3.
Overlap: S3 has priority on contested tickers; Value excludes anything S3
claims; neither liquidates the other's. SPY/UPRO protected from both.
UPRO crash hedge (Value is the defensive leg): enters when DD>10%, SPY<200
SMA, breadth stress>=35% for 3 days, no 30-day cooldown -- scales 30/60/90%
as DD deepens past -10/-15/-20%. Entry: Value sold, UPRO bought in one batch.
Exit: SPY recross, stress<25%/below 3d mean, or 30-day max hold; Value
immediately rebuilt off-schedule. Caveat: SPY_HEDGE_MAX (90%) can exceed
Value's normal 60% if S3 isn't also risk-off; not capped.
Brokerage: InteractiveBrokers Cash account (no margin)
EMAIL: replace YOUR_EMAIL@gmail.com before deploying live
"""
from AlgorithmImports import *
from datetime import date, datetime, timedelta
from collections import defaultdict, deque
import calendar
import json
import math
import numpy as np
STRESS_AMBER = 0.35
STRESS_RED = 0.45
RECOVERY_THRESHOLD = 0.60
FREE_CASH_PCT = 0.025
YOUR_EMAIL = "abc@gmail.com"
S3_BUDGET = 0.40
VALUE_BUDGET = 0.60
DD_THRESHOLD = 0.15
DD_FLOOR = 0.50
STRETCH_WIN_LEN = 126
STALE_REBALANCE_DAYS = 40
HEDGE_ENABLED = True
HEDGE_TOLERANCE = 0.02
SPY_HEDGE_MAX = 0.90
SPY_HEDGE_STEP = 0.30
STRESS_CRASH = 0.35
DD_CRASH = -0.10
STRESS_PERSIST_D = 3
COOLDOWN_DAYS = 30
HWM_SANITY_MULTIPLE = 3.0
MAX_DAILY_EQUITY_JUMP = 0.50
class SectorTopUniverse(FundamentalUniverseSelectionModel):
"""S3 momentum universe; has overlap priority, doesn't defer to Value."""
def __init__(self, algo, blacklist=None):
self.algo = algo
self.blacklist = set(blacklist or [])
super().__init__(self._select)
def _select(self, fundamentals):
buckets = defaultdict(list)
for f in fundamentals:
if not f.has_fundamental_data: continue
if f.symbol.Value in self.blacklist: continue
if f.company_reference.primary_exchange_id not in ("NYS", "NAS", "ASE"): continue
if not f.price or f.price <= 5: continue
if not f.market_cap or f.market_cap < 5_000_000_000: continue
sector = f.asset_classification.morningstar_sector_code
if sector: buckets[sector].append(f)
symbols = []
for _, stocks in buckets.items():
stocks.sort(key=lambda x: x.market_cap, reverse=True)
symbols.extend(s.symbol for s in stocks[:100])
self.algo._s3_universe_symbols_today = set(s.Value for s in symbols)
return symbols
class TrendPredictor:
"""OLS log-price slope trend proxy; positive slope => uptrend."""
def __init__(self, lookback: int = 21) -> None:
self._lookback = max(2, int(lookback))
@property
def lookback(self) -> int:
return self._lookback
def predict(self, closes: list) -> tuple:
n = len(closes)
if n < 2:
return 0.0, False
log_prices = []
for p in closes:
if p is None:
return 0.0, False
price = float(p)
if price <= 0:
return 0.0, False
log_prices.append(math.log(price))
x_mean = (n - 1) / 2.0
y_mean = sum(log_prices) / float(n)
num = 0.0
den = 0.0
for i in range(n):
dx = float(i) - x_mean
dy = log_prices[i] - y_mean
num += dx * dy
den += dx * dx
if den == 0.0:
return 0.0, False
slope = num / den
trend_score = slope * float(n)
predicted_up = slope > 0.0
return float(trend_score), bool(predicted_up)
class ValMomentum2sleeveAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 5, 1)
self.SetEndDate(2026, 5, 1)
self.SetCash(100_000)
self.SetBrokerageModel(
BrokerageName.InteractiveBrokersBrokerage, AccountType.Cash)
self.Settings.FreePortfolioValuePercentage = FREE_CASH_PCT
self.anchor = self.AddEquity("SPY", Resolution.Daily).Symbol
self.SetBenchmark("SPY")
self.hedge_instrument = self.AddEquity("UPRO", Resolution.Daily).Symbol
self.SetSecurityInitializer(
lambda s: (s.SetFeeModel(InteractiveBrokersFeeModel()),
s.SetFillModel(ImmediateFillModel()),
s.SetSlippageModel(ConstantSlippageModel(0.001))))
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.TotalReturn
self.lookbacks = [21, 63, 126, 189, 252]
self.stock_count = 10
self.max_weight = 0.20
self.band_len = 189
self.hist_len = 126
self.allow_universe = True
self.was_risk_off = False
self.risk_off_date = None
self.max_stress = 0.0
self.current_band_idx: dict = {}
self.BOTTOM_LEVELS = {0, 1, 2, 3, 4}
self.symbols: set = set()
self.ma: dict = {}
self.adx: dict = {}
self.close_win: dict = {}
self.stretch_ema: dict = {}
self.stretch_win: dict = {}
self.band_hist: dict = {}
self.adx_limit = 35
self.adx_period = 14
self._hwm = 0.0
self._prev_value = None
self._daily_rets = deque(maxlen=252)
self._s3_prev_qty = {}
self._s3_prev_sleeve_value = None
self._s3_sleeve_peak = 0.0
self._s3_inception_value = None
self._last_rebalance_month = None
self._last_rebalance_date = None
self._pending_weights = None
self._pending_weights_date = None
self._pending_weights_since = {} # Symbol -> date first became pending, for stuck-order alerts
self._pending_weights_is_retry = False # False = first dispatch (MOO); True = retry (MarketOrder)
self._s3_universe_symbols_today = set()
self._s3_claimed = set()
self.value_coarse_count = 1000
self.value_portfolio_size = 15
self.value_max_weight = 0.15
self.value_vol_lookback = 21
self.value_dd_scale_threshold = 0.10
self.value_dd_scale_factor = 0.5
self.value_trend_predictor = TrendPredictor(lookback=21)
self._value_equity_peak = 0.0
self._latest_coarse_count = 0
self._latest_fine_pass_count = 0
self._latest_ml_pass_count = 0
self._value_working_set = []
self._pe_by_symbol = {}
self._last_value_rebalance_month = None
self._last_value_rebalance_date = None
self._value_claimed = set()
self._value_prev_qty = {}
self._value_prev_sleeve_value = None
self._value_sleeve_peak = 0.0
self._value_inception_value = None
self.AddUniverse(self._ValueCoarseSelection, self._ValueFineSelection)
self._hedge_active = False
self._hedge_entry_price = None
self._hedge_entry_date = None
self._hedge_entry_stress = None
self._hedge_entry_dd = None
self._hedge_last_target = 0.0
self._hedge_trades = []
self.current_hedge_target = 0.0
self.last_hedge_exit = None
self.spy_sma200 = self.SMA(self.anchor, 200, Resolution.Daily)
self.WarmUpIndicator(self.anchor, self.spy_sma200, Resolution.Daily)
self.stress_window = RollingWindow[float](STRESS_PERSIST_D)
self._state_key = "two_sleeve_state" if self.LiveMode else "two_sleeve_state_backtest"
self._LoadState()
self.Debug(
f"[DIAG] Post-LoadState: hwm={self._hwm:,.0f} "
f"value_equity_peak={self._value_equity_peak:,.0f} "
f"starting_equity={self.Portfolio.TotalPortfolioValue:,.0f} "
f"hedge_active={self._hedge_active}")
self.SetWarmUp(300)
self.SetUniverseSelection(
SectorTopUniverse(self, blacklist={"GME", "AMC"}))
# 30min before close for extra order-submission buffer (see docstring).
self.Schedule.On(
self.DateRules.EveryDay(self.anchor),
self.TimeRules.BeforeMarketClose(self.anchor, 30),
self._MaybeRebalance)
self.Schedule.On(
self.DateRules.EveryDay(self.anchor),
self.TimeRules.AfterMarketOpen(self.anchor, 10),
self._DailyHedgeCheck)
# Dispatches/retries S3's pending orders on a real wall-clock
# schedule -- NOT from OnData, whose daily-resolution slice for US
# equities in live trading arrives at/after the close, so a market-
# open guard checked there could never actually pass (see docstring).
self.Schedule.On(
self.DateRules.EveryDay(self.anchor),
self.TimeRules.AfterMarketOpen(self.anchor, 15),
self._DispatchPendingWeights)
self.Schedule.On(
self.DateRules.EveryDay(self.anchor),
self.TimeRules.AfterMarketOpen(self.anchor, 30),
self._MaybeRebalanceValue)
self.Schedule.On(
self.DateRules.EveryDay(self.anchor),
self.TimeRules.BeforeMarketClose(self.anchor, 1),
self._DailySnapshot)
def OnSecuritiesChanged(self, changes):
for sec in changes.RemovedSecurities:
s = sec.Symbol
if s == self.anchor or s == self.hedge_instrument: continue
self.symbols.discard(s)
for d in [self.ma, self.adx, self.close_win,
self.stretch_ema, self.stretch_win, self.band_hist,
self.current_band_idx]:
d.pop(s, None)
for sec in changes.AddedSecurities:
s = sec.Symbol
if s == self.anchor or s == self.hedge_instrument: continue
if s.Value not in self._s3_universe_symbols_today:
continue
self.symbols.add(s)
self.ma[s] = self.EMA(s, self.band_len, Resolution.Daily)
self.adx[s] = self.ADX(s, self.adx_period, Resolution.Daily)
self.stretch_ema[s] = self.EMA(s, self.band_len, Resolution.Daily)
self.close_win[s] = RollingWindow[float](self.band_len)
self.stretch_win[s] = RollingWindow[float](STRETCH_WIN_LEN)
self.band_hist[s] = RollingWindow[int](self.hist_len)
if self.LiveMode:
try:
self.WarmUpIndicator(s, self.ma[s], Resolution.Daily)
self.WarmUpIndicator(s, self.adx[s], Resolution.Daily)
self.WarmUpIndicator(s, self.stretch_ema[s], Resolution.Daily)
except Exception as e:
self.Debug(f"[S3] warmup indicator failed for {s.Value}: {e}")
def OnOrderEvent(self, orderEvent):
if orderEvent.Status in (OrderStatus.Invalid, OrderStatus.Canceled):
msg = (f"[ORDER FAILED] {orderEvent.Symbol.Value} qty={orderEvent.Quantity} "
f"status={orderEvent.Status} msg={orderEvent.Message}")
self.Debug(msg)
if self.LiveMode:
self.Log(msg)
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] Order Failed {self.Time:%d %b %Y}", msg)
def _DispatchPendingWeights(self):
"""Submits/retries S3's pending weights on a wall-clock schedule,
not from OnData (see docstring: daily-bar timing issue)."""
if self._pending_weights is None or self.IsWarmingUp:
return
if not self.Securities[self.anchor].Exchange.DateTimeIsOpen(self.Time):
return
today = self.Time.date()
if self._pending_weights_date == today:
return
self._pending_weights_date = today
targets = dict(self._pending_weights)
equity = self.Portfolio.TotalPortfolioValue
protected = {self.anchor, self.hedge_instrument}
use_market_order = self._pending_weights_is_retry
# Held-but-unwanted positions fold into targets (weight 0.0)
# so a failed liquidation gets the same daily retry as a buy.
for pos in list(self.Portfolio.Values):
sym = pos.Symbol
if (pos.Invested and sym not in targets
and sym not in protected
and sym.Value not in self._value_claimed):
targets.setdefault(sym, 0.0)
still_pending = {}
for sym, w in targets.items():
if not self.Securities.ContainsKey(sym):
still_pending[sym] = w
continue
price = self.Securities[sym].Price
if price <= 0:
still_pending[sym] = w
continue
cur = self.Portfolio[sym].Quantity if self.Portfolio.ContainsKey(sym) else 0
if w <= 0:
if cur != 0:
if use_market_order:
self.MarketOrder(sym, -cur)
else:
self.MarketOnOpenOrder(sym, -cur)
still_pending[sym] = 0.0
continue
target_qty = int(equity * w / price)
delta = target_qty - cur
if abs(delta) > 0:
drift = abs(delta * price) / equity
if drift >= 0.02:
if use_market_order:
self.MarketOrder(sym, delta)
else:
self.MarketOnOpenOrder(sym, delta)
still_pending[sym] = w
if still_pending:
self._pending_weights = still_pending
self._pending_weights_is_retry = True
for sym in still_pending:
self._pending_weights_since.setdefault(sym, today)
# Prune resolved symbols so the persisted dict doesn't
# grow unbounded across cycles.
self._pending_weights_since = {
s: d for s, d in self._pending_weights_since.items()
if s in still_pending}
stuck = [s.Value for s, since in self._pending_weights_since.items()
if s in still_pending and (today - since).days >= 5]
if stuck:
msg = f"[ALERT] Orders stuck 5+ days without executing: {stuck}"
self.Debug(msg)
if self.LiveMode:
self.Log(msg)
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] Stuck Orders {self.Time:%d %b %Y}", msg)
self._SaveState()
else:
self._pending_weights = None
self._pending_weights_is_retry = False
self._pending_weights_since = {}
self._SaveState()
def OnData(self, data: Slice):
for s in list(self.symbols):
if not data.ContainsKey(s): continue
bar = data[s]
if bar is None: continue
close = bar.Close
self.close_win[s].Add(close)
if not self.close_win[s].IsReady or not self.ma[s].IsReady: continue
dev = np.std(list(self.close_win[s]))
if dev <= 0: continue
mid = self.ma[s].Current.Value
stretch = abs(close - mid) / dev
self.stretch_ema[s].Update(self.Time, stretch)
self.stretch_win[s].Add(stretch)
bands = [
mid - dev * 1.618, mid - dev * 1.382, mid - dev,
mid - dev * 0.809, mid - dev * 0.5, mid - dev * 0.382,
mid,
mid + dev * 0.382, mid + dev * 0.5, mid + dev * 0.809,
mid + dev, mid + dev * 1.382, mid + dev * 1.618
]
self.current_band_idx[s] = self._band_index(close, bands)
def _band_index(self, price, bands):
for i in range(len(bands) - 1):
if bands[i] <= price < bands[i + 1]: return i
return len(bands) - 2
def _DailyHedgeCheck(self):
if self.IsWarmingUp: return
if not self.Securities[self.anchor].Exchange.DateTimeIsOpen(self.Time): return
if not self.spy_sma200.IsReady: return
eq = self.Portfolio.TotalPortfolioValue
dd = (eq - self._hwm) / self._hwm if self._hwm > 0 else 0.0
idxs = list(self.current_band_idx.values())
if len(idxs) == 0: return
bottom_frac = sum(i in self.BOTTOM_LEVELS for i in idxs) / len(idxs)
self.stress_window.Add(bottom_frac)
if self.stress_window.Count < STRESS_PERSIST_D: return
stress_persistent = all(v >= STRESS_CRASH for v in self.stress_window)
spy_price = float(self.Securities[self.anchor].Price)
sma200 = float(self.spy_sma200.Current.Value)
trend_down = spy_price < sma200
if isinstance(self.last_hedge_exit, date):
in_cooldown = (self.Time.date() - self.last_hedge_exit).days < COOLDOWN_DAYS
else:
in_cooldown = False
crash_env = HEDGE_ENABLED and trend_down and stress_persistent and dd <= DD_CRASH and not in_cooldown
current_upro_w = self.Portfolio[self.hedge_instrument].HoldingsValue / eq if eq > 0 else 0.0
if crash_env:
target = 0.0
if dd <= DD_CRASH: target = SPY_HEDGE_STEP
if dd <= DD_CRASH - 0.05: target = 2 * SPY_HEDGE_STEP
if dd <= DD_CRASH - 0.10: target = SPY_HEDGE_MAX
if target != self._hedge_last_target:
upro_price = float(self.Securities[self.hedge_instrument].Price)
self.Debug(f"[HEDGE ENTER] Swapping Value → UPRO {target:.0%} "
f"dd={dd:.1%} stress={bottom_frac:.2f} "
f"spy={spy_price:.2f} sma200={sma200:.2f} upro={upro_price:.2f} "
f"value_positions={len(self._value_claimed)}")
targets = [PortfolioTarget(sym, 0.0)
for sym in self.Securities.Keys
if sym.Value in self._value_claimed]
targets.append(PortfolioTarget(self.hedge_instrument, target))
self.SetHoldings(targets)
self._value_claimed = set()
self._hedge_last_target = target
self.current_hedge_target = target
if self.LiveMode:
self.Log(f"[HEDGE ENTER] Value→UPRO {target:.0%} dd={dd:.1%}")
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] Hedge Entered {self.Time:%d %b %Y}",
f"Value sleeve swapped to UPRO {target:.0%}\n"
f"DD={dd:.1%} Stress={bottom_frac:.1%}\n"
f"Portfolio: GBP{eq:,.0f}")
if not self._hedge_active:
self._hedge_active = True
self._hedge_entry_price = float(self.Securities[self.hedge_instrument].Price)
self._hedge_entry_date = self.Time
self._hedge_entry_stress = bottom_frac
self._hedge_entry_dd = dd
self._value_prev_qty = {}
self._value_prev_sleeve_value = None
self._SaveState()
return
stress_mean = np.mean([float(x) for x in self.stress_window])
max_duration = False
if HEDGE_ENABLED and self._hedge_active and self._hedge_entry_date is not None:
max_duration = (self.Time.date() - self._hedge_entry_date.date()).days >= 30
exit_signal = (
spy_price > sma200 or
bottom_frac < 0.25 or
bottom_frac < stress_mean or
max_duration
)
if exit_signal and current_upro_w > HEDGE_TOLERANCE:
exit_price = float(self.Securities[self.hedge_instrument].Price)
ret = ((exit_price - self._hedge_entry_price) / self._hedge_entry_price
if self._hedge_entry_price else 0.0)
self.Debug(f"[HEDGE EXIT] Swapping UPRO → Value "
f"px:{self._hedge_entry_price}→{exit_price:.2f} "
f"ret:{ret:+.2%} "
f"dd:{(self._hedge_entry_dd if self._hedge_entry_dd is not None else 0):.1%}→{dd:.1%} "
f"stress:{(self._hedge_entry_stress if self._hedge_entry_stress is not None else 0):.2f}→{bottom_frac:.2f}")
self.Liquidate(self.hedge_instrument)
if self.LiveMode:
self.Log(f"[HEDGE EXIT] UPRO→Value ret={ret:+.2%} dd={dd:.1%}")
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] Hedge Exited {self.Time:%d %b %Y}",
f"UPRO swapped back to Value sleeve\n"
f"Return: {ret:+.2%}\n"
f"Portfolio: GBP{eq:,.0f}")
self._hedge_trades.append({
"entry_date": self._hedge_entry_date,
"exit_date": self.Time,
"entry_price": self._hedge_entry_price,
"exit_price": exit_price,
"return": ret,
"entry_stress": self._hedge_entry_stress,
"exit_stress": bottom_frac,
"entry_dd": self._hedge_entry_dd,
"exit_dd": dd,
})
self._hedge_active = False
self._hedge_entry_price = None
self._hedge_entry_date = None
self._hedge_entry_stress = None
self._hedge_entry_dd = None
self._hedge_last_target = 0.0
self.current_hedge_target = 0.0
self.last_hedge_exit = self.Time.date()
self._SaveState()
self._RebalanceValue()
def _FourthMondayTradingDay(self, year, month):
"""4th Monday of (year, month), rolled to next open trading day."""
cal = calendar.Calendar()
mondays = [d for d in cal.itermonthdates(year, month)
if d.month == month and d.weekday() == 0]
target = mondays[3]
exch = self.Securities[self.anchor].Exchange
while not exch.Hours.IsDateOpen(target):
target += timedelta(days=1)
return target
def _MaybeRebalance(self):
"""Fires _Rebalance() on the 4th Monday, or via a staleness
catch-up; guarded to at most once per calendar month."""
if self.IsWarmingUp: return
today = self.Time.date()
key = (today.year, today.month)
if self._last_rebalance_month == key:
return
target = self._FourthMondayTradingDay(today.year, today.month)
stale = (self._last_rebalance_date is not None and
(today - self._last_rebalance_date).days > STALE_REBALANCE_DAYS)
if today >= target or stale:
if stale and today < target:
gap = (today - self._last_rebalance_date).days
msg = f"[S3] STALE REBALANCE catch-up: {gap}d since last rebalance ({self._last_rebalance_date})"
self.Debug(msg)
if self.LiveMode: self.Log(msg)
self._Rebalance()
self._last_rebalance_month = key
self._last_rebalance_date = today
self._SaveState()
def _SaveState(self):
state = {
"last_rebalance_month": list(self._last_rebalance_month) if self._last_rebalance_month else None,
"last_rebalance_date": self._last_rebalance_date.isoformat() if self._last_rebalance_date else None,
"allow_universe": self.allow_universe,
"was_risk_off": self.was_risk_off,
"max_stress": self.max_stress,
"risk_off_date": self.risk_off_date.isoformat() if self.risk_off_date else None,
"s3_claimed": list(self._s3_claimed),
"hwm": self._hwm,
"last_value_rebalance_month": list(self._last_value_rebalance_month) if self._last_value_rebalance_month else None,
"last_value_rebalance_date": self._last_value_rebalance_date.isoformat() if self._last_value_rebalance_date else None,
"value_claimed": list(self._value_claimed),
"value_equity_peak": self._value_equity_peak,
"s3_sleeve_peak": self._s3_sleeve_peak,
"s3_inception_value": self._s3_inception_value,
"value_sleeve_peak": self._value_sleeve_peak,
"value_inception_value": self._value_inception_value,
"hedge_active": self._hedge_active,
"hedge_entry_price": self._hedge_entry_price,
"hedge_entry_date": self._hedge_entry_date.isoformat() if self._hedge_entry_date else None,
"hedge_entry_stress": self._hedge_entry_stress,
"hedge_entry_dd": self._hedge_entry_dd,
"hedge_last_target": self._hedge_last_target,
"current_hedge_target": self.current_hedge_target,
"last_hedge_exit": self.last_hedge_exit.isoformat() if self.last_hedge_exit else None,
# Retry queue (see OnData/_LoadState): symbols stored by ticker.
"pending_weights": ({s.Value: w for s, w in self._pending_weights.items()}
if self._pending_weights else None),
"pending_weights_date": (self._pending_weights_date.isoformat()
if self._pending_weights_date else None),
"pending_weights_since": {s.Value: d.isoformat()
for s, d in self._pending_weights_since.items()},
"pending_weights_is_retry": self._pending_weights_is_retry,
}
try:
self.ObjectStore.Save(self._state_key, json.dumps(state))
except Exception as e:
self.Debug(f"[STATE] save failed: {e}")
if self.LiveMode: self.Log(f"[STATE] save failed: {e}")
def _LoadState(self):
if not self.LiveMode:
self.Debug("[STATE] Backtest mode -- ignoring any saved ObjectStore "
"state, starting fresh")
return
try:
if not self.ObjectStore.ContainsKey(self._state_key):
self.Debug("[STATE] no saved state found -- starting fresh")
return
state = json.loads(self.ObjectStore.Read(self._state_key))
lrm = state.get("last_rebalance_month")
self._last_rebalance_month = tuple(lrm) if lrm else None
lrd = state.get("last_rebalance_date")
self._last_rebalance_date = date.fromisoformat(lrd) if lrd else None
self.allow_universe = state.get("allow_universe", True)
self.was_risk_off = state.get("was_risk_off", False)
self.max_stress = state.get("max_stress", 0.0)
rod = state.get("risk_off_date")
self.risk_off_date = datetime.fromisoformat(rod) if rod else None
self._s3_claimed = set(state.get("s3_claimed", []))
starting_equity = self.Portfolio.TotalPortfolioValue
loaded_hwm = state.get("hwm", self._hwm)
if starting_equity > 0 and loaded_hwm > HWM_SANITY_MULTIPLE * starting_equity:
msg = (f"[STATE] Discarding implausible S3 hwm={loaded_hwm:,.0f} "
f"(> {HWM_SANITY_MULTIPLE:.0f}x starting equity={starting_equity:,.0f}); "
f"resetting to current equity")
self.Debug(msg)
self.Log(msg)
self._hwm = starting_equity
else:
self._hwm = loaded_hwm
lvm = state.get("last_value_rebalance_month")
self._last_value_rebalance_month = tuple(lvm) if lvm else None
lvd = state.get("last_value_rebalance_date")
self._last_value_rebalance_date = date.fromisoformat(lvd) if lvd else None
self._value_claimed = set(state.get("value_claimed", []))
loaded_value_peak = state.get("value_equity_peak", self._value_equity_peak)
if starting_equity > 0 and loaded_value_peak > HWM_SANITY_MULTIPLE * starting_equity:
msg = (f"[STATE] Discarding implausible value_equity_peak={loaded_value_peak:,.0f} "
f"(> {HWM_SANITY_MULTIPLE:.0f}x starting equity={starting_equity:,.0f}); "
f"resetting to current equity")
self.Debug(msg)
self.Log(msg)
self._value_equity_peak = starting_equity
else:
self._value_equity_peak = loaded_value_peak
self._s3_sleeve_peak = state.get("s3_sleeve_peak", self._s3_sleeve_peak)
self._s3_inception_value = state.get("s3_inception_value", self._s3_inception_value)
self._value_sleeve_peak = state.get("value_sleeve_peak", self._value_sleeve_peak)
self._value_inception_value = state.get("value_inception_value", self._value_inception_value)
self._hedge_active = state.get("hedge_active", False)
self._hedge_entry_price = state.get("hedge_entry_price")
hed = state.get("hedge_entry_date")
self._hedge_entry_date = datetime.fromisoformat(hed) if hed else None
self._hedge_entry_stress = state.get("hedge_entry_stress")
self._hedge_entry_dd = state.get("hedge_entry_dd")
self._hedge_last_target = state.get("hedge_last_target", 0.0)
self.current_hedge_target = state.get("current_hedge_target", 0.0)
lhe = state.get("last_hedge_exit")
self.last_hedge_exit = date.fromisoformat(lhe) if lhe else None
# Re-subscribe via AddEquity so pending tickers are tradable again.
pw = state.get("pending_weights")
if pw:
self._pending_weights = {
self.AddEquity(ticker, Resolution.Daily).Symbol: w
for ticker, w in pw.items()}
else:
self._pending_weights = None
pwd = state.get("pending_weights_date")
self._pending_weights_date = date.fromisoformat(pwd) if pwd else None
self._pending_weights_since = {
self.AddEquity(ticker, Resolution.Daily).Symbol: date.fromisoformat(d)
for ticker, d in state.get("pending_weights_since", {}).items()}
self._pending_weights_is_retry = state.get("pending_weights_is_retry", False)
msg = (f"[STATE] restored: s3_month={self._last_rebalance_month} "
f"value_month={self._last_value_rebalance_month} "
f"allow_universe={self.allow_universe} "
f"s3_claimed={len(self._s3_claimed)} value_claimed={len(self._value_claimed)} "
f"hwm={self._hwm:,.0f} value_equity_peak={self._value_equity_peak:,.0f} "
f"s3_sleeve_peak={self._s3_sleeve_peak:,.0f} "
f"value_sleeve_peak={self._value_sleeve_peak:,.0f} "
f"pending_weights={len(self._pending_weights) if self._pending_weights else 0} "
f"hedge_active={self._hedge_active}")
self.Debug(msg)
self.Log(msg)
except Exception as e:
self.Debug(f"[STATE] load failed, starting fresh: {e}")
self.Log(f"[STATE] load failed, starting fresh: {e}")
def _Rebalance(self):
if self.IsWarmingUp: return
if not self.Securities[self.anchor].Exchange.DateTimeIsOpen(self.Time): return
idxs = list(self.current_band_idx.values())
if len(idxs) < 50: return
bottom_frac = sum(i in self.BOTTOM_LEVELS for i in idxs) / len(idxs)
self.max_stress = max(self.max_stress, bottom_frac)
if bottom_frac >= STRESS_RED:
if not self.was_risk_off:
self.risk_off_date = self.Time
self.allow_universe = False
self.was_risk_off = True
msg = f"[S3][STRESS-RED] RISK-OFF bottom_frac={bottom_frac:.1%}"
self.Debug(msg)
if self.LiveMode:
self.Log(msg)
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] S3 Risk-Off {self.Time:%d %b %Y}",
f"Breadth stress {bottom_frac:.1%} >= {STRESS_RED:.0%}\n"
f"S3 liquidated. Value sleeve unaffected.\n"
f"Portfolio: GBP{self.Portfolio.TotalPortfolioValue:,.0f}")
elif bottom_frac >= STRESS_AMBER and self.allow_universe:
msg = f"[S3][STRESS-AMBER] bottom_frac={bottom_frac:.1%}"
self.Debug(msg)
if self.LiveMode:
self.Log(msg)
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] S3 Amber {self.Time:%d %b %Y}",
f"Stress {bottom_frac:.1%} approaching {STRESS_RED:.0%}\n"
f"Portfolio: GBP{self.Portfolio.TotalPortfolioValue:,.0f}")
elif self.was_risk_off:
denom = max(self.max_stress, 0.10)
imp = (self.max_stress - bottom_frac) / denom
doff = (self.Time - self.risk_off_date).days if self.risk_off_date else 0
if imp >= RECOVERY_THRESHOLD or bottom_frac < 0.15 or doff > 180:
trig = ("60pct" if imp >= RECOVERY_THRESHOLD
else "stress<15" if bottom_frac < 0.15 else "180d")
msg = f"[S3][RECOVERY] trigger={trig} stress={bottom_frac:.1%}"
self.Debug(msg)
if self.LiveMode:
self.Log(msg)
self.Notify.Email(YOUR_EMAIL,
f"[ALERT] S3 Recovery {self.Time:%d %b %Y}",
f"Breadth recovered. trigger={trig}\n"
f"Re-entering market.")
for s in self.symbols:
if s in self.band_hist:
self.band_hist[s] = RollingWindow[int](self.hist_len)
for s in self.symbols:
if s in self.stretch_win:
self.stretch_win[s] = RollingWindow[float](STRETCH_WIN_LEN)
self.allow_universe = True
self.was_risk_off = False
self.max_stress = 0.0
self.risk_off_date = None
else:
self.Debug(f"[S3][RISK-OFF] stress={bottom_frac:.1%} imp={imp:.1%} days={doff}")
else:
self.allow_universe = True
self._SaveState()
if not self.allow_universe:
for k in list(self.Portfolio):
if (k.Value.Invested
and k.Key != self.anchor
and k.Key != self.hedge_instrument
and k.Key.Value not in self._value_claimed):
self.Liquidate(k.Key)
self._s3_claimed = set()
self._SaveState()
self.Debug(f"[S3][RISK-OFF] S3 liquidated. Value sleeve intact. stress={bottom_frac:.1%}")
return
target_exposure = float(np.interp(
bottom_frac, [0.25, STRESS_RED], [1.0, 0.0]))
target_exposure = float(round(target_exposure, 2))
if self._hwm > 0 and bottom_frac < STRESS_AMBER:
dd = (self.Portfolio.TotalPortfolioValue - self._hwm) / self._hwm
if dd < -DD_THRESHOLD:
dd_scale = max(DD_FLOOR, 1.0 + dd)
target_exposure *= dd_scale
self.Debug(f"[S3][DD BREAKER] dd={dd:.1%} scale={dd_scale:.2f} "
f"exposure→{target_exposure:.2f}")
if self.LiveMode:
self.Log(f"[S3][DD BREAKER] dd={dd:.1%} scale={dd_scale:.2f} "
f"exposure→{target_exposure:.2f}")
hist = self.History(
list(self.symbols), max(self.lookbacks) + 1, Resolution.Daily)
if hist.empty: return
closes = hist["close"].unstack(0)
momentum = {}
for s in self.symbols:
if s not in closes: continue
px = closes[s]
if len(px) < max(self.lookbacks) + 1: continue
if not self.adx[s].IsReady or self.adx[s].Current.Value > self.adx_limit: continue
mom = np.mean([px.iloc[-1] / px.iloc[-lb - 1] - 1 for lb in self.lookbacks])
if not self.ma[s].IsReady: continue
if self.Securities[s].Price <= self.ma[s].Current.Value: continue
fundamentals = self.Securities[s].Fundamentals
if fundamentals is None or fundamentals.MarketCap < 5_000_000_000: continue
if mom > 0: momentum[s] = mom
if not momentum:
for k in list(self.Portfolio):
if (k.Value.Invested
and k.Key != self.anchor
and k.Key.Value not in self._value_claimed
and k.Key in self.symbols):
self.Liquidate(k.Key)
self._s3_claimed = set()
self._SaveState()
return
top = sorted(momentum, key=momentum.get, reverse=True)[:self.stock_count]
scaled = {}
for s in top:
if not self.ma[s].IsReady or not self.stretch_ema[s].IsReady: continue
dev = np.std(list(self.close_win[s]))
if dev <= 0: continue
mid = self.ma[s].Current.Value
lm = self.stretch_ema[s].Current.Value
lm2 = lm / 2.0
lm3 = lm2 * 0.38196601
lm4 = lm * 1.38196601
lm5 = lm * 1.61803399
lm6 = (lm + lm2) / 2.0
bands = [
mid-dev*lm5, mid-dev*lm4, mid-dev*lm, mid-dev*lm6, mid-dev*lm2,
mid-dev*lm3, mid,
mid+dev*lm3, mid+dev*lm2, mid+dev*lm6, mid+dev*lm, mid+dev*lm4, mid+dev*lm5
]
price = self.Securities[s].Price
idx = self._band_index(price, bands)
self.band_hist[s].Add(idx)
hist_idx = list(self.band_hist[s])
historical_h = max(hist_idx) if hist_idx else idx
scale = (1.0 if historical_h <= 0
else 0.0 if idx >= historical_h
else max(0.2, 1.0 - idx / historical_h))
if self.stretch_win[s].IsReady:
sw = list(self.stretch_win[s])
cur_s = sw[0]
peak_s = max(sw)
if idx >= 10 and peak_s > 0 and cur_s < peak_s * 0.80:
scale = min(scale, 0.2)
self.Debug(f"[S3][ANTICIPATION] {s.Value}")
scaled[s] = (momentum[s] * self.adx[s].Current.Value) * scale
if not scaled:
for k in list(self.Portfolio):
if (k.Value.Invested
and k.Key != self.anchor
and k.Key.Value not in self._value_claimed
and k.Key in self.symbols):
self.Liquidate(k.Key)
self._s3_claimed = set()
self._SaveState()
return
total_scaled = sum(scaled.values())
raw_weights = {s: v / total_scaled for s, v in scaled.items()}
capped = {s: min(self.max_weight, w) for s, w in raw_weights.items()}
cur_sum = sum(capped.values())
final_weights = {}
if cur_sum > 0:
for s, w in capped.items():
final_weights[s] = (w / cur_sum) * S3_BUDGET * target_exposure
s3_targets_dict = {s: w * (0.95 if self.LiveMode else 1.0)
for s, w in final_weights.items() if w > 0}
self._pending_weights = s3_targets_dict
self._pending_weights_is_retry = False
self._pending_weights_date = None
self._pending_weights_since = {}
self._s3_claimed = set(s.Value for s in s3_targets_dict.keys())
handoff = self._s3_claimed & self._value_claimed
if handoff:
msg = f"[HANDOFF] S3 claiming from Value sleeve: {sorted(handoff)}"
self.Debug(msg)
if self.LiveMode: self.Log(msg)
self._value_claimed -= handoff
self._SaveState()
self.Debug(f"[S3] REBAL {self.Time:%Y-%m-%d} stress={bottom_frac:.1%} "
f"exp={target_exposure:.0%} pos={len(final_weights)} [MOO queued]")
if self.LiveMode:
self.Log(f"[S3] REBAL {self.Time:%Y-%m-%d} stress={bottom_frac:.1%} "
f"exp={target_exposure:.0%} pos={len(final_weights)} "
f"stocks={[s.Value for s in final_weights]}")
def _ValueCoarseSelection(self, coarse: list) -> list:
filtered = []
for x in coarse:
if not x.has_fundamental_data:
continue
if x.price is None or x.price <= 5:
continue
filtered.append(x)
filtered.sort(key=lambda c: c.dollar_volume, reverse=True)
selected = filtered[: self.value_coarse_count]
self._latest_coarse_count = len(selected)
return [c.symbol for c in selected]
def _ValueFineSelection(self, fine: list) -> list:
passed = []
pe_by_symbol = {}
for f in fine:
if f.symbol.Value in self._s3_claimed:
continue
pe = self._GetFloat(
f,
["ValuationRatios.PERatio", "ValuationRatios.PE", "PERatio", "PE"],
default=float("nan"))
debt_to_equity = self._GetFloat(
f,
["OperationRatios.DebtEquityRatio", "OperationRatios.TotalDebtEquityRatio",
"FinancialStatements.balance_sheet.TotalDebtEquityRatio", "DebtEquityRatio"],
default=float("nan"))
dividend_yield = self._GetFloat(
f,
["ValuationRatios.ForwardDividendYield", "ValuationRatios.DividendYield", "DividendYield"],
default=float("nan"))
roi = self._GetFloat(
f,
["OperationRatios.ROIC", "OperationRatios.roi", "OperationRatios.ReturnOnInvestment",
"ROIC", "ROI"],
default=float("nan"))
if not self._IsFinite(pe) or not self._IsFinite(debt_to_equity) or \
not self._IsFinite(dividend_yield) or not self._IsFinite(roi):
continue
if pe < 5.0 or pe > 15.0: continue
if debt_to_equity >= 1.0: continue
if dividend_yield <= 0.01: continue
if roi <= 0.10: continue
passed.append(f)
pe_by_symbol[f.symbol] = float(pe)
passed.sort(key=lambda ff: pe_by_symbol.get(ff.symbol, float("inf")))
passed = passed[: self.value_portfolio_size + 5]
self._value_working_set = [ff.symbol for ff in passed]
self._pe_by_symbol = pe_by_symbol
self._latest_fine_pass_count = len(self._value_working_set)
return self._value_working_set
def _ValueMonthStartTradingDay(self, year, month):
"""First open trading day of (year, month) for self.anchor."""
cal = calendar.Calendar()
days = [d for d in cal.itermonthdates(year, month) if d.month == month]
exch = self.Securities[self.anchor].Exchange
for d in days:
if exch.Hours.IsDateOpen(d):
return d
return days[0]
def _MaybeRebalanceValue(self):
"""Same pattern as S3's _MaybeRebalance (see above)."""
if self.IsWarmingUp: return
today = self.Time.date()
key = (today.year, today.month)
if self._last_value_rebalance_month == key:
return
target = self._ValueMonthStartTradingDay(today.year, today.month)
stale = (self._last_value_rebalance_date is not None and
(today - self._last_value_rebalance_date).days > STALE_REBALANCE_DAYS)
if today >= target or stale:
if stale and today < target:
gap = (today - self._last_value_rebalance_date).days
msg = f"[VALUE] STALE REBALANCE catch-up: {gap}d since last rebalance ({self._last_value_rebalance_date})"
self.Debug(msg)
if self.LiveMode: self.Log(msg)
self._RebalanceValue()
self._last_value_rebalance_month = key
self._last_value_rebalance_date = today
self._SaveState()
def _RebalanceValue(self) -> None:
if self.IsWarmingUp:
return
if self._hedge_active:
self.Debug("[VALUE] Skipping scheduled rebalance -- hedge active (Value is in UPRO)")
return
eq_now = self.Portfolio.TotalPortfolioValue
if (self._prev_value is not None and self._prev_value > 0 and
eq_now > self._prev_value * (1.0 + MAX_DAILY_EQUITY_JUMP)):
msg = (f"[GUARD] Suspicious equity jump {self._prev_value:,.0f} -> {eq_now:,.0f} "
f"at Value rebalance {self.Time:%Y-%m-%d} -- skipping value_equity_peak "
f"update (likely a bad/stale price print)")
self.Debug(msg)
if self.LiveMode: self.Log(msg)
else:
self._value_equity_peak = max(self._value_equity_peak, eq_now)
current_dd = self._ValuePortfolioDrawdown()
candidates = list(self._value_working_set)
if len(candidates) == 0:
self.Debug("[VALUE] Rebalance: no fine candidates; skipping.")
return
scores = {}
vols = {}
ml_pass = []
history = self.History(candidates, self.value_trend_predictor.lookback, Resolution.Daily)
if history.empty:
self.Debug("[VALUE] Rebalance: History returned empty; skipping.")
return
for symbol in candidates:
try:
if symbol not in history.index.get_level_values(0):
continue
sym_hist = history.loc[symbol]
if "close" not in sym_hist.columns:
continue
closes_series = sym_hist["close"].dropna()
closes = [float(x) for x in closes_series.values.tolist()]
if len(closes) < self.value_trend_predictor.lookback:
continue
score, up = self.value_trend_predictor.predict(closes)
scores[symbol] = float(score)
vols[symbol] = self._CalculateVolatility(closes[-self.value_vol_lookback:])
if up:
ml_pass.append(symbol)
except Exception:
continue
self._latest_ml_pass_count = len(ml_pass)
selected = sorted(ml_pass, key=lambda s: scores.get(s, float("-inf")), reverse=True)
if len(selected) < self.value_portfolio_size:
remaining = [s for s in candidates if s not in selected and s in scores]
remaining.sort(key=lambda s: scores.get(s, float("-inf")), reverse=True)
need = self.value_portfolio_size - len(selected)
selected.extend(remaining[:need])
selected = selected[: self.value_portfolio_size]
selected.sort(key=lambda s: self._pe_by_symbol.get(s, float("inf")))
selected = selected[: self.value_portfolio_size]
self.Debug(
f"[VALUE] Universe funnel: coarse={self._latest_coarse_count}, "
f"fine_pass={self._latest_fine_pass_count}, "
f"ml_pass={self._latest_ml_pass_count}, "
f"selected={len(selected)}")
if self.LiveMode:
selected_str = ",".join(
f"{s.Value}(PE={self._pe_by_symbol.get(s, float('nan')):.2f})" for s in selected)
self.Log("[VALUE] Selected symbols: " + selected_str)
selected_tickers = set(s.Value for s in selected)
for k in list(self.Portfolio):
if (k.Value.Invested
and k.Key != self.anchor
and k.Key.Value not in selected_tickers
and k.Key.Value not in self._s3_claimed
and k.Key.Value in self._value_claimed):
self.Liquidate(k.Key, "Removed from value selection")
if len(selected) == 0:
self._value_claimed = set()
self._SaveState()
return
inv_vols = {}
for symbol in selected:
vol = vols.get(symbol, float("inf"))
if vol is None or vol <= 0 or not math.isfinite(vol):
continue
inv_vols[symbol] = 1.0 / vol
if not inv_vols:
weight = 1.0 / float(len(selected))
raw_weights = {s: weight for s in selected}
else:
total_inv_vol = sum(inv_vols.values())
raw_weights = {s: inv_vols[s] / total_inv_vol for s in inv_vols.keys()}
capped = {s: min(self.value_max_weight, w) for s, w in raw_weights.items()}
cur_sum = sum(capped.values())
scale = 1.0
if current_dd > self.value_dd_scale_threshold:
self.Debug(
f"[VALUE] Drawdown {current_dd:.2%} exceeds {self.value_dd_scale_threshold:.2%}, "
f"scaling exposure by {self.value_dd_scale_factor}")
scale = self.value_dd_scale_factor
final_weights = {}
if cur_sum > 0:
for s, w in capped.items():
final_weights[s] = (w / cur_sum) * VALUE_BUDGET * scale
for symbol, target_weight in final_weights.items():
if symbol not in self.Securities: continue
if not self.Securities[symbol].IsTradable: continue
if target_weight <= 0: continue
self.SetHoldings(symbol, target_weight)
self._value_claimed = set(s.Value for s in final_weights.keys())
self._SaveState()
def _GetFloat(self, root: object, paths: list, default: float) -> float:
for path in paths:
value = self._TryGetAttrPath(root, path)
numeric = self._CoerceToFloat(value)
if numeric is None:
continue
return float(numeric)
return float(default)
def _TryGetAttrPath(self, root: object, path: str) -> object:
current = root
for part in path.split("."):
if current is None or not hasattr(current, part):
return None
current = getattr(current, part)
return current
def _CoerceToFloat(self, value: object) -> float:
if value is None:
return None
if hasattr(value, "Value"):
try:
return float(getattr(value, "Value"))
except Exception:
return None
try:
return float(value)
except Exception:
return None
def _IsFinite(self, x: float) -> bool:
try:
return x is not None and math.isfinite(float(x))
except Exception:
return False
def _CalculateVolatility(self, closes: list) -> float:
if len(closes) < 2:
return float("inf")
returns = []
for i in range(1, len(closes)):
if closes[i - 1] <= 0 or closes[i] <= 0:
continue
r = math.log(closes[i] / closes[i - 1])
returns.append(r)
if len(returns) < 2:
return float("inf")
return float(np.std(returns))
def _ValuePortfolioDrawdown(self) -> float:
if self._value_equity_peak <= 0:
return 0.0
equity = self.Portfolio.TotalPortfolioValue
dd = (self._value_equity_peak - equity) / self._value_equity_peak
return max(0.0, float(dd))
def _SleeveReturnAndSnapshot(self, claimed_tickers: set, prev_qty: dict, prev_value):
"""Marks yesterday's qty at today's prices, so a same-day rebalance/
handoff/hedge swap isn't misread as a market move. Returns
(daily_return, current_value, current_qty_snapshot)."""
current_qty = {}
current_value = 0.0
for k in self.Portfolio.Values:
if k.Invested and k.Symbol.Value in claimed_tickers:
current_qty[k.Symbol] = k.Quantity
current_value += k.HoldingsValue
if not prev_qty or prev_value is None or prev_value <= 0:
return 0.0, current_value, current_qty
holdforward = 0.0
for sym, qty in prev_qty.items():
if self.Securities.ContainsKey(sym):
holdforward += qty * self.Securities[sym].Price
daily_return = holdforward / prev_value - 1.0
return daily_return, current_value, current_qty
def _DailySnapshot(self):
if self.IsWarmingUp: return
eq = self.Portfolio.TotalPortfolioValue
peaks_before = (self._hwm, self._value_equity_peak)
suspect_jump = (self._prev_value is not None and self._prev_value > 0 and
eq > self._prev_value * (1.0 + MAX_DAILY_EQUITY_JUMP))
if suspect_jump:
top_holdings = sorted(
((k.Symbol.Value, k.HoldingsValue) for k in self.Portfolio.Values if k.Invested),
key=lambda t: -abs(t[1])
)[:5]
msg = (f"[GUARD] Suspicious equity jump {self._prev_value:,.0f} -> {eq:,.0f} "
f"({(eq / self._prev_value - 1):+.1%}) on {self.Time:%Y-%m-%d} -- "
f"skipping HWM/peak update this bar (likely a bad/stale price print). "
f"Top holdings by value: {top_holdings}")
self.Debug(msg)
if self.LiveMode: self.Log(msg)
else:
self._hwm = max(self._hwm, eq)
self._value_equity_peak = max(self._value_equity_peak, eq)
if (self._hwm, self._value_equity_peak) != peaks_before:
self._SaveState()
dd = (eq - self._hwm) / self._hwm if self._hwm > 0 else 0.0
dr = (eq - self._prev_value) / self._prev_value if self._prev_value else 0.0
self._prev_value = eq
self._daily_rets.append(dr)
sh = ""
if len(self._daily_rets) >= 20:
r = np.array(self._daily_rets)
sig = np.std(r) * np.sqrt(252)
sh = f" Sh={np.mean(r)*252/sig if sig>0 else 0:+.2f}"
mode = "BULL" if self.allow_universe else "RISK-OFF"
hedge = "|HEDGED" if self._hedge_active else ""
self.Debug(f"[SNAP] {self.Time:%Y-%m-%d} Eq={eq:,.0f} DD={dd:.2%} "
f"D={dr:+.2%}{sh} [S3:{mode}{hedge}] "
f"S3pos={len(self._s3_claimed)} ValPos={len(self._value_claimed)} "
f"Cash={self.Portfolio.Cash/eq:.1%}")
sleeve_peaks_before = (self._s3_sleeve_peak, self._s3_inception_value,
self._value_sleeve_peak, self._value_inception_value)
s3_ret, s3_val, self._s3_prev_qty = self._SleeveReturnAndSnapshot(
self._s3_claimed, self._s3_prev_qty, self._s3_prev_sleeve_value)
self._s3_prev_sleeve_value = s3_val
if s3_val > 0:
self._s3_sleeve_peak = max(self._s3_sleeve_peak, s3_val)
if self._s3_inception_value is None:
self._s3_inception_value = s3_val
s3_dd = ((s3_val - self._s3_sleeve_peak) / self._s3_sleeve_peak
if self._s3_sleeve_peak > 0 else 0.0)
s3_cum = ((s3_val / self._s3_inception_value - 1.0)
if self._s3_inception_value else 0.0)
if self._hedge_active:
value_debug_str = "HEDGED (parked in UPRO)"
else:
value_ret, value_val, self._value_prev_qty = self._SleeveReturnAndSnapshot(
self._value_claimed, self._value_prev_qty, self._value_prev_sleeve_value)
self._value_prev_sleeve_value = value_val
if value_val > 0:
self._value_sleeve_peak = max(self._value_sleeve_peak, value_val)
if self._value_inception_value is None:
self._value_inception_value = value_val
value_dd = ((value_val - self._value_sleeve_peak) / self._value_sleeve_peak
if self._value_sleeve_peak > 0 else 0.0)
value_cum = ((value_val / self._value_inception_value - 1.0)
if self._value_inception_value else 0.0)
value_debug_str = (f"{value_val:,.0f} day={value_ret:+.2%} "
f"dd={value_dd:+.2%} cum={value_cum:+.2%}")
sleeve_peaks_after = (self._s3_sleeve_peak, self._s3_inception_value,
self._value_sleeve_peak, self._value_inception_value)
if sleeve_peaks_after != sleeve_peaks_before:
self._SaveState()
self.Debug(f"[SLEEVE] S3 {s3_val:,.0f} day={s3_ret:+.2%} dd={s3_dd:+.2%} "
f"cum={s3_cum:+.2%} | Value {value_debug_str}")
if not self.LiveMode: return
positions = sorted(
[(k.Key.Value, k.Value.HoldingsValue, k.Value.UnrealizedProfitPercent)
for k in self.Portfolio if k.Value.Invested],
key=lambda x: -x[1])
pos_lines = "\n".join(
f" {s:<8} GBP {v:>8,.0f} {p:>+.1%}"
for s, v, p in positions)
subject = (f"{'[UP]' if dr>=0 else '[DN]'} EOD {self.Time:%d %b %Y} "
f"{dr:+.2%} GBP{eq:,.0f} [S3:{mode}]")
s3_sleeve_line = (
f"S3 sleeve: GBP {s3_val:>10,.0f} Day {s3_ret:+.2%} "
f"DD {s3_dd:+.2%} Since-track {s3_cum:+.2%} "
f"({len(self._s3_claimed)} pos)")
if self._hedge_active:
upro_pos = self.Portfolio[self.hedge_instrument]
value_sleeve_line = (
f"Value sleeve: HEDGED into UPRO -- GBP {upro_pos.HoldingsValue:>10,.0f} "
f"Unrealized {upro_pos.UnrealizedProfitPercent:+.2%}")
else:
value_sleeve_line = (
f"Value sleeve: GBP {value_val:>10,.0f} Day {value_ret:+.2%} "
f"DD {value_dd:+.2%} Since-track {value_cum:+.2%} "
f"({len(self._value_claimed)} pos)")
body = (f"S3: {mode} | S3 positions: {len(self._s3_claimed)} | "
f"Value positions: {len(self._value_claimed)}\n"
f"Portfolio: GBP{eq:,.0f}\n"
f"Day: {dr:+.2%} DD: {dd:+.2%}{sh}\n\n"
f"{s3_sleeve_line}\n{value_sleeve_line}\n\n{pos_lines}")
self.Notify.Email(YOUR_EMAIL, subject, body)
def OnWarmupFinished(self):
self.Debug("[INIT] Warmup complete")
if self.LiveMode: self.Log("[INIT] Warmup complete")
eq = self.Portfolio.TotalPortfolioValue
self._hwm = max(self._hwm, eq)
self._prev_value = eq
self._value_equity_peak = max(self._value_equity_peak, eq)
actual_upro_w = (self.Portfolio[self.hedge_instrument].HoldingsValue / eq) if eq > 0 else 0.0
if self._hedge_active and actual_upro_w <= HEDGE_TOLERANCE:
msg = (f"[INIT] Reconciliation: hedge_active loaded True but no actual "
f"UPRO holding (upro_w={actual_upro_w:.2%}) -- resetting hedge state")
self.Debug(msg)
if self.LiveMode: self.Log(msg)
self._hedge_active = False
self._hedge_entry_price = None
self._hedge_entry_date = None
self._hedge_entry_stress = None
self._hedge_entry_dd = None
self._hedge_last_target = 0.0
self.current_hedge_target = 0.0
self._SaveState()
if self.Portfolio.TotalHoldingsValue != 0:
s3_inv = sum(1 for s in self.symbols
if s in self.Portfolio and self.Portfolio[s].Invested)
if s3_inv >= 5:
self.allow_universe = True
msg = f"[INIT] {s3_inv} S3 positions found -- inferring BULL mode"
self.Debug(msg)
if self.LiveMode: self.Log(msg)
managed = self.symbols | {self.anchor, self.hedge_instrument}
unknown = [k.Key for k in self.Portfolio
if k.Value.Invested and k.Key not in managed]
protected_orphans = []
for sym in unknown:
self.symbols.add(sym)
if sym not in self.ma:
self.ma[sym] = self.EMA(sym, self.band_len, Resolution.Daily)
self.adx[sym] = self.ADX(sym, self.adx_period, Resolution.Daily)
self.stretch_ema[sym] = self.EMA(sym, self.band_len, Resolution.Daily)
self.close_win[sym] = RollingWindow[float](self.band_len)
self.stretch_win[sym] = RollingWindow[float](STRETCH_WIN_LEN)
self.band_hist[sym] = RollingWindow[int](self.hist_len)
if self.LiveMode:
try:
self.WarmUpIndicator(sym, self.ma[sym], Resolution.Daily)
self.WarmUpIndicator(sym, self.adx[sym], Resolution.Daily)
self.WarmUpIndicator(sym, self.stretch_ema[sym], Resolution.Daily)
except Exception as e:
self.Debug(f"[INIT] warmup indicator failed for {sym.Value}: {e}")
msg = f"[INIT] Adopted orphan: {sym.Value}"
self.Debug(msg)
if self.LiveMode: self.Log(msg)
if sym.Value not in self._s3_claimed and sym.Value not in self._value_claimed:
self._s3_claimed.add(sym.Value)
self._value_claimed.add(sym.Value)
protected_orphans.append(sym.Value)
if protected_orphans:
msg = (f"[INIT] Reconciliation: no persisted ownership found for "
f"{protected_orphans} -- provisionally protected under both "
f"sleeves until their next respective rebalance")
self.Debug(msg)
if self.LiveMode: self.Log(msg)
self._SaveState()
self.Debug(f"[INIT] S3 symbols={len(self.symbols)} "
f"Invested={sum(1 for k in self.Portfolio if k.Value.Invested)} "
f"s3_claimed={len(self._s3_claimed)} value_claimed={len(self._value_claimed)}")
def OnEndOfAlgorithm(self):
eq = self.Portfolio.TotalPortfolioValue
self.Debug(f"[END] Eq={eq:,.2f} Ret={(eq/100_000-1)*100:+.2f}%")