BRIEF = """
IMPLEMENTATION BRIEF - Task 104
Paper: Leung (2024), "Statistical Arbitrage via Single-view and Multi-view Spectral Clustering on Mixed
Frequency Data", SSRN 4975855. Headline strategy replicated: MV-CRSp(25)-All-BvLS-OU (Table 6 #1 / Table 8
last row): multi-view CO-REGULARIZED spectral clustering (Algorithm 4, lambda = 1/2, K = 25) on the "All"
feature set (25 views), distance screening to 50 pairs, simple long-short spread of normalized prices,
Ornstein-Uhlenbeck (Bertram 2010) optimal entry/exit thresholds.

1. UNIVERSE
- S&P 500 constituents, point in time: QuantConnect "US ETF Constituents" dataset, SPY holdings (daily
  files, history starts June 2009). Common stock only (Morningstar security_type ST00000001) via
  self.fundamentals(symbols) inside the selection callback.
- Dual-class shares (paper 3.7): group constituents by Morningstar company_id; when a company has more than
  one listed class in the index, keep the class with the earliest ipo_date (longest history), drop the rest.
- Eligibility for one formation window: constituent on the formation end date AND complete feature history
  (no missing day, no NaN in any of the 25 views) over the 252 formation days; rolling FF regressions need
  >= 126 daily observations (paper's minimum).
- Sample: paper 2000-12-31..2022. Platform: SPY constituents start June 2009 -> backtest start 2009-06-01,
  no end date. Formation/rolling-beta history before that date is seeded with history() (QC daily and
  minute data reach back to 1998).

2. FEATURES (Table 2), all at daily cadence, per asset, T = 252 formation days, N ~ 500
- P: normalized price P_t / P_1 (adjusted daily closes; P_1 = close on the first formation day).
- Intraday (from 1-minute bars of day t, regular session, log returns R_m between consecutive minute
  closes, M = number of returns): RV = sum R^2; Skew = sqrt(M) sum R^3 / (sum R^2)^1.5;
  Kurt = M sum R^4 / (sum R^2)^2; SVDown = sum (min(R,0))^2; SVUp = sum (max(R,0))^2;
  Beta_idx = sum R_i R_idx / sum R_idx^2 for 13 reference ETFs (SPY, DIA, IWM, QQQ, XLB, XLE, XLF, XLI,
  XLK, XLP, XLU, XLV, XLY), minute returns aligned on common timestamps.
- Low frequency: rolling 252-trading-day OLS (min 126 obs) of daily log return on daily log returns of
  Mkt, SMB, HML, UMD -> beta_mkt, beta_smb, beta_hml, beta_umd; ivol = std of residuals; tvol = std of the
  daily log returns in the window.
- Total 25 views: P, RV, Skew, Kurt, SVDown, SVUp, 13 Betas, 4 betas, ivol, tvol. View 1 = P.
- Data source: QC US Equities minute + daily bars (ADJUSTED). Each morning one batched history(minute)
  request for the previous session over all constituents + the 13 ETFs, features computed with numpy /
  pandas and appended to per-view T x N tables kept in the algorithm. New constituents are seeded once on
  first sight with a 252-day minute history and 504-day daily history (chunked requests).
- Fama-French / Carhart daily factor returns: NOT available on the platform (see REQUESTS).

3. CLUSTERING (every 21 trading days, on the morning after the formation end day)
- Standardize each view X_l by ROW (per day, cross-sectional z-score).
- Per view: RBF similarity with local scaling, kbar = 7: W_ij = exp(-||x_i - x_j||^2 / (sigma_i sigma_j)),
  sigma_i = Euclidean distance from asset i (its T-vector in that view) to its 7th nearest neighbour;
  W_ii = 0. Normalized Laplacian L = D^-1/2 W D^-1/2.
- Algorithm 4 (Kumar & Daume 2011 pairwise co-regularization): maximize sum_v tr(U_v' L_v U_v)
  + lambda sum_{v != v'} tr(U_v U_v' U_v' U_v'') s.t. U_v'U_v = I, lambda = 1/2. Alternating maximization:
  initialize U_v = top-K eigenvectors of L_v; then cycle over views, U_v <- top-K eigenvectors of
  L_v + lambda sum_{v' != v} U_v' U_v'^T; stop when the objective improves by < 1e-6 relative or after 20
  sweeps. V = U_1 (price view), rows normalized to unit norm, K-means (K = 25, k-means++, 10 restarts,
  fixed seed). Clusters of size 1 discarded.

4. SCREENING (3.3.5) -> 50 pairs
- Within each cluster compute MSE_ij = (1/T) sum_t (P_ti - P_tj)^2 on normalized prices; keep the
  ceil(50/K) = 2 lowest-MSE pairs per cluster; from those keep the 50 lowest overall (fewer if not enough).
  An asset may appear in two pairs of the same cluster (paper requires clusters, not pairs, to be disjoint).

5. SIGNAL AND TRADING RULE (per pair, per cycle)
- spread_t = P_t1 - P_t2 (normalized prices, P_1 = first formation day of that cycle; asset order fixed by
  symbol string).
- OU fit on the 252 formation spreads (Bertram 2010, zero-mean OU dX = -alpha X dt + eta dB, exact
  discretization MLE): b = sum X_t X_t-1 / sum X_t-1^2, alpha = -ln(b), eta^2 = 2 alpha s^2 / (1 - b^2),
  s^2 = mean squared residual (dt = 1 day; a* is invariant to the time unit). Pairs with b <= 0 or b >= 1
  (no mean reversion) are not traded in that cycle.
- c = 2 x 0.0005 x ||beta||_1 = 0.002. a* = argmax_{a < -c/2} alpha(2a + c) / (2 pi Erfi(a sqrt(alpha)/eta))
  (scipy bounded scalar optimisation on [-8 eta/sqrt(2 alpha), -c/2]); m* = -a*.
- Trading period = the 126 trading days after formation. Each day at the close: if no position and
  spread <= a*: open LONG spread (long asset 1, short asset 2, equal dollars); if spread >= m*: open SHORT
  spread. Long spread closes when spread >= m*, short spread closes when spread <= a*. Because the close
  level equals the opposite entry level, a close and the opposite open occur on the same day (one net
  order per leg). Re-opening allowed until the end of the trading period; open trades are force-closed at
  the close of the last trading day of the period (paper Table 10 "unconvergent"). Delistings are closed
  by the engine.
- Realization on the platform: signal sampled at before_market_close(SPY, 16) from the latest minute
  prices (close proxy) and executed with market-on-close orders (fills at the official close).

6. PORTFOLIO CONSTRUCTION / SIZING
- 6 overlapping cycles x 50 pairs = 300 slots of committed capital (paper Remark 2). Paper: $1 long / $1
  short per trade (2x gross per unit capital). Scaled to the 1x gross rule: each open trade holds
  +0.5/300 of portfolio value long and -0.5/300 short (weights fixed in shares at entry, theta fixed).
  Maximum gross exposure = 1.0 when all 300 slots are active. Orders of a symbol across trades are netted
  into one MOC order per symbol per day.
- Starting cash $10,000,000 (smallest leg weight 0.001667 -> ~$16,700 per leg).
- Fees/slippage: platform defaults (orders skill). The paper's 5 bp per leg is a reporting convention;
  it enters only through c in the OU threshold.

7. TIMING
- Trading-day counter on SPY's calendar. Cycle k starts on trading day 21k after the backtest start:
  formation = the 252 trading days ending the previous day, trading = the next 126 trading days.
- Daily universe selection (ETF constituents) at the platform's early-morning fire: feature update,
  seeding, and (every 21st day) clustering + OU fits. Universe returns the names of all pairs under
  monitoring (all active cycles), minute resolution. Signal/orders at before_market_close(SPY, 16).

8. DATASETS
- US ETF Constituents (SPY) - universe, from 2009-06.
- US Equities minute/daily bars (ADJUSTED) - prices, realized measures; from 1998.
- US Fundamental Data (Morningstar) - security_type, company_id, ipo_date for share-class filter.
- Kenneth French daily factors - MISSING on the platform (REQUESTS).

DECISIONS
1. Variant: MV-CRSp(25)-All-BvLS-OU (Table 6 #1). Rejected: K = 10, other feature sets, Alg 3 (MV-Sp),
   single-view, SprN / optimal-distribution / copula rules, cointegration signals.
2. Sample start 2009-06-01 (SPY constituents dataset start) instead of 2000-12-31; method unchanged.
3. Universe = SPY ETF holdings as S&P 500 proxy; common stock only; dual-class -> earliest ipo_date per
   company_id (paper: "longest history"). Rejected: is_primary_share (not what the paper states).
4. Eligible assets per cycle: constituents on the formation end date with complete 252-day features; new
   joiners are seeded with history so they are eligible as soon as their data is complete. Rejected: only
   names that were constituents throughout the window.
5. Table 2 typos: RV = sum of squared 1-minute log returns (printed as squared return differences);
   Kurt denominator (sum R^2)^2 (Amaya et al 2015; printed 3/2).
6. Intraday returns: regular-session minute closes only, no overnight return; days with < 2 returns or
   zero RV give NaN and exclude the asset from that cycle.
7. 1-minute bars (paper's stated frequency). Footnote 11 says 5-minute gives no material difference; if
   the daily minute-history pipeline proves too slow for the full run I will report it, not switch
   silently.
8. Rolling FF regression on raw factor returns (paper: "log returns of Mkt, SMB, HML, UMD"; Mkt = Mkt-RF +
   RF if that is how the data arrive); log returns of the asset. Depends on REQUEST 1.
9. Standardization by row = cross-sectional z-score per day (paper text). Rejected: per-asset time series
   standardization.
10. Local scaling sigma_i = distance to 7th nearest OTHER asset in the same view; W_ii = 0.
11. Algorithm 4 iteration scheme (paper gives only the objective): alternating eigen-updates initialized
    from single-view solutions, convergence 1e-6 relative or 20 sweeps (Kumar & Daume 2011 practice).
12. K-means: sklearn k-means++, n_init = 10, random_state = 0.
13. Screening: assets may repeat across pairs within a cluster; clusters of size 1 discarded; if fewer than
    50 pairs exist, trade all available.
14. Spread asset order by symbol string (rule is symmetric, order irrelevant to PnL).
15. OU fit: zero-mean OU as in the paper's SDE (spread not demeaned), exact-discretization MLE; invalid fits
    (b <= 0 or b >= 1) -> pair not traded that cycle. Rejected: demeaning the spread or fitting an
    intercept.
16. Entry/exit: long spread at <= a*, exit at m* = -a*; short at >= m*, exit at a*; the exit coincides with
    the opposite entry, so the position flips the same day (Bertram's cycle). Rejected: waiting one day
    after a close before re-opening, or exit at zero (that is the two-std rule).
17. Signal at the day's close is realized as a 15:44 ET sample of minute prices + MOC orders (fill at the
    official close). Rejected: daily bars + next-open execution (1-day lag).
18. Sizing: committed capital 300 slots, +0.5/300 / -0.5/300 of portfolio value per leg at entry
    (1x gross rule; paper $1/$1). Position weights fixed at entry (paper's theta fixed). Rejected:
    employed-capital re-leveraging (not implementable as a live book).
19. Fees: platform defaults instead of 5 bp per leg; c = 0.002 kept in the OU threshold.
20. Force-close at the close of the last day of the 126-day trading period; a symbol removed from SPY
    during a trading period keeps being traded until the period ends (delisting closes it).
21. Cycle calendar: literal 21-trading-day spacing, 252/126 trading-day windows counted on SPY's calendar
    (paper: "each month has 21 business days").

CLIENT DECISIONS (approved brief, override the above where they differ)
C1. FF factors: use the platform's Fama-French dataset
    (https://www.quantconnect.com/docs/v2/writing-algorithms/datasets/quantconnect/fama-french): market,
    size, value, momentum (up-minus-down) factors for the four rolling betas and ivol. Keep all 25 views.
C2. Execution: DAILY data. Spreads from daily closes; decide at 08:00 ET next trading day; market orders
    filling at that day's open. Forced close at trading-period end also fills at the next open. Minute
    data only for realized estimators via the morning history request; do NOT subscribe traded names at
    minute resolution.
C3. Cycle schedule: new cycle on the first trading day of each calendar month; formation = the 252 trading
    days before that day; trading period = six calendar months; six books overlap.
C4. Universe: keep every SPY constituent (no security-type filter). Only removal: dual-class rule, keep the
    class with the longest PRICE history on the platform; ipo_date only as tie-break.
Approved as written otherwise: 2009-06-01 start, no end date, W_ii = 0, +0.5/300 / -0.5/300 per leg,
1-minute bars for realized estimators.

BUILD NOTES (post-approval, recorded for the final report)
B1. Decisions 17, 20 and 21 above are superseded by C2 and C3 (daily closes, next-open market orders,
    first-trading-day-of-month cycles, 6-calendar-month trading periods).
B2. Start date moved from 2009-06-01 to 2009-08-01: the US ETF Constituents (SPY) data begins 2009-08-01
    on the platform (verified in the research notebook); formation history before that date comes from
    price history requests, so the first cycle still forms on 2009-08-03 with a full 252-day window.
B3. The ETF-constituents selector only fires when the constituents file changes (monthly); selection is
    forced to run every trading day via universe_settings.schedule.on(date_rules.every_day(SPY)) so the
    daily session update, seeding and month-start cycle formation run on every trading day.
B4. Algorithm 4 (co-regularized spectral clustering) is capped at 20 alternating sweeps; every cycle hit
    the cap (tol 1e-6 not reached), the embedding after 20 sweeps is used.
B5. A cycle whose 6-calendar-month end falls on a non-trading day is closed at the next trading day's
    open, which coincides with the new month's formation day -> 7 books overlap for that single day.
B6. Data issue (research notebook, cell 1-2): DIA's minute series is flat-lined on 2022-10-03 (390 bars,
    identical close), so the intraday beta vs DIA is undefined for every stock that session. DIA is a
    benchmark view, not a traded asset, so it is not blocked; a feature cell that is missing for EVERY
    aligned stock on a session is treated as a benchmark glitch and the previous session's value is
    carried forward (1 cell of 252 x 24 in the affected cycles, logged as glitch_cells_filled). Stock-
    specific gaps still exclude the stock from that cycle, as before.
B7. Delisting of a leg mid-trade: LEAN liquidates only the delisted leg (tag "Liquidate from delisting").
    The first full run (86d20f88ae1c0c03190f3e5d1fee3a34) left the surviving leg open, unhedged, until the
    cycle's forced close (up to 6 months; ~50 delisting events in the sample). Fixed: at the next 08:00
    decision the pair is closed (surviving leg ordered at the open) and retired for the rest of the cycle.
    Verified 2009-12-23 (ESV delisted: SLB, RDC, CHK legs closed the same morning).
C5. (client, after the 12-hour cut-off) Final backtest window 2021-09-01 to 2026-09-01; the 252-day
    formation window before the start comes from history requests as before. Diagnostic self.log lines and
    counters removed. Security initializer replaced by self.settings.seed_initial_prices = True.

REQUESTS (answered by C1)
1. Fama-French / Carhart daily factor returns (Mkt, SMB, HML, UMD from Kenneth French's Data Library)
   are needed for the 4 beta_FF views, ivol and the paper's regression features. I searched the platform's
   dataset list (Nasdaq Data Link, FRED, Cash Indices, Composite Factor Bundle): none provides these daily
   factor series (Nasdaq Data Link's Kenneth French feed is discontinued and needs a client API key).
   Options for the client to choose: (a) supply the daily FF3 + Momentum files in the Object Store (give me
   the keys); (b) authorise an in-algorithm download from Kenneth French's website; (c) ETF proxies
   (Mkt = SPY, SMB = IWM - SPY, HML = IWD - IWF; UMD has no ETF before 2013 -> MTUM only from 2013, so
   UMD would have to be dropped or built from stock momentum portfolios); (d) drop the beta_FF views
   (21 views) and keep ivol/tvol computed against the remaining factors. I will not proceed on this
   input without your decision.
"""

