Overall Statistics
Total Orders
102
Average Win
1.87%
Average Loss
-1.52%
Compounding Annual Return
0.588%
Drawdown
9.900%
Expectancy
0.112
Start Equity
10000000
End Equity
10297412.33
Net Profit
2.974%
Sharpe Ratio
-0.896
Sortino Ratio
-0.592
Probabilistic Sharpe Ratio
0.890%
Loss Rate
50%
Win Rate
50%
Profit-Loss Ratio
1.22
Alpha
-0.037
Beta
-0.001
Annual Standard Deviation
0.041
Annual Variance
0.002
Information Ratio
-0.642
Tracking Error
0.148
Treynor Ratio
32.108
Total Fees
$14442.36
Estimated Strategy Capacity
$110000000.00
Lowest Capacity Asset
AXP R735QTJ8XC9X
Portfolio Turnover
2.77%
Drawdown Recovery
1491
# region imports
from AlgorithmImports import *
from stat_arb import PairStatArb
# endregion


class SpectralStatArb(QCAlgorithm):

    def initialize(self):
        self.backtest_years = float(self.get_parameter("backtest-years") or 5)
        self.start_offset_days = int(float(self.get_parameter("start-offset-days") or 0))
        self.set_start_date(self.end_date - timedelta(int(self.backtest_years * 365)) + timedelta(self.start_offset_days))
        self.set_cash(float(self.get_parameter("initial-cash") or 100_000))
        self.settings.automatic_indicator_warm_up = True
        self.settings.seed_initial_prices = True
        self._formation_date_rule = FuncDateRule("semiannual_formation", self._semiannual_formation_dates)
        self.universe_settings.resolution = Resolution.DAILY
        self.universe_settings.schedule.on(self._formation_date_rule)
        self._universe = self.add_universe(lambda fundamental: [
            f.symbol
            for f in sorted(
                [f for f in fundamental if f.has_fundamental_data and not np.isnan(f.market_cap) and f.market_cap > 5e9],
                key=lambda f: f.dollar_volume
            )[-100:]
        ])
        self.laplacian_lookback = 252
        self.entry_z_score = float(self.get_parameter("entry-z-score") or 2.0)
        self.exit_z_score = float(self.get_parameter("exit-z-score") or 0.0)
        self.max_adf_p_value = float(self.get_parameter("max-adf-p-value") or 0.05)
        self.max_half_life = float(self.get_parameter("max-half-life") or 126)
        self._pair_spread = None
        self.set_warm_up(timedelta(45))

    def _semiannual_formation_dates(self, start, end):
        return [
            datetime(year, month, 1)
            for year in range(start.year, end.year + 1)
            for month in (1, 7)
            if start <= datetime(year, month, 1) <= end
        ]

    def on_warmup_finished(self):
        training_time_rule = self.time_rules.at(6, 0)
        trading_time_rule = self.time_rules.at(8, 0)
        self.train(self._formation_date_rule, training_time_rule, self._train)
        self.schedule.on(self.date_rules.every_day("SPY"), trading_time_rule, self._rebalance)
        if self.live_mode:
            self._train()
            self._rebalance()
        else:
            self.train(self.date_rules.today, training_time_rule, self._train)
            self.schedule.on(self.date_rules.today, trading_time_rule, self._rebalance)

    def on_securities_changed(self, changes):
        for security in changes.added_securities:
            security.log_return = self.logr(security, 1)
            security.log_return.window.size = self.laplacian_lookback
            security.log_return.reset()
            history = self.history[TradeBar](security, self.laplacian_lookback + 1, Resolution.DAILY)
            for bar in history:
                security.log_return.update(bar.end_time, bar.close)
        for security in changes.removed_securities:
            if self._pair_spread and security.symbol in self._pair_spread.symbols:
                self._drop_pair("security_removed")
            self.deregister_indicator(security.log_return)

    def _train(self):
        pair_spread = PairStatArb(
            self.laplacian_lookback,
            self.max_adf_p_value,
            self.max_half_life
        ).train(self, self._universe)
        self._drop_pair("formation_retrain")
        if pair_spread is None:
            self.debug(f"{self.time.date()} train no_pair")
            return
        self._pair_spread = pair_spread
        self.register_indicator(pair_spread.symbol_a, self._pair_spread, Resolution.DAILY)
        self.register_indicator(pair_spread.symbol_b, self._pair_spread, Resolution.DAILY)
        self.debug(
            f"{self.time.date()} train selected pair={pair_spread.symbol_a.value}/{pair_spread.symbol_b.value} "
            f"beta={pair_spread.hedge_ratio:.4f} distance={pair_spread.distance:.6f} "
            f"half_life={pair_spread.half_life:.2f} adf_p={pair_spread.adf_p_value:.4f}"
        )

    def _rebalance(self):
        pair = self._pair_spread
        if pair is None or not pair.is_ready:
            return
        asset_a, asset_b = pair.symbols
        security_a = self.securities[asset_a]
        security_b = self.securities[asset_b]
        if security_a.invested != security_b.invested:
            self.debug(f"{self.time.date()} one_leg_guard {asset_a.value}/{asset_b.value} holdings={security_a.holdings.quantity}/{security_b.holdings.quantity}")
            self._liquidate_pair(pair, "one_leg_guard")
            return
        half_life = float(pair.half_life)
        if not np.isfinite(half_life) or half_life > self.max_half_life:
            return
        z_score = float(pair.z_score)
        if not np.isfinite(z_score):
            return
        invested = security_a.invested or security_b.invested
        if not invested and abs(z_score) > self.entry_z_score:
            gross = 1 + abs(pair.hedge_ratio)
            if not np.isfinite(gross) or gross <= 0:
                return
            direction = np.sign(z_score)
            weight_a = -direction / gross
            weight_b = direction * pair.hedge_ratio / gross
            quantity_a = self.calculate_order_quantity(asset_a, weight_a)
            quantity_b = self.calculate_order_quantity(asset_b, weight_b)
            if quantity_a == 0 or quantity_b == 0:
                self.debug(
                    f"{self.time.date()} skip_entry {asset_a.value}/{asset_b.value} "
                    f"z={z_score:.2f} beta={pair.hedge_ratio:.4f} "
                    f"weights={weight_a:.3f}/{weight_b:.3f} quantities={quantity_a}/{quantity_b} "
                    f"prices={security_a.price:.2f}/{security_b.price:.2f}"
                )
                return
            tag = f"entry {asset_a.value}/{asset_b.value}"
            self.debug(
                f"{self.time.date()} {tag} z={z_score:.2f} beta={pair.hedge_ratio:.4f} "
                f"half_life={half_life:.2f} weights={weight_a:.3f}/{weight_b:.3f} "
                f"quantities={quantity_a}/{quantity_b} prices={security_a.price:.2f}/{security_b.price:.2f}"
            )
            self.set_holdings([PortfolioTarget(asset_a, weight_a), PortfolioTarget(asset_b, weight_b)], False, tag=tag)
        elif invested and (
            security_a.holdings.quantity < 0 and z_score <= self.exit_z_score
            or security_a.holdings.quantity > 0 and z_score >= -self.exit_z_score
        ):
            tag = f"exit {asset_a.value}/{asset_b.value}"
            self.debug(f"{self.time.date()} {tag} z={z_score:.2f} holdings={security_a.holdings.quantity}/{security_b.holdings.quantity}")
            self._liquidate_pair(pair, tag)

    def _drop_pair(self, reason: str):
        if self._pair_spread:
            self.debug(f"{self.time.date()} drop_pair reason={reason} pair={self._pair_spread.symbol_a.value}/{self._pair_spread.symbol_b.value}")
            self._liquidate_pair(self._pair_spread, f"drop {reason} {self._pair_spread.symbol_a.value}/{self._pair_spread.symbol_b.value}")
            self.deregister_indicator(self._pair_spread)
        self._pair_spread = None

    def _liquidate_pair(self, pair, tag: str):
        for symbol in pair.symbols:
            self.liquidate(symbol, tag=tag)
