fithia2/libs/backtest/artifacts.py

"""Write all output artifacts for a backtest run."""
from __future__ import annotations

import csv
import datetime as dt
import json
from collections import defaultdict
from pathlib import Path
from typing import Any

import pyarrow as pa
import pyarrow.parquet as pq

from libs.backtest.domain import (
    BacktestConfig,
    DailyPortfolioState,
    ExperimentManifest,
    ExperimentResult,
    FilledTrade,
    MetricsBundle,
    OpenPosition,
)
from libs.backtest.snapshots import resolve_snapshot_path
from libs.common.ids import sha256_checksum_str
from libs.common.logging import get_logger

logger = get_logger(__name__)


def _resolve_trade_sleeve(trade: FilledTrade, candidate_map: dict[str, Any] | None = None) -> str:
    if (
        trade.engine_id == "cash_parking"
        or trade.event_type == "cash_parking"
        or str(trade.exit_reason.value).upper() == "PARKING"
    ):
        return "parking"
    candidate = (candidate_map or {}).get(trade.trade_id)
    if candidate is not None:
        features = getattr(candidate, "features", {}) or {}
        sleeve = str(features.get("trade_sleeve") or "").strip().lower()
        if sleeve in {"core", "idle_alpha", "form4", "ownership", "risk_off_alpha", "parking"}:
            return sleeve
    return "core"


def create_run_directory(output_root: str | Path, run_id: str) -> Path:
    """Create the full run directory tree. Returns the run root path."""
    run_dir = Path(output_root) / run_id
    for subdir in ["logs", "metrics", "artifacts", "plots", "notes"]:
        (run_dir / subdir).mkdir(parents=True, exist_ok=True)
    logger.info("run_directory_created", path=str(run_dir))
    return run_dir


def write_metadata(
    run_dir: Path,
    run_id: str,
    started_at: dt.datetime,
    finished_at: dt.datetime,
    git_hash: str,
    total_trading_days: int,
    total_candidates_seen: int,
    total_orders_rejected: int,
    requested_snapshot_id: str | None = None,
    canonical_snapshot_id: str | None = None,
    split_name: str | None = None,
) -> Path:
    """Write metadata.json."""
    meta = {
        "run_id": run_id,
        "started_at": started_at.isoformat(),
        "finished_at": finished_at.isoformat(),
        "elapsed_seconds": (finished_at - started_at).total_seconds(),
        "git_commit_hash": git_hash,
        "total_trading_days": total_trading_days,
        "total_candidates_seen": total_candidates_seen,
        "total_orders_rejected": total_orders_rejected,
    }
    if requested_snapshot_id is not None:
        meta["requested_snapshot_id"] = requested_snapshot_id
    if canonical_snapshot_id is not None:
        meta["canonical_snapshot_id"] = canonical_snapshot_id
    if split_name is not None:
        meta["split_name"] = split_name
    out = run_dir / "metadata.json"
    out.write_text(json.dumps(meta, indent=2))
    return out


def write_metrics_summary(run_dir: Path, metrics: MetricsBundle) -> Path:
    """Write metrics/metrics_summary.json."""
    out = run_dir / "metrics" / "metrics_summary.json"
    out.write_text(metrics.model_dump_json(indent=2))
    return out


