ironcondor 0.5.0

//! The `#[pyclass] Bundle` handle and its lazy DataFrame accessors, plus
//! `load_bundle`.
//!
//! A [`Bundle`] is always **backed by an on-disk directory** (`bundle.path`) —
//! the finalized `<output_dir>/<run_id>/` `run()` published, or an existing
//! bundle opened by [`load_bundle`] ([docs/06 §6](../../docs/06-python-bindings.md)).
//! It carries no decoded data: the accessors are **lazy**, reading the on-disk
//! Parquet / JSON on call, so a caller who only wants [`Bundle::metrics`] never
//! decodes the four tables.
//!
//! # DataFrames via pandas (an optional soft dependency)
//!
//! The four table accessors return **pandas DataFrames** built by calling
//! `pandas.read_parquet(<table path>)` on demand — the on-disk format is plain
//! Parquet, so a `polars` / `pyarrow` user can also read `bundle.path` directly
//! and skip the accessors entirely ([ADR-0004](../../docs/adr/0004-parquet-result-bundle.md)).
//! `pandas` (and its `pyarrow` engine) is an **optional extra**
//! (`pip install ironcondor[pandas]`), not a hard wheel dependency, so the core
//! `run()` / `load_bundle` / `metrics()` surface works without it; a missing
//! pandas raises a clear `ImportError` naming the extra.
//!
//! # `metrics()` reads only the manifest
//!
//! [`Bundle::metrics`] parses the manifest JSON `metrics` object into a Python
//! `dict` — it reads `manifest.json` only, never the Parquet tables.
//!
//! # `load_bundle` validates through the hardened reader
//!
//! [`load_bundle`] validates the directory through [`crate::read_bundle`] — the
//! read-back **security gate** (schema tag, manifest fields, table schemas,
//! `row_counts` cross-check, `contract_id` round-trip, resource ceilings) — so a
//! hostile directory fails typed **before** a handle is returned; the accessors
//! then still read lazily from disk.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use pyo3::exceptions::PyImportError;
use pyo3::prelude::*;
use pyo3::types::PyModule;

use super::errors::{guard_boundary, to_pyerr};
use crate::ResourceLimits;
use crate::bundle::read_bundle;
use crate::error::BacktestError;

/// A bundle read-back / copy I/O failure, routed through the single mapping seam
/// so it surfaces as `ic.BundleError` (⊂ `IOError` ⊂ `ic.IronCondorError`) —
/// consistent with `load_bundle`, so `except ic.IronCondorError` around any
/// accessor catches a missing/corrupt manifest or a missing table.
fn bundle_err(py: Python<'_>, message: String) -> PyErr {
    to_pyerr(py, BacktestError::Bundle(message))
}

/// A handle to a finalized on-disk result bundle directory.
///
/// Construct one from [`run()`](super::run::run) or [`load_bundle`]; the
/// accessors read the on-disk Parquet / JSON lazily on each call.
#[pyclass(name = "Bundle", module = "ironcondor")]
pub struct Bundle {
    /// The finalized bundle directory (`<output_dir>/<run_id>/`).
    dir: PathBuf,
}

impl Bundle {
    /// Wrap a finalized bundle directory (used by `run()` and `load_bundle`).
    #[must_use]
    pub(crate) fn from_dir(dir: PathBuf) -> Self {
        Self { dir }
    }
}

#[pymethods]
impl Bundle {
    /// The on-disk bundle directory `run()` published (or `load_bundle` opened).
    #[getter]
    fn path(&self) -> String {
        self.dir.display().to_string()
    }