from pathlib import Path
import math
import sys

import numpy as np
import pandas as pd


EPSILON = 1e-12


def collect_failures():
    failures = []
    _test_weight_sizing(failures)
    _test_spread_and_z_score(failures)
    _test_half_life(failures)
    _test_laplacian_partition(failures)
    _test_candidate_selection(failures)
    _test_price_view(failures)
    _test_beta_and_residual_volatility_view(failures)
    _test_standardize_zero_variance_guard(failures)
    _test_single_view_correlation_invariant_to_standardization(failures)
    _test_crossing_count(failures)
    _test_source_patterns(failures)
    return failures


def main():
    failures = collect_failures()
    if failures:
        print("math audit failed")
        for failure in failures:
            print(f"- {failure}")
        sys.exit(1)
    print("math audit passed")


def _test_weight_sizing(failures):
    for pair_weight in (1.0, 0.5, 0.05):
        for beta in (0.25, 0.85, 1.0, 4.0):
            for direction in (-1, 1):
                gross = 1 + abs(beta)
                weight_a = pair_weight * -direction / gross
                weight_b = pair_weight * direction * beta / gross
                _assert_close(abs(weight_a) + abs(weight_b), pair_weight, failures, "pair sleeve gross exposure does not sum to pair_weight")
                _assert(direction * weight_a < 0, failures, "spread direction should short A when z-score is positive and long A when negative")
                _assert(direction * weight_b > 0, failures, "spread direction should long B when z-score is positive and short B when negative")


def _test_spread_and_z_score(failures):
    prices_a = np.array([100.0, 102.0, 101.0, 104.0])
    prices_b = np.array([50.0, 51.0, 50.5, 52.0])
    beta = 0.75
    normalized_a = prices_a / prices_a[0]
    normalized_b = prices_b / prices_b[0]
    spread = normalized_a - beta * normalized_b
    spread_mean = float(np.mean(spread))
    spread_sd = float(np.std(spread))
    live_value = float(106.0 / prices_a[0] - beta * 53.0 / prices_b[0])
    z_score = float((live_value - spread_mean) / spread_sd)
    _assert_close(spread[0], 1 - beta, failures, "initial normalized spread is inconsistent with beta")
    _assert(spread_sd > 0, failures, "spread standard deviation should be positive in audit fixture")
    _assert_close(z_score, (live_value - spread_mean) / spread_sd, failures, "z-score should use fixed formation moments")


def _test_half_life(failures):
    mild_slope = -0.05
    steep_slope = -0.20
    mild_half_life = -math.log(2) / mild_slope
    steep_half_life = -math.log(2) / steep_slope
    _assert(mild_half_life > 0 and steep_half_life > 0, failures, "negative mean-reversion slopes should produce positive half-lives")
    _assert(steep_half_life < mild_half_life, failures, "faster mean reversion should produce a shorter half-life")


def _test_laplacian_partition(failures):
    returns = np.array([[0.01, 0.02, -0.01], [0.02, 0.01, -0.02], [-0.01, -0.02, 0.02], [0.00, 0.01, -0.01]])
    corr = np.corrcoef(returns, rowvar=False)
    weights = np.abs(corr)
    np.fill_diagonal(weights, 0)
    laplacian = np.diag(np.sum(weights, axis=1)) - weights
    _, eigenvectors = np.linalg.eigh(laplacian)
    fiedler = eigenvectors[:, 1]
    if fiedler[np.argmax(np.abs(fiedler))] < 0:
        fiedler = -fiedler
    left = [i for i in range(len(fiedler)) if fiedler[i] <= 0]
    right = [i for i in range(len(fiedler)) if fiedler[i] > 0]
    _assert_close(float(np.max(np.abs(np.sum(laplacian, axis=1)))), 0.0, failures, "graph Laplacian rows should sum to zero")
    _assert(sorted(left + right) == list(range(len(fiedler))), failures, "Fiedler partitions should cover every asset exactly once")


def _test_candidate_selection(failures):
    candidates = [{"pair": ("A", "B"), "distance": 1.00, "adf": 0.01}, {"pair": ("A", "C"), "distance": 1.10, "adf": 0.01}, {"pair": ("C", "D"), "distance": 1.40, "adf": 0.02}, {"pair": ("E", "F"), "distance": 1.60, "adf": 0.01}]
    ranked = sorted(candidates, key=lambda candidate: (candidate["distance"], candidate["adf"]))
    selected = _select_candidates_like_strategy(ranked, pairs_per_cycle=3, max_distance_ratio=1.5)
    _assert([candidate["pair"] for candidate in selected] == [("A", "B"), ("C", "D")], failures, "candidate selection should enforce no overlap and distance cap")


def _select_candidates_like_strategy(ranked_candidates, pairs_per_cycle, max_distance_ratio):
    best_pair = ranked_candidates[0] if ranked_candidates else None
    selected_pairs = []
    selected_symbols = []
    if best_pair:
        selected_pairs.append(best_pair)
        selected_symbols.extend(best_pair["pair"])
    for candidate in ranked_candidates:
        if len(selected_pairs) >= pairs_per_cycle:
            break
        if candidate == best_pair or candidate["pair"][0] in selected_symbols or candidate["pair"][1] in selected_symbols:
            continue
        if best_pair and max_distance_ratio > 0 and candidate["distance"] > best_pair["distance"] * max_distance_ratio:
            continue
        selected_pairs.append(candidate)
        selected_symbols.extend(candidate["pair"])
    return selected_pairs


def _test_price_view(failures):
    prices = pd.DataFrame({"A": [50.0, 55.0, 45.0], "B": [10.0, 8.0, 12.0]})
    normalized = prices / prices.iloc[0]
    _assert_close(normalized["A"].iloc[1], 1.1, failures, "normalized price view should scale by each column's own first formation price")
    _assert_close(normalized["B"].iloc[2], 1.2, failures, "normalized price view should scale each column independently")