# region imports
from AlgorithmImports import *
import numpy as np
import pandas as pd
from scipy.special import erfi
from scipy.optimize import minimize_scalar
from sklearn.cluster import KMeans
# endregion

N_INTRADAY = 18          # RV, Skew, Kurt, SVDown, SVUp, 13 ETF betas
N_LOWFREQ = 6            # beta_mkt, beta_smb, beta_hml, beta_umd, ivol, tvol
N_VIEWS = 1 + N_INTRADAY + N_LOWFREQ   # + normalized price = 25
FORMATION = 252
FF_MIN_OBS = 126
KEEP = 2 * FORMATION + 5


class SymbolData:
    """Per-asset daily records: session date, adjusted close, daily log return and the 24 non-price views."""

    def __init__(self, symbol: Symbol) -> None:
        self.symbol = symbol
        self.dates = []
        self.closes = []
        self.logrets = []
        self.features = []   # np.array(N_INTRADAY + N_LOWFREQ)

    def last_date(self):
        return self.dates[-1] if self.dates else None

    def append(self, date, close: float, intraday: np.ndarray, ff_table: dict) -> None:
        if self.dates and date <= self.dates[-1]:
            return
        logret = np.log(close / self.closes[-1]) if self.closes and self.closes[-1] > 0 and close > 0 else np.nan
        self.dates.append(date)
        self.closes.append(close)
        self.logrets.append(logret)
        if len(intraday) != N_INTRADAY:
            intraday = np.full(N_INTRADAY, np.nan)
        self.features.append(np.concatenate([np.asarray(intraday, dtype=float), self._lowfreq(ff_table)]))
        if len(self.dates) > KEEP:
            del self.dates[0], self.closes[0], self.logrets[0], self.features[0]

    def _lowfreq(self, ff_table: dict) -> np.ndarray:
        """Rolling 252-day OLS of daily log returns on Mkt, SMB, HML, UMD (min 126 obs), ivol and tvol."""
        out = np.full(N_LOWFREQ, np.nan)
        dates = self.dates[-FORMATION:]
        rets = self.logrets[-FORMATION:]
        rows = [(r, ff_table[d]) for d, r in zip(dates, rets) if d in ff_table and np.isfinite(r)]
        if len(rows) < FF_MIN_OBS:
            return out
        y = np.array([r for r, _ in rows])
        f = np.array([x for _, x in rows])
        x = np.column_stack([np.ones(len(y)), f])
        coef, _, _, _ = np.linalg.lstsq(x, y, rcond=None)
        resid = y - x @ coef
        out[:4] = coef[1:]
        out[4] = np.std(resid, ddof=1)
        out[5] = np.std(y, ddof=1)
        return out