def write_snapshot_provenance(
    run_dir: Path,
    config: BacktestConfig,
) -> dict[str, str]:
    """Persist the snapshot manifest and a stable fingerprint for reproducibility checks."""
    requested_snapshot_id = config.requested_snapshot_id or config.dataset_snapshot_id
    canonical_snapshot_id = config.canonical_snapshot_id or config.dataset_snapshot_id
    try:
        snapshot_dir = resolve_snapshot_path(requested_snapshot_id)
    except Exception:
        return {}
    if snapshot_dir is None:
        return {}
    manifest_path = snapshot_dir / "manifest.json"
    if not manifest_path.exists():
        return {}

    manifest_text = manifest_path.read_text()
    manifest_data = json.loads(manifest_text)
    manifest_snapshot_id = manifest_data.get("snapshot_id")
    manifest_output_dir = manifest_data.get("output_dir")
    snapshot_dir_name = snapshot_dir.name
    snapshot_id_matches_manifest = manifest_snapshot_id == snapshot_dir_name
    output_dir_matches_manifest = (
        True if not manifest_output_dir else Path(str(manifest_output_dir)).name == snapshot_dir_name
    )
    if not snapshot_id_matches_manifest or not output_dir_matches_manifest:
        logger.warning(
            "snapshot_manifest_metadata_mismatch",
            dataset_snapshot_id=config.dataset_snapshot_id,
            source_manifest_path=str(manifest_path.resolve()),
            manifest_snapshot_id=manifest_snapshot_id,
            manifest_output_dir=manifest_output_dir,
            snapshot_dir_name=snapshot_dir_name,
        )
    manifest_copy = run_dir / "snapshot_manifest.json"
    manifest_copy.write_text(json.dumps(manifest_data, indent=2))

    fingerprint = {
        "dataset_snapshot_id": config.dataset_snapshot_id,
        "requested_snapshot_id": requested_snapshot_id,
        "canonical_snapshot_id": canonical_snapshot_id,
        "source_manifest_path": str(manifest_path.resolve()),
        "snapshot_dir_name": snapshot_dir_name,
        "manifest_snapshot_id": manifest_snapshot_id,
        "manifest_output_dir": manifest_output_dir,
        "snapshot_id_matches_manifest": snapshot_id_matches_manifest,
        "output_dir_matches_manifest": output_dir_matches_manifest,
        "manifest_sha256": sha256_checksum_str(manifest_text),
        "snapshot_created_at_utc": manifest_data.get("created_at_utc"),
        "code_commit_hash": manifest_data.get("code_commit_hash"),
        "feature_version": manifest_data.get("feature_version"),
        "parser_version": manifest_data.get("parser_version"),
        "label_version": manifest_data.get("label_version"),
        "row_counts": manifest_data.get("row_counts"),
        "total_rows": manifest_data.get("total_rows"),
    }
    fingerprint_path = run_dir / "snapshot_fingerprint.json"
    fingerprint_path.write_text(json.dumps(fingerprint, indent=2))
    return {
        "snapshot_manifest": str(manifest_copy),
        "snapshot_fingerprint": str(fingerprint_path),
    }


def write_trade_blotter(
    run_dir: Path,
    trades: list[FilledTrade],
    candidate_map: dict[str, Any] | None = None,
) -> Path | None:
    """Write artifacts/trade_blotter.parquet."""
    if not trades:
        return None
    rows = [
        {
            "trade_id": t.trade_id,
            "position_id": t.position_id,
            "event_id": t.event_id,
            "symbol": t.symbol,
            "source_symbol": t.source_symbol,
            "event_date": t.event_date.isoformat() if t.event_date else None,
            "event_type": t.event_type,
            "trade_sleeve": _resolve_trade_sleeve(t, candidate_map),
            "score": t.score,
            "timing_class": t.timing_class,
            "engine_id": t.engine_id,
            "entry_timing_policy": t.entry_timing_policy,
            "trade_symbol_mode": t.trade_symbol_mode,
            "shadow_only": t.shadow_only,
            "parent_position_id": t.parent_position_id,
            "is_add_on": t.is_add_on,
            "entry_date": t.entry_date.isoformat(),
            "exit_date": t.exit_date.isoformat(),
            "entry_price": t.entry_price,
            "exit_price": t.exit_price,
            "exit_reason": t.exit_reason.value,
            "shares": t.shares,
            "commission": t.commission,
            "slippage_bps": t.slippage_bps,
            "gross_pnl": t.gross_pnl,
            "net_pnl": t.net_pnl,
            "pnl_pct": t.pnl_pct,
            "r_multiple": t.r_multiple,
            "holding_days": t.holding_days,
        }
        for t in trades
    ]
    out = run_dir / "artifacts" / "trade_blotter.parquet"
    _write_parquet(rows, out)
    return out


def write_daily_equity_curve(
    run_dir: Path,
    equity_curve: list[DailyPortfolioState],
) -> Path | None:
    """Write artifacts/daily_equity_curve.parquet."""
    if not equity_curve:
        return None
    rows = [
        {
            "date": s.date.isoformat(),
            "equity": s.equity,
            "cash_available": s.cash_available,
            "gross_exposure": s.gross_exposure,
            "net_exposure": s.net_exposure,
            "unrealized_pnl": s.unrealized_pnl,
            "realized_pnl": s.realized_pnl,
            "open_position_count": len(s.open_positions),
            "daily_new_risk_used": s.daily_new_risk_used,
            "peak_equity": s.peak_equity,
            "current_drawdown_pct": s.current_drawdown_pct,
            "raw_cash": s.raw_cash,
            "parking_value": s.parking_value,
            "idle_alpha_exposure": s.idle_alpha_exposure,
            "primary_exposure": s.primary_exposure,
        }
        for s in equity_curve
    ]
    out = run_dir / "artifacts" / "daily_equity_curve.parquet"
    _write_parquet(rows, out)
    return out