def _test_beta_and_residual_volatility_view(failures):
    rng = np.random.default_rng(0)
    n, window = 200, 63
    spy = pd.Series(rng.normal(0, 0.01, n))
    stock_returns = pd.DataFrame({"A": 1.5 * spy + rng.normal(0, 0.005, n)})
    spy_returns = spy.reindex(stock_returns.index).dropna()
    beta = stock_returns.loc[spy_returns.index].rolling(window).cov(spy_returns).div(spy_returns.rolling(window).var(), axis=0).dropna()
    residual_vol = stock_returns.loc[beta.index].sub(beta.mul(spy_returns.loc[beta.index], axis=0)).rolling(window).std().dropna()
    manual_beta = np.cov(stock_returns["A"].iloc[-window:], spy.iloc[-window:], ddof=1)[0, 1] / np.var(spy.iloc[-window:], ddof=1)
    _assert_close(beta["A"].iloc[-1], manual_beta, failures, "rolling beta view should equal rolling covariance over rolling variance")
    _assert(residual_vol["A"].mean() < stock_returns["A"].std(), failures, "residual volatility should be smaller than total volatility for a market-driven series")


def _test_standardize_zero_variance_guard(failures):
    constant = np.array([2.0, 2.0, 2.0])
    standardized = (constant - float(np.mean(constant))) / float(np.std(constant)) if float(np.std(constant)) > 0 else constant * 0
    _assert(np.all(standardized == 0), failures, "standardizing a zero-variance view should return zeros instead of dividing by zero")


def _test_single_view_correlation_invariant_to_standardization(failures):
    rng = np.random.default_rng(1)
    raw = pd.DataFrame(rng.normal(0, 1, (50, 4)), columns=["A", "B", "C", "D"])
    plain_corr = np.abs(raw.corr().fillna(0).values)
    standardized = (raw - float(np.mean(raw.values))) / float(np.std(raw.values))
    standardized_corr = np.abs(standardized.corr().fillna(0).values)
    _assert(np.allclose(plain_corr, standardized_corr), failures, "a single standardized view should match the raw view's correlation so default feature-mode=returns is unaffected")


def _test_crossing_count(failures):
    entry_z_score = 2.0
    z_scores = [0.5, 2.5, 2.6, 0.1, -2.2, -0.3, 2.1]
    crossing_count = 0
    above_entry = False
    for z_score in z_scores:
        is_above = abs(z_score) > entry_z_score
        crossing_count += 1 if is_above and not above_entry else 0
        above_entry = is_above
    _assert(crossing_count == 3, failures, "crossing count should only increment on rising edges above the entry threshold")


def _test_source_patterns(failures):
    stat_arb = _read("stat_arb.py")
    pair_spread = _read("pair_spread.py")
    main = _read("main.py")
    _require("normalized = aligned / aligned.iloc[0]", stat_arb, failures, "formation prices should be normalized by first close")
    _require("mse = float(np.mean(np.square(normalized.iloc[:, 0] - normalized.iloc[:, 1])))", stat_arb, failures, "distance should be normalized-price MSE")
    _require("hedge_ratio = float(np.polyfit(independent, dependent, 1)[0])", stat_arb, failures, "hedge ratio should be OLS slope")
    _require("spread = dependent - hedge_ratio * independent", stat_arb, failures, "formation spread should match documented spread")
    _require("adfuller(spread)", stat_arb, failures, "spread stationarity should be tested with ADF")
    _require("half_life = float(-np.log(2) / slope)", stat_arb, failures, "half-life formula should match audit")
    _require('(pair.adf_p_value, pair.half_life, pair.distance) if self._ranking_mode == "stationarity" else (pair.distance, pair.adf_p_value)', stat_arb, failures, "ranking should prioritize distance then ADF by default, or ADF then half-life in stationarity mode")
    _require("candidate.symbol_a in selected_symbols or candidate.symbol_b in selected_symbols", stat_arb, failures, "selection should prevent overlapping symbols")
    _require("candidate.distance > best_pair.distance * self._max_distance_ratio", stat_arb, failures, "selection should enforce distance ratio cap")
    _require("price_a / self._price_scale_by_symbol[self.symbol_a] - self.hedge_ratio * price_b / self._price_scale_by_symbol[self.symbol_b]", pair_spread, failures, "live spread should use formation normalizers")
    _require("(self.value - self._spread_mean) / self.spread_sd", pair_spread, failures, "live z-score should use formation moments")
    _require("time_a != time_b or time_a == self._last_time", pair_spread, failures, "indicator should update only on matched timestamps")
    _require("above_entry = abs(self.z_score) > self._entry_z_score", pair_spread, failures, "crossing count should track the entry z-score threshold")
    _require("self.crossing_count += 1 if above_entry and not self._above_entry else 0", pair_spread, failures, "crossing count should only increment on rising edges")
    _require("weight_a = pair_weight * -direction / gross", main, failures, "entry weight A should short positive z-score spread")
    _require("weight_b = pair_weight * direction * pair.hedge_ratio / gross", main, failures, "entry weight B should hedge spread with beta")
    _require("quantity_a == 0 or quantity_b == 0", main, failures, "entry should skip one-legged quantity rounding")
    _require("return prices / prices.iloc[0]", stat_arb, failures, "price view should normalize by each security's first formation price")
    _require("returns.loc[spy_returns.index].rolling(window).cov(spy_returns).div(spy_returns.rolling(window).var(), axis=0).dropna()", stat_arb, failures, "beta view should be rolling covariance over rolling variance versus SPY")
    _require("returns.sub(beta.mul(spy_returns, axis=0)).rolling(window).std().dropna()", stat_arb, failures, "residual volatility view should be rolling std of market-model residuals")
    _require("(values - float(np.mean(array))) / std if std > 0 else values * 0", stat_arb, failures, "standardization should guard against zero-variance views")
    _require('self._spy = self.add_equity("SPY", Resolution.DAILY)', main, failures, "SPY should be tracked explicitly as the market-beta reference")
    _require('self.feature_mode = self.get_parameter("feature-mode") or "returns"', main, failures, "default feature-mode should be returns to preserve current behavior")
    _require('self.cluster_mode = self.get_parameter("cluster-mode") or "single-view"', main, failures, "default cluster-mode should be single-view to preserve current behavior")


def _read(path):
    return Path(path).read_text(encoding="utf-8")


def _require(pattern, text, failures, message):
    if pattern not in text:
        failures.append(message)


def _assert(condition, failures, message):
    if not condition:
        failures.append(message)


def _assert_close(actual, expected, failures, message):
    if abs(actual - expected) > EPSILON:
        failures.append(f"{message}: expected {expected} found {actual}")


if __name__ == "__main__":
    main()
from argparse import ArgumentParser
from collections import defaultdict
from csv import DictReader
from dataclasses import dataclass
import decimal


@dataclass
class ParsedOrder:
    time: str
    symbol: str
    quantity: decimal.Decimal
    value: decimal.Decimal
    tag: str