    /// `fills.parquet` as a `pandas.DataFrame` (columns exactly the frozen
    /// schema). Lazily read on call.
    ///
    /// # Errors
    ///
    /// Raises `ImportError` if pandas is not installed, or `ic.BundleError` (⊂
    /// `IOError`) if the table is missing or unreadable.
    fn fills<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        self.read_table(py, "fills.parquet")
    }

    /// `equity_curve.parquet` as a `pandas.DataFrame`. Lazily read on call.
    ///
    /// # Errors
    ///
    /// Raises `ImportError` if pandas is not installed, or `ic.BundleError` (⊂
    /// `IOError`) if the table is missing or unreadable.
    fn equity_curve<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        self.read_table(py, "equity_curve.parquet")
    }

    /// `positions.parquet` as a `pandas.DataFrame`. Lazily read on call.
    ///
    /// # Errors
    ///
    /// Raises `ImportError` if pandas is not installed, or `ic.BundleError` (⊂
    /// `IOError`) if the table is missing or unreadable.
    fn positions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        self.read_table(py, "positions.parquet")
    }

    /// `greeks_attribution.parquet` as a `pandas.DataFrame`. Lazily read on call.
    ///
    /// # Errors
    ///
    /// Raises `ImportError` if pandas is not installed, or `ic.BundleError` (⊂
    /// `IOError`) if the table is missing or unreadable.
    fn greeks_attribution<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        self.read_table(py, "greeks_attribution.parquet")
    }

    /// The manifest's `metrics` object as a `dict` (Sharpe, max drawdown,
    /// per-leg, …). Reads `manifest.json` only, never the Parquet tables.
    ///
    /// # Errors
    ///
    /// Raises `ic.BundleError` (⊂ `IOError`) if `manifest.json` is missing,
    /// unreadable, not JSON, or carries no `metrics` object.
    fn metrics<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        guard_boundary(|| {
            // Re-read the manifest bounded by the default resource ceiling. The
            // handle retains only its directory, not the `ResourceLimits` used
            // at load, so a directory swapped under a live handle after load
            // cannot force an unbounded allocation here (F34) — the filesystem
            // size is rejected before the read, then `Read::take` caps it.
            let limits = ResourceLimits::default();
            let buf = self.read_manifest_bounded(py, &limits)?;
            let value: serde_json::Value = serde_json::from_slice(&buf)
                .map_err(|e| bundle_err(py, format!("manifest.json is not valid JSON: {e}")))?;
            let metrics = value
                .get("metrics")
                .ok_or_else(|| bundle_err(py, "manifest.json has no metrics object".to_string()))?;
            let metrics_json = serde_json::to_string(metrics)
                .map_err(|e| bundle_err(py, format!("cannot re-encode manifest metrics: {e}")))?;
            // Parse into native Python objects via the stdlib json module — a
            // metrics object becomes a plain dict.
            let json = py.import("json")?;
            json.call_method1("loads", (metrics_json,))
        })
    }

    /// **Deprecated-by-design copy alias.** `run()` already wrote the bundle;
    /// `write(dir)` only **copies** the finalized directory to `dir` (honouring
    /// `overwrite`) — it never re-runs the engine
    /// ([docs/06 §6](../../docs/06-python-bindings.md)). Returns a [`Bundle`]
    /// handle to the copy.
    ///
    /// # Errors
    ///
    /// Raises `ic.BundleError` (⊂ `IOError`) if `dir` already exists and
    /// `overwrite` is false, or on any copy I/O failure.
    #[pyo3(signature = (dir, overwrite = false))]
    fn write(&self, py: Python<'_>, dir: String, overwrite: bool) -> PyResult<Self> {
        guard_boundary(|| {
            // Emit a DeprecationWarning so callers migrate off the copy alias.
            emit_deprecation(
                py,
                "Bundle.write() is a deprecated copy alias: run() already wrote the bundle at \
                 bundle.path; write(dir) only copies it and never re-runs the engine.",
            )?;

            let dest = PathBuf::from(&dir);

            // Refuse to copy a bundle onto itself, or into a directory that
            // CONTAINS it: canonicalise both and reject BEFORE any deletion. The
            // old path deleted the destination then copied from `self.dir`, so
            // `dest == bundle.path` (or an ancestor) irreversibly destroyed the
            // very bundle the copy was meant to preserve (F35).
            let source = self.dir.canonicalize().map_err(|e| {
                bundle_err(
                    py,
                    format!("cannot resolve bundle {}: {e}", self.dir.display()),
                )
            })?;
            if let Some(dest_canon) = canonicalize_dest(&dest)
                && source.starts_with(&dest_canon)
            {
                return Err(bundle_err(
                    py,
                    format!(
                        "refusing to write bundle onto itself: destination {dir} resolves to the \
                         source bundle {} (or a directory containing it)",
                        self.dir.display()
                    ),
                ));
            }

            let dest_exists = dest
                .try_exists()
                .map_err(|e| bundle_err(py, format!("cannot stat {dir}: {e}")))?;
            if dest_exists && !overwrite {
                return Err(bundle_err(
                    py,
                    format!("destination {dir} already exists (pass overwrite=True to replace it)"),
                ));
            }

            // Stage the copy into a sibling temp directory, then publish it with
            // an atomic rename — a reader never sees a half-populated
            // destination, and the copy reads only the source, never a live
            // destination (mirrors the Rust writer's atomic publish).
            let parent = dest
                .parent()
                .filter(|p| !p.as_os_str().is_empty())
                .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
            std::fs::create_dir_all(&parent)
                .map_err(|e| bundle_err(py, format!("cannot create {}: {e}", parent.display())))?;
            let staging = unique_sibling(&parent, &dest, "stage");
            std::fs::create_dir(&staging).map_err(|e| {
                bundle_err(
                    py,
                    format!("cannot create staging dir {}: {e}", staging.display()),
                )
            })?;
            if let Err(err) = self.copy_files_into(py, &staging) {
                let _ = std::fs::remove_dir_all(&staging);
                return Err(err);
            }

            // Publish atomically. On overwrite, move the existing destination
            // aside to a sibling backup first, swap, then drop the backup — a
            // failure leaves either the old or the new bundle, never a partial
            // one; on failure the source (untouched throughout) always survives.
            if dest_exists {
                let backup = unique_sibling(&parent, &dest, "backup");
                if let Err(e) = std::fs::rename(&dest, &backup) {
                    let _ = std::fs::remove_dir_all(&staging);
                    return Err(bundle_err(py, format!("cannot replace {dir}: {e}")));
                }
                if let Err(e) = std::fs::rename(&staging, &dest) {
                    let _ = std::fs::rename(&backup, &dest);
                    let _ = std::fs::remove_dir_all(&staging);
                    return Err(bundle_err(py, format!("cannot publish copy to {dir}: {e}")));
                }
                let _ = std::fs::remove_dir_all(&backup);
            } else if let Err(e) = std::fs::rename(&staging, &dest) {
                let _ = std::fs::remove_dir_all(&staging);
                return Err(bundle_err(py, format!("cannot publish copy to {dir}: {e}")));
            }

            Ok(Self::from_dir(dest))
        })
    }
}