def write_position_timeline(
    run_dir: Path,
    trades: list[FilledTrade],
    candidate_map: dict[str, Any] | None = None,
    open_positions: list[OpenPosition] | None = None,
) -> Path | None:
    """Write artifacts/position_timeline.parquet (one row per position)."""
    rows = []
    for t in trades:
        rows.append(
                {
                    "position_id": t.position_id,
                    "event_id": t.event_id,
                    "symbol": t.symbol,
                    "source_symbol": t.source_symbol,
                    "event_date": t.event_date.isoformat() if t.event_date else None,
                    "trade_sleeve": _resolve_trade_sleeve(t, candidate_map),
                    "timing_class": t.timing_class,
                    "engine_id": t.engine_id,
                    "entry_timing_policy": t.entry_timing_policy,
                    "trade_symbol_mode": t.trade_symbol_mode,
                    "shadow_only": t.shadow_only,
                    "parent_position_id": t.parent_position_id,
                    "is_add_on": t.is_add_on,
                    "entry_date": t.entry_date.isoformat(),
                    "exit_date": t.exit_date.isoformat(),
                    "entry_price": t.entry_price,
                    "exit_price": t.exit_price,
                "exit_reason": t.exit_reason.value,
                "shares": t.shares,
                "net_pnl": t.net_pnl,
                "r_multiple": t.r_multiple,
                "holding_days": t.holding_days,
                "status": "closed",
            }
        )
    if open_positions:
        for p in open_positions:
            rows.append(
                {
                    "position_id": p.position_id,
                    "event_id": p.plan.candidate.event_id,
                    "symbol": p.plan.candidate.symbol,
                    "source_symbol": p.plan.candidate.source_symbol,
                    "event_date": p.plan.event_date.isoformat() if p.plan.event_date else None,
                    "trade_sleeve": str((p.plan.candidate.features or {}).get("trade_sleeve") or "core"),
                    "timing_class": p.plan.timing_class,
                    "engine_id": p.plan.engine_id,
                    "entry_timing_policy": p.plan.entry_timing_policy,
                    "trade_symbol_mode": p.plan.candidate.trade_symbol_mode,
                    "shadow_only": p.plan.shadow_only,
                    "parent_position_id": p.parent_position_id,
                    "is_add_on": p.is_add_on,
                    "entry_date": p.entry_date.isoformat(),
                    "exit_date": None,
                    "entry_price": p.entry_price,
                    "exit_price": None,
                    "exit_reason": None,
                    "shares": p.shares_total,
                    "net_pnl": None,
                    "r_multiple": None,
                    "holding_days": p.days_held,
                    "status": p.status.value,
                }
            )
    if not rows:
        return None
    out = run_dir / "artifacts" / "position_timeline.parquet"
    _write_parquet(rows, out)
    return out


def write_attribution_by_event_type(
    run_dir: Path,
    trades: list[FilledTrade],
    candidate_map: dict[str, Any],
) -> Path:
    """Write metrics/attribution_by_event_type.csv."""
    bucket_data: dict[str, dict[str, float | int]] = defaultdict(
        lambda: {"count": 0, "wins": 0, "net_pnl": 0.0, "avg_r": 0.0, "_r_sum": 0.0}
    )
    for t in trades:
        cand = candidate_map.get(t.trade_id)
        et = getattr(cand, "event_type", "unknown") if cand else "unknown"
        d = bucket_data[et]
        d["count"] = int(d["count"]) + 1
        if t.net_pnl > 0:
            d["wins"] = int(d["wins"]) + 1
        d["net_pnl"] = float(d["net_pnl"]) + t.net_pnl
        d["_r_sum"] = float(d["_r_sum"]) + t.r_multiple

    out = run_dir / "metrics" / "attribution_by_event_type.csv"
    with open(out, "w", newline="") as f:
        writer = csv.DictWriter(
            f, fieldnames=["event_type", "count", "wins", "win_rate", "net_pnl", "avg_r"]
        )
        writer.writeheader()
        for et, d in sorted(bucket_data.items()):
            count = int(d["count"])
            wins = int(d["wins"])
            writer.writerow(
                {
                    "event_type": et,
                    "count": count,
                    "wins": wins,
                    "win_rate": wins / count if count > 0 else 0.0,
                    "net_pnl": round(float(d["net_pnl"]), 4),
                    "avg_r": round(float(d["_r_sum"]) / count if count > 0 else 0.0, 4),
                }
            )
    return out


