From bb61a6056d9652c452f2aa1a67a67f56fb516806 Mon Sep 17 00:00:00 2001 From: AuroNyx <162569106+Atlas2005@users.noreply.github.com> Date: Sun, 12 Jul 2026 09:50:22 +0800 Subject: [PATCH] feat(evaluation): add full-A strategy and ML OOS ablation --- scripts/benchmark_buy_and_hold_benchmark.py | 53 ++ .../run_full_a_strategy_ml_oos_ablation_v1.py | 72 +++ src/quantpilot_core/evaluation/__init__.py | 18 + .../full_a_strategy_ml_oos_ablation.py | 559 ++++++++++++++++++ .../evaluation/ml_factor_training.py | 43 +- .../ml_ranking_robustness_walkforward.py | 181 +++++- .../real_data_walk_forward_smoke.py | 67 ++- ...test_full_a_strategy_ml_oos_ablation_v1.py | 492 +++++++++++++++ ...st_ml_ranking_robustness_walkforward_v1.py | 16 + 9 files changed, 1480 insertions(+), 21 deletions(-) create mode 100644 scripts/benchmark_buy_and_hold_benchmark.py create mode 100644 scripts/run_full_a_strategy_ml_oos_ablation_v1.py create mode 100644 src/quantpilot_core/evaluation/full_a_strategy_ml_oos_ablation.py create mode 100644 tests/evaluation/test_full_a_strategy_ml_oos_ablation_v1.py diff --git a/scripts/benchmark_buy_and_hold_benchmark.py b/scripts/benchmark_buy_and_hold_benchmark.py new file mode 100644 index 0000000..88fdbd4 --- /dev/null +++ b/scripts/benchmark_buy_and_hold_benchmark.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Manual complexity check for the Full-A buy-and-hold benchmark. + +Run with ``PYTHONPATH=src python scripts/benchmark_buy_and_hold_benchmark.py``. +It prints timings for the retired repeated-mask reference and the grouped +implementation; it deliberately makes no timing assertion. +""" +from __future__ import annotations + +import time +from types import SimpleNamespace + +import pandas as pd + +from quantpilot_core.evaluation.real_data_walk_forward_smoke import _buy_and_hold_benchmark + + +def reference_benchmark(frame: pd.DataFrame, windows: tuple[SimpleNamespace, ...], initial_cash: float) -> float: + ordered = frame.sort_values(["date", "symbol"], kind="stable") + symbols = tuple(sorted(str(symbol) for symbol in ordered["symbol"].dropna().unique())) + valid = [] + for symbol in symbols: + group = ordered.loc[ordered["symbol"] == symbol] + start_rows = group.loc[group["date"] >= windows[0].test_start] + end_rows = group.loc[group["date"] <= windows[-1].test_end] + if not start_rows.empty and not end_rows.empty: + start_price, end_price = float(start_rows.iloc[0]["close"]), float(end_rows.iloc[-1]["close"]) + if start_price > 0 and end_price > 0: + valid.append((start_price, end_price)) + return round(sum((initial_cash / len(valid) / start) * end for start, end in valid), 6) + + +def main() -> None: + symbol_count, date_count = 500, 80 + dates = pd.bdate_range("2024-01-02", periods=date_count).date.astype(str) + frame = pd.DataFrame([ + {"date": date, "symbol": f"{index:06d}.SZ", "close": 10.0 + index / 1000.0 + day / 100.0} + for day, date in enumerate(dates) + for index in range(symbol_count) + ]).sample(frac=1.0, random_state=7).reset_index(drop=True) + windows = (SimpleNamespace(test_start=dates[20], test_end=dates[-1]),) + + started = time.perf_counter(); reference = reference_benchmark(frame, windows, 1_000_000.0); reference_seconds = time.perf_counter() - started + started = time.perf_counter(); optimized = _buy_and_hold_benchmark(frame, windows, 1_000_000.0); optimized_seconds = time.perf_counter() - started + + assert reference == optimized["benchmark_final_equity"] + print(f"symbols={symbol_count} rows={len(frame)}") + print(f"reference repeated full-column comparisons: {symbol_count}; seconds={reference_seconds:.6f}") + print(f"grouped benchmark full-column comparisons: 0; seconds={optimized_seconds:.6f}") + + +if __name__ == "__main__": + main() diff --git a/scripts/run_full_a_strategy_ml_oos_ablation_v1.py b/scripts/run_full_a_strategy_ml_oos_ablation_v1.py new file mode 100644 index 0000000..e5880d2 --- /dev/null +++ b/scripts/run_full_a_strategy_ml_oos_ablation_v1.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python +"""Manual-only, offline-after-snapshot Full-A strategy/ML OOS ablation.""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from quantpilot_core.evaluation import FullAStrategyMLOOSAblationConfig, run_full_a_strategy_ml_oos_ablation_v1 +from quantpilot_core.real_data_provider import ProviderError, SnapshotDailyBarProvider + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Manual-only Full-A rule/ML OOS ablation (no broker, no DeepSeek).") + parser.add_argument("--snapshot-root", type=Path, required=True) + parser.add_argument("--start-date", default="2023-01-01"); parser.add_argument("--end-date", default="2024-12-31") + parser.add_argument("--artifact-path", default="artifacts/full_a_strategy_ml_oos_ablation/latest_report.json") + parser.add_argument("--symbols", default=None); parser.add_argument("--initial-cash", type=float, default=1_000_000.0) + parser.add_argument("--fold-count", type=int, default=5); parser.add_argument("--min-fold-count", type=int, default=3) + parser.add_argument("--train-window-days", type=int, default=60) + parser.add_argument("--validation-window-days", type=int, default=20); parser.add_argument("--test-window-days", type=int, default=20) + parser.add_argument("--max-windows-per-fold", type=int, default=1); parser.add_argument("--min-symbols-required", type=int, default=20) + parser.add_argument("--target-position-count", type=int, default=10); parser.add_argument("--fixed-rule-modes", default=None) + parser.add_argument("--no-ml", action="store_true"); parser.add_argument("--no-dynamic-selector", action="store_true") + parser.add_argument("--no-progress", action="store_true", help="Suppress flushed JSON progress events on stderr.") + return parser + + +def _config_from_args(args: argparse.Namespace, *, provider: SnapshotDailyBarProvider, symbols: tuple[str, ...], modes: tuple[str, ...]) -> FullAStrategyMLOOSAblationConfig: + return FullAStrategyMLOOSAblationConfig( + symbols=symbols, start_date=args.start_date, end_date=args.end_date, provider=provider, + initial_cash=args.initial_cash, fold_count=args.fold_count, min_fold_count=args.min_fold_count, + train_window_days=args.train_window_days, validation_window_days=args.validation_window_days, + test_window_days=args.test_window_days, max_windows_per_fold=args.max_windows_per_fold, + min_symbols_required=args.min_symbols_required, target_position_count=args.target_position_count, + fixed_rule_modes=modes, include_ml_candidate=not args.no_ml, + include_dynamic_selector=not args.no_dynamic_selector, artifact_path=args.artifact_path, + ) + + +def main() -> int: + parser = _build_parser() + args = parser.parse_args() + try: provider = SnapshotDailyBarProvider(args.snapshot_root) + except ProviderError as exc: parser.error(str(exc)) + symbols = tuple(item.strip() for item in args.symbols.split(",") if item.strip()) if args.symbols else provider.daily_symbol_union(args.start_date, args.end_date) + modes = tuple(item.strip() for item in args.fixed_rule_modes.split(",") if item.strip()) if args.fixed_rule_modes else FullAStrategyMLOOSAblationConfig().fixed_rule_modes + def progress(event: dict[str, object]) -> None: + print(json.dumps(event, sort_keys=True), file=sys.stderr, flush=True) + + try: + report = run_full_a_strategy_ml_oos_ablation_v1( + _config_from_args(args, provider=provider, symbols=symbols, modes=modes), + progress_callback=None if args.no_progress else progress, + ) + except KeyboardInterrupt: + print("Full-A ablation interrupted; no completed artifact was written.", file=sys.stderr, flush=True) + return 130 + print("QuantPilot Full-A strategy/ML OOS ablation v1 (manual-only; offline; no broker; no DeepSeek)") + print(f"provider: {report.provider}; snapshot digest: {report.snapshot_provenance.get('snapshot_digest')}") + print(f"folds: {len(report.common_folds)}; candidates: {len(report.per_candidate_metrics)}") + for name, row in report.per_candidate_metrics.items(): print(f"{name}: excess={row.get('strategy_excess_return')} return={row.get('total_return')} drawdown={row.get('max_drawdown')}") + if args.no_dynamic_selector: + print(f"dynamic selector: disabled; recommendation: {report.recommendation}") + else: + print(f"dynamic selections: {len(report.selection_history)}; recommendation: {report.recommendation}") + print(f"artifact: {report.artifact_path}") + return 0 if report.run_status == "completed" else 2 + + +if __name__ == "__main__": raise SystemExit(main()) diff --git a/src/quantpilot_core/evaluation/__init__.py b/src/quantpilot_core/evaluation/__init__.py index d1ff2c6..2238ee2 100644 --- a/src/quantpilot_core/evaluation/__init__.py +++ b/src/quantpilot_core/evaluation/__init__.py @@ -46,6 +46,16 @@ TurnoverAwareRebalanceOptimizationReport, run_turnover_aware_rebalance_optimization_v1, ) +from quantpilot_core.evaluation.full_a_strategy_ml_oos_ablation import ( + DEFAULT_FIXED_RULE_MODES, + DEFAULT_FULL_A_STRATEGY_ML_OOS_ABLATION_ARTIFACT_PATH, + DynamicSelectorPolicy, + FullAStrategyMLOOSAblationConfig, + FullAStrategyMLOOSAblationReport, + ProvisionalCandidateThresholds, + run_full_a_strategy_ml_oos_ablation_v1, + score_dynamic_candidates, +) from quantpilot_core.evaluation.a_share_market_reality_execution import ( DEFAULT_A_SHARE_MARKET_REALITY_EXECUTION_REPORT_ARTIFACT_PATH, AShareMarketRealityExecutionConfig, @@ -82,6 +92,8 @@ __all__ = [ "DEFAULT_FACTOR_RANKING_BASELINE_ARTIFACT_PATH", + "DEFAULT_FIXED_RULE_MODES", + "DEFAULT_FULL_A_STRATEGY_ML_OOS_ABLATION_ARTIFACT_PATH", "DEFAULT_A_SHARE_MARKET_REALITY_EXECUTION_REPORT_ARTIFACT_PATH", "DEFAULT_A_SHARE_TRADABILITY_METADATA_ENRICHMENT_REPORT_ARTIFACT_PATH", "DEFAULT_FACTOR_RANKING_SWEEP_INTEGRATION_ARTIFACT_PATH", @@ -99,6 +111,10 @@ "FactorRankingSweepIntegrationConfig", "FactorRankingSweepIntegrationReport", "FactorScore", + "DynamicSelectorPolicy", + "FullAStrategyMLOOSAblationConfig", + "FullAStrategyMLOOSAblationReport", + "ProvisionalCandidateThresholds", "AShareMarketRealityExecutionConfig", "AShareMarketRealityExecutionReport", "AShareTradabilityMetadataEnrichmentConfig", @@ -142,6 +158,8 @@ "run_real_data_walk_forward_scaleup_v1", "run_real_data_walk_forward_smoke", "run_factor_ranking_baseline_v1", + "run_full_a_strategy_ml_oos_ablation_v1", "run_ml_factor_training_v1", "run_profitability_smoke_test", + "score_dynamic_candidates", ] diff --git a/src/quantpilot_core/evaluation/full_a_strategy_ml_oos_ablation.py b/src/quantpilot_core/evaluation/full_a_strategy_ml_oos_ablation.py new file mode 100644 index 0000000..bc59c24 --- /dev/null +++ b/src/quantpilot_core/evaluation/full_a_strategy_ml_oos_ablation.py @@ -0,0 +1,559 @@ +"""Offline, common-fold Full-A rule/ML OOS ablation. + +This module is intentionally an orchestrator. It does not own a portfolio, +account, execution model, factor implementation, or ML training path. +""" +from __future__ import annotations + +import json +import time +from dataclasses import asdict, dataclass, field, replace +from pathlib import Path +from typing import Any, Callable, Mapping + +import pandas as pd + +from quantpilot_core.data_provider_normalization import canonicalize_a_share_symbol +from quantpilot_core.evaluation.ml_ranking_robustness_walkforward import ( + MLRankingRobustnessWalkForwardConfig, build_ml_ranking_walkforward_folds, + detect_target_horizon_trading_days, run_ml_ranking_robustness_walkforward_v1, +) +from quantpilot_core.evaluation.real_data_walk_forward_smoke import ( + FACTOR_RANKING_BASELINE_MODES, SUPPORTED_REAL_DATA_SCALEUP_RANKING_MODES, + RealDataWalkForwardScaleupConfig, RealDataWalkForwardSmokeConfig, + TurnoverAwareRebalanceConfig, _bars_to_price_frame, _data_quality_warnings, + _load_bars, _provider_name, _run_scaleup_with_loaded_price_frame, +) +from quantpilot_core.real_data_provider.contracts import DailyBarProvider + + +DEFAULT_FULL_A_STRATEGY_ML_OOS_ABLATION_ARTIFACT_PATH = Path("artifacts/full_a_strategy_ml_oos_ablation/latest_report.json") +DEFAULT_FIXED_RULE_MODES = ( + "equal_weight_baseline", "momentum_20d", "momentum_60d", "low_volatility", + "low_volatility_v1", "low_volatility_with_trend_filter", + "low_volatility_with_liquidity_filter", "defensive_composite_v1", + "momentum_reversal_guarded", +) + + +@dataclass(frozen=True) +class ProvisionalCandidateThresholds: + min_successful_fold_ratio: float = 0.60 + min_positive_return_fold_ratio: float = 0.50 + min_positive_excess_fold_ratio: float = 0.50 + min_aggregate_excess_return: float = 0.0 + max_drawdown: float = -0.30 + min_worst_fold_return: float = -0.20 + max_rejected_trade_ratio: float = 0.20 + max_cost_to_turnover_ratio: float = 0.05 + + +@dataclass(frozen=True) +class DynamicSelectorPolicy: + excess_return_weight: float = 1.0 + stability_weight: float = 0.25 + drawdown_weight: float = 0.50 + turnover_weight: float = 0.05 + cost_weight: float = 0.05 + fallback_candidate_id: str = "equal_weight_baseline" + + +@dataclass(frozen=True) +class FullAStrategyMLOOSAblationConfig: + symbols: tuple[str, ...] = () + start_date: str = "2023-01-01" + end_date: str = "2024-12-31" + provider: str | DailyBarProvider = "all_a_share_snapshot" + initial_cash: float = 1_000_000.0 + fold_count: int = 5 + min_fold_count: int = 3 + train_window_days: int = 60 + validation_window_days: int = 20 + test_window_days: int = 20 + max_windows_per_fold: int = 1 + min_symbols_required: int = 20 + target_position_count: int = 10 + max_position_weight: float = 0.10 + reserve_cash_weight: float = 0.02 + min_order_lot: int = 100 + fixed_rule_modes: tuple[str, ...] = DEFAULT_FIXED_RULE_MODES + include_ml_candidate: bool = True + include_dynamic_selector: bool = True + cost_multipliers: tuple[float, ...] = (1.0, 1.5, 2.0) + turnover_aware_rebalance: TurnoverAwareRebalanceConfig = field(default_factory=TurnoverAwareRebalanceConfig) + provisional_thresholds: ProvisionalCandidateThresholds = field(default_factory=ProvisionalCandidateThresholds) + selector_policy: DynamicSelectorPolicy = field(default_factory=DynamicSelectorPolicy) + baseline_candidate_id: str = "equal_weight_baseline" + artifact_path: str | Path | None = DEFAULT_FULL_A_STRATEGY_ML_OOS_ABLATION_ARTIFACT_PATH + metadata: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class FullAStrategyMLOOSAblationReport: + run_status: str + failure_stage: str | None + provider: str + snapshot_provenance: Mapping[str, Any] + date_range: tuple[str, str] + symbols_requested: tuple[str, ...] + valid_symbols: tuple[str, ...] + skipped_symbols: tuple[str, ...] + common_folds: tuple[Mapping[str, Any], ...] + purge_embargo_evidence: Mapping[str, Any] + candidate_definitions: tuple[Mapping[str, Any], ...] + per_candidate_metrics: Mapping[str, Mapping[str, Any]] + per_candidate_fold_metrics: Mapping[str, tuple[Mapping[str, Any], ...]] + fixed_rule_results: Mapping[str, Mapping[str, Any]] + ml_results: Mapping[str, Any] | None + dynamic_selector_results: Mapping[str, Any] | None + candidate_deltas: Mapping[str, Mapping[str, Any]] + cost_sensitivity: tuple[Mapping[str, Any], ...] + turnover_execution_metrics: Mapping[str, Mapping[str, Any]] + selection_history: tuple[Mapping[str, Any], ...] + recommendation: str + recommendation_reason: str + rejection_reasons: Mapping[str, tuple[str, ...]] + notes: tuple[str, ...] + artifact_path: str | None = None + + +def score_dynamic_candidates(validation_metrics: Mapping[str, Mapping[str, Any]], policy: DynamicSelectorPolicy | None = None) -> tuple[str | None, Mapping[str, float], str]: + """Choose from validation evidence only; lexical candidate ID breaks ties.""" + policy = policy or DynamicSelectorPolicy() + scores: dict[str, float] = {} + for candidate_id, row in validation_metrics.items(): + if row.get("available", True) is False or row.get("strategy_excess_return") is None: + continue + excess = float(row.get("strategy_excess_return") or 0.0) + win_rate = float(row.get("positive_return_fold_ratio") or 0.0) + drawdown = abs(float(row.get("max_drawdown") or 0.0)) + turnover = float(row.get("turnover") or 0.0) + cost = float(row.get("cost_total") or 0.0) + scale = max(1.0, turnover) + scores[candidate_id] = round(policy.excess_return_weight * excess + policy.stability_weight * win_rate - policy.drawdown_weight * drawdown - policy.turnover_weight * (turnover / scale) - policy.cost_weight * (cost / scale), 12) + if not scores: + return None, scores, "no_available_validation_candidate" + selected = sorted(scores, key=lambda name: (-scores[name], name))[0] + return selected, scores, "highest_validation_score_stable_tie_break" + + +def run_full_a_strategy_ml_oos_ablation_v1(config: FullAStrategyMLOOSAblationConfig | None = None, *, price_frame: pd.DataFrame | None = None, model_backend_factory: Callable[[], Any] | None = None, lightgbm_importer: Callable[[str], Any] | None = None, progress_callback: Callable[[Mapping[str, Any]], None] | None = None, **kwargs: Any) -> FullAStrategyMLOOSAblationReport: + payload = config or FullAStrategyMLOOSAblationConfig(**kwargs) + _validate(payload) + provider_name = _provider_name(payload.provider) + requested = tuple(canonicalize_a_share_symbol(symbol) for symbol in payload.symbols) + provenance = _provenance(payload.provider, len(requested)) + warnings: tuple[str, ...] = () + # The cache exists only for this run. The benchmark is independent of + # ranking mode and execution costs, so a fold's immutable price-window, + # initial-cash, and benchmark-mode inputs are sufficient within this + # single immutable source frame. + benchmark_cache: dict[tuple[str, str, float, str], Mapping[str, Any]] = {} + if price_frame is None: + smoke = _smoke_config(payload) + try: + loaded = _load_bars(smoke) # one batch/provider load for every candidate + price_frame = _bars_to_price_frame(loaded.bars) + warnings = loaded.warnings + _data_quality_warnings(price_frame, smoke) + except Exception as exc: + return _write(_unavailable(payload, provider_name, requested, provenance, "data_load", str(exc)), payload.artifact_path) + frame = price_frame.copy() + frame["date"] = pd.to_datetime(frame["date"]).dt.date.astype(str) + if requested: + frame = frame.loc[frame["symbol"].isin(set(requested))].copy() + frame = frame.sort_values(["date", "symbol"], kind="stable").reset_index(drop=True) + valid = tuple(sorted(str(item) for item in frame["symbol"].dropna().unique())) if not frame.empty else () + requested = requested or valid + skipped = tuple(symbol for symbol in requested if symbol not in set(valid)) + if len(valid) < payload.min_symbols_required: + return _write(_unavailable(payload, provider_name, requested, provenance, "data_validation", f"valid symbols below minimum: {len(valid)} < {payload.min_symbols_required}", valid, skipped), payload.artifact_path) + ml_config = _ml_config(payload, requested) + folds = build_ml_ranking_walkforward_folds(frame, ml_config) + if len(folds) < payload.min_fold_count: + return _write(_unavailable(payload, provider_name, requested, provenance, "fold_plan", "not enough chronological folds", valid, skipped), payload.artifact_path) + tracker = _progress_tracker(payload, folds, progress_callback) + tracker.emit("provider_load_complete", candidate=None, fold_index=None, cost_multiplier=None, advance=True) + common_folds = tuple(_fold_evidence(fold, frame, payload) for fold in folds) + tracker.emit("common_fold_evidence_complete", candidate=None, fold_index=None, cost_multiplier=None, advance=True) + rule_folds: dict[str, tuple[Mapping[str, Any], ...]] = {} + for mode in payload.fixed_rule_modes: + rows = [] + for fold_index, fold in enumerate(folds, start=1): + rows.append(_run_rule_fold(mode, fold, frame, payload, provider_name, warnings, benchmark_cache=benchmark_cache)) + tracker.emit("fixed_rule_fold_complete", candidate=mode, fold_index=fold_index, cost_multiplier=1.0, advance=True) + rule_folds[mode] = tuple(rows) + ml_report = run_ml_ranking_robustness_walkforward_v1(ml_config, price_frame=frame, model_backend_factory=model_backend_factory, lightgbm_importer=lightgbm_importer) if payload.include_ml_candidate else None + if payload.include_ml_candidate: + tracker.emit("ml_stage_complete", candidate="lightgbm_ml_ranking", fold_index=None, cost_multiplier=None, advance=True) + candidate_folds: dict[str, tuple[Mapping[str, Any], ...]] = dict(rule_folds) + if ml_report is not None: + benchmark_reference = rule_folds.get(payload.baseline_candidate_id, next(iter(rule_folds.values()), ())) + ml_rows_by_fold = {str(row.get("fold_id")): row for row in ml_report.fold_results} + candidate_folds["lightgbm_ml_ranking"] = tuple( + _ml_fold( + ml_rows_by_fold.get(fold["fold_id"], _missing_ml_fold(fold)), + common_fold=fold, + benchmark_reference=benchmark_reference[index] if index < len(benchmark_reference) else None, + ) + for index, fold in enumerate(folds) + ) + metrics = {name: _aggregate(rows) for name, rows in candidate_folds.items()} + base_validation = _validation_folds_for_cost( + folds, frame, payload, provider_name, warnings, 1.0, ml_report, benchmark_cache=benchmark_cache, + on_fold_complete=lambda mode, index: tracker.emit("validation_candidate_fold_complete", candidate=mode, fold_index=index, cost_multiplier=1.0, advance=True), + ) + history, dynamic_rows = _dynamic_rows(folds, candidate_folds, base_validation, payload) + if payload.include_dynamic_selector: + candidate_folds["dynamic_selector"] = dynamic_rows + metrics["dynamic_selector"] = _aggregate(dynamic_rows) + tracker.emit("dynamic_selector_complete", candidate="dynamic_selector", fold_index=None, cost_multiplier=1.0, advance=True) + cost_rows: list[Mapping[str, Any]] = [] + base_selections = tuple(row.get("selected_candidate") for row in history) + for multiplier in payload.cost_multipliers: + multiplier = float(multiplier) + tracker.emit("cost_multiplier_started", candidate=None, fold_index=None, cost_multiplier=multiplier, advance=False) + test_folds = _test_folds_for_cost( + folds, frame, payload, provider_name, warnings, multiplier, ml_report, rule_folds, + benchmark_cache=benchmark_cache, + on_fold_complete=lambda mode, index, multiplier=multiplier: tracker.emit("test_candidate_fold_complete", candidate=mode, fold_index=index, cost_multiplier=multiplier, advance=True), + ) + validation_folds = base_validation if multiplier == 1.0 else _validation_folds_for_cost( + folds, frame, payload, provider_name, warnings, multiplier, ml_report, + benchmark_cache=benchmark_cache, + on_fold_complete=lambda mode, index, multiplier=multiplier: tracker.emit("validation_candidate_fold_complete", candidate=mode, fold_index=index, cost_multiplier=multiplier, advance=True), + ) + for name, rows in test_folds.items(): + cost_rows.append({"candidate_id": name, "cost_multiplier": multiplier, **_aggregate(rows)}) + if payload.include_dynamic_selector: + scenario_history, scenario_dynamic_rows = _dynamic_rows(folds, test_folds, validation_folds, payload) + selections = tuple(row.get("selected_candidate") for row in scenario_history) + cost_rows.append({ + "candidate_id": "dynamic_selector", + "cost_multiplier": multiplier, + "selection_changed_vs_base": selections != base_selections, + "selected_candidates": selections, + **_aggregate(scenario_dynamic_rows), + }) + tracker.emit("dynamic_selector_complete", candidate="dynamic_selector", fold_index=None, cost_multiplier=multiplier, advance=True) + deltas = _deltas(metrics, payload.baseline_candidate_id) + rejections = {name: tuple(_rejection_reasons(row, payload.provisional_thresholds)) for name, row in metrics.items()} + recommendation = _recommend(metrics, rejections) + result = _write(FullAStrategyMLOOSAblationReport( + run_status="completed", failure_stage=None, provider=provider_name, snapshot_provenance=provenance, + date_range=(payload.start_date, payload.end_date), symbols_requested=requested, valid_symbols=valid, skipped_symbols=skipped, + common_folds=common_folds, purge_embargo_evidence={"target_horizon_trading_days": detect_target_horizon_trading_days(ml_config.target_label), "purge_or_embargo_days": detect_target_horizon_trading_days(ml_config.target_label), "source": "derived_from_target_label_and_applied_by_ml_runner", "all_decisions_before_test": True}, + candidate_definitions=tuple({"candidate_id": name, "kind": "fixed_rule" if name in rule_folds else "ml" if name == "lightgbm_ml_ranking" else "dynamic_selector"} for name in candidate_folds), + per_candidate_metrics=metrics, per_candidate_fold_metrics=candidate_folds, fixed_rule_results={name: metrics[name] for name in rule_folds}, + ml_results=None if ml_report is None else {"run_status": ml_report.run_status, "fold_count": ml_report.fold_count, "backend_note": "existing_ml_ranking_robustness_walkforward_v1", "metrics": metrics.get("lightgbm_ml_ranking")}, + dynamic_selector_results=None if not payload.include_dynamic_selector else {"metrics": metrics["dynamic_selector"], "policy": asdict(payload.selector_policy)}, + candidate_deltas=deltas, cost_sensitivity=tuple(cost_rows), turnover_execution_metrics={name: {key: value for key, value in metric.items() if key in {"turnover", "cost_total", "rejected_trade_ratio", "filled_trades", "rejected_trades"}} for name, metric in metrics.items()}, + selection_history=history, recommendation=recommendation, recommendation_reason="provisional_cost_after_fill_oos_thresholds_by_compounded_excess_return" if recommendation != "no_candidate" else "no_candidate_met_provisional_cost_after_fill_oos_thresholds", rejection_reasons=rejections, + notes=("manual_only", "offline_after_snapshot_creation", "no_deepseek", "no_broker", "common_fold_plan", "single_provider_batch_load", *warnings), + ), payload.artifact_path) + tracker.emit("artifact_write_complete", candidate=None, fold_index=None, cost_multiplier=None, advance=True) + return result + + +class _ProgressTracker: + """Emit deterministic, caller-owned progress without changing run output.""" + + def __init__(self, total_work_units: int, fold_count: int, callback: Callable[[Mapping[str, Any]], None] | None) -> None: + self.total_work_units = max(1, int(total_work_units)) + self.fold_count = int(fold_count) + self.callback = callback + self.completed_work_units = 0 + self.started_at = time.monotonic() + + def emit( + self, + phase: str, + *, + candidate: str | None, + fold_index: int | None, + cost_multiplier: float | None, + advance: bool, + ) -> None: + if advance: + self.completed_work_units += 1 + if self.callback is None: + return + completed = min(self.completed_work_units, self.total_work_units) + self.callback({ + "phase": phase, + "candidate": candidate, + "fold_index": fold_index, + "fold_count": None if fold_index is None else self.fold_count, + "cost_multiplier": cost_multiplier, + "completed_work_units": completed, + "total_work_units": self.total_work_units, + "percentage": round(100.0 * completed / self.total_work_units, 6), + "elapsed_seconds": round(time.monotonic() - self.started_at, 6), + }) + + +def _progress_tracker( + config: FullAStrategyMLOOSAblationConfig, + folds: tuple[Mapping[str, str], ...], + callback: Callable[[Mapping[str, Any]], None] | None, +) -> _ProgressTracker: + fold_count = len(folds) + rule_units = len(config.fixed_rule_modes) * fold_count + non_base_cost_count = sum(float(multiplier) != 1.0 for multiplier in config.cost_multipliers) + dynamic_units = (1 + len(config.cost_multipliers)) if config.include_dynamic_selector else 0 + # provider/frame readiness, common evidence, base rules, base validation, + # optional ML, non-base rule test/validation runs, selector runs, artifact. + total = 1 + 1 + rule_units + rule_units + (1 if config.include_ml_candidate else 0) + total += non_base_cost_count * rule_units * 2 + dynamic_units + 1 + return _ProgressTracker(total, fold_count, callback) + + +def _run_rule_fold(mode: str, fold: Mapping[str, str], frame: pd.DataFrame, config: FullAStrategyMLOOSAblationConfig, provider_name: str, warnings: tuple[str, ...], cost_multiplier: float = 1.0, benchmark_cache: dict[tuple[str, str, float, str], Mapping[str, Any]] | None = None) -> Mapping[str, Any]: + # Last train_window_days immediately precede the fixed common test block. + dates = tuple(sorted(frame["date"].unique())) + test_start = dates.index(fold["test_start"]) + start = dates[max(0, test_start - config.train_window_days)] + portion = frame.loc[(frame["date"] >= start) & (frame["date"] <= fold["test_end"])].copy() + scaleup = RealDataWalkForwardScaleupConfig(symbols=tuple(config.symbols) or tuple(sorted(frame["symbol"].unique())), start_date=start, end_date=fold["test_end"], initial_cash=config.initial_cash, train_window_days=config.train_window_days, test_window_days=config.test_window_days, max_windows=1, min_symbols_required=config.min_symbols_required, artifact_path=None, provider=provider_name, advisory_mode="disabled", target_position_count=config.target_position_count, max_position_weight=config.max_position_weight, reserve_cash_weight=config.reserve_cash_weight, ranking_mode=mode, min_order_lot=config.min_order_lot, cost_multiplier=cost_multiplier, turnover_aware_rebalance=config.turnover_aware_rebalance) + smoke = RealDataWalkForwardSmokeConfig(symbols=scaleup.symbols, start_date=start, end_date=fold["test_end"], initial_cash=config.initial_cash, train_window_days=config.train_window_days, test_window_days=config.test_window_days, max_windows=1, min_symbols_required=config.min_symbols_required, artifact_path=None, provider=provider_name, advisory_mode="disabled") + benchmark_key = ( + str(fold["test_start"]), str(fold["test_end"]), float(scaleup.initial_cash), str(scaleup.benchmark_mode), + ) + report = _run_scaleup_with_loaded_price_frame( + scaleup, smoke_config=smoke, provider_name=provider_name, price_frame=portion, data_warnings=warnings, + benchmark_cache=benchmark_cache, benchmark_cache_key=benchmark_key, + ) + return {"fold_id": fold["fold_id"], "status": "completed" if report.windows_run else "failed", "training_range": (fold["training_start"], fold["training_end"]), "validation_range": (fold["validation_start"], fold["validation_end"]), "test_range": (fold["test_start"], fold["test_end"]), "decision_timestamp": fold["validation_end"], "total_return": report.total_return, "benchmark_total_return": report.benchmark_total_return, "strategy_excess_return": report.strategy_excess_return, "max_drawdown": report.max_drawdown, "turnover": report.turnover, "cost_total": report.cost_total, "rejected_trade_ratio": report.rejected_trade_ratio, "filled_trades": report.filled_trades, "rejected_trades": report.rejected_trades, "execution_windows": report.per_window_metrics} + + +def _ml_fold( + row: Mapping[str, Any], + *, + common_fold: Mapping[str, str] | None = None, + benchmark_reference: Mapping[str, Any] | None = None, +) -> Mapping[str, Any]: + result = dict(row.get("ml_result") or {}) + expected_range = None if common_fold is None else (common_fold["test_start"], common_fold["test_end"]) + actual_range = row.get("actual_trading_evaluation_range") + reference_range = _execution_range((benchmark_reference or {}).get("execution_windows", ())) + mismatch = ( + row.get("status") == "completed" + and common_fold is not None + and ( + tuple(row.get("test_range") or ()) != expected_range + or (reference_range != (None, None) and tuple(actual_range or ()) != reference_range) + or ( + benchmark_reference is not None + and result.get("benchmark_total_return") != benchmark_reference.get("benchmark_total_return") + ) + ) + ) + if mismatch: + return { + "fold_id": row["fold_id"], "status": "failed", + "training_range": row.get("training_range"), "validation_range": row.get("validation_range"), + "test_range": row.get("test_range"), "decision_timestamp": (row.get("validation_range") or (None, None))[1], + "ineligibility_reason": "ml_common_fold_or_benchmark_mismatch", + "actual_trading_evaluation_range": actual_range, + "expected_trading_evaluation_range": reference_range, + "execution_windows": row.get("ml_execution_windows", ()), + } + ranking_evidence = dict(row.get("ml_ranking_evidence") or {}) + if benchmark_reference is not None: + reference_selected = _candidate_symbols_from_execution_windows(benchmark_reference.get("execution_windows", ())) + ml_selected = tuple(ranking_evidence.get("ml_selected_symbols", ())) + overlap = len(set(reference_selected) & set(ml_selected)) + ranking_evidence.update({ + "equal_weight_selected_symbols": reference_selected, + "selection_overlap_count": overlap, + "selection_overlap_ratio": round(overlap / len(ml_selected), 6) if ml_selected else None, + }) + return { + "fold_id": row["fold_id"], "status": row.get("status"), + "training_range": row.get("training_range"), "validation_range": row.get("validation_range"), + "test_range": row.get("test_range"), "decision_timestamp": (row.get("validation_range") or (None, None))[1], + "failure_reason": row.get("failure_reason"), + "actual_trading_evaluation_range": actual_range, + "execution_windows": row.get("ml_execution_windows", ()), + "ranking_evidence": ranking_evidence or None, + **result, "prediction_count": row.get("prediction_count"), "model_backend": row.get("model_backend"), + } + + +def _execution_range(windows: Any) -> tuple[str | None, str | None]: + rows = tuple(windows or ()) + if not rows: + return None, None + return str(rows[0].get("test_start")), str(rows[-1].get("test_end")) + + +def _candidate_symbols_from_execution_windows(windows: Any) -> tuple[str, ...]: + rows = tuple(windows or ()) + return tuple(str(symbol) for symbol in (rows[0].get("candidate_symbols", ()) if rows else ())) + + +def _missing_ml_fold(fold: Mapping[str, str]) -> Mapping[str, Any]: + return { + "fold_id": fold["fold_id"], "status": "failed", + "training_range": (fold["training_start"], fold["training_end"]), + "validation_range": (fold["validation_start"], fold["validation_end"]), + "test_range": (fold["test_start"], fold["test_end"]), + "failure_reason": "missing_ml_common_fold_evidence", + } + + +def _test_folds_for_cost(folds, frame, config, provider_name, warnings, multiplier, ml_report, base_rule_folds, *, benchmark_cache=None, on_fold_complete: Callable[[str, int], None] | None = None): + rows = {} + for mode in config.fixed_rule_modes: + if multiplier == 1.0: + rows[mode] = base_rule_folds[mode] + continue + mode_rows = [] + for index, fold in enumerate(folds, start=1): + mode_rows.append(_run_rule_fold(mode, fold, frame, config, provider_name, warnings, multiplier, benchmark_cache)) + if on_fold_complete is not None: + on_fold_complete(mode, index) + rows[mode] = tuple(mode_rows) + if ml_report is not None: + rows["lightgbm_ml_ranking"] = _ml_cost_folds(ml_report, multiplier, validation=False) + return rows + + +def _validation_folds_for_cost(folds, frame, config, provider_name, warnings, multiplier, ml_report, *, benchmark_cache=None, on_fold_complete: Callable[[str, int], None] | None = None): + """Build only pre-test selector evidence for one cost assumption.""" + rows = {} + for mode in config.fixed_rule_modes: + mode_rows = [] + for index, fold in enumerate(folds, start=1): + validation_fold = {**fold, "test_start": fold["validation_start"], "test_end": fold["validation_end"]} + mode_rows.append(_run_rule_fold(mode, validation_fold, frame, config, provider_name, warnings, multiplier, benchmark_cache)) + if on_fold_complete is not None: + on_fold_complete(mode, index) + rows[mode] = tuple(mode_rows) + if ml_report is not None: + rows["lightgbm_ml_ranking"] = _ml_cost_folds(ml_report, multiplier, validation=True) + return rows + + +def _ml_cost_folds(ml_report: Any, multiplier: float, *, validation: bool) -> tuple[Mapping[str, Any], ...]: + fold_details = {str(row["fold_id"]): row for row in ml_report.fold_results} + output = [] + for scenario in ml_report.cost_sensitivity_results: + if float(scenario.get("cost_multiplier", -1)) != multiplier: + continue + detail = fold_details.get(str(scenario.get("fold_id")), {}) + result = scenario.get("validation_result") if validation else scenario + if not isinstance(result, Mapping) or result.get("available", True) is False: + output.append({ + "fold_id": scenario.get("fold_id"), + "status": "failed", + "training_range": detail.get("training_range"), + "validation_range": detail.get("validation_range"), + "test_range": detail.get("test_range"), + "decision_timestamp": (detail.get("validation_range") or (None, None))[1], + "ineligibility_reason": "missing_pretest_ml_validation_portfolio_evidence" if validation else None, + }) + continue + output.append({ + "fold_id": scenario.get("fold_id"), + "status": detail.get("status", "completed"), + "training_range": detail.get("training_range"), + "validation_range": detail.get("validation_range"), + "test_range": detail.get("test_range"), + "decision_timestamp": (detail.get("validation_range") or (None, None))[1], + **dict(result), + }) + return tuple(output) + + +def _dynamic_rows(folds: tuple[Mapping[str, str], ...], candidate_folds: Mapping[str, tuple[Mapping[str, Any], ...]], validation_folds: Mapping[str, tuple[Mapping[str, Any], ...]], config: FullAStrategyMLOOSAblationConfig) -> tuple[tuple[Mapping[str, Any], ...], tuple[Mapping[str, Any], ...]]: + history: list[Mapping[str, Any]] = []; output: list[Mapping[str, Any]] = [] + for index, fold in enumerate(folds): + evidence = {} + eligibility = {} + for name, rows in validation_folds.items(): + row = next((item for item in rows if item.get("fold_id") == fold["fold_id"]), None) + metric = _aggregate((row,)) if row is not None else {"available": False} + reason = None if metric.get("available") else (row or {}).get("ineligibility_reason", "missing_pretest_validation_portfolio_evidence") + eligibility[name] = {"eligible": reason is None, "reason": reason} + if reason is None: + evidence[name] = metric + selected, scores, reason = score_dynamic_candidates(evidence, config.selector_policy) + if selected is None and config.selector_policy.fallback_candidate_id in candidate_folds: + selected, reason = config.selector_policy.fallback_candidate_id, "fallback_candidate_no_prior_completed_validation_evidence" + source = next((row for row in candidate_folds.get(selected or "", ()) if row.get("fold_id") == fold["fold_id"]), None) + history.append({"fold_id": fold["fold_id"], "evidence_timestamp": fold["validation_end"], "selected_candidate": selected, "candidate_scores": scores, "candidate_eligibility": eligibility, "selection_reason": reason, "uses_test_outcomes": False}) + output.append({**dict(source or {"fold_id": fold["fold_id"], "status": "failed"}), "selected_candidate": selected, "selection_scores": scores, "selection_reason": reason}) + return tuple(history), tuple(output) + + +def _aggregate(rows: tuple[Mapping[str, Any], ...]) -> Mapping[str, Any]: + complete = [row for row in rows if row.get("status") == "completed"] + def values(key: str) -> list[float]: return [float(row[key]) for row in complete if row.get(key) is not None] + def total(key: str) -> float | None: return round(sum(values(key)), 6) if values(key) else None + def mean(key: str) -> float | None: + series = values(key) + return round(sum(series) / len(series), 6) if series else None + def median(key: str) -> float | None: + series = sorted(values(key)) + if not series: return None + middle = len(series) // 2 + return round(series[middle] if len(series) % 2 else (series[middle - 1] + series[middle]) / 2, 6) + def compounded(key: str) -> float | None: + series = values(key) + if not series: return None + product = 1.0 + for value in series: product *= 1.0 + value + return round(product - 1.0, 6) + returns, excess, drawdowns = values("total_return"), values("strategy_excess_return"), values("max_drawdown") + turnover, costs = values("turnover"), values("cost_total") + rejected, filled = values("rejected_trades"), values("filled_trades") + return {"available": bool(complete), "successful_fold_ratio": round(len(complete) / len(rows), 6) if rows else 0.0, "positive_return_fold_ratio": round(sum(value > 0 for value in returns) / len(returns), 6) if returns else None, "positive_excess_fold_ratio": round(sum(value > 0 for value in excess) / len(excess), 6) if excess else None, "mean_fold_return": mean("total_return"), "median_fold_return": median("total_return"), "sum_fold_return": total("total_return"), "compounded_fold_return": compounded("total_return"), "mean_benchmark_return": mean("benchmark_total_return"), "median_benchmark_return": median("benchmark_total_return"), "sum_benchmark_return": total("benchmark_total_return"), "compounded_benchmark_return": compounded("benchmark_total_return"), "mean_excess_return": mean("strategy_excess_return"), "median_excess_return": median("strategy_excess_return"), "sum_excess_return": total("strategy_excess_return"), "compounded_excess_return": compounded("strategy_excess_return"), "total_return": compounded("total_return"), "benchmark_total_return": compounded("benchmark_total_return"), "strategy_excess_return": compounded("strategy_excess_return"), "max_drawdown": min(drawdowns) if drawdowns else None, "worst_fold_return": min(returns) if returns else None, "turnover": total("turnover") or 0.0, "cost_total": total("cost_total") or 0.0, "cost_to_turnover_ratio": round((sum(costs) / sum(turnover)), 8) if sum(turnover) else None, "rejected_trade_ratio": round(sum(rejected) / (sum(rejected) + sum(filled)), 6) if sum(rejected) + sum(filled) else 0.0, "filled_trades": int(sum(filled)), "rejected_trades": int(sum(rejected))} + + +def _fold_evidence(fold: Mapping[str, str], frame: pd.DataFrame, config: FullAStrategyMLOOSAblationConfig) -> Mapping[str, Any]: + return {**dict(fold), "decision_timestamp": fold["validation_end"], "purge_or_embargo_days": detect_target_horizon_trading_days(_ml_config(config, tuple(config.symbols)).target_label), "train_rows": int(frame[frame.date.between(fold["training_start"], fold["training_end"])].shape[0]), "validation_rows": int(frame[frame.date.between(fold["validation_start"], fold["validation_end"])].shape[0]), "test_rows": int(frame[frame.date.between(fold["test_start"], fold["test_end"])].shape[0]), "decision_before_test": fold["validation_end"] < fold["test_start"]} + + +def _deltas(metrics: Mapping[str, Mapping[str, Any]], baseline: str) -> Mapping[str, Mapping[str, Any]]: + base = metrics.get(baseline, {}) + keys = ("total_return", "strategy_excess_return", "max_drawdown", "positive_return_fold_ratio", "turnover", "cost_total", "rejected_trade_ratio") + return {name: {f"{key}_delta": round(float(row[key]) - float(base[key]), 6) if row.get(key) is not None and base.get(key) is not None else None for key in keys} for name, row in metrics.items()} + + +def _recommend(metrics: Mapping[str, Mapping[str, Any]], rejections: Mapping[str, tuple[str, ...]]) -> str: + """Return the eligible candidate with the documented compounded OOS basis.""" + eligible = [name for name in metrics if not rejections.get(name)] + if not eligible: + return "no_candidate" + return sorted( + eligible, + key=lambda name: (-float(metrics[name].get("compounded_excess_return") or -1e9), name), + )[0] + + +def _rejection_reasons(metric: Mapping[str, Any], t: ProvisionalCandidateThresholds) -> list[str]: + result=[] + if metric.get("successful_fold_ratio") is None or float(metric["successful_fold_ratio"]) < t.min_successful_fold_ratio: result.append("below_min_successful_fold_ratio") + if metric.get("positive_return_fold_ratio") is None or float(metric["positive_return_fold_ratio"]) < t.min_positive_return_fold_ratio: result.append("below_min_positive_return_fold_ratio") + if metric.get("positive_excess_fold_ratio") is None or float(metric["positive_excess_fold_ratio"]) < t.min_positive_excess_fold_ratio: result.append("below_min_positive_excess_fold_ratio") + if metric.get("compounded_excess_return") is None or float(metric["compounded_excess_return"]) < t.min_aggregate_excess_return: result.append("below_min_aggregate_excess_return") + if metric.get("max_drawdown") is None or float(metric["max_drawdown"]) < t.max_drawdown: result.append("max_drawdown_breach") + if metric.get("worst_fold_return") is None or float(metric["worst_fold_return"]) < t.min_worst_fold_return: result.append("worst_fold_return_breach") + if metric.get("rejected_trade_ratio") is None or float(metric["rejected_trade_ratio"]) > t.max_rejected_trade_ratio: result.append("rejected_trade_ratio_breach") + if metric.get("cost_to_turnover_ratio") is None or float(metric["cost_to_turnover_ratio"]) > t.max_cost_to_turnover_ratio: result.append("cost_to_turnover_ratio_breach") + return result + + +def _smoke_config(c: FullAStrategyMLOOSAblationConfig) -> RealDataWalkForwardSmokeConfig: return RealDataWalkForwardSmokeConfig(symbols=c.symbols, start_date=c.start_date, end_date=c.end_date, initial_cash=c.initial_cash, train_window_days=c.train_window_days, test_window_days=c.test_window_days, max_windows=c.max_windows_per_fold, provider=c.provider, advisory_mode="disabled", min_symbols_required=c.min_symbols_required, artifact_path=None) +def _ml_config(c: FullAStrategyMLOOSAblationConfig, symbols: tuple[str, ...]) -> MLRankingRobustnessWalkForwardConfig: return MLRankingRobustnessWalkForwardConfig(symbols=symbols, start_date=c.start_date, end_date=c.end_date, provider=c.provider, initial_cash=c.initial_cash, fold_count=c.fold_count, min_fold_count=c.min_fold_count, train_window_days=c.train_window_days, validation_window_days=c.validation_window_days, test_window_days=c.test_window_days, max_windows_per_fold=c.max_windows_per_fold, min_symbols_required=c.min_symbols_required, target_position_count=c.target_position_count, max_position_weight=c.max_position_weight, reserve_cash_weight=c.reserve_cash_weight, min_order_lot=c.min_order_lot, cost_multipliers=c.cost_multipliers, turnover_aware_rebalance=c.turnover_aware_rebalance, artifact_path=None, metadata=dict(c.metadata)) +def _provenance(provider: Any, requested: int) -> Mapping[str, Any]: return provider.snapshot_provenance(requested_symbols=requested) if hasattr(provider, "snapshot_provenance") else {"provider": _provider_name(provider), "external_calls_occurred": False if _provider_name(provider) == "all_a_share_snapshot" else None} +def _validate(c: FullAStrategyMLOOSAblationConfig) -> None: + unsupported=[mode for mode in c.fixed_rule_modes if mode not in SUPPORTED_REAL_DATA_SCALEUP_RANKING_MODES] + if unsupported: raise ValueError(f"unsupported fixed_rule_modes: {', '.join(unsupported)}") + if c.min_fold_count > c.fold_count: raise ValueError("min_fold_count must be no greater than fold_count") + if not c.fixed_rule_modes and not c.include_ml_candidate: raise ValueError("at least one fixed or ML candidate is required") + if any(value <= 0 for value in c.cost_multipliers): raise ValueError("cost_multipliers must be positive") +def _unavailable(c, provider, requested, provenance, stage, reason, valid=(), skipped=()): return FullAStrategyMLOOSAblationReport("unavailable", stage, provider, provenance, (c.start_date,c.end_date), requested, tuple(valid), tuple(skipped), (), {}, (), {}, {}, None, None, {}, (), {}, (), "no_candidate", reason, {}, ("manual_only", "offline_after_snapshot_creation", reason)) +def _write(report: FullAStrategyMLOOSAblationReport, artifact_path: str | Path | None) -> FullAStrategyMLOOSAblationReport: + if artifact_path is None: return report + path=Path(artifact_path); path.parent.mkdir(parents=True, exist_ok=True); path.write_text(json.dumps(asdict(replace(report, artifact_path=str(path))), default=str, indent=2, sort_keys=True)+"\n", encoding="utf-8"); return replace(report, artifact_path=str(path)) diff --git a/src/quantpilot_core/evaluation/ml_factor_training.py b/src/quantpilot_core/evaluation/ml_factor_training.py index 9eb657b..98ebe33 100644 --- a/src/quantpilot_core/evaluation/ml_factor_training.py +++ b/src/quantpilot_core/evaluation/ml_factor_training.py @@ -454,7 +454,7 @@ def run_ml_ranking_scaleup_evaluation_v1( str(window_bounds["evaluation_frame_end"]), ) prediction_map = { - f"{record.date}|{record.symbol}": float(record.prediction_score) + (str(record.date), str(record.symbol)): float(record.prediction_score) for record in valid_records } base_scaleup_config = _scaleup_config_for_ml_evaluation( @@ -470,6 +470,7 @@ def run_ml_ranking_scaleup_evaluation_v1( **dict(base_scaleup_config.metadata), "ml_prediction_map": prediction_map, "ml_prediction_record_count": len(valid_records), + "ml_ranking_input_source": "model_prediction_records_by_test_start", }, ) rule_config = replace(base_scaleup_config, ranking_mode=payload.same_window_rule_ranking_mode) @@ -971,6 +972,35 @@ def _train_and_predict( model_backend_factory: Callable[[], Any] | None, lightgbm_importer: Callable[[str], Any] | None, ) -> tuple[Mapping[str, Any], Mapping[tuple[str, str], float], tuple[Mapping[str, Any], ...]]: + """Train once and return held-out test predictions (legacy API).""" + + status, test_predictions, _validation_predictions, importance = _train_and_predict_with_validation( + dataset, + config, + model_backend_factory=model_backend_factory, + lightgbm_importer=lightgbm_importer, + ) + return status, test_predictions, importance + + +def _train_and_predict_with_validation( + dataset: MLFactorDataset, + config: MLFactorTrainingConfig, + *, + model_backend_factory: Callable[[], Any] | None, + lightgbm_importer: Callable[[str], Any] | None, +) -> tuple[ + Mapping[str, Any], + Mapping[tuple[str, str], float], + Mapping[tuple[str, str], float], + tuple[Mapping[str, Any], ...], +]: + """Train once and expose both pre-test validation and test predictions. + + The established callers retain ``_train_and_predict``. Walk-forward + orchestration may additionally consume validation predictions to evaluate + a selector without ever consulting test-period outcomes. + """ backend_factory = model_backend_factory lightgbm_available = False fallback_reason = None @@ -998,6 +1028,7 @@ def _train_and_predict( return ( {"model_backend": model_backend, "lightgbm_available": False, "model_trained": False, "fallback_reason": fallback_reason}, {}, + {}, (), ) train_rows = _rows_for_split(dataset, "train", config.target_label) @@ -1012,6 +1043,7 @@ def _train_and_predict( "fallback_reason": "insufficient_training_or_prediction_rows", }, {}, + {}, (), ) model = backend_factory() @@ -1026,6 +1058,11 @@ def _train_and_predict( (str(row["date"]), str(row["symbol"])): _finite_or_none(score) for row, score in zip(test_rows, predicted) } + validation_predicted = model.predict(x_validation) + validation_predictions = { + (str(row["date"]), str(row["symbol"])): _finite_or_none(score) + for row, score in zip(validation_rows, validation_predicted) + } importance_values = getattr(model, "feature_importances_", None) if importance_values is None: importance = tuple({"feature": feature, "importance": None} for feature in dataset.features) @@ -1040,8 +1077,12 @@ def _train_and_predict( "lightgbm_available": lightgbm_available, "model_trained": True, "fallback_reason": None, + "training_row_count": len(train_rows), + "validation_row_count": len(validation_rows), + "prediction_row_count": len(test_rows), }, predictions, + validation_predictions, importance, ) diff --git a/src/quantpilot_core/evaluation/ml_ranking_robustness_walkforward.py b/src/quantpilot_core/evaluation/ml_ranking_robustness_walkforward.py index 4063252..38dc7a4 100644 --- a/src/quantpilot_core/evaluation/ml_ranking_robustness_walkforward.py +++ b/src/quantpilot_core/evaluation/ml_ranking_robustness_walkforward.py @@ -23,7 +23,7 @@ _normalize_price_frame, _provider_name, _slice_evaluation_frame, - _train_and_predict, + _train_and_predict_with_validation, ) from quantpilot_core.evaluation.real_data_walk_forward_smoke import ( DEFAULT_REAL_DATA_SCALEUP_SYMBOLS, @@ -357,20 +357,23 @@ def _run_one_fold( dataset = replace(dataset, train_validation_test_split=split) purged_dataset, audit = apply_label_boundary_purge(dataset, split, target_label=config.target_label, horizon=horizon) try: - status, predictions, _ = _train_and_predict( + status, predictions, validation_predictions, _ = _train_and_predict_with_validation( purged_dataset, training_config, model_backend_factory=model_backend_factory, lightgbm_importer=lightgbm_importer, ) except ImportError: - status, predictions = ( + status, predictions, validation_predictions = ( {"model_backend": "lightgbm_unavailable", "lightgbm_available": False, "model_trained": False, "fallback_reason": "lightgbm_not_installed"}, {}, + {}, ) records = build_ml_prediction_records(purged_dataset, predictions) valid_records = tuple(record for record in records if _is_finite_number(record.prediction_score)) + validation_records = build_ml_prediction_records(purged_dataset, validation_predictions) + valid_validation_records = tuple(record for record in validation_records if _is_finite_number(record.prediction_score)) invalid_prediction_count = len(records) - len(valid_records) if not status.get("model_trained") or not valid_records: reason = f"model_training_skipped:{status.get('fallback_reason')}" if not status.get("model_trained") else "no_finite_prediction_score_records" @@ -392,7 +395,33 @@ def _run_one_fold( data_warnings=data_warnings, cost_multiplier=float(multiplier), ) - scenario = _scenario_row(fold_index, fold, multiplier, ml_report, rule_report) + validation_report = None + if valid_validation_records: + validation_report, _ = _evaluate_prediction_records( + fold=fold, + records=valid_validation_records, + price_frame=price_frame, + config=config, + provider_name=provider_name, + data_warnings=data_warnings, + cost_multiplier=float(multiplier), + evaluation_end=fold["validation_end"], + ) + scenario = _scenario_row( + fold_index, + fold, + multiplier, + ml_report, + rule_report, + validation_result=_portfolio_result( + validation_report, + evaluation_end=fold["validation_end"], + ), + ) + scenario = { + **scenario, + "ranking_evidence": _ranking_evidence(fold, valid_records, status, ml_report, rule_report), + } scenario_rows.append(scenario) if float(multiplier) == 1.0: base_row = _fold_row_from_scenario( @@ -420,17 +449,24 @@ def _evaluate_prediction_records( provider_name: str, data_warnings: tuple[str, ...], cost_multiplier: float, + evaluation_end: str | None = None, ) -> tuple[Any, Any]: dates = tuple(sorted(str(value) for value in pd.to_datetime(price_frame["date"]).dt.date.dropna().unique())) - first_prediction = min(record.date for record in records) - first_index = dates.index(first_prediction) - start_index = max(0, first_index - int(config.train_window_days) + 1) - end_date = fold["test_end"] - end_index = min(len(dates) - 1, dates.index(end_date) + int(config.test_window_days)) + # The scale-up executor builds its first test window immediately after its + # train slice. Anchor that slice to the declared fold boundary instead of + # the first ML prediction, and never extend the frame beyond the fold. + # This is the same construction used by Full-A's fixed-rule folds. + test_start = fold["validation_start"] if evaluation_end is not None else fold["test_start"] + start_index = max(0, dates.index(test_start) - int(config.train_window_days)) + end_date = evaluation_end or fold["test_end"] + end_index = dates.index(end_date) eval_start = dates[start_index] eval_end = dates[end_index] eval_frame = _slice_evaluation_frame(price_frame, eval_start, eval_end) - prediction_map = {f"{record.date}|{record.symbol}": float(record.prediction_score) for record in records} + prediction_map = { + (str(record.date), str(record.symbol)): float(record.prediction_score) + for record in records + } base_config = RealDataWalkForwardScaleupConfig( symbols=config.symbols, start_date=eval_start, @@ -438,7 +474,7 @@ def _evaluate_prediction_records( initial_cash=config.initial_cash, train_window_days=config.train_window_days, test_window_days=config.test_window_days, - max_windows=config.max_windows_per_fold, + max_windows=1 if evaluation_end is not None else config.max_windows_per_fold, min_symbols_required=config.min_symbols_required, allow_partial_universe=config.allow_partial_universe, artifact_path=None, @@ -459,6 +495,7 @@ def _evaluate_prediction_records( "fold_id": fold["fold_id"], "cost_multiplier": cost_multiplier, "ml_prediction_map": prediction_map, + "ml_ranking_input_source": "model_prediction_records_by_test_start", }, ) _validate_scaleup_config(base_config) @@ -469,7 +506,7 @@ def _evaluate_prediction_records( initial_cash=config.initial_cash, train_window_days=config.train_window_days, test_window_days=config.test_window_days, - max_windows=config.max_windows_per_fold, + max_windows=1 if evaluation_end is not None else config.max_windows_per_fold, provider=provider_name, advisory_mode="disabled", allow_partial_universe=config.allow_partial_universe, @@ -494,7 +531,39 @@ def _evaluate_prediction_records( return ml_report, rule_report -def _scenario_row(fold_index: int, fold: Mapping[str, str], multiplier: float, ml_report: Any, rule_report: Any) -> Mapping[str, Any]: +def _portfolio_result( + report: Any | None, + *, + evaluation_end: str | None = None, +) -> Mapping[str, Any] | None: + if report is None or not report.windows_run: + return None + return { + "available": True, + "total_return": report.total_return, + "benchmark_total_return": report.benchmark_total_return, + "strategy_excess_return": report.strategy_excess_return, + "max_drawdown": report.max_drawdown, + "cost_total": report.cost_total, + "turnover": report.turnover, + "filled_trades": report.filled_trades, + "rejected_trades": report.rejected_trades, + "rejected_trade_ratio": report.rejected_trade_ratio, + "evaluation_end": _last_window_test_end(report.per_window_metrics), + "execution_windows": _execution_windows_from_scaleup(report), + "pretest_validation_evidence": evaluation_end is not None, + } + + +def _scenario_row( + fold_index: int, + fold: Mapping[str, str], + multiplier: float, + ml_report: Any, + rule_report: Any, + *, + validation_result: Mapping[str, Any] | None, +) -> Mapping[str, Any]: execution = _execution_metrics_from_scaleup(ml_report) rule_execution = _execution_metrics_from_scaleup(rule_report) return { @@ -509,6 +578,8 @@ def _scenario_row(fold_index: int, fold: Mapping[str, str], multiplier: float, m "cost_total": ml_report.cost_total if ml_report.windows_run else None, "turnover": ml_report.turnover if ml_report.windows_run else None, "trade_count": ml_report.filled_trades if ml_report.windows_run else None, + "filled_trades": ml_report.filled_trades if ml_report.windows_run else None, + "rejected_trades": ml_report.rejected_trades if ml_report.windows_run else None, "attempted_order_count": execution["attempted_order_count"], "fill_ratio": execution["fill_ratio"], "partial_fill_count": execution["partial_fill_count"], @@ -532,9 +603,84 @@ def _scenario_row(fold_index: int, fold: Mapping[str, str], multiplier: float, m "turnover_aware_attribution": getattr(ml_report, "turnover_aware_attribution", {}), "rule_turnover_aware_attribution": getattr(rule_report, "turnover_aware_attribution", {}), "ml_beats_rule": _gt(ml_report.strategy_excess_return, rule_report.strategy_excess_return), + "validation_result": validation_result, } +def _ranking_evidence( + fold: Mapping[str, str], + records: tuple[MLFactorPredictionRecord, ...], + status: Mapping[str, Any], + ml_report: Any, + rule_report: Any, +) -> Mapping[str, Any]: + selection_date = str(fold["test_start"]) + scored = sorted( + ( + (str(record.symbol), float(record.prediction_score)) + for record in records + if str(record.date) == selection_date and _is_finite_number(record.prediction_score) + ), + key=lambda item: (-item[1], item[0]), + ) + values = [score for _, score in scored] + mean = sum(values) / len(values) if values else None + std = ( + round(math.sqrt(sum((value - mean) ** 2 for value in values) / len(values)), 12) + if mean is not None else None + ) + ml_selected = _candidate_symbols_from_scaleup(ml_report) + rule_selected = _candidate_symbols_from_scaleup(rule_report) + overlap = len(set(ml_selected) & set(rule_selected)) + fallback_reason = _ml_ranking_fallback_reason(ml_report) + return { + "model_backend": status.get("model_backend"), + "training_row_count": status.get("training_row_count"), + "prediction_count": len(records), + "prediction_selection_date": selection_date, + "prediction_min": min(values) if values else None, + "prediction_max": max(values) if values else None, + "prediction_stddev": std, + "unique_prediction_count": len(set(values)), + "top_predicted_symbols": tuple({"symbol": symbol, "score": score} for symbol, score in scored[:10]), + "bottom_predicted_symbols": tuple({"symbol": symbol, "score": score} for symbol, score in sorted(scored, key=lambda item: (item[1], item[0]))[:10]), + "same_window_rule_selected_symbols": rule_selected, + "ml_selected_symbols": ml_selected, + "selection_overlap_count": overlap, + "selection_overlap_ratio": round(overlap / len(ml_selected), 6) if ml_selected else None, + "actual_filled_order_symbols": _filled_symbols_from_scaleup(ml_report), + "fallback_used": fallback_reason is not None, + "fallback_reason": fallback_reason, + "ranking_input_source": "model_prediction_records_by_test_start", + } + + +def _candidate_symbols_from_scaleup(report: Any) -> tuple[str, ...]: + windows = tuple(getattr(report, "per_window_metrics", ()) or ()) + return tuple(str(symbol) for symbol in (windows[0].get("candidate_symbols", ()) if windows else ())) + + +def _filled_symbols_from_scaleup(report: Any) -> tuple[str, ...]: + symbols: list[str] = [] + for window in tuple(getattr(report, "per_window_metrics", ()) or ()): + for outcome in tuple(window.get("execution_outcomes", ()) or ()): + if str(outcome.get("status")) == "filled" and int(outcome.get("filled_quantity", 0)) > 0: + symbols.append(str(outcome.get("symbol"))) + return tuple(symbols) + + +def _ml_ranking_fallback_reason(report: Any) -> str | None: + if getattr(report, "windows_run", 0): + return None + for value in ( + *tuple(getattr(report, "data_quality_warnings", ()) or ()), + *tuple(getattr(report, "notes", ()) or ()), + ): + if str(value).startswith("ml_ranking_input_unavailable:"): + return str(value) + return "existing_scaleup_evaluator_returned_no_windows" + + def _fold_row_from_scenario( fold_index: int, fold: Mapping[str, str], @@ -550,7 +696,7 @@ def _fold_row_from_scenario( "fold_id": fold["fold_id"], "fold_index": fold_index, "status": "completed" if ml_report.windows_run else "failed", - "failure_reason": None if ml_report.windows_run else "existing_scaleup_evaluator_returned_no_windows", + "failure_reason": None if ml_report.windows_run else _ml_ranking_fallback_reason(ml_report), "training_range": (fold["training_start"], fold["training_end"]), "validation_range": (fold["validation_start"], fold["validation_end"]), "test_range": (fold["test_start"], fold["test_end"]), @@ -559,7 +705,9 @@ def _fold_row_from_scenario( "model_trained": bool(status.get("model_trained")), "prediction_count": prediction_count, "invalid_prediction_count": invalid_prediction_count, - "ml_result": {key: scenario.get(key) for key in ("total_return", "benchmark_total_return", "strategy_excess_return", "max_drawdown", "cost_total", "turnover", "trade_count", "attempted_order_count", "fill_ratio", "partial_fill_count", "rejected_order_count", "deferred_order_count", "execution_rejection_reasons", "rejected_trade_ratio", "turnover_aware_attribution")}, + "ml_result": {key: scenario.get(key) for key in ("total_return", "benchmark_total_return", "strategy_excess_return", "max_drawdown", "cost_total", "turnover", "trade_count", "filled_trades", "rejected_trades", "attempted_order_count", "fill_ratio", "partial_fill_count", "rejected_order_count", "deferred_order_count", "execution_rejection_reasons", "rejected_trade_ratio", "turnover_aware_attribution")}, + "ml_ranking_evidence": scenario.get("ranking_evidence"), + "validation_result": scenario.get("validation_result"), "ml_execution_windows": _execution_windows_from_scaleup(ml_report), "same_window_rule_baseline": { "ranking_mode": rule_report.ranking_mode, @@ -659,6 +807,9 @@ def _execution_windows_from_scaleup(report: Any) -> tuple[Mapping[str, Any], ... rows.append( { "run_label": row.get("run_label"), + "test_start": row.get("test_start"), + "test_end": row.get("test_end"), + "candidate_symbols": row.get("candidate_symbols", ()), "execution_reality": row.get("execution_reality"), "attempted_order_count": row.get("attempted_order_count"), "execution_fill_ratio": row.get("execution_fill_ratio"), diff --git a/src/quantpilot_core/evaluation/real_data_walk_forward_smoke.py b/src/quantpilot_core/evaluation/real_data_walk_forward_smoke.py index 4f8e105..ef68f66 100644 --- a/src/quantpilot_core/evaluation/real_data_walk_forward_smoke.py +++ b/src/quantpilot_core/evaluation/real_data_walk_forward_smoke.py @@ -7,7 +7,7 @@ from datetime import date from importlib import import_module from pathlib import Path -from typing import Any, Callable, Mapping, Sequence +from typing import Any, Callable, Hashable, Mapping, MutableMapping, Sequence import pandas as pd @@ -814,6 +814,9 @@ def _run_scaleup_with_loaded_price_frame( provider_name: str, price_frame: pd.DataFrame, data_warnings: tuple[str, ...], + benchmark_metrics: Mapping[str, Any] | None = None, + benchmark_cache: MutableMapping[Hashable, Mapping[str, Any]] | None = None, + benchmark_cache_key: Hashable | None = None, ) -> RealDataWalkForwardScaleupReport: if price_frame.empty: unavailable = _unavailable_report(smoke_config, provider_name, "provider returned no usable OHLCV rows", data_warnings) @@ -833,9 +836,20 @@ def _run_scaleup_with_loaded_price_frame( unavailable = _unavailable_report(smoke_config, provider_name, "not enough trading dates for configured windows", data_warnings) return _scaleup_report_from_smoke(unavailable, config) + ml_input_issue = _ml_ranking_input_issue(price_frame, windows, config) + if ml_input_issue is not None: + unavailable = _unavailable_report(smoke_config, provider_name, ml_input_issue, data_warnings) + return _scaleup_report_from_smoke(unavailable, config) + per_window, final_account = _run_scaleup_rebalance_windows(price_frame, windows, config) final_equity = _final_equity(per_window) - benchmark = _benchmark_metrics(price_frame, windows, float(config.initial_cash), config.benchmark_mode) + benchmark = benchmark_metrics + if benchmark is None and benchmark_cache is not None and benchmark_cache_key is not None: + benchmark = benchmark_cache.get(benchmark_cache_key) + if benchmark is None: + benchmark = _benchmark_metrics(price_frame, windows, float(config.initial_cash), config.benchmark_mode) + if benchmark_cache is not None and benchmark_cache_key is not None: + benchmark_cache[benchmark_cache_key] = benchmark strategy_return = _total_return(config.initial_cash, final_equity) valid_symbols = tuple(sorted(str(symbol) for symbol in price_frame["symbol"].dropna().unique())) expected_symbols = tuple(canonicalize_a_share_symbol(symbol) for symbol in config.symbols) @@ -974,7 +988,9 @@ def _run_scaleup_rebalance_windows( start_market_rows = _first_market_rows_for_window(price_frame, window.test_start, window.test_end, config.metadata) end_prices = _latest_prices_for_window(price_frame, window.test_start, window.test_end) starting_equity = _account_equity(account, start_prices) - ranked_candidates = _rank_scaleup_candidates(train_prices, start_prices, config) + ranked_candidates = _rank_scaleup_candidates( + train_prices, start_prices, config, ranking_date=str(window.test_start), + ) selected = tuple(symbol for _, symbol, _ in ranked_candidates[: int(config.target_position_count)]) policy_notes = _turnover_aware_selection_notes() if config.turnover_aware_rebalance.enabled: @@ -1143,12 +1159,14 @@ def _rank_scaleup_candidates( train_prices: pd.DataFrame, start_prices: Mapping[str, float], config: RealDataWalkForwardScaleupConfig, + *, + ranking_date: str | None = None, ) -> tuple[tuple[int, str, float | None], ...]: if train_prices.empty or config.ranking_mode == "equal_weight_baseline": return tuple((rank, symbol, 0.0) for rank, symbol in enumerate(sorted(start_prices), start=1)) if config.ranking_mode == "ml_prediction_score": prediction_map = config.metadata.get("ml_prediction_map", {}) - as_of_date = str(pd.Timestamp(train_prices["date"].max()).date()) + as_of_date = ranking_date or str(pd.Timestamp(train_prices["date"].max()).date()) scores: list[tuple[float, str, float | None]] = [] for symbol in sorted(start_prices): score = prediction_map.get((as_of_date, symbol)) @@ -1186,6 +1204,35 @@ def _rank_scaleup_candidates( return tuple((rank, symbol, score) for rank, (score, symbol) in enumerate(ranked, start=1)) +def _ml_ranking_input_issue( + price_frame: pd.DataFrame, + windows: Sequence[Any], + config: RealDataWalkForwardScaleupConfig, +) -> str | None: + """Reject absent or tied ML inputs rather than executing a lexical fallback.""" + if config.ranking_mode != "ml_prediction_score": + return None + prediction_map = config.metadata.get("ml_prediction_map", {}) + for window in windows: + start_prices = _first_prices_for_window(price_frame, window.test_start, window.test_end) + scores = [] + for symbol in sorted(start_prices): + value = prediction_map.get((str(window.test_start), symbol)) + if value is None: + value = prediction_map.get(f"{window.test_start}|{symbol}") + try: + numeric = float(value) + except (TypeError, ValueError): + continue + if not pd.isna(numeric) and numeric not in (float("inf"), float("-inf")): + scores.append(numeric) + if len(scores) < int(config.target_position_count): + return f"ml_ranking_input_unavailable:insufficient_finite_scores:{len(scores)}" + if len(set(scores)) <= 1: + return "ml_ranking_input_unavailable:constant_or_tied_scores" + return None + + def _ranking_score(group: pd.DataFrame, ranking_mode: str) -> float: prices = group.sort_values(["date", "symbol"], kind="stable")["close"].astype(float) if prices.empty: @@ -2545,9 +2592,19 @@ def _buy_and_hold_benchmark( end_date = str(windows[-1].test_end) if windows else str(frame["date"].max()) ordered = frame.sort_values(["date", "symbol"], kind="stable") symbols = tuple(sorted(str(symbol) for symbol in ordered["symbol"].dropna().unique())) + # `ordered` establishes the exact row ordering used by the previous + # per-symbol boolean mask implementation. Group it once rather than + # scanning the full symbol column for every benchmark constituent. + grouped = ordered.groupby("symbol", sort=False) valid: list[tuple[str, float, float]] = [] for symbol in symbols: - group = ordered.loc[ordered["symbol"] == symbol].copy() + try: + group = grouped.get_group(symbol) + except KeyError: + # Keep the old behavior for unusual non-string source labels: + # `symbols` contains their string representation, while the old + # equality comparison would have found no matching rows. + continue start_rows = group.loc[group["date"] >= start_date] end_rows = group.loc[group["date"] <= end_date] if start_rows.empty or end_rows.empty: diff --git a/tests/evaluation/test_full_a_strategy_ml_oos_ablation_v1.py b/tests/evaluation/test_full_a_strategy_ml_oos_ablation_v1.py new file mode 100644 index 0000000..d51a035 --- /dev/null +++ b/tests/evaluation/test_full_a_strategy_ml_oos_ablation_v1.py @@ -0,0 +1,492 @@ +from __future__ import annotations + +from dataclasses import replace +import importlib.util +from pathlib import Path +import sys +from types import SimpleNamespace + +import pandas as pd +import pytest + +from quantpilot_core.evaluation.full_a_strategy_ml_oos_ablation import ( + DynamicSelectorPolicy, + FullAStrategyMLOOSAblationConfig, + ProvisionalCandidateThresholds, + _aggregate, + _dynamic_rows, + _fold_evidence, + _ml_cost_folds, + _ml_fold, + _recommend, + _rejection_reasons, + _run_rule_fold, + _validate, + run_full_a_strategy_ml_oos_ablation_v1, + score_dynamic_candidates, +) +from quantpilot_core.evaluation.real_data_walk_forward_smoke import _buy_and_hold_benchmark +from quantpilot_core.evaluation.real_data_walk_forward_smoke import RealDataWalkForwardScaleupConfig, _rank_scaleup_candidates +from quantpilot_core.evaluation.ml_ranking_robustness_walkforward import build_ml_ranking_walkforward_folds +from quantpilot_core.evaluation.ml_ranking_robustness_walkforward import MLRankingRobustnessWalkForwardConfig + + +def _cli_module(): + path = Path(__file__).parents[2] / "scripts" / "run_full_a_strategy_ml_oos_ablation_v1.py" + spec = importlib.util.spec_from_file_location("full_a_ablation_cli_test", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _frame() -> pd.DataFrame: + dates = pd.bdate_range("2023-01-02", periods=140) + return pd.DataFrame( + [ + {"date": date.date().isoformat(), "symbol": symbol, "open": 10 + index, "high": 11 + index, "low": 9 + index, "close": 10.5 + index, "volume": 100000, "amount": 1_000_000} + for index, date in enumerate(dates) + for symbol in ("000001.SZ", "000002.SZ") + ] + ) + + +def _ranking_frame() -> pd.DataFrame: + dates = pd.bdate_range("2023-01-02", periods=180) + symbols = ("000001.SZ", "000002.SZ", "000003.SZ", "600000.SH") + return pd.DataFrame( + [ + {"date": date.date().isoformat(), "symbol": symbol, "open": 10 + symbol_index + index * 0.01, "high": 11 + symbol_index + index * 0.01, "low": 9 + symbol_index + index * 0.01, "close": 10.5 + symbol_index + index * 0.01, "volume": 100000 + symbol_index * 1000, "amount": 1_000_000 + symbol_index * 1000} + for index, date in enumerate(dates) + for symbol_index, symbol in enumerate(symbols) + ] + ) + + +def _reference_buy_and_hold_benchmark(frame: pd.DataFrame, windows, initial_cash: float) -> dict[str, object]: + """The pre-optimization per-symbol scan, retained only as test oracle.""" + if frame.empty: + return {"benchmark_final_equity": None, "benchmark_total_return": None, "benchmark_notes": ("simple_equal_weight_benchmark_unavailable:no_price_rows",)} + start_date = str(windows[0].test_start) if windows else str(frame["date"].min()) + end_date = str(windows[-1].test_end) if windows else str(frame["date"].max()) + ordered = frame.sort_values(["date", "symbol"], kind="stable") + symbols = tuple(sorted(str(symbol) for symbol in ordered["symbol"].dropna().unique())) + valid = [] + for symbol in symbols: + group = ordered.loc[ordered["symbol"] == symbol].copy() + start_rows = group.loc[group["date"] >= start_date] + end_rows = group.loc[group["date"] <= end_date] + if start_rows.empty or end_rows.empty: + continue + start_price = float(start_rows.iloc[0]["close"]) + end_price = float(end_rows.iloc[-1]["close"]) + if start_price > 0 and end_price > 0: + valid.append((symbol, start_price, end_price)) + if not valid: + return {"benchmark_final_equity": None, "benchmark_total_return": None, "benchmark_notes": ("simple_equal_weight_benchmark_unavailable:no_valid_symbol_prices",)} + final_equity = round(sum((initial_cash / len(valid) / start_price) * end_price for _, start_price, end_price in valid), 6) + return { + "benchmark_final_equity": final_equity, + "benchmark_total_return": round((final_equity - initial_cash) / initial_cash, 6), + "benchmark_notes": ("simple_equal_weight_close_to_close_no_costs", f"benchmark_start:{start_date}", f"benchmark_end:{end_date}", f"benchmark_symbols:{','.join(symbol for symbol, _, _ in valid)}"), + } + + +def test_grouped_benchmark_matches_reference_for_unsorted_missing_and_duplicate_rows() -> None: + frame = pd.DataFrame([ + {"date": "2023-01-04", "symbol": "000002.SZ", "close": 22.0}, + {"date": "2023-01-02", "symbol": "000001.SZ", "close": 10.0}, + {"date": "2023-01-03", "symbol": "000001.SZ", "close": 12.0}, + {"date": "2023-01-02", "symbol": "000002.SZ", "close": 20.0}, + {"date": "2023-01-04", "symbol": "000001.SZ", "close": 15.0}, + {"date": "2023-01-04", "symbol": "000001.SZ", "close": 16.0}, + {"date": "2023-01-03", "symbol": "000003.SZ", "close": 0.0}, + {"date": "2023-01-04", "symbol": "000003.SZ", "close": 5.0}, + {"date": "2023-01-04", "symbol": "000004.SZ", "close": 7.0}, + ]) + windows = (SimpleNamespace(test_start="2023-01-03", test_end="2023-01-04"),) + assert _buy_and_hold_benchmark(frame, windows, 1_000.0) == _reference_buy_and_hold_benchmark(frame, windows, 1_000.0) + + +def test_cli_exposes_min_fold_count_without_changing_defaults() -> None: + cli = _cli_module() + defaults = cli._build_parser().parse_args(["--snapshot-root", "/tmp/snapshot"]) + assert defaults.fold_count == 5 + assert defaults.min_fold_count == 3 + assert FullAStrategyMLOOSAblationConfig().fold_count == 5 + assert FullAStrategyMLOOSAblationConfig().min_fold_count == 3 + + probe = cli._build_parser().parse_args(["--snapshot-root", "/tmp/snapshot", "--fold-count", "1", "--min-fold-count", "1"]) + accepted = cli._config_from_args(probe, provider=object(), symbols=("000001.SZ",), modes=("equal_weight_baseline",)) + assert accepted.fold_count == accepted.min_fold_count == 1 + _validate(accepted) + + invalid = cli._build_parser().parse_args(["--snapshot-root", "/tmp/snapshot", "--fold-count", "1"]) + with pytest.raises(ValueError, match="min_fold_count must be no greater than fold_count"): + _validate(cli._config_from_args(invalid, provider=object(), symbols=("000001.SZ",), modes=("equal_weight_baseline",))) + + +@pytest.mark.parametrize(("extra_args", "expects_progress"), [((), True), (("--no-progress",), False)]) +def test_cli_keyboard_interrupt_and_progress_selection_are_preserved(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], extra_args: tuple[str, ...], expects_progress: bool) -> None: + cli = _cli_module() + captured = {} + + class FakeSnapshotProvider: + def __init__(self, root): + self.root = root + + def daily_symbol_union(self, start_date, end_date): + return ("000001.SZ",) + + def interrupted(config, *, progress_callback): + captured["progress_callback"] = progress_callback + raise KeyboardInterrupt + + monkeypatch.setattr(cli, "SnapshotDailyBarProvider", FakeSnapshotProvider) + monkeypatch.setattr(cli, "run_full_a_strategy_ml_oos_ablation_v1", interrupted) + monkeypatch.setattr(sys, "argv", ["full-a", "--snapshot-root", "/tmp/snapshot", *extra_args]) + + assert cli.main() == 130 + assert (captured["progress_callback"] is not None) is expects_progress + assert "Full-A ablation interrupted; no completed artifact was written." in capsys.readouterr().err + + +def test_cli_reports_disabled_dynamic_selector_without_positive_selection_count(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + cli = _cli_module() + + class FakeSnapshotProvider: + def __init__(self, root): + self.root = root + + def daily_symbol_union(self, start_date, end_date): + return ("000001.SZ",) + + report = SimpleNamespace(provider="snapshot", snapshot_provenance={}, common_folds=(), per_candidate_metrics={}, selection_history=(object(),), recommendation="no_candidate", artifact_path=None, run_status="completed") + monkeypatch.setattr(cli, "SnapshotDailyBarProvider", FakeSnapshotProvider) + monkeypatch.setattr(cli, "run_full_a_strategy_ml_oos_ablation_v1", lambda *args, **kwargs: report) + monkeypatch.setattr(sys, "argv", ["full-a", "--snapshot-root", "/tmp/snapshot", "--no-dynamic-selector"]) + + assert cli.main() == 0 + output = capsys.readouterr().out + assert "dynamic selector: disabled" in output + assert "dynamic selections: 1" not in output + + +def test_common_folds_are_chronological_and_decided_before_test() -> None: + config = FullAStrategyMLOOSAblationConfig(symbols=("000001.SZ", "000002.SZ"), fold_count=2, min_fold_count=1, train_window_days=20, validation_window_days=10, test_window_days=10) + folds = build_ml_ranking_walkforward_folds(_frame(), MLRankingRobustnessWalkForwardConfig(symbols=config.symbols, fold_count=2, min_fold_count=1, train_window_days=20, validation_window_days=10, test_window_days=10)) + assert folds + evidence = _fold_evidence(folds[0], _frame(), config) + assert evidence["training_end"] < evidence["validation_start"] <= evidence["validation_end"] < evidence["test_start"] + assert evidence["decision_before_test"] is True + + +def test_selector_uses_validation_metrics_and_has_stable_ties() -> None: + metrics = {"z_rule": {"strategy_excess_return": 0.1, "positive_return_fold_ratio": 0.5, "max_drawdown": -0.1, "turnover": 10, "cost_total": 1}, "a_rule": {"strategy_excess_return": 0.1, "positive_return_fold_ratio": 0.5, "max_drawdown": -0.1, "turnover": 10, "cost_total": 1}} + selected, scores, reason = score_dynamic_candidates(metrics, DynamicSelectorPolicy()) + assert selected == "a_rule" + assert scores["a_rule"] == scores["z_rule"] + assert reason == "highest_validation_score_stable_tie_break" + + +def _eligible_metric(*, drawdown: float = -0.10, compounded_excess_return: float = 0.10) -> dict[str, float]: + return { + "successful_fold_ratio": 1.0, + "positive_return_fold_ratio": 1.0, + "positive_excess_fold_ratio": 1.0, + "compounded_excess_return": compounded_excess_return, + "max_drawdown": drawdown, + "worst_fold_return": -0.05, + "rejected_trade_ratio": 0.01, + "cost_to_turnover_ratio": 0.01, + } + + +@pytest.mark.parametrize( + ("drawdown", "breaches"), + [(-0.10, False), (-0.30, False), (-0.31, True), (-0.50, True)], +) +def test_drawdown_threshold_is_a_negative_lower_bound(drawdown: float, breaches: bool) -> None: + reasons = _rejection_reasons(_eligible_metric(drawdown=drawdown), ProvisionalCandidateThresholds()) + assert ("max_drawdown_breach" in reasons) is breaches + + +def test_dynamic_selector_can_be_recommended_and_no_candidate_remains_possible() -> None: + metrics = { + "dynamic_selector": _eligible_metric(compounded_excess_return=0.20), + "fixed_rule": _eligible_metric(compounded_excess_return=0.10), + } + assert _recommend(metrics, {name: () for name in metrics}) == "dynamic_selector" + assert _recommend(metrics, {name: ("max_drawdown_breach",) for name in metrics}) == "no_candidate" + + +def test_recommendation_rejects_high_return_candidate_with_excessive_drawdown() -> None: + metrics = { + "unsafe": _eligible_metric(drawdown=-0.50, compounded_excess_return=0.90), + "safe": _eligible_metric(compounded_excess_return=0.10), + } + rejections = {name: tuple(_rejection_reasons(row, ProvisionalCandidateThresholds())) for name, row in metrics.items()} + assert "max_drawdown_breach" in rejections["unsafe"] + assert _recommend(metrics, rejections) == "safe" + + +def _fold(fold_id: str = "fold_01") -> dict[str, str]: + return { + "fold_id": fold_id, + "training_start": "2023-01-02", + "training_end": "2023-02-01", + "validation_start": "2023-02-02", + "validation_end": "2023-02-10", + "test_start": "2023-02-13", + "test_end": "2023-02-24", + } + + +def _portfolio_row(fold_id: str, excess: float, *, status: str = "completed") -> dict[str, object]: + return { + "fold_id": fold_id, + "status": status, + "total_return": excess, + "benchmark_total_return": 0.0, + "strategy_excess_return": excess, + "max_drawdown": -0.05, + "turnover": 10.0, + "cost_total": 0.1, + "filled_trades": 1, + "rejected_trades": 0, + } + + +def test_dynamic_selection_uses_validation_not_test_outcomes() -> None: + folds = (_fold(),) + tests = {"a_rule": (_portfolio_row("fold_01", -0.90),), "z_rule": (_portfolio_row("fold_01", 0.90),)} + validation = {"a_rule": (_portfolio_row("fold_01", 0.30),), "z_rule": (_portfolio_row("fold_01", 0.10),)} + history, selected_rows = _dynamic_rows(folds, tests, validation, FullAStrategyMLOOSAblationConfig()) + assert history[0]["selected_candidate"] == "a_rule" + assert history[0]["uses_test_outcomes"] is False + assert history[0]["evidence_timestamp"] < folds[0]["test_start"] + assert selected_rows[0]["strategy_excess_return"] == -0.90 + + +def test_validation_change_can_change_selection_and_ml_eligibility_is_recorded() -> None: + folds = (_fold(),) + tests = {"fixed": (_portfolio_row("fold_01", 0.1),), "lightgbm_ml_ranking": (_portfolio_row("fold_01", 0.2),)} + validation = {"fixed": (_portfolio_row("fold_01", 0.1),), "lightgbm_ml_ranking": (_portfolio_row("fold_01", 0.3),)} + history, _ = _dynamic_rows(folds, tests, validation, FullAStrategyMLOOSAblationConfig(fixed_rule_modes=("equal_weight_baseline",))) + assert history[0]["selected_candidate"] == "lightgbm_ml_ranking" + assert history[0]["candidate_eligibility"]["lightgbm_ml_ranking"] == {"eligible": True, "reason": None} + + missing_ml = {"fixed": validation["fixed"], "lightgbm_ml_ranking": (_portfolio_row("fold_01", 0.0, status="failed"),)} + history, _ = _dynamic_rows(folds, tests, missing_ml, FullAStrategyMLOOSAblationConfig(fixed_rule_modes=("equal_weight_baseline",))) + assert history[0]["selected_candidate"] == "fixed" + assert history[0]["candidate_eligibility"]["lightgbm_ml_ranking"]["reason"] == "missing_pretest_validation_portfolio_evidence" + + +def test_fold_aggregation_names_sum_and_compounded_returns_separately() -> None: + metrics = _aggregate((_portfolio_row("fold_01", 0.10), _portfolio_row("fold_02", 0.20))) + assert metrics["mean_fold_return"] == 0.15 + assert metrics["median_fold_return"] == 0.15 + assert metrics["sum_fold_return"] == 0.30 + assert metrics["compounded_fold_return"] == 0.32 + assert metrics["compounded_excess_return"] == 0.32 + assert metrics["total_return"] == metrics["compounded_fold_return"] + + +def test_purge_is_derived_from_the_actual_ml_label_horizon() -> None: + config = FullAStrategyMLOOSAblationConfig(symbols=("000001.SZ", "000002.SZ")) + evidence = _fold_evidence(_fold(), _frame(), config) + assert not hasattr(config, "purge_or_embargo_days") + assert evidence["purge_or_embargo_days"] == 20 + + +def test_selector_ignores_full_period_aggregate_fields() -> None: + evidence = _portfolio_row("fold_01", 0.10) + evidence["full_period_strategy_excess_return"] = 999.0 + selected, scores, _ = score_dynamic_candidates({"candidate": evidence}) + clean_selected, clean_scores, _ = score_dynamic_candidates({"candidate": _portfolio_row("fold_01", 0.10)}) + assert selected == clean_selected == "candidate" + assert scores == clean_scores + + +class _DeterministicRegressor: + def fit(self, x, y, eval_set=None): + self.feature_importances_ = [1.0] * len(x.columns) + return self + + def predict(self, x): + return [float(index) for index in range(len(x))] + + +class _ConstantRegressor: + def fit(self, x, y, eval_set=None): + self.feature_importances_ = [1.0] * len(x.columns) + return self + + def predict(self, x): + return [1.0] * len(x) + + +def test_ml_prediction_map_uses_rebalance_date_and_preserves_symbol_association() -> None: + train = pd.DataFrame([ + {"date": "2023-01-02", "symbol": "000001.SZ", "close": 10.0}, + {"date": "2023-01-02", "symbol": "000002.SZ", "close": 10.0}, + {"date": "2023-01-03", "symbol": "000001.SZ", "close": 10.1}, + {"date": "2023-01-03", "symbol": "000002.SZ", "close": 10.1}, + ]) + config = RealDataWalkForwardScaleupConfig( + symbols=("000001.SZ", "000002.SZ"), ranking_mode="ml_prediction_score", + metadata={"ml_prediction_map": {("2023-01-04", "000001.SZ"): 1.0, ("2023-01-04", "000002.SZ"): 9.0}}, + ) + ranked = _rank_scaleup_candidates(train, {"000001.SZ": 10.2, "000002.SZ": 10.2}, config, ranking_date="2023-01-04") + assert tuple(symbol for _, symbol, _ in ranked) == ("000002.SZ", "000001.SZ") + + +def test_reversed_ml_scores_change_executor_selection_and_constant_scores_are_unavailable() -> None: + config = FullAStrategyMLOOSAblationConfig( + symbols=("000001.SZ", "000002.SZ", "000003.SZ", "600000.SH"), fold_count=2, min_fold_count=1, + train_window_days=20, validation_window_days=10, test_window_days=10, max_windows_per_fold=1, + min_symbols_required=4, target_position_count=2, fixed_rule_modes=("equal_weight_baseline",), + include_ml_candidate=True, include_dynamic_selector=False, cost_multipliers=(1.0,), artifact_path=None, + ) + reversed_report = run_full_a_strategy_ml_oos_ablation_v1(config, price_frame=_ranking_frame(), model_backend_factory=_DeterministicRegressor) + fixed_rows = {row["fold_id"]: row for row in reversed_report.per_candidate_fold_metrics["equal_weight_baseline"]} + ml_rows = reversed_report.per_candidate_fold_metrics["lightgbm_ml_ranking"] + assert all(row["ranking_evidence"]["ml_selected_symbols"] != row["ranking_evidence"]["equal_weight_selected_symbols"] for row in ml_rows) + assert all(tuple(row["ranking_evidence"]["ml_selected_symbols"]) == tuple(item["symbol"] for item in row["ranking_evidence"]["top_predicted_symbols"][:2]) for row in ml_rows) + assert all(set(row["ranking_evidence"]["actual_filled_order_symbols"]) == set(row["ranking_evidence"]["ml_selected_symbols"]) for row in ml_rows) + assert all(row["execution_windows"][0]["candidate_symbols"] != fixed_rows[row["fold_id"]]["execution_windows"][0]["candidate_symbols"] for row in ml_rows) + assert all(row["benchmark_total_return"] == fixed_rows[row["fold_id"]]["benchmark_total_return"] for row in ml_rows) + + constant_report = run_full_a_strategy_ml_oos_ablation_v1(config, price_frame=_ranking_frame(), model_backend_factory=_ConstantRegressor) + constant_rows = constant_report.per_candidate_fold_metrics["lightgbm_ml_ranking"] + assert all(row["status"] == "failed" for row in constant_rows) + assert all(row["ranking_evidence"]["fallback_used"] is True for row in constant_rows) + assert all(row["ranking_evidence"]["fallback_reason"] == "ml_ranking_input_unavailable:constant_or_tied_scores" for row in constant_rows) + + +def test_orchestrator_reports_dynamic_cost_sensitivity_and_reuses_injected_frame(monkeypatch: pytest.MonkeyPatch) -> None: + def fail_load(*args, **kwargs): + raise AssertionError("injected price frame must avoid provider loading") + + monkeypatch.setattr("quantpilot_core.evaluation.full_a_strategy_ml_oos_ablation._load_bars", fail_load) + report = run_full_a_strategy_ml_oos_ablation_v1( + FullAStrategyMLOOSAblationConfig( + symbols=("000001.SZ", "000002.SZ"), + fold_count=2, + min_fold_count=1, + train_window_days=20, + validation_window_days=10, + test_window_days=10, + min_symbols_required=2, + target_position_count=2, + fixed_rule_modes=("equal_weight_baseline",), + cost_multipliers=(1.0, 1.5), + artifact_path=None, + ), + price_frame=_frame(), + model_backend_factory=_DeterministicRegressor, + ) + dynamic_rows = [row for row in report.cost_sensitivity if row["candidate_id"] == "dynamic_selector"] + assert report.run_status == "completed" + assert {row["cost_multiplier"] for row in dynamic_rows} == {1.0, 1.5} + assert all("selection_changed_vs_base" in row for row in dynamic_rows) + assert all(row["uses_test_outcomes"] is False for row in report.selection_history) + assert all("candidate_eligibility" in row for row in report.selection_history) + fixed_by_fold = {row["fold_id"]: row for row in report.per_candidate_fold_metrics["equal_weight_baseline"]} + ml_rows = report.per_candidate_fold_metrics["lightgbm_ml_ranking"] + assert all(row["test_range"] == fixed_by_fold[row["fold_id"]]["test_range"] for row in ml_rows) + assert all(row["benchmark_total_return"] == fixed_by_fold[row["fold_id"]]["benchmark_total_return"] for row in ml_rows) + assert all({"filled_trades", "rejected_trades", "rejected_trade_ratio", "turnover", "cost_total", "execution_windows"} <= set(row) for row in ml_rows) + assert all(row["filled_trades"] > 0 for row in ml_rows if row["turnover"] or row["cost_total"]) + + +def test_ml_fold_adapter_preserves_execution_evidence_and_rejects_common_fold_mismatch() -> None: + row = { + "fold_id": "fold_01", "status": "completed", + "training_range": ("2023-01-02", "2023-02-01"), + "validation_range": ("2023-02-02", "2023-02-10"), + "test_range": ("2023-02-13", "2023-02-24"), + "actual_trading_evaluation_range": ("2023-02-13", "2023-02-24"), + "ml_result": {"benchmark_total_return": 0.1, "turnover": 100.0, "cost_total": 1.0, "filled_trades": 2, "rejected_trades": 1, "rejected_trade_ratio": 0.333333}, + "ml_execution_windows": ({"test_start": "2023-02-13", "test_end": "2023-02-24"},), + } + fixed = {"benchmark_total_return": 0.1, "execution_windows": ({"test_start": "2023-02-13", "test_end": "2023-02-24"},)} + adapted = _ml_fold(row, common_fold=_fold(), benchmark_reference=fixed) + assert adapted["filled_trades"] == 2 + assert adapted["rejected_trades"] == 1 + assert adapted["execution_windows"] == row["ml_execution_windows"] + + mismatched = _ml_fold({**row, "test_range": ("2023-02-14", "2023-02-24")}, common_fold=_fold(), benchmark_reference=fixed) + assert mismatched["status"] == "failed" + assert mismatched["ineligibility_reason"] == "ml_common_fold_or_benchmark_mismatch" + + unavailable = _ml_fold({"fold_id": "fold_01", "status": "failed", "failure_reason": "missing"}, common_fold=_fold(), benchmark_reference=fixed) + assert unavailable["status"] == "failed" + assert unavailable["failure_reason"] == "missing" + assert "filled_trades" not in unavailable + + +def test_ml_cost_fold_adapter_preserves_actual_execution_counts() -> None: + detail = {"fold_id": "fold_01", "status": "completed", "training_range": ("a", "b"), "validation_range": ("c", "d"), "test_range": ("e", "f")} + scenario = {"fold_id": "fold_01", "cost_multiplier": 1.0, "benchmark_total_return": 0.1, "turnover": 100.0, "cost_total": 1.0, "filled_trades": 2, "rejected_trades": 1, "rejected_trade_ratio": 0.333333, "execution_windows": ({"test_start": "e", "test_end": "f"},)} + report = SimpleNamespace(fold_results=(detail,), cost_sensitivity_results=(scenario,)) + output = _ml_cost_folds(report, 1.0, validation=False) + assert output[0]["filled_trades"] == 2 + assert output[0]["rejected_trades"] == 1 + assert output[0]["execution_windows"] == scenario["execution_windows"] + + +def test_rule_candidates_reuse_fold_benchmarks_across_cost_multipliers(monkeypatch: pytest.MonkeyPatch) -> None: + import quantpilot_core.evaluation.real_data_walk_forward_smoke as scaleup_module + + original = scaleup_module._benchmark_metrics + calls = [] + + def counted(*args, **kwargs): + calls.append((args[2], args[3])) + return original(*args, **kwargs) + + monkeypatch.setattr(scaleup_module, "_benchmark_metrics", counted) + report = run_full_a_strategy_ml_oos_ablation_v1( + FullAStrategyMLOOSAblationConfig( + symbols=("000001.SZ", "000002.SZ"), fold_count=2, min_fold_count=1, + train_window_days=20, validation_window_days=10, test_window_days=10, + min_symbols_required=2, target_position_count=2, + fixed_rule_modes=("equal_weight_baseline", "momentum_20d"), + include_ml_candidate=False, include_dynamic_selector=False, + cost_multipliers=(1.0, 1.5), artifact_path=None, + ), + price_frame=_frame(), + ) + # Two OOS folds plus two validation folds, not once per candidate or cost. + assert len(calls) == len(report.common_folds) * 2 + + isolated_cache = {} + fold = dict(report.common_folds[0]) + _run_rule_fold("equal_weight_baseline", fold, _frame(), config := FullAStrategyMLOOSAblationConfig( + symbols=("000001.SZ", "000002.SZ"), train_window_days=20, test_window_days=10, + min_symbols_required=2, target_position_count=2, artifact_path=None, + ), "all_a_share_snapshot", (), benchmark_cache=isolated_cache) + _run_rule_fold("equal_weight_baseline", fold, _frame(), replace(config, initial_cash=2_000_000.0), "all_a_share_snapshot", (), benchmark_cache=isolated_cache) + assert len(isolated_cache) == 2 + + +def test_progress_events_are_monotonic_and_do_not_change_report_values() -> None: + config = FullAStrategyMLOOSAblationConfig( + symbols=("000001.SZ", "000002.SZ"), fold_count=2, min_fold_count=1, + train_window_days=20, validation_window_days=10, test_window_days=10, + min_symbols_required=2, target_position_count=2, + fixed_rule_modes=("equal_weight_baseline",), include_ml_candidate=False, + include_dynamic_selector=True, cost_multipliers=(1.0, 1.5), artifact_path=None, + ) + events = [] + with_progress = run_full_a_strategy_ml_oos_ablation_v1(config, price_frame=_frame(), progress_callback=events.append) + without_progress = run_full_a_strategy_ml_oos_ablation_v1(config, price_frame=_frame()) + + assert with_progress == without_progress + assert events[-1]["phase"] == "artifact_write_complete" + assert events[-1]["percentage"] == 100.0 + assert [event["completed_work_units"] for event in events] == sorted(event["completed_work_units"] for event in events) + assert all(event["fold_count"] in {None, 2} for event in events) + assert any(event["candidate"] == "equal_weight_baseline" and event["fold_index"] == 1 for event in events) + assert any(event["cost_multiplier"] == 1.5 for event in events) diff --git a/tests/evaluation/test_ml_ranking_robustness_walkforward_v1.py b/tests/evaluation/test_ml_ranking_robustness_walkforward_v1.py index 254131d..6543357 100644 --- a/tests/evaluation/test_ml_ranking_robustness_walkforward_v1.py +++ b/tests/evaluation/test_ml_ranking_robustness_walkforward_v1.py @@ -266,6 +266,22 @@ def test_cost_sensitivity_scenarios_are_reported(tmp_path: Path) -> None: ) +def test_ml_validation_portfolio_evidence_is_capped_before_the_test_period(tmp_path: Path) -> None: + report = run_ml_ranking_robustness_walkforward_v1( + config(tmp_path, fold_count=3), + price_frame=fixture_frame(days=170), + model_backend_factory=CountingRegressor, + ) + + for row in report.fold_results: + if row["status"] != "completed": + continue + validation = row["validation_result"] + assert validation is not None + assert validation["pretest_validation_evidence"] is True + assert validation["evaluation_end"] <= row["validation_range"][1] + + def test_concentration_diagnostics_use_existing_evaluation_artifacts(tmp_path: Path) -> None: report = run_ml_ranking_robustness_walkforward_v1( config(tmp_path, fold_count=3),