def main():
    parser = ArgumentParser(description="Compute pair-level P&L from a QuantConnect orders CSV.")
    parser.add_argument("orders_csv")
    parser.add_argument("--logs", nargs="*", default=[])
    args = parser.parse_args()
    orders = _read_orders(args.orders_csv)
    groups, grouping_issues = _group_orders(orders)
    closed_trades, open_trades, lifecycle_issues = _match_trades(groups)
    selected_pairs, duplicate_share_class_pairs_from_logs, duplicate_share_class_pair_count, skipped_entries, skip_reason_count = _read_logs(args.logs)
    issues = grouping_issues + lifecycle_issues
    pnl_by_pair = defaultdict(decimal.Decimal)
    count_by_pair = defaultdict(int)
    for trade in closed_trades:
        pnl_by_pair[trade["pair"]] += trade["pnl"]
        count_by_pair[trade["pair"]] += 1
    one_leg_groups = [issue for issue in issues if "expected 2 orders" in issue or "symbols" in issue]
    duplicate_share_class_pairs = [pair for pair in pnl_by_pair if _is_duplicate_share_class_pair(pair)]
    print(f"orders={len(orders)} groups={len(groups)} closed_trades={len(closed_trades)} open_trades={len(open_trades)} issues={len(issues)} one_leg_groups={len(one_leg_groups)}")
    if pnl_by_pair:
        total_pnl = sum(pnl_by_pair.values(), decimal.Decimal("0"))
        top_pair = max(pnl_by_pair, key=lambda pair: abs(pnl_by_pair[pair]))
        print(f"total_closed_pnl={total_pnl:.2f} top_abs_pair={top_pair} top_abs_pnl={pnl_by_pair[top_pair]:.2f} other_pnl={total_pnl - pnl_by_pair[top_pair]:.2f} duplicate_share_class_pairs={len(duplicate_share_class_pairs)}")
    if args.logs:
        print(f"logs_selected_pairs={len(selected_pairs)} logs_duplicate_share_class_pairs={len(duplicate_share_class_pairs_from_logs)} logs_skipped_entries={len(skipped_entries)}")
        if skip_reason_count:
            print("skip_entry_reasons")
            for reason, count in sorted(skip_reason_count.items()):
                print(f"{reason},count={count}")
        if duplicate_share_class_pair_count:
            print("selected_duplicate_share_class_pairs")
            for pair, count in sorted(duplicate_share_class_pair_count.items()):
                print(f"{pair},count={count}")
    print("closed_pnl_by_pair")
    for pair in sorted(pnl_by_pair, key=lambda key: pnl_by_pair[key]):
        print(f"{pair},trades={count_by_pair[pair]},pnl={pnl_by_pair[pair]:.2f}")
    if open_trades:
        print("open_trades")
        for pair, trade in sorted(open_trades.items()):
            quantities = ",".join([symbol + ":" + str(quantity) for symbol, quantity in sorted(trade["quantities"].items())])
            print(f"{pair},opened={trade['start_time']},quantities={quantities},cash_flow={-trade['value_sum']:.2f}")
    if issues:
        print("issues")
        for issue in issues:
            print(issue)


def _read_orders(path):
    orders = []
    with open(path, newline="", encoding="utf-8-sig") as csv_file:
        for row in DictReader(csv_file):
            if row["Status"].strip().lower() != "filled":
                continue
            orders.append(ParsedOrder(row["Time"].strip(), row["Symbol"].strip(), decimal.Decimal(row["Quantity"].strip()), decimal.Decimal(row["Value"].strip()), row["Tag"].strip()))
    return orders


def _read_logs(paths):
    selected_pairs = []
    duplicate_share_class_pairs = []
    duplicate_share_class_pair_count = defaultdict(int)
    skipped_entries = []
    skip_reason_count = defaultdict(int)
    for path in paths:
        with open(path, encoding="utf-8-sig") as log_file:
            for line in log_file:
                selected_pair = _selected_pair_from_log(line)
                if selected_pair:
                    selected_pairs.append(selected_pair)
                    if _is_duplicate_share_class_pair(selected_pair):
                        duplicate_share_class_pairs.append(selected_pair)
                        duplicate_share_class_pair_count[selected_pair] += 1
                skipped_entry = _skipped_entry_from_log(line)
                if skipped_entry:
                    skipped_entries.append(skipped_entry)
                    skip_reason_count[skipped_entry["reason"]] += 1
    return selected_pairs, duplicate_share_class_pairs, duplicate_share_class_pair_count, skipped_entries, skip_reason_count


def _selected_pair_from_log(line):
    if "train selected" not in line:
        return ""
    pair = _token_value(line, "pair=")
    return _normalize_pair(pair)


def _skipped_entry_from_log(line):
    if "skip_entry" not in line:
        return {}
    pair = _pair_after_token(line, "skip_entry")
    quantities = _token_value(line, "quantities=")
    reason = "zero_quantity" if "/0" in quantities or quantities.startswith("0/") or quantities == "0" else "unknown"
    return {"pair": _normalize_pair(pair), "quantities": quantities, "reason": reason}


def _group_orders(orders):
    groups_by_key = {}
    untagged_by_time = defaultdict(list)
    issues = []
    for order in orders:
        pair = _pair_from_tag(order.tag)
        event = _event_from_tag(order.tag)
        if pair:
            groups_by_key.setdefault((order.time, pair, event), []).append(order)
        else:
            untagged_by_time[order.time].append(order)
    for time, rows in untagged_by_time.items():
        symbols = _unique_symbols(rows)
        event = _event_from_rows(rows)
        if len(symbols) == 2:
            groups_by_key.setdefault((time, "/".join(symbols), event), []).extend(rows)
        else:
            pair = "UNRESOLVED"
            groups_by_key.setdefault((time, pair, event), []).extend(rows)
            issues.append(f"{time} {pair} symbols={','.join(symbols)} expected 2 symbols")
    groups = []
    for (time, pair, event), rows in groups_by_key.items():
        symbols = _unique_symbols(rows)
        pair_symbols = pair.split("/") if pair != "UNRESOLVED" else symbols
        if len(rows) != 2:
            issues.append(f"{time} {pair} {event} expected 2 orders found {len(rows)}")
        if sorted(pair_symbols) != symbols:
            issues.append(f"{time} {pair} {event} symbols={','.join(symbols)} do not match pair")
        groups.append({"time": time, "pair": pair, "event": event, "orders": rows})
    groups.sort(key=lambda group: (group["time"], _event_rank(group["event"]), group["pair"]))
    return groups, issues


def _pair_from_tag(tag):
    for token in reversed(tag.split()):
        if "/" in token:
            symbols = [symbol.strip() for symbol in token.split("/") if symbol.strip()]
            if len(symbols) == 2:
                return "/".join(sorted(symbols))
    return ""


def _pair_after_token(line, marker):
    tokens = line.split()
    for i, token in enumerate(tokens):
        if token == marker and i + 1 < len(tokens):
            return tokens[i + 1]
    return ""


def _token_value(line, prefix):
    for token in line.replace(",", " ").split():
        if token.startswith(prefix):
            return token[len(prefix):]
    return ""