impl Bundle {
    /// Read `manifest.json` into memory, **bounded by** `limits.max_manifest_bytes`.
    ///
    /// The filesystem size is rejected before a byte is read, then the read
    /// itself is capped with [`std::io::Read::take`], so a directory swapped
    /// under a live handle after load cannot drive an unbounded allocation
    /// (F34). Mirrors the hardened reader's `read_manifest` bounded read
    /// (`src/bundle/reader.rs`).
    ///
    /// A crossed ceiling is a [`BacktestError::TapeTooLarge`] (→ `ic.DataError`),
    /// consistent with the read-back gate; any other I/O failure is
    /// `ic.BundleError`.
    fn read_manifest_bounded(&self, py: Python<'_>, limits: &ResourceLimits) -> PyResult<Vec<u8>> {
        use std::io::Read as _;
        let path = self.dir.join("manifest.json");
        let meta = std::fs::metadata(&path)
            .map_err(|e| bundle_err(py, format!("cannot stat {}: {e}", path.display())))?;
        if !meta.is_file() {
            return Err(bundle_err(
                py,
                format!("{} is not a regular file", path.display()),
            ));
        }
        let len = meta.len();
        if len > limits.max_manifest_bytes {
            return Err(to_pyerr(
                py,
                BacktestError::TapeTooLarge {
                    limit: "max_manifest_bytes",
                    value: len,
                    cap: limits.max_manifest_bytes,
                },
            ));
        }
        let file = std::fs::File::open(&path)
            .map_err(|e| bundle_err(py, format!("cannot open {}: {e}", path.display())))?;
        let cap = usize::try_from(len).unwrap_or(0);
        let mut buf = Vec::with_capacity(cap);
        file.take(limits.max_manifest_bytes)
            .read_to_end(&mut buf)
            .map_err(|e| bundle_err(py, format!("cannot read {}: {e}", path.display())))?;
        Ok(buf)
    }

    /// Copy every regular file in the finalized bundle directory into `dest_dir`
    /// (a flat directory — manifest + the four Parquet tables, no
    /// subdirectories). Reads only `self.dir`; never touches `dest_dir`'s
    /// siblings.
    fn copy_files_into(&self, py: Python<'_>, dest_dir: &Path) -> PyResult<()> {
        let entries = std::fs::read_dir(&self.dir).map_err(|e| {
            bundle_err(
                py,
                format!("cannot read bundle {}: {e}", self.dir.display()),
            )
        })?;
        for entry in entries {
            let entry =
                entry.map_err(|e| bundle_err(py, format!("cannot read bundle entry: {e}")))?;
            let file_type = entry
                .file_type()
                .map_err(|e| bundle_err(py, format!("cannot stat bundle entry: {e}")))?;
            if file_type.is_file() {
                let target = dest_dir.join(entry.file_name());
                std::fs::copy(entry.path(), &target).map_err(|e| {
                    bundle_err(py, format!("cannot copy {:?}: {e}", entry.file_name()))
                })?;
            }
        }
        Ok(())
    }

