feat(SH01): bootstrap storia full-history + last_block_only (punto-10: regime live non robusto)

Ri-validazione sh01_trainwindow_validate.py (rolling train_window == regime live):
- tw=8760 (live 365g): BTC FULL +82% ma fee-2x -42% (6/9 anni), ETH Sharpe -0.02,
  trade-rate 21.7-26.4% vs 9.8% validato -> NON robusto. Diagnosi sweep confermata
  (LogReg over-confident su train corto, th 0.58 inerte).
- progressione MONOTONA con la memoria (2y: Sh 2.05; 3y: 3.09; expanding: ROBUSTO
  8/9) -> l'edge di SH01 e' la memoria lunga.
- sweep soglia nel regime corto: instabile/incoerente fra asset -> NON ri-tunare.

Fix (decisione utente): ripristinare in live il regime validato.
- ml_wf_entries(last_block_only=True): fitta/predice solo l'ultimo blocco del WF
  (confini deterministici start+k*retrain) -> entries IDENTICHE per costruzione
  al tail del WF completo (parity test esatto); 0.6s/tick su 73k barre vs ~140 fit.
- runner._with_history: tick degli sleeve ml su parquet locale + feed live
  (dedup timestamp, gap-guard con WARN una-tantum se parquet stantio).
- _defs.py: params {last_block_only: true} sugli sleeve SHAPE (solo path live;
  il backtest canonico resta WF completo).

Effetto atteso live: trade-rate SH01 ~25% -> ~10% delle barre, selettivita'
ripristinata. Manutenzione: parquet fresco via download_all (oggi 2026-05-28,
margine ~11 mesi col feed 365g).

Test: 87/87 (4 nuovi: parity last-block esatta, merge storia, gap fallback).
Diario docs/diary/2026-06-07-sh01-trainwindow.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/portfolio/runner.py

@@ -116,6 +116,32 @@ def rebalance_allocations(ledger: PortfolioLedger, workers: dict, weights: dict[
     ledger.save()
 
 
+_OHLCV = ["timestamp", "open", "high", "low", "close", "volume"]
+
+
+def _with_history(hist: pd.DataFrame | None, live: pd.DataFrame,
+                  warned: set | None = None, asset: str = "") -> pd.DataFrame:
+    """Bootstrap storia per SH01 (punto-10, 2026-06-07): parquet locale PRIMA del
+    feed live (dedup sul timestamp). La ri-validazione ha mostrato che l'edge SH01
+    richiede il training EXPANDING full-history: col solo lookback live (365g) la
+    LogReg e' over-confident e la strategia NON e' robusta. Se c'e' un gap fra
+    parquet e feed (parquet stantio oltre il lookback) si usa il SOLO feed con
+    WARN: meglio il regime corto dichiarato che una serie con un buco."""
+    if hist is None or live is None or live.empty:
+        return live
+    lo = int(live["timestamp"].iloc[0])
+    h = hist[hist["timestamp"] < lo]
+    if h.empty:
+        return live
+    if int(h["timestamp"].iloc[-1]) < lo - 2 * 3_600_000:    # buco > 1 barra 1h
+        if warned is not None and asset not in warned:
+            warned.add(asset)
+            print(f"[runner] WARN: gap fra parquet e feed live per {asset} "
+                  f"(parquet stantio? rilanciare download_all) — SH01 senza bootstrap")
+        return live
+    return pd.concat([h[_OHLCV], live[_OHLCV]], ignore_index=True)
+
+
 def _resample(df: pd.DataFrame, tf: str) -> pd.DataFrame:
     """Resampla candele 1h -> 4h/1d mantenendo timestamp ms reale (come get_df del backtest)."""
     if tf == "1h":
@@ -247,6 +273,21 @@ def run(config_path: str = "portfolios.yml"):
     if ps_family:
         print(f"[runner] position_size globale={position_size} override famiglia={ps_family}")
 
+    # bootstrap storia full per gli sleeve ML (SH01): parquet locale + feed live.
+    # L'edge SH01 richiede train expanding full-history (sh01_trainwindow_validate);
+    # il path live fitta solo l'ultimo blocco (last_block_only nei params SHAPE).
+    ml_hist: dict[str, pd.DataFrame] = {}
+    ml_gap_warned: set[str] = set()
+    for a in sorted({s.asset for s in live_specs if s.kind == "ml"}):
+        try:
+            from src.data.downloader import load_data
+            ml_hist[a] = load_data(a, "1h")
+            last = pd.to_datetime(ml_hist[a]["timestamp"].iloc[-1], unit="ms", utc=True)
+            print(f"[runner] bootstrap storia SH01 {a}: {len(ml_hist[a])} barre parquet "
+                  f"(fino a {last:%Y-%m-%d %H:%M})")
+        except Exception as e:
+            print(f"[runner] WARN bootstrap storia {a} fallito: {e} — SH01 col solo feed")
+
     # lookback (giorni) richiesto per ogni asset = max sui worker che lo usano
     asset_days: dict[str, int] = {}
     for s in live_specs:
@@ -322,9 +363,13 @@ def run(config_path: str = "portfolios.yml"):
                     w.tick(res[s.a], res[s.b])
                 elif s.kind in _MULTI_KINDS:
                     w.tick(res)
+                elif s.kind == "ml":
+                    # SH01: storia full (parquet bootstrap + feed) -> il walk-forward
+                    # interno fitta solo l'ultimo blocco (last_block_only nei params).
+                    w.tick(_with_history(ml_hist.get(s.asset), res[s.asset],
+                                         ml_gap_warned, s.asset))
                 else:
-                    # single (fade/dip) e ml (SH01): StrategyWorker. SH01 retraina dentro
-                    # generate_signals (walk-forward) -> nessun training esterno.
+                    # single (fade/dip): StrategyWorker su feed live.
                     w.tick(res[s.asset])
 
             ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()})