def write_attribution_by_sector(
    run_dir: Path,
    trades: list[FilledTrade],
    candidate_map: dict[str, Any],
) -> Path:
    """Write metrics/attribution_by_sector.csv."""
    bucket_data: dict[str, dict[str, float | int]] = defaultdict(
        lambda: {"count": 0, "wins": 0, "net_pnl": 0.0, "_r_sum": 0.0}
    )
    for t in trades:
        cand = candidate_map.get(t.trade_id)
        sector = getattr(cand, "sector", "UNKNOWN") if cand else "UNKNOWN"
        d = bucket_data[sector]
        d["count"] = int(d["count"]) + 1
        if t.net_pnl > 0:
            d["wins"] = int(d["wins"]) + 1
        d["net_pnl"] = float(d["net_pnl"]) + t.net_pnl
        d["_r_sum"] = float(d["_r_sum"]) + t.r_multiple

    out = run_dir / "metrics" / "attribution_by_sector.csv"
    with open(out, "w", newline="") as f:
        writer = csv.DictWriter(
            f, fieldnames=["sector", "count", "wins", "win_rate", "net_pnl", "avg_r"]
        )
        writer.writeheader()
        for sector, d in sorted(bucket_data.items()):
            count = int(d["count"])
            wins = int(d["wins"])
            writer.writerow(
                {
                    "sector": sector,
                    "count": count,
                    "wins": wins,
                    "win_rate": wins / count if count > 0 else 0.0,
                    "net_pnl": round(float(d["net_pnl"]), 4),
                    "avg_r": round(float(d["_r_sum"]) / count if count > 0 else 0.0, 4),
                }
            )
    return out


def write_attribution_by_engine(
    run_dir: Path,
    trades: list[FilledTrade],
    per_engine_metrics: dict[str, dict[str, Any]] | None = None,
) -> Path:
    """Write metrics/attribution_by_engine.csv."""
    bucket_data: dict[str, dict[str, float | int | bool]] = defaultdict(
        lambda: {"count": 0, "wins": 0, "net_pnl": 0.0, "_r_sum": 0.0, "shadow_only": False}
    )
    for t in trades:
        engine_id = t.engine_id or "default"
        d = bucket_data[engine_id]
        d["count"] = int(d["count"]) + 1
        if t.net_pnl > 0:
            d["wins"] = int(d["wins"]) + 1
        d["net_pnl"] = float(d["net_pnl"]) + t.net_pnl
        d["_r_sum"] = float(d["_r_sum"]) + t.r_multiple
        d["shadow_only"] = bool(t.shadow_only)

    if per_engine_metrics:
        for engine_id, summary in per_engine_metrics.items():
            d = bucket_data.setdefault(
                engine_id,
                {"count": 0, "wins": 0, "net_pnl": 0.0, "_r_sum": 0.0, "shadow_only": False},
            )
            d["shadow_only"] = bool(summary.get("shadow_only", d["shadow_only"]))

    out = run_dir / "metrics" / "attribution_by_engine.csv"
    with open(out, "w", newline="") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=[
                "engine_id",
                "shadow_only",
                "count",
                "wins",
                "win_rate",
                "net_pnl",
                "avg_r",
            ],
        )
        writer.writeheader()
        for engine_id, d in sorted(bucket_data.items()):
            count = int(d["count"])
            wins = int(d["wins"])
            writer.writerow(
                {
                    "engine_id": engine_id,
                    "shadow_only": bool(d["shadow_only"]),
                    "count": count,
                    "wins": wins,
                    "win_rate": wins / count if count > 0 else 0.0,
                    "net_pnl": round(float(d["net_pnl"]), 4),
                    "avg_r": round(float(d["_r_sum"]) / count if count > 0 else 0.0, 4),
                }
            )
    return out


def write_score_bucket_report(
    run_dir: Path,
    score_bucket_hit_rate: dict[str, float],
    trades: list[FilledTrade],
    candidate_map: dict[str, Any],
) -> Path:
    """Write metrics/score_bucket_report.csv."""
    bucket_counts: dict[str, int] = defaultdict(int)
    for t in trades:
        cand = candidate_map.get(t.trade_id)
        bucket = getattr(cand, "score_bucket", "unknown") if cand else "unknown"
        bucket_counts[bucket] += 1

    out = run_dir / "metrics" / "score_bucket_report.csv"
    with open(out, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["score_bucket", "trade_count", "win_rate"])
        writer.writeheader()
        for bucket in sorted(set(list(score_bucket_hit_rate.keys()) + list(bucket_counts.keys()))):
            writer.writerow(
                {
                    "score_bucket": bucket,
                    "trade_count": bucket_counts.get(bucket, 0),
                    "win_rate": round(score_bucket_hit_rate.get(bucket, 0.0), 4),
                }
            )
    return out


def write_plots(run_dir: Path, generate: bool = False) -> Path:
    """Create plots directory. generate=True logs a warning (matplotlib not available)."""
    plots_dir = run_dir / "plots"
    plots_dir.mkdir(exist_ok=True)
    if generate:
        logger.warning(
            "plots_not_implemented",
            message="generate_plots=True is a no-op; matplotlib is not in dependencies.",
        )
    return plots_dir


def write_run_notes(run_dir: Path, notes: str = "") -> Path:
    """Write notes/run_notes.md."""
    out = run_dir / "notes" / "run_notes.md"
    out.write_text(notes or "# Run Notes\n\n_No notes provided._\n")
    return out