def _normalize_pair(pair):
    if "/" not in pair:
        return ""
    symbols = [symbol.strip() for symbol in pair.split("/") if symbol.strip()]
    if len(symbols) != 2:
        return ""
    return "/".join(sorted(symbols))


def _event_from_tag(tag):
    clean_tag = tag.strip().lower()
    if clean_tag.startswith("entry "):
        return "entry"
    if clean_tag.startswith("exit ") or clean_tag == "liquidated":
        return "exit"
    if clean_tag.startswith("drop "):
        return "drop"
    return "unknown"


def _event_from_rows(rows):
    events = []
    for order in rows:
        event = _event_from_tag(order.tag)
        if event not in events:
            events.append(event)
    return events[0] if len(events) == 1 else "unknown"


def _event_rank(event):
    rank_by_event = {"drop": 0, "exit": 0, "unknown": 1, "entry": 2}
    return rank_by_event[event] if event in rank_by_event else 1


def _unique_symbols(rows):
    symbols = []
    for order in rows:
        if order.symbol not in symbols:
            symbols.append(order.symbol)
    return sorted(symbols)


def _is_duplicate_share_class_pair(pair):
    symbols = pair.split("/")
    if len(symbols) != 2:
        return False
    root_a = symbols[0].split(".")[0]
    root_b = symbols[1].split(".")[0]
    return root_a == root_b and symbols[0] != symbols[1]


def _match_trades(groups):
    open_trades = {}
    closed_trades = []
    issues = []
    for group in groups:
        pair = group["pair"]
        if pair == "UNRESOLVED":
            continue
        quantities = defaultdict(decimal.Decimal)
        value_sum = decimal.Decimal("0")
        for order in group["orders"]:
            quantities[order.symbol] += order.quantity
            value_sum += order.value
        if group["event"] == "entry":
            if pair in open_trades:
                issues.append(f"{group['time']} {pair} entry while trade already open")
                continue
            open_trades[pair] = {"pair": pair, "start_time": group["time"], "quantities": dict(quantities), "value_sum": value_sum}
            continue
        if group["event"] in ("drop", "exit"):
            if pair not in open_trades:
                issues.append(f"{group['time']} {pair} {group['event']} without open trade")
                continue
            trade = open_trades[pair]
            for symbol, quantity in quantities.items():
                trade["quantities"][symbol] = trade["quantities"].get(symbol, decimal.Decimal("0")) + quantity
            trade["value_sum"] += value_sum
            if all(quantity == 0 for quantity in trade["quantities"].values()):
                trade["end_time"] = group["time"]
                trade["pnl"] = -trade["value_sum"]
                closed_trades.append(trade)
                del open_trades[pair]
            else:
                issues.append(f"{group['time']} {pair} {group['event']} left nonzero quantities={trade['quantities']}")
            continue
        if pair not in open_trades:
            open_trades[pair] = {"pair": pair, "start_time": group["time"], "quantities": dict(quantities), "value_sum": value_sum}
            continue
        trade = open_trades[pair]
        for symbol, quantity in quantities.items():
            trade["quantities"][symbol] = trade["quantities"].get(symbol, decimal.Decimal("0")) + quantity
        trade["value_sum"] += value_sum
        if all(quantity == 0 for quantity in trade["quantities"].values()):
            trade["end_time"] = group["time"]
            trade["pnl"] = -trade["value_sum"]
            closed_trades.append(trade)
            del open_trades[pair]
        else:
            issues.append(f"{group['time']} {pair} adjusted existing trade quantities={trade['quantities']}")
    return closed_trades, open_trades, issues


if __name__ == "__main__":
    main()
from AlgorithmImports import *


class PairSpread(PythonIndicator):

    def __init__(self, symbol_a, symbol_b, hedge_ratio: float, price_scale_a: float, price_scale_b: float, spread_mean: float, spread_sd: float, half_life: float, distance: float, adf_p_value: float, name: str = "PairSpread"):
        super().__init__()
        self.name = name
        self.symbol_a = symbol_a
        self.symbol_b = symbol_b
        self.symbols = (symbol_a, symbol_b)
        self.hedge_ratio = float(hedge_ratio)
        self.distance = float(distance)
        self.adf_p_value = float(adf_p_value)
        self.value = 0.0
        self.spread = 0.0
        self.z_score = 0.0
        self.half_life = float(half_life)
        self.spread_sd = float(spread_sd)
        self.time = datetime.min
        self.current = IndicatorDataPoint(self.time, self.value)
        self._price_scale_by_symbol = {symbol_a: float(price_scale_a), symbol_b: float(price_scale_b)}
        self._spread_mean = float(spread_mean)
        self._prices = {}
        self._last_time = None
        self._is_ready = False

    def update(self, input_: BaseData):
        price = float(input_.value)
        if input_.symbol not in self.symbols or price <= 0:
            return self.is_ready
        self._prices[input_.symbol] = (input_.end_time, price)
        if self.symbol_a not in self._prices or self.symbol_b not in self._prices:
            return self.is_ready
        time_a, price_a = self._prices[self.symbol_a]
        time_b, price_b = self._prices[self.symbol_b]
        if time_a != time_b or time_a == self._last_time:
            return self.is_ready
        self._last_time = time_a
        self.time = time_a
        self.value = float(price_a / self._price_scale_by_symbol[self.symbol_a] - self.hedge_ratio * price_b / self._price_scale_by_symbol[self.symbol_b])
        self.spread = self.value
        self.z_score = float((self.value - self._spread_mean) / self.spread_sd)
        self.current = IndicatorDataPoint(self.time, self.value)
        self._is_ready = True
        return self.is_ready

    @property
    def is_ready(self) -> bool:
        return self._is_ready
from AlgorithmImports import *
from pair_spread import PairSpread
from statsmodels.tsa.stattools import adfuller