    /// Read one bundle table into a `pandas.DataFrame` via `pandas.read_parquet`.
    ///
    /// Shared by the four table accessors; wrapped in [`guard_boundary`] so an
    /// unexpected panic (e.g. from within pandas) becomes `ic.EngineError`
    /// rather than crossing the FFI boundary.
    fn read_table<'py>(&self, py: Python<'py>, file: &str) -> PyResult<Bound<'py, PyAny>> {
        guard_boundary(|| {
            let path = self.dir.join(file);
            if !path.is_file() {
                return Err(bundle_err(
                    py,
                    format!("bundle table {} is missing", path.display()),
                ));
            }
            let path_str = path.to_str().ok_or_else(|| {
                bundle_err(
                    py,
                    format!("bundle path {} is not valid UTF-8", path.display()),
                )
            })?;
            // A missing pandas is a missing *optional dependency*, so it stays an
            // `ImportError` (not `ic.BundleError`) — install the extra, don't
            // treat it as a corrupt bundle.
            let pandas = import_pandas(py)?;
            pandas.call_method1("read_parquet", (path_str,))
        })
    }
}

/// Open an existing on-disk bundle in `dir`, validating it through the hardened
/// read-back gate ([`crate::read_bundle`]) before returning a handle.
///
/// The full validation (schema tag, manifest fields, table schemas, `row_counts`
/// cross-check, `contract_id` round-trip, resource ceilings) runs with the GIL
/// released; on success the accessors still read the tables lazily from disk.
///
/// # Errors
///
/// Raises `ic.BundleError` (⊂ `IOError`) if the directory is not a valid
/// `ironcondor.bundle.v1` bundle (a wrong schema tag, a malformed manifest /
/// table, a non-round-trippable `contract_id`, …) and `ic.DataError` (⊂
/// `IOError`) for a crossed resource ceiling — both descend from
/// `ic.IronCondorError`.
#[pyfunction]
pub fn load_bundle(py: Python<'_>, dir: String) -> PyResult<Bundle> {
    // `guard_boundary` outside the `detach` region: a panic in the GIL-released
    // reader is caught in Rust (GIL re-attached) before any Python re-entry.
    guard_boundary(|| {
        let path = PathBuf::from(&dir);
        let limits = ResourceLimits::default();
        // Validate (decode + check) with the GIL released (`detach`, the pyo3
        // 0.29 name for `allow_threads`) — pure-Rust I/O.
        py.detach(|| read_bundle(&path, &limits))
            .map_err(|e| to_pyerr(py, e))?;
        Ok(Bundle::from_dir(path))
    })
}

/// Import `pandas`, mapping a missing install to a clear `ImportError` naming the
/// optional extra.
fn import_pandas(py: Python<'_>) -> PyResult<Bound<'_, PyModule>> {
    py.import("pandas").map_err(|_| {
        PyImportError::new_err(
            "pandas is required for Bundle DataFrame accessors; install the optional extra with \
             `pip install ironcondor[pandas]` (pulls pandas + pyarrow). The bundle is plain \
             Parquet + JSON at bundle.path, so polars / pyarrow can read it directly instead.",
        )
    })
}

/// Resolve `dest` to a canonical path for the same-path identity check (F35),
/// tolerating a not-yet-existing destination by canonicalising its existing
/// parent and re-joining the final component. Returns `None` when neither `dest`
/// nor its parent resolves — then it cannot alias the always-existing source.
fn canonicalize_dest(dest: &Path) -> Option<PathBuf> {
    if let Ok(canon) = dest.canonicalize() {
        return Some(canon);
    }
    let parent = dest.parent()?;
    let name = dest.file_name()?;
    let parent = if parent.as_os_str().is_empty() {
        Path::new(".")
    } else {
        parent
    };
    Some(parent.canonicalize().ok()?.join(name))
}

/// A unique hidden sibling path under `parent` for atomic staging / backup
/// (F35). The process id plus a monotonic counter keep concurrent `write()`
/// calls collision-free without reading a wall clock.
fn unique_sibling(parent: &Path, dest: &Path, tag: &str) -> PathBuf {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let base = dest
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("bundle");
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    let pid = std::process::id();
    parent.join(format!(".{base}.{tag}.{pid}.{n}.tmp"))
}

/// Emit a Python `DeprecationWarning` with `message`.
fn emit_deprecation(py: Python<'_>, message: &str) -> PyResult<()> {
    let warnings = py.import("warnings")?;
    let category = py.get_type::<pyo3::exceptions::PyDeprecationWarning>();
    warnings.call_method1("warn", (message, category))?;
    Ok(())
}