def write_per_engine_metrics(
    run_dir: Path,
    per_engine_metrics: dict[str, dict[str, Any]],
) -> Path:
    """Write metrics/per_engine_metrics.json."""
    out = run_dir / "metrics" / "per_engine_metrics.json"
    out.write_text(json.dumps(per_engine_metrics, indent=2, default=str))
    return out


def write_sleeve_decomposition(
    run_dir: Path,
    trades: list[FilledTrade],
    equity_curve: list[DailyPortfolioState],
    candidate_map: dict[str, Any],
    config: Any = None,
) -> Path:
    """Write metrics/sleeve_decomposition.json with sleeve breakdown."""
    initial_equity = equity_curve[0].equity if equity_curve else 1.0
    final_equity = equity_curve[-1].equity if equity_curve else initial_equity
    total_return = (final_equity / initial_equity - 1.0) * 100.0

    sleeve_data: dict[str, dict[str, Any]] = {
        "core": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
        "idle_alpha": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
        "form4": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
        "ownership": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
        "risk_off_alpha": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
        "parking": {"trade_count": 0, "wins": 0, "net_pnl": 0.0},
    }
    for t in trades:
        sleeve = _resolve_trade_sleeve(t, candidate_map)
        if sleeve not in sleeve_data:
            sleeve = "core"
        d = sleeve_data[sleeve]
        d["trade_count"] = int(d["trade_count"]) + 1
        d["net_pnl"] = float(d["net_pnl"]) + t.net_pnl
        if t.net_pnl > 0:
            d["wins"] = int(d["wins"]) + 1

    total_pnl = sum(d["net_pnl"] for d in sleeve_data.values())
    result: dict[str, Any] = {}
    for sleeve, d in sleeve_data.items():
        count = int(d["trade_count"])
        wins = int(d["wins"])
        pnl = float(d["net_pnl"])
        result[sleeve] = {
            "trade_count": count,
            "net_pnl": round(pnl, 4),
            "win_rate": round(wins / count, 4) if count > 0 else None,
            "contribution_pct": round(pnl / total_pnl * 100.0, 2) if total_pnl != 0 else 0.0,
        }

    # Compute composite amplification vs core-only estimated return
    core_pnl = float(sleeve_data["core"]["net_pnl"])
    core_only_return = core_pnl / initial_equity * 100.0
    result["composite_amplification"] = (
        round(total_return / core_only_return, 3) if core_only_return != 0 else None
    )
    result["composite_total_return_pct"] = round(total_return, 2)

    # Avg idle fraction from equity curve if available
    idle_vals = [
        ((s.raw_cash or 0.0) + (s.parking_value or 0.0)) / s.equity * 100.0
        for s in equity_curve
        if s.equity > 0 and s.raw_cash is not None
    ]
    result["avg_idle_fraction_pct"] = round(sum(idle_vals) / len(idle_vals), 2) if idle_vals else None

    # IA synergy diagnostics: injected vs blocked engines
    if config is not None and getattr(config, "idle_alpha_sleeve_preset", None):
        from libs.backtest.domain import IDLE_ALPHA_SLEEVE_PRESETS
        preset = IDLE_ALPHA_SLEEVE_PRESETS.get(config.idle_alpha_sleeve_preset, {})
        ia_preset_ids = {e["engine_id"] for e in preset.get("strategy_engines", [])}
        # Only count engines with post_allocation_idle_only=True as "actually injected"
        ia_actual_engine_ids = {
            e.engine_id for e in config.strategy_engines
            if getattr(e, "post_allocation_idle_only", False)
        }
        ia_injected = []
        ia_blocked = []
        for pid in ia_preset_ids:
            if pid in ia_actual_engine_ids:
                ia_injected.append(pid)
            elif f"{pid}__ia_sleeve" in ia_actual_engine_ids:
                ia_injected.append(f"{pid}__ia_sleeve")
            else:
                ia_blocked.append(pid)
        result["ia_engines_injected"] = sorted(ia_injected)
        result["ia_engines_blocked"] = sorted(ia_blocked)
        result["ia_dedup_mode"] = getattr(config, "idle_alpha_dedup_mode", "skip")
        # Per IA engine PnL (only true IA/Phase2 trades)
        ia_pnl: dict[str, float] = {}
        for t in trades:
            if t.engine_id in ia_injected:
                ia_pnl[t.engine_id] = ia_pnl.get(t.engine_id, 0.0) + t.net_pnl
        result["per_ia_engine_pnl"] = {k: round(v, 4) for k, v in sorted(ia_pnl.items())}

    out = run_dir / "metrics" / "sleeve_decomposition.json"
    out.write_text(json.dumps(result, indent=2))
    return out


