| 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()