class PairStatArb:

    def __init__(self, laplacian_lookback: int, max_adf_p_value: float = 0.05, max_half_life: float = 126):
        self._laplacian_lookback = laplacian_lookback
        self._max_adf_p_value = max_adf_p_value
        self._max_half_life = max_half_life

    def train(self, algorithm, universe):
        ready_securities = sorted(
            [algorithm.securities[s] for s in universe.selected if algorithm.securities[s].log_return.window.is_ready],
            key=lambda security: security.symbol.value
        )
        if len(ready_securities) < 2:
            algorithm.debug(f"{algorithm.time.date()} train insufficient_ready ready={len(ready_securities)}")
            return None
        df_returns = pd.DataFrame({
            security.symbol.value: [float(point.value) for point in security.log_return.window]
            for security in ready_securities
        })
        weights = np.abs(df_returns.corr().fillna(0).values)
        np.fill_diagonal(weights, 0)
        laplacian = np.diag(np.sum(weights, axis=1)) - weights
        _, eigenvectors = np.linalg.eigh(laplacian)
        fiedler_vector = eigenvectors[:, 1]
        if fiedler_vector[np.argmax(np.abs(fiedler_vector))] < 0:
            fiedler_vector = -fiedler_vector
        clusters = [
            [ready_securities[i].symbol for i in range(len(ready_securities)) if fiedler_vector[i] <= 0],
            [ready_securities[i].symbol for i in range(len(ready_securities)) if fiedler_vector[i] > 0]
        ]
        history = algorithm.history([security.symbol for security in ready_securities], self._laplacian_lookback, Resolution.DAILY)
        if history.empty:
            algorithm.debug(f"{algorithm.time.date()} train empty_history ready={len(ready_securities)}")
            return None
        prices = history["close"].unstack(level=0)
        best_pair = None
        best_mse = np.inf
        for cluster in clusters:
            for i in range(len(cluster)):
                for j in range(i + 1, len(cluster)):
                    pair_spread = self._try_create_pair(algorithm, prices, cluster[i], cluster[j])
                    if pair_spread and pair_spread.distance < best_mse:
                        best_pair = pair_spread
                        best_mse = pair_spread.distance
        algorithm.debug(
            f"{algorithm.time.date()} train candidates ready={len(ready_securities)} "
            f"clusters={len(clusters[0])}/{len(clusters[1])} "
            f"best={best_pair.symbol_a.value + '/' + best_pair.symbol_b.value if best_pair else 'None'}"
        )
        return best_pair

    def _try_create_pair(self, algorithm, prices, asset_a, asset_b):
        if asset_a not in prices or asset_b not in prices:
            return None
        aligned = pd.concat([prices[asset_a], prices[asset_b]], axis=1).dropna()
        if len(aligned) < self._laplacian_lookback or (aligned <= 0).values.any():
            return None
        normalized = aligned / aligned.iloc[0]
        best_pair = None
        best_adf = np.inf
        mse = float(np.mean(np.square(normalized.iloc[:, 0] - normalized.iloc[:, 1])))
        for dependent_index, independent_index in ((0, 1), (1, 0)):
            dependent = normalized.iloc[:, dependent_index]
            independent = normalized.iloc[:, independent_index]
            if np.all(independent == independent.iloc[0]):
                continue
            hedge_ratio = float(np.polyfit(independent, dependent, 1)[0])
            if not np.isfinite(hedge_ratio) or hedge_ratio < 0.25 or hedge_ratio > 4:
                continue
            if not self._has_executable_sleeve(algorithm, aligned, dependent_index, independent_index, hedge_ratio):
                continue
            spread = dependent - hedge_ratio * independent
            if np.all(spread == spread.iloc[0]):
                continue
            spread_values = np.array(spread, dtype=float)
            if not np.isfinite(spread_values).all():
                continue
            spread_sd = float(np.std(spread_values))
            if spread_sd <= 0:
                continue
            lagged = spread_values[:-1]
            if len(lagged) <= 1 or np.all(lagged == lagged[0]):
                continue
            slope = float(np.polyfit(lagged, spread_values[1:] - lagged, 1)[0])
            if not np.isfinite(slope) or slope >= 0:
                continue
            half_life = float(-np.log(2) / slope)
            if not np.isfinite(half_life) or half_life > self._max_half_life:
                continue
            adf_result = adfuller(spread)
            adf = float(adf_result[0])
            adf_p_value = float(adf_result[1])
            if adf_p_value >= self._max_adf_p_value or adf >= best_adf:
                continue
            if dependent_index == 0:
                best_pair = PairSpread(asset_a, asset_b, hedge_ratio, float(aligned.iloc[0, 0]), float(aligned.iloc[0, 1]), float(np.mean(spread_values)), spread_sd, half_life, mse, adf_p_value)
            else:
                best_pair = PairSpread(asset_b, asset_a, hedge_ratio, float(aligned.iloc[0, 1]), float(aligned.iloc[0, 0]), float(np.mean(spread_values)), spread_sd, half_life, mse, adf_p_value)
            best_adf = adf
        if best_pair is None:
            return None
        pair_history = algorithm.history[TradeBar]([best_pair.symbol_a, best_pair.symbol_b], 2, Resolution.DAILY)
        for bars in pair_history:
            if best_pair.symbol_a in bars and best_pair.symbol_b in bars:
                best_pair.update(bars[best_pair.symbol_a])
                best_pair.update(bars[best_pair.symbol_b])
        return best_pair

    def _has_executable_sleeve(self, algorithm, aligned, dependent_index: int, independent_index: int, hedge_ratio: float) -> bool:
        price_a = float(aligned.iloc[-1, dependent_index])
        price_b = float(aligned.iloc[-1, independent_index])
        gross = 1 + abs(hedge_ratio)
        if not np.isfinite(gross) or gross <= 0 or price_a <= 0 or price_b <= 0:
            return False
        portfolio_value = float(algorithm.portfolio.total_portfolio_value)
        return portfolio_value / gross >= price_a and portfolio_value * abs(hedge_ratio) / gross >= price_b
from csv import DictReader
from pathlib import Path
import sys
import pair_attribution as pa
import math_audit


EXPORT_DIR = Path("validation_exports")
SOURCE_AUDIT = Path("source_audit.md")
COMPLETION_AUDIT = Path("completion_audit.md")
REQUIRED_PERFORMANCE_COLUMNS = ["net_profit", "cagr", "sharpe", "drawdown", "fees", "turnover", "orders"]
REQUIRED_PAIR_COUNTS = {"100000": ["1", "5", "10", "20"], "10000000": ["1", "5", "10", "20"]}
CLEANUP_BACKTEST = "cloud_logcleanup_100k_pairs2_ratio15"
FORBIDDEN_CLEANUP_LOG_PATTERNS = ["train candidate", "train candidates", "train start"]
REQUIRED_SOURCE_AUDIT_TERMS = ["Leung", "Gatev", "Vidyamurthy", "Avellaneda", "Rotondi", "QuantConnect", "math_audit.py", "validate_plan.py"]
REQUIRED_COMPLETION_AUDIT_TERMS = ["Success Criteria", "Implementation Plan", "Mathematical Audit", "QuantConnect And LEAN Audit", "Validation Matrix", "Guardrails", "Residual Scope"]


def main():
    failures = []
    performance_rows = _read_csv(EXPORT_DIR / "performance_summary.csv", failures)
    attribution_rows = _read_csv(EXPORT_DIR / "attribution_summary.csv", failures)
    pair_rows = _read_csv(EXPORT_DIR / "pair_pnl_summary.csv", failures)
    open_rows = _read_csv(EXPORT_DIR / "open_trades_summary.csv", failures)
    if not performance_rows or not attribution_rows or not pair_rows or not open_rows:
        _finish(failures)
    performance_by_name = {row["backtest"]: row for row in performance_rows}
    attribution_by_name = {row["backtest"]: row for row in attribution_rows}
    backtests = [name for name in performance_by_name if name in attribution_by_name]
    _check_required_pair_counts(performance_rows, failures)
    _check_shifted_start(performance_rows, failures)
    _check_performance_rows(performance_rows, failures)
    _check_attribution_rows(attribution_rows, failures)
    _check_pair_rows(pair_rows, backtests, failures)
    _check_open_rows(open_rows, backtests, failures)
    _check_exports_exist(backtests, failures)
    _check_order_exports(backtests, failures)
    _check_cleanup_run(failures)
    _check_math_audit(failures)
    _check_source_audit(failures)
    _check_completion_audit(failures)
    _finish(failures)