def write_non_core_allocator_shadow_candidates(
    run_dir: Path,
    rows: list[dict[str, Any]],
) -> Path | None:
    """Write artifacts/non_core_allocator_shadow_candidates.parquet."""
    if not rows:
        return None
    out = run_dir / "artifacts" / "non_core_allocator_shadow_candidates.parquet"
    _write_parquet(rows, out)
    return out


def write_non_core_allocator_shadow_summary(
    run_dir: Path,
    rows: list[dict[str, Any]],
) -> Path | None:
    """Write metrics/non_core_allocator_shadow.json."""
    if not rows:
        return None

    per_sleeve: dict[str, dict[str, Any]] = defaultdict(
        lambda: {
            "candidate_count": 0,
            "complete_count": 0,
            "shadow_selected_count": 0,
            "live_selected_count": 0,
            "disagreement_count": 0,
            "marginal_score_sum": 0.0,
            "shadow_selected_cash": 0.0,
        }
    )
    parking_distribution: dict[str, int] = defaultdict(int)
    decision_distribution: dict[str, int] = defaultdict(int)
    complete_total = 0
    disagreement_total = 0
    for row in rows:
        sleeve = str(row.get("allocator_v2_family") or "unknown")
        d = per_sleeve[sleeve]
        d["candidate_count"] += 1
        complete = bool(row.get("allocator_v2_complete"))
        if complete:
            d["complete_count"] += 1
            complete_total += 1
        marginal_score = float(row.get("allocator_v2_marginal_score") or 0.0)
        d["marginal_score_sum"] += marginal_score
        shadow_selected = bool(row.get("allocator_v2_shadow_selected"))
        live_selected = bool(row.get("allocator_v2_live_selected"))
        if shadow_selected:
            d["shadow_selected_count"] += 1
            d["shadow_selected_cash"] += float(row.get("allocator_v2_requested_cash_est") or 0.0)
        if live_selected:
            d["live_selected_count"] += 1
        if bool(row.get("allocator_v2_live_vs_shadow_disagree")):
            d["disagreement_count"] += 1
            disagreement_total += 1
        parking_symbol = str(row.get("allocator_v2_parking_symbol") or "").strip().lower()
        if parking_symbol:
            parking_distribution[parking_symbol] += 1
        decision = str(row.get("allocator_v2_shadow_decision") or "unknown")
        decision_distribution[decision] += 1

    total_candidates = len(rows)
    summary = {
        "candidate_count": total_candidates,
        "complete_count": complete_total,
        "complete_pct": round(complete_total / total_candidates * 100.0, 2) if total_candidates else None,
        "live_vs_shadow_disagreement_count": disagreement_total,
        "parking_symbol_distribution": dict(sorted(parking_distribution.items())),
        "shadow_decision_distribution": dict(sorted(decision_distribution.items())),
        "sleeves": {},
    }
    for sleeve, d in sorted(per_sleeve.items()):
        count = int(d["candidate_count"])
        summary["sleeves"][sleeve] = {
            "candidate_count": count,
            "complete_count": int(d["complete_count"]),
            "complete_pct": round(int(d["complete_count"]) / count * 100.0, 2) if count else None,
            "shadow_selected_count": int(d["shadow_selected_count"]),
            "live_selected_count": int(d["live_selected_count"]),
            "disagreement_count": int(d["disagreement_count"]),
            "avg_marginal_score": round(float(d["marginal_score_sum"]) / count, 6) if count else None,
            "shadow_selected_cash": round(float(d["shadow_selected_cash"]), 4),
        }

    out = run_dir / "metrics" / "non_core_allocator_shadow.json"
    out.write_text(json.dumps(summary, indent=2))
    return out