def session_features(stock_lc: pd.DataFrame, etf_lc: pd.DataFrame) -> pd.DataFrame:
    """Intraday views for one session from minute log-close matrices (time x symbol).
    Returns DataFrame symbol x N_INTRADAY."""
    r = stock_lc.ffill().diff().iloc[1:]
    e = etf_lc.ffill().diff().iloc[1:].reindex(r.index)
    mask = r.notna().values.astype(float)
    r0 = r.fillna(0.0).values
    e0 = e.fillna(0.0).values
    m = mask.sum(axis=0)
    rv = (r0 ** 2).sum(axis=0)
    with np.errstate(divide="ignore", invalid="ignore"):
        skew = np.sqrt(m) * (r0 ** 3).sum(axis=0) / rv ** 1.5
        kurt = m * (r0 ** 4).sum(axis=0) / rv ** 2
        sv_down = (np.minimum(r0, 0.0) ** 2).sum(axis=0)
        sv_up = (np.maximum(r0, 0.0) ** 2).sum(axis=0)
        betas = (r0.T @ e0) / (mask.T @ (e0 ** 2))
    out = np.column_stack([rv, skew, kurt, sv_down, sv_up, betas])
    bad = (m < 2) | (rv <= 0)
    out[bad, :] = np.nan
    return pd.DataFrame(out, index=stock_lc.columns)


def row_standardize(x: np.ndarray) -> np.ndarray:
    """Cross-sectional z-score of a T x N view matrix, one row (day) at a time."""
    mu = x.mean(axis=1, keepdims=True)
    sd = x.std(axis=1, keepdims=True)
    sd[sd == 0] = 1.0
    return (x - mu) / sd


def local_scaling_affinity(x: np.ndarray, kbar: int = 7) -> np.ndarray:
    """RBF affinity with Zelnik-Manor/Perona local scaling on N x T asset vectors; W_ii = 0."""
    sq = (x ** 2).sum(axis=1)
    d2 = np.maximum(sq[:, None] + sq[None, :] - 2.0 * x @ x.T, 0.0)
    d = np.sqrt(d2)
    sigma = np.sort(d, axis=1)[:, min(kbar, d.shape[0] - 1)]
    sigma[sigma == 0] = 1e-12
    w = np.exp(-d2 / (sigma[:, None] * sigma[None, :]))
    np.fill_diagonal(w, 0.0)
    return w


def normalized_laplacian(w: np.ndarray) -> np.ndarray:
    deg = w.sum(axis=1)
    dinv = np.where(deg > 0, 1.0 / np.sqrt(np.where(deg > 0, deg, 1.0)), 0.0)
    return dinv[:, None] * w * dinv[None, :]


def top_k_eigvecs(m: np.ndarray, k: int) -> np.ndarray:
    _, vecs = np.linalg.eigh(m)
    return vecs[:, -k:]


def coregularized_spectral(ls: List[np.ndarray], k: int, lam: float = 0.5, tol: float = 1e-6,
                           max_sweeps: int = 20) -> Tuple[np.ndarray, int]:
    """Kumar & Daume (2011) pairwise co-regularized multi-view spectral embedding (paper Algorithm 4).
    Alternating maximization of sum_v tr(U_v' L_v U_v) + lam sum_{v != v'} ||U_v' U_v'||_F^2.
    Returns the embedding of view 1 and the number of sweeps run."""
    us = [top_k_eigvecs(l, k) for l in ls]

    def objective() -> float:
        val = sum(np.trace(u.T @ l @ u) for u, l in zip(us, ls))
        for i in range(len(us)):
            for j in range(len(us)):
                if i != j:
                    val += lam * np.sum((us[i].T @ us[j]) ** 2)
        return float(val)

    prev = objective()
    sweeps = 0
    for sweeps in range(1, max_sweeps + 1):
        total = sum(u @ u.T for u in us)
        for v in range(len(us)):
            total -= us[v] @ us[v].T
            us[v] = top_k_eigvecs(ls[v] + lam * total, k)
            total += us[v] @ us[v].T
        cur = objective()
        if abs(cur - prev) <= tol * max(abs(prev), 1e-12):
            break
        prev = cur
    return us[0], sweeps


def ou_thresholds(x: np.ndarray, c: float = 0.002) -> Optional[Tuple[float, float, float, float]]:
    """Zero-mean OU exact-discretization MLE on the spread and Bertram (2010) optimal entry a* < -c/2,
    exit m* = -a*. Returns (a*, m*, alpha, eta) or None when the fit shows no mean reversion."""
    x0, x1 = x[:-1], x[1:]
    den = float(np.sum(x0 ** 2))
    if den <= 0:
        return None
    b = float(np.sum(x1 * x0) / den)
    if b <= 0 or b >= 1:
        return None
    alpha = -np.log(b)
    s2 = float(np.mean((x1 - b * x0) ** 2))
    eta = np.sqrt(2.0 * alpha * s2 / (1.0 - b ** 2))
    if not np.isfinite(eta) or eta <= 0:
        return None
    lo, hi = -8.0 * eta / np.sqrt(2.0 * alpha), -c / 2.0
    if lo >= hi:
        return None

    def neg_mu(a: float) -> float:
        return -alpha * (2.0 * a + c) / (2.0 * np.pi * erfi(a * np.sqrt(alpha) / eta))

    res = minimize_scalar(neg_mu, bounds=(lo, hi), method="bounded", options={"xatol": 1e-10})
    a_star = float(res.x)
    if not np.isfinite(a_star) or a_star >= hi:
        return None
    return a_star, -a_star, float(alpha), float(eta)