def _read_csv(path, failures):
    if not path.exists():
        failures.append(f"missing {path}")
        return []
    with path.open(newline="", encoding="utf-8") as csv_file:
        return list(DictReader(csv_file))


def _check_required_pair_counts(rows, failures):
    seen = {}
    for row in rows:
        if row["start_offset_days"] != "0":
            continue
        seen.setdefault(row["initial_cash"], []).append(row["pairs_per_cycle"])
    for initial_cash, required_counts in REQUIRED_PAIR_COUNTS.items():
        missing = [count for count in required_counts if count not in seen.get(initial_cash, [])]
        if missing:
            failures.append(f"missing pair-count rows for initial_cash={initial_cash}: {sorted(missing)}")


def _check_shifted_start(rows, failures):
    if not any(row["start_offset_days"] != "0" for row in rows):
        failures.append("missing shifted-start validation row")


def _check_performance_rows(rows, failures):
    for row in rows:
        for column in REQUIRED_PERFORMANCE_COLUMNS:
            if not row.get(column):
                failures.append(f"{row['backtest']} missing performance column {column}")


def _check_attribution_rows(rows, failures):
    for row in rows:
        name = row["backtest"]
        for column in ["issues", "one_leg_groups", "logs_skipped_entries"]:
            if int(row[column]) != 0:
                failures.append(f"{name} expected {column}=0 found {row[column]}")
        for column in ["total_closed_pnl", "duplicate_excluded_closed_pnl", "top_abs_pair", "other_pnl", "open_pairs"]:
            if not row.get(column):
                failures.append(f"{name} missing attribution column {column}")
        if int(row["closed_trades"]) <= 0:
            failures.append(f"{name} has no closed trades")
        if int(row["open_trades"]) <= 0:
            failures.append(f"{name} has no open-trade exposure row")


def _check_pair_rows(rows, backtests, failures):
    rows_by_backtest = {}
    for row in rows:
        rows_by_backtest.setdefault(row["backtest"], []).append(row)
    for name in backtests:
        if not rows_by_backtest.get(name):
            failures.append(f"{name} missing pair-level P&L rows")
    if not any(row.get("duplicate_share_class_pair") == "True" for row in rows):
        failures.append("missing duplicate share-class pair attribution rows")


def _check_open_rows(rows, backtests, failures):
    rows_by_backtest = {}
    for row in rows:
        rows_by_backtest.setdefault(row["backtest"], []).append(row)
    for name in backtests:
        if not rows_by_backtest.get(name):
            failures.append(f"{name} missing open exposure rows")


def _check_exports_exist(backtests, failures):
    for name in backtests:
        for suffix in ("orders.csv", "logs.txt"):
            path = EXPORT_DIR / f"{name}_{suffix}"
            if not path.exists():
                failures.append(f"{name} missing raw export {path.name}")


def _check_order_exports(backtests, failures):
    for name in backtests:
        orders_path = EXPORT_DIR / f"{name}_orders.csv"
        logs_path = EXPORT_DIR / f"{name}_logs.txt"
        orders = _read_csv(orders_path, failures)
        if orders:
            _check_order_rows(name, orders, failures)
        _check_raw_attribution(name, orders_path, logs_path, failures)


def _check_order_rows(name, rows, failures):
    for row in rows:
        if row.get("Status") != "Filled":
            failures.append(f"{name} order is not filled at {row.get('Time')} {row.get('Symbol')}")
        if row.get("Type") != "Market On Open":
            failures.append(f"{name} expected Market On Open order at {row.get('Time')} {row.get('Symbol')} found {row.get('Type')}")
        if not row.get("Tag") or "/" not in row.get("Tag", ""):
            failures.append(f"{name} order missing pair tag at {row.get('Time')} {row.get('Symbol')}")


def _check_raw_attribution(name, orders_path, logs_path, failures):
    if not orders_path.exists() or not logs_path.exists():
        return
    orders = pa._read_orders(orders_path)
    groups, grouping_issues = pa._group_orders(orders)
    closed_trades, open_trades, lifecycle_issues = pa._match_trades(groups)
    _, _, _, skipped_entries, _ = pa._read_logs([logs_path])
    issues = grouping_issues + lifecycle_issues
    one_leg_groups = [issue for issue in issues if "expected 2 orders" in issue or "symbols" in issue]
    if issues:
        failures.append(f"{name} raw attribution expected 0 issues found {len(issues)}")
    if one_leg_groups:
        failures.append(f"{name} raw attribution expected 0 one-leg groups found {len(one_leg_groups)}")
    if skipped_entries:
        failures.append(f"{name} raw attribution expected 0 skipped entries found {len(skipped_entries)}")
    if not closed_trades or not open_trades:
        failures.append(f"{name} raw attribution missing closed or open trades")


def _check_cleanup_run(failures):
    orders_path = EXPORT_DIR / f"{CLEANUP_BACKTEST}_orders.csv"
    logs_path = EXPORT_DIR / f"{CLEANUP_BACKTEST}_logs.txt"
    if not orders_path.exists() or not logs_path.exists():
        failures.append(f"{CLEANUP_BACKTEST} missing cleanup proof exports")
        return
    log_text = logs_path.read_text(encoding="utf-8-sig")
    for pattern in FORBIDDEN_CLEANUP_LOG_PATTERNS:
        if pattern in log_text:
            failures.append(f"{CLEANUP_BACKTEST} cleanup log still contains '{pattern}'")
    if "train selected" not in log_text:
        failures.append(f"{CLEANUP_BACKTEST} cleanup log missing train selected attribution lines")
    orders = pa._read_orders(orders_path)
    raw_orders = _read_csv(orders_path, failures)
    _check_order_rows(CLEANUP_BACKTEST, raw_orders, failures)
    if len(orders) != 94:
        failures.append(f"{CLEANUP_BACKTEST} expected 94 filled orders found {len(orders)}")
    _check_raw_attribution(CLEANUP_BACKTEST, orders_path, logs_path, failures)


def _check_math_audit(failures):
    for failure in math_audit.collect_failures():
        failures.append(f"math audit: {failure}")


def _check_source_audit(failures):
    if not SOURCE_AUDIT.exists():
        failures.append(f"missing {SOURCE_AUDIT}")
        return
    text = SOURCE_AUDIT.read_text(encoding="utf-8")
    for term in REQUIRED_SOURCE_AUDIT_TERMS:
        if term not in text:
            failures.append(f"{SOURCE_AUDIT} missing source audit term {term}")