def write_non_core_allocator_shadow_report(
    run_dir: Path,
    rows: list[dict[str, Any]],
) -> Path | None:
    """Write notes/non_core_allocator_shadow_report.md."""
    if not rows:
        return None

    summary_path = write_non_core_allocator_shadow_summary(run_dir, rows)
    summary: dict[str, Any] = {}
    if summary_path and summary_path.exists():
        summary = json.loads(summary_path.read_text())

    def _fmt_pct(value: Any) -> str:
        if value is None:
            return "n/a"
        return f"{float(value):.2f}%"

    def _fmt_num(value: Any, digits: int = 2) -> str:
        if value is None:
            return "n/a"
        return f"{float(value):.{digits}f}"

    def _fmt_cash(value: Any) -> str:
        if value is None:
            return "n/a"
        return f"${float(value):,.2f}"

    lines: list[str] = [
        "# Non-Core Allocator Shadow Report",
        "",
        "## Summary",
        f"- Candidates: {summary.get('candidate_count', len(rows))}",
        f"- Complete scoring: {_fmt_pct(summary.get('complete_pct'))}",
        f"- Live vs shadow disagreements: {summary.get('live_vs_shadow_disagreement_count', 0)}",
        "",
        "## Parking Context",
    ]

    parking_distribution = summary.get("parking_symbol_distribution", {}) or {}
    if parking_distribution:
        for symbol, count in sorted(parking_distribution.items()):
            lines.append(f"- `{symbol}`: {count}")
    else:
        lines.append("- No parking benchmark observations recorded.")

    lines.extend(["", "## Sleeve Summary"])
    sleeve_summary = summary.get("sleeves", {}) or {}
    if sleeve_summary:
        lines.extend(
            [
                "",
                "| Sleeve | Candidates | Complete % | Shadow Selected | Live Selected | Disagreements | Avg Score | Shadow Cash |",
                "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
            ]
        )
        for sleeve, data in sorted(sleeve_summary.items()):
            lines.append(
                "| {sleeve} | {candidate_count} | {complete_pct} | {shadow_selected_count} | "
                "{live_selected_count} | {disagreement_count} | {avg_score} | {shadow_cash} |".format(
                    sleeve=sleeve,
                    candidate_count=int(data.get("candidate_count", 0)),
                    complete_pct=_fmt_pct(data.get("complete_pct")),
                    shadow_selected_count=int(data.get("shadow_selected_count", 0)),
                    live_selected_count=int(data.get("live_selected_count", 0)),
                    disagreement_count=int(data.get("disagreement_count", 0)),
                    avg_score=_fmt_num(data.get("avg_marginal_score"), 4),
                    shadow_cash=_fmt_cash(data.get("shadow_selected_cash")),
                )
            )
    else:
        lines.append("")
        lines.append("- No per-sleeve summary available.")

    def _candidate_sort_key(row: dict[str, Any]) -> tuple[float, float, str, str]:
        return (
            -float(row.get("allocator_v2_marginal_score") or 0.0),
            -float(row.get("allocator_v2_requested_cash_est") or 0.0),
            str(row.get("date") or ""),
            str(row.get("symbol") or ""),
        )

    shadow_only = sorted(
        [
            row for row in rows
            if bool(row.get("allocator_v2_shadow_selected"))
            and not bool(row.get("allocator_v2_live_selected"))
        ],
        key=_candidate_sort_key,
    )
    live_only = sorted(
        [
            row for row in rows
            if bool(row.get("allocator_v2_live_selected"))
            and not bool(row.get("allocator_v2_shadow_selected"))
        ],
        key=_candidate_sort_key,
    )
    blocked_budget = sorted(
        [row for row in rows if str(row.get("allocator_v2_shadow_decision") or "") == "blocked_by_budget"],
        key=_candidate_sort_key,
    )
    blocked_better = sorted(
        [row for row in rows if str(row.get("allocator_v2_shadow_decision") or "") == "blocked_by_better_opportunity"],
        key=_candidate_sort_key,
    )

    def _append_candidate_section(title: str, candidate_rows: list[dict[str, Any]], *, limit: int = 15) -> None:
        lines.extend(["", f"## {title}"])
        if not candidate_rows:
            lines.append("")
            lines.append("- None.")
            return
        lines.extend(
            [
                "",
                "| Date | Sleeve | Symbol | Engine | Score | Cash Est | Parking | Decision |",
                "| --- | --- | --- | --- | ---: | ---: | --- | --- |",
            ]
        )
        for row in candidate_rows[:limit]:
            lines.append(
                "| {date} | {sleeve} | `{symbol}` | `{engine}` | {score} | {cash} | `{parking}` | `{decision}` |".format(
                    date=str(row.get("date") or ""),
                    sleeve=str(row.get("allocator_v2_family") or ""),
                    symbol=str(row.get("symbol") or ""),
                    engine=str(row.get("engine_id") or ""),
                    score=_fmt_num(row.get("allocator_v2_marginal_score"), 4),
                    cash=_fmt_cash(row.get("allocator_v2_requested_cash_est")),
                    parking=str(row.get("allocator_v2_parking_symbol") or ""),
                    decision=str(row.get("allocator_v2_shadow_decision") or ""),
                )
            )

    _append_candidate_section("Top Shadow-Selected But Live-Skipped", shadow_only)
    _append_candidate_section("Top Live-Selected But Shadow-Skipped", live_only)
    _append_candidate_section("Top Budget-Blocked Candidates", blocked_budget, limit=10)
    _append_candidate_section("Top Better-Opportunity Blocks", blocked_better, limit=10)

    disagreement_by_date: dict[str, int] = defaultdict(int)
    for row in rows:
        if bool(row.get("allocator_v2_live_vs_shadow_disagree")):
            disagreement_by_date[str(row.get("date") or "")] += 1
    lines.extend(["", "## Disagreement Hotspots"])
    if disagreement_by_date:
        lines.extend(["", "| Date | Disagreements |", "| --- | ---: |"])
        for date, count in sorted(disagreement_by_date.items(), key=lambda item: (-item[1], item[0]))[:15]:
            lines.append(f"| {date} | {count} |")
    else:
        lines.append("")
        lines.append("- No disagreement dates.")

    out = run_dir / "notes" / "non_core_allocator_shadow_report.md"
    out.write_text("\n".join(lines) + "\n")
    return out