class Pair:
    def __init__(self, s1: Symbol, s2: Symbol, p1: float, p2: float, a: float, m: float) -> None:
        self.s1, self.s2, self.p1, self.p2, self.a, self.m = s1, s2, p1, p2, a, m
        self.direction = 0      # +1 long spread (long s1 / short s2), -1 short spread, 0 flat
        self.q1 = 0
        self.q2 = 0
        self.dead = False       # a leg was delisted: pair retired for the rest of the cycle

    def delisted(self, algorithm: QCAlgorithm) -> bool:
        for s in (self.s1, self.s2):
            if not algorithm.securities.contains_key(s) or algorithm.securities[s].is_delisted:
                return True
        return False


class Cycle:
    def __init__(self, start, end, pairs: List[Pair]) -> None:
        self.start, self.end, self.pairs = start, end, pairs


class SpectralClusteringStatArb(QCAlgorithm):
    """Leung (2024) MV-CRSp(25)-All-BvLS-OU: multi-view co-regularized spectral clustering on 25 daily
    feature views, distance screening to 50 pairs, Bertram (2010) OU optimal entry/exit thresholds."""

    ETF_TICKERS = ["SPY", "DIA", "IWM", "QQQ", "XLB", "XLE", "XLF", "XLI", "XLK", "XLP", "XLU", "XLV", "XLY"]
    CHUNK = 40

    def initialize(self) -> None:
        self.set_start_date(2021, 9, 1)
        self.set_end_date(2026, 9, 1)
        self.set_cash(10_000_000)
        self.settings.min_absolute_portfolio_target_percentage = 0
        self.settings.minimum_order_margin_portfolio_percentage = 0
        self.settings.seed_initial_prices = True
        self.universe_settings.resolution = Resolution.DAILY

        self._etfs = [self.add_equity(t, Resolution.DAILY).symbol for t in self.ETF_TICKERS]
        self._spy = self._etfs[0]
        self._ff = self.add_data(FamaFrench, "FF", Resolution.DAILY).symbol
        self.universe_settings.schedule.on(self.date_rules.every_day(self._spy))
        self._universe = self.add_universe(self.universe.etf(self._spy, self.universe_settings, self._select))

        self._data = {}            # Symbol -> SymbolData (constituents)
        self._etf_lc = {}          # session date -> minute log-close DataFrame of the 13 ETFs
        self._ff_table = {}        # session date -> np.array([Mkt, SMB, HML, UMD])
        self._sessions = []        # SPY session dates (ascending)
        self._last_processed = None
        self._seeded = False
        self._cycles = []          # active Cycle objects (up to 6 overlapping)
        self._last_cycle_month = None
        self._class_choice = {}    # company_id -> Symbol kept for dual-class companies
        self._k = 25
        self._lam = 0.5
        self._n_pairs = 50
        self._kbar = 7
        self._cost = 0.002
        self._slots = 6 * self._n_pairs

        self.schedule.on(self.date_rules.every_day(self._spy), self.time_rules.at(8, 0), self._trade)

    # ------------------------------------------------------------------ data pipeline
    def _select(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
        symbols = [c.symbol for c in constituents]
        self._update_calendar()
        prev = self._sessions[-1]
        if prev != self._last_processed:
            self._update_ff()
            if not self._seeded:
                self._seed_etfs()
                self._seeded = True
            else:
                active = {s for c in self._cycles for p in c.pairs for s in (p.s1, p.s2)}
                self._update_session(prev, [s for s in self._data if s in active or s in set(symbols)])
            self._last_processed = prev
        new = [s for s in symbols if s not in self._data]
        if new:
            self._seed_symbols(new)
        d = self._next_trading_day()
        if d.month != prev.month and self._last_cycle_month != (d.year, d.month):
            self._last_cycle_month = (d.year, d.month)
            cycle = self._form_cycle(d, symbols)
            if cycle is not None:
                self._cycles.append(cycle)
        return sorted({s for c in self._cycles for p in c.pairs for s in (p.s1, p.s2)})

    def _next_trading_day(self):
        hours = self.securities[self._spy].exchange.hours
        d = self.time.date()
        while not hours.is_date_open(datetime.combine(d, time(0, 0))):
            d += timedelta(days=1)
        return d

    def _update_calendar(self) -> None:
        bars = self.history(TradeBar, [self._spy], 2 * FORMATION + 10, Resolution.DAILY)
        dates = [t.date() for t in bars.index.get_level_values("time")] if not bars.empty else []
        if not self._sessions:
            self._sessions = dates
        else:
            for d in dates:
                if d > self._sessions[-1]:
                    self._sessions.append(d)
        self._sessions = self._sessions[-(2 * FORMATION + 10):]

    def _update_ff(self) -> None:
        n = 10 if self._ff_table else 4 * FORMATION
        for pt in self.history[FamaFrench](self._ff, n, Resolution.DAILY):
            vals = [pt.market_excess_return, pt.risk_free_rate, pt.smb, pt.hml, pt.momentum]
            if any(v is None for v in vals):
                continue
            mkt, rf, smb, hml, umd = [float(v) for v in vals]
            self._ff_table[pt.time.date()] = np.array([mkt + rf, smb, hml, umd])

    def _minute_lc(self, symbols: List[Symbol], start: datetime, end: datetime) -> pd.DataFrame:
        """Minute log-close matrix (time x symbol) over [start, end], regular session only."""
        frames = []
        for i in range(0, len(symbols), self.CHUNK):
            df = self.history(TradeBar, symbols[i:i + self.CHUNK], start, end, Resolution.MINUTE)
            if not df.empty:
                frames.append(df["close"].unstack(0))
        if not frames:
            return pd.DataFrame()
        lc = pd.concat(frames, axis=1).reindex(columns=symbols)   # fixed column set (missing -> NaN)
        return np.log(lc[lc > 0])

    def _seed_etfs(self) -> None:
        start = self._sessions[-FORMATION]
        lc = self._minute_lc(self._etfs, datetime.combine(start, time(0, 0)), self.time)
        for d, sub in lc.groupby(lc.index.date):
            self._etf_lc[d] = sub
        self._etf_lc = {d: v for d, v in self._etf_lc.items() if d in set(self._sessions[-FORMATION:])}

    def _update_session(self, prev, symbols: List[Symbol]) -> None:
        start = datetime.combine(prev, time(0, 0))
        etf = self._minute_lc(self._etfs, start, self.time)
        etf = etf[etf.index.date == prev] if not etf.empty else etf
        self._etf_lc[prev] = etf if not etf.empty else None
        for d in [d for d in self._etf_lc if d < self._sessions[-FORMATION]]:
            del self._etf_lc[d]
        if not symbols:
            return
        lc = self._minute_lc(symbols, start, self.time)
        lc = lc[lc.index.date == prev] if not lc.empty else lc
        closes = self.history(TradeBar, symbols, 1, Resolution.DAILY)
        self._append_session(prev, symbols, lc, closes)

    def _append_session(self, d, symbols: List[Symbol], lc: pd.DataFrame, daily: pd.DataFrame) -> None:
        etf = self._etf_lc.get(d)
        feats = session_features(lc, etf) if (etf is not None and not lc.empty) else None
        close_map = {}
        if not daily.empty:
            sub = daily.reset_index()
            sub = sub[sub["time"].dt.date == d]
            close_map = dict(zip(sub["symbol"], sub["close"]))
        for s in symbols:
            close = close_map.get(s)
            if close is None or not np.isfinite(close):
                continue
            f = feats.loc[s].values if (feats is not None and s in feats.index) else np.full(N_INTRADAY, np.nan)
            self._data[s].append(d, float(close), f, self._ff_table)

    def _seed_symbols(self, symbols: List[Symbol]) -> None:
        """Seed new constituents with 504 daily closes and 252 sessions of intraday features."""
        for s in symbols:
            self._data[s] = SymbolData(s)
        sessions = self._sessions[-FORMATION:]
        start = datetime.combine(sessions[0], time(0, 0))
        for i in range(0, len(symbols), self.CHUNK):
            chunk = symbols[i:i + self.CHUNK]
            daily = self.history(TradeBar, chunk, 2 * FORMATION + 5, Resolution.DAILY)
            if daily.empty:
                continue
            daily = daily.reset_index()
            daily["date"] = daily["time"].dt.date
            lc = self._minute_lc(chunk, start, self.time)
            by_day = {d: sub for d, sub in lc.groupby(lc.index.date)} if not lc.empty else {}
            for d, sub in daily.groupby("date"):
                if d > self._sessions[-1]:
                    continue
                close_map = dict(zip(sub["symbol"], sub["close"]))
                if d in by_day and self._etf_lc.get(d) is not None:
                    feats = session_features(by_day[d], self._etf_lc[d])
                else:
                    feats = None
                for s in chunk:
                    close = close_map.get(s)
                    if close is None or not np.isfinite(close):
                        continue
                    f = feats.loc[s].values if (feats is not None and s in feats.index) else np.full(N_INTRADAY, np.nan)
                    self._data[s].append(d, float(close), f, self._ff_table)

    # ------------------------------------------------------------------ formation
    def _dedup_share_classes(self, symbols: List[Symbol]) -> List[Symbol]:
        """Paper 3.7 / client C4: one class per company, keep the class with the longest platform price
        history; earliest ipo_date as tie-break."""
        groups = {}
        ipo = {}
        for f in self.fundamentals(symbols):
            cid = f.company_reference.company_id
            if not cid:
                continue
            groups.setdefault(cid, []).append(f.symbol)
            ipo[f.symbol] = f.security_reference.ipo_date
        drop = set()
        for cid, members in groups.items():
            if len(members) < 2:
                continue
            if cid not in self._class_choice or self._class_choice[cid] not in members:
                hist = self.history(TradeBar, members, datetime(1998, 1, 1), self.time, Resolution.DAILY)
                first = {}
                if not hist.empty:
                    first = hist.reset_index().groupby("symbol")["time"].min().to_dict()
                members.sort(key=lambda s: (first.get(s, datetime.max), ipo.get(s, datetime.max), str(s)))
                self._class_choice[cid] = members[0]
            drop.update(s for s in members if s != self._class_choice[cid])
        return [s for s in symbols if s not in drop]

    def _form_cycle(self, d, symbols: List[Symbol]) -> Optional[Cycle]:
        sessions = self._sessions[-FORMATION:]
        candidates = self._dedup_share_classes(symbols)
        aligned = [s for s in sorted(candidates, key=str)
                   if (sd := self._data.get(s)) is not None and len(sd.dates) >= FORMATION
                   and sd.dates[-FORMATION:] == sessions]
        if len(aligned) < 2 * self._k:
            return None
        all_feats = np.stack([np.array(self._data[s].features[-FORMATION:]) for s in aligned], axis=2)  # T x 24 x N
        # A feature cell missing for EVERY stock on a session is a benchmark-ETF data glitch (e.g. a flat-lined
        # minute series, zero intraday variance -> beta undefined), not a stock property: carry the previous
        # session's value forward for that cell. Stock-specific gaps still exclude the stock below.
        univ_nan = np.all(np.isnan(all_feats), axis=2)                                        # T x 24
        for t, j in zip(*np.nonzero(univ_nan)):
            if t > 0:
                all_feats[t, j, :] = all_feats[t - 1, j, :]
        ok = np.all(np.isfinite(all_feats), axis=(0, 1))                                      # N
        eligible = [s for s, good in zip(aligned, ok) if good]
        if len(eligible) < 2 * self._k:
            return None
        closes = np.array([self._data[s].closes[-FORMATION:] for s in eligible]).T          # T x N
        prices = closes / closes[0:1, :]
        feats = all_feats[:, :, ok]                                                            # T x 24 x N
        views = [prices] + [feats[:, j, :] for j in range(N_INTRADAY + N_LOWFREQ)]
        ls = [normalized_laplacian(local_scaling_affinity(row_standardize(v).T, self._kbar)) for v in views]
        u, _ = coregularized_spectral(ls, self._k, self._lam)
        norms = np.linalg.norm(u, axis=1, keepdims=True)
        norms[norms == 0] = 1.0
        labels = KMeans(n_clusters=self._k, init="k-means++", n_init=10, random_state=0).fit_predict(u / norms)

        per_cluster = int(np.ceil(self._n_pairs / self._k))
        candidates_pairs = []
        for c in range(self._k):
            idx = np.where(labels == c)[0]
            if len(idx) < 2:
                continue
            pm = prices[:, idx]
            mse = []
            for a in range(len(idx)):
                for b in range(a + 1, len(idx)):
                    mse.append((float(np.mean((pm[:, a] - pm[:, b]) ** 2)), idx[a], idx[b]))
            mse.sort()
            candidates_pairs.extend(mse[:per_cluster])
        candidates_pairs.sort()
        selected = candidates_pairs[:self._n_pairs]

        pairs = []
        for _, i, j in selected:
            fit = ou_thresholds(prices[:, i] - prices[:, j], self._cost)
            if fit is None:
                continue
            a_star, m_star, _, _ = fit
            pairs.append(Pair(eligible[i], eligible[j], closes[0, i], closes[0, j], a_star, m_star))
        end = (pd.Timestamp(d) + pd.DateOffset(months=6)).date()
        return Cycle(d, end, pairs)

    # ------------------------------------------------------------------ trading
    def _trade(self) -> None:
        """08:00 ET: apply the OU rule to yesterday's closing spreads; market orders fill at today's open."""
        today = self.time.date()
        if not self._sessions or self._last_processed != self._sessions[-1]:
            return
        prev = self._sessions[-1]
        deltas = {}
        slot_value = self.portfolio.total_portfolio_value * 0.5 / self._slots
        for cycle in list(self._cycles):
            if today >= cycle.end:
                for p in cycle.pairs:
                    self._close_pair(p, deltas)
                self._cycles.remove(cycle)
                continue
            for p in cycle.pairs:
                if p.dead:
                    continue
                if p.delisted(self):
                    # the engine liquidates the delisted leg; close the surviving leg the same morning
                    self._close_pair(p, deltas)
                    p.dead = True
                    continue
                sd1, sd2 = self._data.get(p.s1), self._data.get(p.s2)
                if sd1 is None or sd2 is None or sd1.last_date() != prev or sd2.last_date() != prev:
                    continue
                x = sd1.closes[-1] / p.p1 - sd2.closes[-1] / p.p2
                if (p.direction == 1 and x >= p.m) or (p.direction == -1 and x <= p.a):
                    self._close_pair(p, deltas)
                if p.direction == 0:
                    if x <= p.a:
                        self._open_pair(p, 1, slot_value, sd1.closes[-1], sd2.closes[-1], deltas)
                    elif x >= p.m:
                        self._open_pair(p, -1, slot_value, sd1.closes[-1], sd2.closes[-1], deltas)
        for s, q in deltas.items():
            if q != 0 and self.securities.contains_key(s) and self.securities[s].is_tradable:
                self.market_order(s, q)

    def _open_pair(self, p: Pair, direction: int, slot_value: float, c1: float, c2: float, deltas: dict) -> None:
        q1 = int(np.floor(slot_value / c1)) * direction
        q2 = -int(np.floor(slot_value / c2)) * direction
        if q1 == 0 or q2 == 0:
            return
        p.direction, p.q1, p.q2 = direction, q1, q2
        deltas[p.s1] = deltas.get(p.s1, 0) + q1
        deltas[p.s2] = deltas.get(p.s2, 0) + q2

    def _close_pair(self, p: Pair, deltas: dict) -> None:
        if p.direction == 0:
            return
        deltas[p.s1] = deltas.get(p.s1, 0) - p.q1
        deltas[p.s2] = deltas.get(p.s2, 0) - p.q2
        p.direction, p.q1, p.q2 = 0, 0, 0

NOTES = """
PAPER: Leung, Raymond C. W. (Sept 2024). "Statistical Arbitrage via Single-view and Multi-view Spectral
Clustering on Mixed Frequency Data". SSRN 4975855. 42 pages. Supplemental Materials = [Leung, 2024].

--- p1-3 (Abstract, Intro) ---
- Two steps of stat arb: (1) identify clusters of similar assets; (2) monitor clustered assets for price
  deviations, long-short trade.
- Features: daily closing prices PLUS realized estimators from high-frequency intraday data (realized
  volatility, realized beta, others) at daily cadence. Realized variance = sum of squared intraday returns.
- Clustering: single-view spectral clustering (concatenate features into one long vector per asset, stack to
  one matrix) vs multi-view spectral clustering (collection of feature matrices, one per feature/view).
- Trading rules studied: distance method 2-std rule (Gatev et al 2006) extended to other parametric
  distributions; bivariate cointegration (Vidyamurthy 2004, Chiu & Wong 2011, Huck & Afawubo 2015);
  copula (Xie et al 2016, Rad et al 2016); optimal entry-exit under Ornstein-Uhlenbeck spread (Bertram 2010).
- Main text: S&P 500 equities 2000-2022. Also ETFs 2012-2022, continuous futures 2009-2022, forex 2010-2022
  (details in Supplemental Materials).
- Classical distance method: normalized price index P_{t,i}/P_{1,i} over formation period; pairs with
  similar indices in same cluster; trading period: long underpriced / short overpriced when gap opens,
  profit when gap closes.
- BEST STRATEGY (paper's headline, S&P500): multi-view clustering on many realized quantities + closing
  prices, with OU optimal entry/exit rule (Bertram 2010). Sharpe 1.166 gross, 0.830 net of TC. Max DD 4.6%
  on daily excess returns. Distance method + 2std: Sharpe 0.571 gross, 0.338 net, MDD 18.8%.
- Best net Sharpe by clustering: distance 0.187; single-view 0.495; multi-view 0.830.

--- p4-6 (Lit review, Methodology 3, 3.1, 3.2, start 3.3) ---
- Strategy = (a) features set; (b) clustering algorithm; (c) signal; (d) trading rule. Table 1 summarizes.
- Trading conditions endogenous (not a fixed rebalancing period).
- 3.1 Formation/trading periods: N assets. Formation period = 12x21 = 252 business days. Trading period =
  6x21 = 126 business days following formation. T_form, T_trade sets of days. Figure 1 timeline (p24).
  (Overlap/staggering of periods not stated here - check Fig 1 / Sec 3.6.)
- 3.2 Features: normalized price index  P_{t,i} := P_{t,i}/P_{1,i}, normalized to the first day of the
  formation period (closing prices). Table 2 (p32) lists individual features; Table 3 (p33) lists reference
  indices for intraday beta (e.g. SPY); Table 4 (p34) lists feature SETS.
- Realized estimators from intraday data: realized variance, skewness, kurtosis, semivariances, intraday
  realized betas vs reference assets. Day t has M equispaced intraday prices {P_{t_m,i}}_{m=0..M}, with
  t_m - t_{m-1} = 1 MINUTE exclusively. Log return R_{t_m,i} = log P_{t_m,i} - log P_{t_{m-1},i}.
  a+ = max(a,0), a- = min(a,0). Beta_{t,index,i} = realized covariance / realized variance of index
  (intraday, day t only). E.g. Beta_{t,i,SPY} vs SPDR S&P500 ETF.
- Low-frequency features: daily rolling regressions, lookback 252 trading days (min 126 obs), daily log
  returns of asset on daily log returns of Mkt, SMB, HML, UMD (Kenneth French website) -> betas
  beta_{t,mkt,i}, beta_{t,smb,i}, beta_{t,hml,i}, beta_{t,umd,i}. Idiosyncratic vol = std of residuals.
  Total vol = std of daily returns over window. Upper-case Beta = intraday day-t; lower-case beta =
  252-day rolling.
- Notation: p features, x_{t,i;l} = l-th feature of asset i day t. X_l is T x N matrix (T=252), columns
  x_{i;l} (T x 1 per asset).
- 3.3 Clustering: three procedures: distance method (Gatev: pairwise grouping by small MSE of normalized
  price processes = bivariate clustering), single-view spectral, multi-view spectral. Clusters need not
  cover all assets (may be non-exhaustive); (continues p7).
- Footnote 4: distance method = (i) distance clustering (pairs by closeness of MSE of normalized prices)
  + (ii) two-std trading rule (normalized price difference deviates >= 2 std). Paper separates these.

--- p7-9 (3.3.1 distance, 3.3.2 single-view, 3.3.3 multi-view) ---
- Clusters pairwise disjoint; need not cover all assets. Bivariate cluster = exactly 2 assets; multivariate
  cluster = >= 3 assets. Clusters with only one asset are DISCARDED.
- 3.3.1 Distance clustering: concatenate p features into pT x N matrix Z = [X_1; X_2; ...; X_p], columns z_i
  (pT x 1). MSE_ij := (1/T) ||z_i - z_j||^2 (Euclidean). With p=1 and price -> Gatev. Select the 50 pairs
  with lowest MSE => 50 clusters of 2 assets each. Footnote 5: literature uses 20 pairs; 50 chosen for
  broad representation (see Remark 1, p10). (Whether an asset can appear in several pairs: not stated;
  but clusters must be pairwise disjoint per p7 -> implies greedy disjoint selection? ambiguous.)
- 3.3.2 Single-view: standardize each X_l BY ROW (each row = one day across N assets -> mean 0, std 1),
  stack to pT x N matrix Z. K-means (Algorithm 1) is a benchmark. Spectral clustering (Ng et al 2001 /
  Shi-Malik 2000): similarity W with RBF + local scaling (Zelnik-Manor & Perona 2004), kbar = 7:
  sigma_i := ||z_i - z_{kbar}|| = distance of i to its 7th nearest neighbour.
  W_ij = exp(-||z_i - z_j||^2 / (sigma_i sigma_j))   (eq 3).
  Algorithm 2: L := D^{-1/2} W D^{-1/2}, d_ii = sum_j w_ij; top K eigenvectors u_1..u_K of L;
  U = [u_1..u_K] (N x K); row-normalize U to unit Euclidean norm; K-means on rows y_i into K clusters.
- 3.3.3 Multi-view: use set of standardized matrices {X_1..X_p} (each standardized by row as above), one
  similarity matrix per view: W_l,ij = exp(-||x_{i;l} - x_{j;l}||^2 / (sigma_{i;l} sigma_{j;l})), kbar=7
  (eq 4). View 1 = normalized prices (most informative view).
  Algorithm 3 (co-trained multi-view spectral, Ng et al + Blum-Mitchell co-training):
    1: L_v = D_v^{-1/2} W_v D_v^{-1/2} per view; U_v = N x K top-K eigenvectors of L_v.
    2: S_v = sym( W_v * sum_{v' != v} U_{v'} U_{v'}^T ), sym(A) = (A + A^T)/2.
    3: Use S_1..S_p as new similarity matrices, compute normalized Laplacians, new U_1..U_p.
    4: V := U_1 (view 1). 5: row normalize V. 6: K-means on V -> C_1..C_K.
    (Number of co-training iterations: ONE pass as written in Algorithm 3.)
  Algorithm 4 (co-regularized multi-view spectral, Kumar & Daume 2011): maximize
    sum_v tr(U_v^T L_v U_v) + lambda * sum_{v != v'} tr(U_v U_v^T U_v' U_v'^T), s.t. U_v^T U_v = I.
    lambda fixed = 1/2 (no CV). V := U_1, row normalize, K-means. lambda=0 reduces to Alg 3.

--- p10-12 (3.3.4 cluster sizes, 3.3.5 screening, 3.4 signals, 3.5 trading rules start) ---
- 3.3.4: K in {10, 25} only (sensitivity check; no optimality stance).
- 3.3.5 Distance method screening (distill multivariate clusters to 50 pairs): given clusters C_1..C_K with
  |C_k| >= 2: for each cluster, for all i != j in C_k compute MSE_ij on NORMALIZED PRICES over formation
  period. Rules: (1) within each cluster select ceil(50/K) pairs with lowest MSE within cluster; (2) from
  these K*ceil(50/K) pairs select the 50 lowest MSE. K=10 -> 5 per cluster; K=25 -> 2 per cluster.
  Figure 3 illustrates. Remark 1: 50 pairs so that K=25 gives 2 pairs per cluster.
- 3.4.1 Bivariate signals (assets labelled 1 and 2):
  * Simple long-short (Gatev): undirected spread_t = P_{t,1} - P_{t,2} (normalized prices). Direction fixed
    by the opening rule.
  * Bivariate cointegration (Vidyamurthy; Rad et al 2016 Sec 4.2; Engle-Granger 2-step): on normalized
    prices in formation period, OLS P1 on P2 and P2 on P1; compute ADF stat of both regressions' residuals;
    pick regression with LOWEST ADF stat. If first: spread_t = P_{t,1} - beta_hat P_{t,2}. ADF only a
    heuristic to pick direction (no cointegration test gate).
  * Bivariate copula (Rad et al 2016 Sec 4.3): daily returns in formation; fit marginals by MLE (Gaussian,
    logistic, generalized logistic, GEV); copulas (Gaussian, Student-t, Clayton, Gumbel, Frank, Joe); pick
    (F1,F2,C) triple with highest AIC. Details in Supplemental Materials.
- 3.4.2 Multivariate cointegration (cluster of n=|C_k| assets): Johansen VECM with ONE lag, no trend, no
  constant; AIC selects rank r>=1 (r<n); cointegrating vectors normalized via Phillips triangular
  representation (one entry = 1), then each vector rescaled by Euclidean norm of the other entries
  (footnote 6: D = diag(1/||b_l||)). Select ONE vector: u_lt = beta_l^T P_t over formation; ADF stat with
  lag 1, no trend, no constant; pick smallest ADF stat -> spread_t = beta_{l*}^T P_t.
- 3.5 Trading rules. Spread percentile thresholds: spreads of the form spread_t = beta^T P_t (n=2 for
  simple LS and biv. coint.; n=|C_k| for multivariate). Two-std rule (Gatev): fit Gaussian on formation
  spread -> mu_hat, sigma_hat. In trading period observe spread_s = beta^T P_s daily. OPEN first s_op where
  spread >= Phi^{-1}_{mu,sigma}(0.975): take position of beta_i DOLLARS in asset i (i.e. long asset1 $1,
  short asset2 $beta_hat... wait sign: spread high -> position +beta_i dollars in asset i?? as written:
  "take the position of beta_i dollars into asset i" when spread >= upper; CLOSE first s > s_op with
  spread_s <= 0. Symmetric: spread <= Phi^{-1}(0.025): position -beta_i dollars in asset i; close when
  spread_s >= 0. (NOTE: literally taken this is long the expensive leg; likely paper convention that the
  "position" is the spread exposure; mean reversion requires SHORTING spread when high -> I must decide:
  economic sense = short spread when high (i.e. -beta_i dollars in asset i), long when low. Check p13
  for clarification.) After close, continue monitoring for re-open until end of trading period.
  Close threshold is spread crossing 0 (NOT mu_hat) -- as written "spread_s <= 0". Hmm, for
  spread = P1 - beta P2 the mean is not zero in general... paper says 0. Check later pages/Supp.
  Benchmark strategy = distance clustering on closing prices + two-std rule.
  Footnote 7: use 0.025 / 0.975 percentiles ("round numbers") for all distributions incl. Gaussian.
- Non-Gaussian extension: fit formation spread to each of a set of distributions (incl. Gaussian) ...
  (continues p13).

--- p13-15 (3.5 cont., 3.6 performance, 3.7 data, 4 results start) ---
- Non-Gaussian thresholds: fit formation spread to each distribution, pick LOWEST AIC; Q_hat quantile fn;
  thresholds Q_hat(0.025), Q_hat(0.975); rest identical to Gaussian case.
- Bivariate mispricing index (copula only; Xie et al 2016, Rad et al 2016): u_{s,i} = F_i(R_{s,i});
  h1(u1|u2), h2(u2|u1) conditional copula CDFs; m_{s,1} = h1 - 0.5, m_{s,2} = h2 - 0.5; cumulative
  M_{s,i} = M_{s-1,i} + m_{s,i}, reset to 0 at start of trading period. Open when M1 > 0.5 and M2 < -0.5:
  short $1 asset 1, long $1 asset 2; unwind when M1 <= 0 and M2 >= 0. Symmetric: M1 < -0.5 and M2 > 0.5:
  long $1 asset1, short $1 asset 2; close when M1 >= 0 and M2 <= 0. Continue monitoring after close.
- ORNSTEIN-UHLENBECK thresholds (Bertram 2010) [used by the HEADLINE strategy]: dX = -alpha X dt + eta dB.
  Enter at X = a, exit at X = m, a < m. Optimal is symmetric m* = -a*. a* maximizes expected mean return
  mu(a, c, alpha, eta) = alpha (2a + c) / (2 pi Erfi(a sqrt(alpha)/eta))  [note: "2a + c" as printed;
  Bertram's formula is mu = alpha(m - a - c)/(pi(Erfi(m sqrt(alpha)/eta) - Erfi(a sqrt(alpha)/eta))); with
  m = -a this gives alpha(-2a - c)/(2 pi (-Erfi(a sqrt(alpha)/eta))) = alpha(2a + c)/(2 pi Erfi(a
  sqrt(alpha)/eta)) with a < 0 the entry level. So enter when spread <= a* < 0 (long spread), exit at -a*;
  symmetric: enter when spread >= -a* (short spread), exit at a*.] Need c > 0. Continue monitoring after
  close. X_t = spread_t (the spread as constructed; OU has zero mean -> spread presumably demeaned? paper
  sets X_t = spread_t directly). c = 2 * 0.0005 * ||beta||_1 (one-way 5bp per asset). MLE (alpha_hat,
  eta_hat) on formation spread. (Bertram MLE: AR(1) regression of X_t on X_{t-1} with dt = 1/252.)
- 3.6 Performance: trade l opened at s_op with share vector theta_l. Long $1 asset1 / short $1 asset2 ->
  theta = (1/P_{sop,1}, -1/P_{sop,2}). Daily MTM PnL_{s,l} = theta^T (P_s - P_{s-1}) (raw prices).
  Employed-capital portfolio excess return R_s = (1/L_s) sum_l PnL_{s,l}, L_s = number of ACTIVE trades on
  day s, summed across all SIX overlapping trading periods (=> formation/trading periods are STAGGERED:
  a new 12-month formation + 6-month trading cycle starts every month; 6 trading periods overlap at any
  time, each with 50 pairs => committed capital denominator 6 x 50 = 300, Remark 2).
  TC: 5bp proportional on each dollar of position per asset at entry and at exit (footnote 8: from
  Frazzini et al 2018 Table II). Employed capital portfolio is the focus (Gatev p806).
- 3.7 Data: R + Python (single-view) + C++ (multi-view). Data: Kibot intraday 1-min + daily (split/div
  adjusted); CRSP daily S&P 500 constituents 2000-2022; WRDS Beta Suite rolling betas. Footnote 10: start
  2000-12-31 (Kibot intraday coverage). Footnote 11: 1-min vs 5-min realized estimators: no material
  difference. Dual-class shares (GOOG/GOOGL, FOX/FOXA, NWSA/NWS): keep the class with longest history,
  drop the other (avoid spurious pair selection).
- ETFs (45 names), futures (22), forex (29 USD, 13 GBP, 15 EUR) in Supplemental Materials; not main text.
- Section 4: main text figures/tables = S&P 500 equities.

--- p16-18 (4.1-4.3 results) ---
- >2000 combinations computed. Abbreviations: DM = distance clustering; SV-KM(K) single-view K-means;
  SV-Sp(K) single-view spectral; MV-Sp(K) multi-view (co-trained, Alg 3) spectral; MV-CRSp(K) multi-view
  co-regularized spectral (Alg 4). Signals: BvLS simple long-short; BvCoint; BvCopl; MvCoint. Rules:
  Spr(norm)/SprN two-std; Spr(optd) optimal distribution percentiles; MisPrcIdx; OrnUhl/OU.
  Any SV/MV clustering + bivariate signal ALWAYS applies distance screening (3.3.5).
- Table 5 benchmark DM on P only: BvLS + OU Sharpe 0.851 gross / 0.536 net (MDD 13.7% net); BvLS + SprN
  0.571 / 0.338.
- Multivariate cointegration signal: worst results -> dropped; all discussion = pairs of two assets.
- Table 6 (key): multi-view methods rank highest. BEST STRATEGY = "MV-CRSp(25)-All-OU": multi-view
  CO-REGULARIZED spectral clustering (Algorithm 4, lambda = 1/2) with K = 25 clusters on the "All"
  features set (Table 2), distance screening to 50 pairs, bivariate simple long-short signal (spread =
  P1 - P2 normalized prices), Ornstein-Uhlenbeck (Bertram) thresholds. Net Sharpe 0.830, MDD 7.0% daily.
  Other top strategies: multi-view on feature subsets, Sharpe 0.70-0.80.
- Table 7 (All features): best DM 0.187 net; best SV 0.495; best MV 0.830. 10th-best MV = 0.611.
- Table 8 highlighted cases: DM-P-BvLS-SprN (Gatev benchmark), DM-All-..., SV-KM(25)-All-SprN,
  SV-Sp(25)-All-SprN, MV-Sp(25)-All-SprN, MV-CRSp(25)-All-SprN, MV-CRSp(25)-All-OU (the "best").
- Table 9: FF3+MOM and FF5+MOM regressions of daily excess returns. MV-CRSp(25)-All-OU alpha 2.4% p.a.
  (t=3.655); MV-CRSp(25)-All-SprN 3.0% (2.276); MV-Sp(25)-All-SprN 2.0% (2.139). DM & SV: no sig alpha.

--- p19-21 (4.3 cont, 4.4, Conclusion, references) ---
- All strategies slightly negative but significant market beta (near market neutral).
- Table 10 per-trade stats: MV-CRSp(25)-All-BvLS-OU win rate 69.2%, mean per-trade daily PnL 38.13bp,
  total PnL per trade 54.32bp. Table 11: number of active trades per day, distinct pairs.
- 4.4 ETFs/futures/forex: same 5bp TC; details in Supplemental Materials. Not the main result.
- Conclusion: simple Gatev pairs no longer works post-2000; multi-view clustering + realized estimators
  does. Universe restricted to S&P 500 constituents due to intraday data availability.
- Refs of interest: Bertram 2010 Physica A 389(11):2234-2243 (OU optimal thresholds); Gatev et al 2006;
  Rad et al 2016; Zelnik-Manor & Perona 2004 (local scaling); Kumar & Daume 2011 (co-regularized
  multi-view spectral); Amaya et al 2015 (realized skewness/kurtosis); Bollerslev et al 2020 (realized
  semicovariances); Barndorff-Nielsen & Shephard 2002; Andersen et al 2001/2003 (realized vol).

--- p22-24 (references only) ---
- Kumar & Daume 2011 ICML "A co-training approach for multi-view spectral clustering" (Alg 3 & 4 source).
- No author code link anywhere in the references / text so far.

--- p25-30 (Figure captions 1-6) ---
- Fig 1: formation 12 months x 21 bdays, trading 6 months x 21 bdays. Two steps in formation: cluster,
  then within each cluster select candidates + construct signal. Each square = 1 month.
- Fig 3: K in {10,25}, NumPairs = 50 for S&P 500.
- Fig 5/6: monthly excess returns obtained by compounding daily excess returns. Sample 2000-2022.

--- p31-33 (Fig 7, Table 1, Table 2) ---
- Fig 7: non-S&P500 classes use K=5 and 10 pairs. TC 5bp per monetary unit per asset at entry and exit.
- Table 1: strategy = (a) features set (Table 4) + (b) clustering + (c) signal + (d) trading rule. SV/MV
  with bivariate signals always apply distance screening.
- Table 2 FEATURE DEFINITIONS (p33):
  P_t raw closing price; normalized price P_t/P_1; R_t daily log return log(P_t/P_{t-1});
  RV_t = sum_{m=1..M} (R_{t_m})^2  [printed as (R_{t_m} - R_{t_{m-1}})^2 - typo; intraday log returns
  squared]; RCov_{t,i,j} = sum_m R_{t_m,i} R_{t_m,j};
  Skew_t = sqrt(M) sum R^3 / (sum R^2)^{3/2};  Kurt_t = M sum R^4 / (sum R^2)^{2} [printed ^{3/2}, Amaya
  et al 2015 definition uses squared];  SVDown_t = sum (R^-)^2; SVUp_t = sum (R^+)^2;
  Beta_{index,t} = RCov_{index,t} / RV_{index,t} (intraday, daily);
  beta_mkt, beta_smb, beta_hml, beta_umd (252-day rolling FF-Carhart regression); ivol (Ang et al 2006 =
  std of residuals); tvol (std of daily returns).
- Table 3 (p34) lists reference indices for realized betas; Table 4 (p34/35) lists feature sets.

--- p34-36 (Table 3, Table 4, Table 5) ---
- Table 3 reference indices for realized betas: SPY, DIA, IWM, QQQ, XLB, XLE, XLF, XLI, XLK, XLP, XLU,
  XLV, XLY (13 ETFs).
- Table 4 feature sets (all include normalized price P). Groups: SK = (Skew, Kurt); SV = (SVDown, SVUp);
  Beta_SPY; Beta_idx = Beta vs {DIA, IWM, QQQ}; Beta_sec = Beta vs {XLB, XLE, XLF, XLI, XLK, XLP, XLU,
  XLV, XLY} (9 sectors); beta_FF = {mkt, smb, hml, umd}; vol = (ivol, tvol).
  Set 14 = "All" = P, RV, SK, SV, Beta_SPY, Beta_idx, Beta_sec, beta_FF, vol.
  => All = 1 (P) + 1 (RV) + 2 (Skew,Kurt) + 2 (SVDown,SVUp) + 1 (Beta_SPY) + 3 (Beta_idx) + 9 (Beta_sec)
     + 4 (beta_FF) + 2 (ivol,tvol) = 25 views/features.
  (Note: Table 4 column headers "P RV | SK SV | Beta_SPY Beta_idx | Beta_sec beta_FF | vol"; set 14 has
  every column marked.)
- Table 5: DM on P results (annualized, 252 bdays). SR = ann mean excess / ann std. MDD on daily excess
  returns. TC Y/N.

--- p37-39 (Tables 6, 7, 8) ---
- Table 6 top-10 (net of TC, all combos): #1 MV-CRSp(25) / All / BvLS / OrnUhl. CONFIRMED headline
  strategy: MV-CRSp(25)-All-BvLS-OU (Table 8 last row). Signal BvLS = simple long-short spread of
  normalized prices; trading rule OrnUhl.
- Table 7: All-features top-10 per clustering family.
- Table 8 highlighted cases (9): DM-P-BvLS-SprN ... MV-CRSp(25)-All-BvLS-OU.

--- p40-42 (Tables 9, 10, 11) ---
- Table 9: factor regressions (Newey-West 6 lags).
- Table 10: trade classification: "convergent" = opens and closes endogenously by the rule; "unconvergent"
  = FORCIBLY CLOSED because still open at END OF THE TRADING PERIOD. => all open trades of a cycle are
  closed at the end of its 6-month trading period. MV-CRSp(25)-All-BvLS-OU: ~221 active trades per day on
  average (out of 300 possible = 6 cycles x 50 pairs), 5191 distinct pairs over 2000-2022.
- Table 11: number of active trades = sum of trades in play on a given day (all overlapping cycles).
END OF PAPER (42 pages). No author code repository mentioned.

--- PLATFORM DOC: Fama-French dataset (client-supplied link) ---
- self.add_data(FamaFrench, "FF", Resolution.DAILY).symbol ; daily, from January 1998, America/New_York.
- QC computes factors itself (Mkt-RF, SMB, HML, RMW, CMA, Momentum, RF) to French definitions; values
  arrive the day AFTER the session they measure (no look-ahead). Every factor nullable (None when missing).
- `value` property = HML; `is_estimate` flag. History: self.history(ff_symbol, n, Resolution.DAILY) or
  self.history[FamaFrench](...). Field names to be introspected in research (attributes list not rendered).
"""