def _check_completion_audit(failures):
    if not COMPLETION_AUDIT.exists():
        failures.append(f"missing {COMPLETION_AUDIT}")
        return
    text = COMPLETION_AUDIT.read_text(encoding="utf-8")
    for term in REQUIRED_COMPLETION_AUDIT_TERMS:
        if term not in text:
            failures.append(f"{COMPLETION_AUDIT} missing completion audit term {term}")
    if "Incomplete" in text or "Pending" in text:
        failures.append(f"{COMPLETION_AUDIT} contains incomplete or pending status")


def _finish(failures):
    if failures:
        print("validation failed")
        for failure in failures:
            print(f"- {failure}")
        sys.exit(1)
    print("validation passed")


if __name__ == "__main__":
    main()
from csv import DictReader
from pathlib import Path
import sys
import pair_attribution as pa


EXPORT_DIR = Path("sharpe1_exports")
PAPER = Path("paper.md")
SOURCE_AUDIT = Path("source_audit.md")
REQUIRED_PERFORMANCE_COLUMNS = ["backtest", "backtest_id", "initial_cash", "pairs_per_cycle", "max_distance_ratio", "start_offset_days", "feature_mode", "cluster_mode", "entry_z_score", "exit_z_score", "net_profit", "sharpe", "drawdown", "fees", "orders"]
REQUIRED_NAME_SUBSTRINGS = ["scaffold_baseline", "price_vol", "price_beta_vol", "multiview", "entry175_exit025", "entry150_exit050", "shift90", "shift180", "shift365", "10m"]
REQUIRED_PAPER_TERMS = ["Literature Alignment And Sharpe 1 Search", "feature-mode", "cluster-mode", "sharpe1_literature_alignment_plan.md"]
REQUIRED_SOURCE_AUDIT_TERMS = ["sharpe1_literature_alignment_plan.md", "multi-view-average", "feature-mode", "validate_sharpe1_plan.py"]
SHARPE_TARGET = 1.0
ROBUSTNESS_SHARPE_TARGET = 0.5
CANDIDATE_FEATURE_MODE = "price-vol"
CANDIDATE_PAIRS_PER_CYCLE_BY_CASH = {"100000": "2", "10000000": "10"}


def main():
    failures = []
    rows = _read_csv(EXPORT_DIR / "performance_summary.csv", failures)
    if not rows:
        _finish(failures)
    _check_required_columns(rows, failures)
    _check_required_categories(rows, failures)
    _check_attribution_exports(failures)
    _check_plan_reports_evidence(failures)
    candidate_rows = [row for row in rows if row["feature_mode"] == CANDIDATE_FEATURE_MODE and row["cluster_mode"] == "single-view" and row["entry_z_score"] == "2.0" and row["exit_z_score"] == "0.0" and row["pairs_per_cycle"] == CANDIDATE_PAIRS_PER_CYCLE_BY_CASH.get(row["initial_cash"])]
    _report_acceptance_decision(candidate_rows, failures)
    _finish(failures)


def _read_csv(path, failures):
    if not path.exists():
        failures.append(f"missing {path}")
        return []
    with path.open(newline="", encoding="utf-8") as csv_file:
        return list(DictReader(csv_file))


def _check_required_columns(rows, failures):
    for row in rows:
        for column in REQUIRED_PERFORMANCE_COLUMNS:
            if not row.get(column):
                failures.append(f"{row.get('backtest', '?')} missing performance column {column}")


def _check_required_categories(rows, failures):
    names = [row["backtest"] for row in rows]
    for substring in REQUIRED_NAME_SUBSTRINGS:
        if not any(substring in name for name in names):
            failures.append(f"missing a required backtest category matching '{substring}'")


def _check_attribution_exports(failures):
    orders_paths = list(EXPORT_DIR.glob("*_orders.csv"))
    if not orders_paths:
        failures.append("no raw orders/logs exports available yet; one-leg-group, top-pair-concentration, and skipped-entry checks are unverified this session")
        return
    for orders_path in orders_paths:
        logs_path = orders_path.with_name(orders_path.name.replace("_orders.csv", "_logs.txt"))
        if not logs_path.exists():
            failures.append(f"{orders_path.name} has no matching logs export")
            continue
        orders = pa._read_orders(orders_path)
        groups, grouping_issues = pa._group_orders(orders)
        _, _, lifecycle_issues = pa._match_trades(groups)
        _, _, _, skipped_entries, _ = pa._read_logs([logs_path])
        issues = grouping_issues + lifecycle_issues
        one_leg_groups = [issue for issue in issues if "expected 2 orders" in issue or "symbols" in issue]
        if one_leg_groups:
            failures.append(f"{orders_path.name} raw attribution expected 0 one-leg groups found {len(one_leg_groups)}")
        if skipped_entries:
            failures.append(f"{orders_path.name} raw attribution expected 0 skipped entries found {len(skipped_entries)}")


def _check_plan_reports_evidence(failures):
    if not PAPER.exists() or not SOURCE_AUDIT.exists():
        failures.append(f"missing {PAPER} or {SOURCE_AUDIT}")
        return
    paper_text = PAPER.read_text(encoding="utf-8")
    source_audit_text = SOURCE_AUDIT.read_text(encoding="utf-8")
    for term in REQUIRED_PAPER_TERMS:
        if term not in paper_text:
            failures.append(f"{PAPER} missing sharpe1 term {term}")
    for term in REQUIRED_SOURCE_AUDIT_TERMS:
        if term not in source_audit_text:
            failures.append(f"{SOURCE_AUDIT} missing sharpe1 term {term}")


def _report_acceptance_decision(candidate_rows, failures):
    if not candidate_rows:
        failures.append(f"no main-run row found for candidate feature_mode={CANDIDATE_FEATURE_MODE}")
        return
    main_rows = [row for row in candidate_rows if row["start_offset_days"] == "0" and row["initial_cash"] == "100000"]
    robustness_rows = [row for row in candidate_rows if row["start_offset_days"] != "0" or row["initial_cash"] != "100000"]
    main_sharpe = float(main_rows[-1]["sharpe"]) if main_rows else float("nan")
    robust_pass_count = len([row for row in robustness_rows if float(row["sharpe"]) >= ROBUSTNESS_SHARPE_TARGET])
    print(f"candidate main-run sharpe={main_sharpe} target={SHARPE_TARGET}")
    print(f"candidate robustness checks passing sharpe>={ROBUSTNESS_SHARPE_TARGET}: {robust_pass_count} of {len(robustness_rows)}")
    if main_sharpe < SHARPE_TARGET:
        failures.append(f"primary target not met: main-run sharpe {main_sharpe} < {SHARPE_TARGET}")
    if robust_pass_count < 2:
        failures.append(f"robustness target not met: only {robust_pass_count} of {len(robustness_rows)} shifted/capital checks reached sharpe>={ROBUSTNESS_SHARPE_TARGET}")


def _finish(failures):
    if failures:
        print("sharpe1 plan validation: acceptance criteria not met")
        for failure in failures:
            print(f"- {failure}")
        sys.exit(1)
    print("sharpe1 plan validation passed")


if __name__ == "__main__":
    main()