def write_all_artifacts(
    run_dir: Path,
    run_id: str,
    manifest: ExperimentManifest,
    config: BacktestConfig,
    metrics: MetricsBundle,
    trades: list[FilledTrade],
    equity_curve: list[DailyPortfolioState],
    open_positions: list[OpenPosition],
    candidate_map: dict[str, Any],
    started_at: dt.datetime,
    finished_at: dt.datetime,
    git_hash: str,
    total_trading_days: int,
    total_candidates_seen: int,
    total_orders_rejected: int,
    split_name: str | None = None,
    per_engine_metrics: dict[str, dict[str, Any]] | None = None,
    non_core_allocator_shadow_rows: list[dict[str, Any]] | None = None,
) -> dict[str, str]:
    """Write all output files. Returns mapping of artifact_name → file_path."""
    from libs.backtest.manifests import save_manifest, save_resolved_config

    paths: dict[str, str] = {}

    # Core files
    paths["manifest"] = str(save_manifest(manifest, run_dir))
    paths["resolved_config"] = str(save_resolved_config(config, run_dir))
    paths.update(write_snapshot_provenance(run_dir, config))
    paths["metadata"] = str(
        write_metadata(
            run_dir, run_id, started_at, finished_at, git_hash,
            total_trading_days, total_candidates_seen, total_orders_rejected,
            requested_snapshot_id=config.requested_snapshot_id or config.dataset_snapshot_id,
            canonical_snapshot_id=config.canonical_snapshot_id or config.dataset_snapshot_id,
            split_name=split_name,
        )
    )

    # Metrics
    if config.reporting.write_metrics_summary:
        paths["metrics_summary"] = str(write_metrics_summary(run_dir, metrics))
    paths["attribution_by_event_type"] = str(
        write_attribution_by_event_type(run_dir, trades, candidate_map)
    )
    paths["attribution_by_sector"] = str(
        write_attribution_by_sector(run_dir, trades, candidate_map)
    )
    paths["attribution_by_engine"] = str(
        write_attribution_by_engine(run_dir, trades, per_engine_metrics)
    )
    paths["score_bucket_report"] = str(
        write_score_bucket_report(
            run_dir, metrics.score_bucket_hit_rate, trades, candidate_map
        )
    )
    if per_engine_metrics:
        paths["per_engine_metrics"] = str(write_per_engine_metrics(run_dir, per_engine_metrics))

    # Trade data
    if config.reporting.write_trade_blotter:
        p = write_trade_blotter(run_dir, trades, candidate_map)
        if p:
            paths["trade_blotter"] = str(p)

    if config.reporting.write_equity_curve:
        p = write_daily_equity_curve(run_dir, equity_curve)
        if p:
            paths["daily_equity_curve"] = str(p)

    p = write_position_timeline(run_dir, trades, candidate_map, open_positions)
    if p:
        paths["position_timeline"] = str(p)

    paths["sleeve_decomposition"] = str(
        write_sleeve_decomposition(run_dir, trades, equity_curve, candidate_map, config)
    )
    if non_core_allocator_shadow_rows:
        p = write_non_core_allocator_shadow_candidates(run_dir, non_core_allocator_shadow_rows)
        if p:
            paths["non_core_allocator_shadow_candidates"] = str(p)
        p = write_non_core_allocator_shadow_summary(run_dir, non_core_allocator_shadow_rows)
        if p:
            paths["non_core_allocator_shadow"] = str(p)
        p = write_non_core_allocator_shadow_report(run_dir, non_core_allocator_shadow_rows)
        if p:
            paths["non_core_allocator_shadow_report"] = str(p)

    # Plots (no-op)
    paths["plots_dir"] = str(write_plots(run_dir, config.reporting.generate_plots))

    # Notes
    paths["run_notes"] = str(write_run_notes(run_dir, manifest.notes or ""))

    return paths


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _write_parquet(rows: list[dict[str, Any]], path: Path) -> None:
    """Write list of row dicts to Parquet."""
    if not rows:
        return
    keys = list(rows[0].keys())
    columns: dict[str, list[Any]] = {k: [] for k in keys}
    for row in rows:
        for k in keys:
            columns[k].append(row.get(k))
    table = pa.table({k: pa.array(v) for k, v in columns.items()})
    pq.write_table(table, str(path))
    logger.debug("parquet_written", path=str(path), rows=len(rows))