# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
# Projet QC : 16239-macro-regimes-28b -- diagnostic reel, 6 ETF, 2019 -> 2026-09
qc_macro_diagnostic_code = '''
# 16239-macro-regimes-28b -- Diagnostic macro composite borne -> allocation mensuelle
# Lane myia-po-2026:CoursIA -- See #16239 livrable 1 (notebook QC-Py-28b)
# Proxies documents : ETF != series macro officielles (PMI/ISM indisponibles dans QC)
# Sans lookahead : chaque rebalancement n'utilise que l'historique PASSE accumule dans OnData.
from AlgorithmImports import *
class MacroRegimeAllocation(QCAlgorithm):
exposition = {
"EXPANSION": 1.00,
"NEUTRE": 0.60,
"RALENTISSEMENT": 0.30,
"STRESS": 0.10,
"WARMUP": 0.60, # diagnostic pas encore calculable : defaut documente
}
def initialize(self):
self.set_start_date(2016, 1, 1)
self.set_end_date(2026, 9, 12)
self.set_cash(100000)
self._tickers = ["SPY", "IEF", "SHY", "HYG", "LQD", "GLD"]
for t in self._tickers:
self.add_equity(t, Resolution.Daily)
self._closes = {t: [] for t in self._tickers}
self._dates = []
self._regimes = []
self.schedule.on(
self.date_rules.month_start("SPY"),
self.time_rules.at(9, 31),
self.rebalance,
)
def on_data(self, data):
if not all(data.bars.contains_key(t) for t in self._tickers):
return
for t in self._tickers:
self._closes[t].append(float(data.bars[t].close))
self._dates.append(self.time)
@staticmethod
def _z_of(series):
window = series[-252:]
mean = sum(window) / len(window)
std = (sum((x - mean) ** 2 for x in window) / len(window)) ** 0.5
return 0.0 if std == 0 else (window[-1] - mean) / std
def _diagnostic(self):
n = len(self._dates)
if n < 452: # 200 j de SMA + 252 j de fenetre z
return None
k = min(n, 800)
c = {t: v[-k:] for t, v in self._closes.items()}
trend = [
c["SPY"][i] / (sum(c["SPY"][i - 199 : i + 1]) / 200) - 1.0
for i in range(199, k)
]
slope = [c["IEF"][i] / c["SHY"][i] for i in range(k)]
credit = [c["HYG"][i] / c["LQD"][i] for i in range(k)]
gold_mom = [c["GLD"][i] / c["GLD"][i - 63] - 1.0 for i in range(63, k)]
z = {
"z_trend": self._z_of(trend),
"z_slope": self._z_of(slope),
"z_credit": self._z_of(credit),
"z_gold": -self._z_of(gold_mom), # inverse : or qui monte = demande refuge
}
def sig(v):
return 1 if v > 0.5 else (-1 if v < -0.5 else 0)
signals = sorted(sig(v) for v in z.values())
composite = (signals[1] + signals[2]) / 2 # mediane de 4 valeurs
if z["z_trend"] < -1.0 and z["z_credit"] < -1.0:
regime = "STRESS"
elif composite >= 0.5:
regime = "EXPANSION"
elif composite <= -0.5:
regime = "RALENTISSEMENT"
else:
regime = "NEUTRE"
return z, composite, regime
def rebalance(self):
result = self._diagnostic()
if result is None:
z, composite, regime = {}, 0.0, "WARMUP"
else:
z, composite, regime = result
self._regimes.append((self.time.date().isoformat(), regime, round(composite, 2)))
self.set_runtime_statistic("Regime", regime)
self.set_runtime_statistic("Median", f"{composite:+.2f}")
self.set_runtime_statistic(
"z(trend/slope/credit/gold)",
f"{z.get('z_trend', 0):+.2f} {z.get('z_slope', 0):+.2f} "
f"{z.get('z_credit', 0):+.2f} {z.get('z_gold', 0):+.2f}",
)
expo = self.exposition[regime]
self.set_holdings(
[PortfolioTarget("SPY", expo), PortfolioTarget("SHY", 1.0 - expo)]
)
self.debug(
f"{self.time.date()} regime={regime} median={composite:+.2f} "
f"z_trend={z.get('z_trend', 0):+.2f} z_slope={z.get('z_slope', 0):+.2f} "
f"z_credit={z.get('z_credit', 0):+.2f} z_gold={z.get('z_gold', 0):+.2f} "
f"-> SPY {expo:.0%} / SHY {1 - expo:.0%}"
)
def on_end_of_algorithm(self):
counts = {}
for _, r, _m in self._regimes:
counts[r] = counts.get(r, 0) + 1
self.set_runtime_statistic(
"Regimes 2016-2026",
" ".join(f"{k}:{v}" for k, v in sorted(counts.items())),
)
self.debug(f"=== Regimes mensuels ({len(self._regimes)} rebalancements) ===")
for d, r, m in self._regimes:
self.debug(f" {d} {r:>15} median={m:+.2f}")
'''