fp_types/

lib.rs

#![forbid(unsafe_code)]
#![warn(rustdoc::broken_intra_doc_links)]

//! Foundational value-type abstractions for **frankenpandas** — the
//! enums, structs, and free functions that every other crate
//! (fp-columnar, fp-index, fp-frame, fp-io, ...) consumes when
//! representing scalar data, dtypes, missing values, and time deltas.
//!
//! The types here intentionally stay tiny and dependency-light
//! (`serde`, `thiserror`) so they can sit at the bottom of the
//! workspace dep graph.
//!
//! ## Core value types
//!
//! - [`DType`]: the dtype enum — `Null`, `Bool`, `Int64`, `Float64`,
//!   `Utf8`, `Categorical`, `Timedelta64`, `Datetime64`, `Period`,
//!   `Interval`, `Sparse`. Drives column / series storage decisions
//!   across the workspace.
//! - [`Scalar`]: the per-cell value enum, parameterized by `DType`.
//!   Each variant holds the actual data (`Int64(i64)`, `Float64(f64)`,
//!   `Utf8(String)`, ...) plus the `Null(NullKind)` variant for
//!   missing values.
//! - [`NullKind`]: distinguishes the three pandas missing-value
//!   "flavors" — `Null` (Python `None` / SQL NULL), `NaN`
//!   (floating-point not-a-number), `NaT` (timedelta / datetime
//!   not-a-time). `Scalar::Null(...)` carries the kind so downstream
//!   code can preserve pandas semantics.
//! - [`SparseDType`]: descriptor for sparse-encoded dtypes (paired
//!   value dtype + fill value).
//!
//! ## Time / duration types
//!
//! - [`Timedelta`]: nanosecond-precision duration with arithmetic
//!   helpers ([`Timedelta::add`], [`Timedelta::sub`],
//!   [`Timedelta::mul_scalar`], [`Timedelta::div_scalar`],
//!   [`Timedelta::div_timedelta`]) that propagate `NaT` per pandas
//!   semantics. [`TimedeltaComponents`] breaks a timedelta into
//!   days/hours/minutes/seconds/nanos for display.
//! - [`Timestamp`]: nanosecond-precision wall-clock timestamp with
//!   optional timezone. Includes floor / ceil / round helpers and
//!   `NaT` propagation.
//!
//! ## Dtype inference + casting
//!
//! - [`infer_dtype`]: derive a [`DType`] from a slice of scalars
//!   (used during DataFrame construction).
//! - [`common_dtype`]: pandas-style dtype promotion for binary ops.
//! - [`cast_scalar`] / [`cast_scalar_owned`]: convert a scalar to a
//!   target dtype with explicit error reporting on impossible casts.
//!
//! ## Missing-value helpers
//!
//! Free fns matching `pd.isna` / `pd.notna` / `pd.fillna` / `pd.dropna`
//! plus the `nan*` aggregations ([`nansum`], [`nanmean`], [`nancount`],
//! [`nanmin`], [`nanmax`], [`nanmedian`], [`nanvar`], [`nanstd`])
//! that mirror pandas' missing-aware reductions.
//!
//! ## Error reporting
//!
//! Errors are explicit enums via `thiserror`: [`TypeError`] for
//! dtype-related failures (incompatible-cast, no-common-dtype) and
//! [`TimedeltaError`] for parse failures.

use serde::{Deserialize, Serialize};
use thiserror::Error;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DType {
    Null,
    Bool,
    /// Nullable boolean extension dtype. Matches pandas `BooleanDtype()`.
    #[serde(rename = "boolean")]
    BoolNullable,
    Int64,
    /// Nullable Int64 extension dtype. Matches pandas `Int64Dtype()` / `pd.NA`.
    #[serde(rename = "Int64")]
    Int64Nullable,
    Float64,
    #[serde(alias = "string", alias = "str")]
    Utf8,
    Categorical,
    Timedelta64,
    /// Nanosecond-precision datetime since Unix epoch. Matches pandas `datetime64[ns]`.
    Datetime64,
    /// Period ordinal. Matches pandas `period[freq]`. Stores ordinal + frequency code.
    Period,
    /// Numeric interval value. Matches pandas `interval[float64]`.
    Interval,
    Sparse,
}

impl DType {
    /// Returns true if the dtype is numeric (integer or floating point).
    #[must_use]
    pub const fn is_numeric(&self) -> bool {
        matches!(self, Self::Int64 | Self::Int64Nullable | Self::Float64)
    }

    /// Returns true if the dtype is an integer type.
    #[must_use]
    pub const fn is_integer(&self) -> bool {
        matches!(self, Self::Int64 | Self::Int64Nullable)
    }

    /// Returns true if the dtype is a floating point type.
    #[must_use]
    pub const fn is_floating(&self) -> bool {
        matches!(self, Self::Float64)
    }

    /// Returns true if the dtype is boolean.
    #[must_use]
    pub const fn is_bool(&self) -> bool {
        matches!(self, Self::Bool | Self::BoolNullable)
    }

    /// Returns true if the dtype is object/string type.
    #[must_use]
    pub const fn is_object(&self) -> bool {
        matches!(self, Self::Utf8)
    }

    /// Returns true if the dtype is datetime.
    #[must_use]
    pub const fn is_datetime(&self) -> bool {
        matches!(self, Self::Datetime64)
    }

    /// Returns true if the dtype is timedelta.
    #[must_use]
    pub const fn is_timedelta(&self) -> bool {
        matches!(self, Self::Timedelta64)
    }

    /// Returns true if the dtype is categorical.
    #[must_use]
    pub const fn is_categorical(&self) -> bool {
        matches!(self, Self::Categorical)
    }

    /// Returns true if the dtype is sparse.
    #[must_use]
    pub const fn is_sparse(&self) -> bool {
        matches!(self, Self::Sparse)
    }

    /// Returns true if the dtype is period.
    #[must_use]
    pub const fn is_period(&self) -> bool {
        matches!(self, Self::Period)
    }

    /// Returns true if the dtype is interval.
    #[must_use]
    pub const fn is_interval(&self) -> bool {
        matches!(self, Self::Interval)
    }

    /// Return the dtype name as a string.
    ///
    /// Matches numpy dtype.name property.
    #[must_use]
    pub const fn name(&self) -> &'static str {
        match self {
            Self::Bool => "bool",
            Self::BoolNullable => "boolean",
            Self::Int64 => "int64",
            Self::Int64Nullable => "Int64",
            Self::Float64 => "float64",
            Self::Utf8 => "object",
            Self::Datetime64 => "datetime64[ns]",
            Self::Timedelta64 => "timedelta64[ns]",
            Self::Categorical => "category",
            Self::Period => "period",
            Self::Interval => "interval",
            Self::Sparse => "Sparse",
            Self::Null => "object",
        }
    }

    /// Return the dtype kind character.
    ///
    /// Matches numpy dtype.kind property.
    #[must_use]
    pub const fn kind(&self) -> char {
        match self {
            Self::Bool | Self::BoolNullable => 'b',
            Self::Int64 | Self::Int64Nullable => 'i',
            Self::Float64 => 'f',
            Self::Utf8 => 'O',
            Self::Datetime64 => 'M',
            Self::Timedelta64 => 'm',
            Self::Categorical => 'O',
            Self::Period => 'O',
            Self::Interval => 'O',
            Self::Sparse => 'O',
            Self::Null => 'O',
        }
    }

    /// Return the dtype itemsize in bytes.
    ///
    /// Matches numpy dtype.itemsize property.
    #[must_use]
    pub const fn itemsize(&self) -> usize {
        match self {
            Self::Bool | Self::BoolNullable => 1,
            Self::Int64
            | Self::Int64Nullable
            | Self::Float64
            | Self::Datetime64
            | Self::Timedelta64
            | Self::Period => 8,
            Self::Utf8 | Self::Categorical | Self::Interval | Self::Sparse | Self::Null => 8,
        }
    }

    /// Returns true if this is an extension dtype (categorical, sparse, period, interval, nullable).
    ///
    /// Matches `pd.api.types.is_extension_array_dtype()`.
    #[must_use]
    pub const fn is_extension(&self) -> bool {
        matches!(
            self,
            Self::Categorical
                | Self::Sparse
                | Self::Period
                | Self::Interval
                | Self::Int64Nullable
                | Self::BoolNullable
        )
    }

    /// Returns true if this is a nullable extension dtype (Int64, boolean).
    ///
    /// Nullable extension dtypes preserve their dtype when nulls are introduced,
    /// unlike numpy dtypes which promote to float64.
    #[must_use]
    pub const fn is_nullable(&self) -> bool {
        matches!(self, Self::Int64Nullable | Self::BoolNullable)
    }

    /// Returns the non-nullable equivalent dtype.
    ///
    /// For nullable extension dtypes, returns the numpy equivalent.
    /// For non-nullable dtypes, returns self.
    #[must_use]
    pub const fn to_non_nullable(&self) -> Self {
        match self {
            Self::Int64Nullable => Self::Int64,
            Self::BoolNullable => Self::Bool,
            other => *other,
        }
    }

    /// Returns the nullable equivalent dtype.
    ///
    /// For numpy int64/bool, returns the nullable extension dtype.
    /// For already-nullable or other dtypes, returns self.
    #[must_use]
    pub const fn to_nullable(&self) -> Self {
        match self {
            Self::Int64 => Self::Int64Nullable,
            Self::Bool => Self::BoolNullable,
            other => *other,
        }
    }

    /// Returns true if this is a signed integer type.
    ///
    /// Matches `pd.api.types.is_signed_integer_dtype()`.
    #[must_use]
    pub const fn is_signed_integer(&self) -> bool {
        matches!(self, Self::Int64 | Self::Int64Nullable)
    }

    /// Returns true if this is a string/object dtype.
    ///
    /// Matches `pd.api.types.is_string_dtype()`.
    #[must_use]
    pub const fn is_string_dtype(&self) -> bool {
        matches!(self, Self::Utf8)
    }

    /// Returns true for any real numeric dtype (integer or float).
    ///
    /// Matches `pd.api.types.is_any_real_numeric_dtype()`.
    #[must_use]
    pub const fn is_any_real_numeric(&self) -> bool {
        self.is_numeric()
    }

    /// Returns true for datetime-like dtypes (datetime, timedelta, period).
    ///
    /// Matches `pd.api.types.is_datetime64_any_dtype()` family.
    #[must_use]
    pub const fn is_datetime_like(&self) -> bool {
        matches!(self, Self::Datetime64 | Self::Timedelta64 | Self::Period)
    }

    /// Return the numpy dtype character code.
    ///
    /// Matches numpy dtype.char property.
    #[must_use]
    pub const fn char(&self) -> char {
        match self {
            Self::Bool | Self::BoolNullable => '?',
            Self::Int64 | Self::Int64Nullable => 'l',
            Self::Float64 => 'd',
            Self::Utf8 => 'O',
            Self::Datetime64 => 'M',
            Self::Timedelta64 => 'm',
            Self::Categorical | Self::Period | Self::Interval | Self::Sparse | Self::Null => 'O',
        }
    }

    /// Return the numpy type number.
    ///
    /// Matches numpy dtype.num property.
    #[must_use]
    pub const fn num(&self) -> i32 {
        match self {
            Self::Bool | Self::BoolNullable => 0,
            Self::Int64 | Self::Int64Nullable => 7,
            Self::Float64 => 12,
            Self::Utf8 => 17,
            Self::Datetime64 => 21,
            Self::Timedelta64 => 22,
            Self::Categorical | Self::Period | Self::Interval | Self::Sparse | Self::Null => 17,
        }
    }

    /// Return the byte order character.
    ///
    /// Matches numpy dtype.byteorder property. Returns '=' (native) for all types.
    #[must_use]
    pub const fn byteorder(&self) -> char {
        '='
    }

    /// Return the numpy dtype string representation.
    ///
    /// Matches numpy dtype.str property.
    #[must_use]
    pub const fn str_repr(&self) -> &'static str {
        match self {
            Self::Bool | Self::BoolNullable => "|b1",
            Self::Int64 | Self::Int64Nullable => "<i8",
            Self::Float64 => "<f8",
            Self::Utf8 => "|O8",
            Self::Datetime64 => "<M8[ns]",
            Self::Timedelta64 => "<m8[ns]",
            Self::Categorical | Self::Period | Self::Interval | Self::Sparse | Self::Null => "|O8",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SparseDType {
    pub value_dtype: DType,
    pub fill_value: Scalar,
}

impl SparseDType {
    /// Construct a pandas-style sparse dtype descriptor.
    ///
    /// This records the logical dense value dtype plus the scalar value that is
    /// elided from storage. The concrete sparse column representation lives in
    /// fp-columnar; this descriptor is the shared public contract.
    pub fn new(value_dtype: DType, fill_value: Scalar) -> Result<Self, TypeError> {
        if matches!(value_dtype, DType::Null | DType::Sparse) {
            return Err(TypeError::InvalidSparseValueDType { dtype: value_dtype });
        }

        let fill_value = if fill_value.is_missing() {
            Scalar::missing_for_dtype(value_dtype)
        } else {
            cast_scalar_owned(fill_value, value_dtype)?
        };

        Ok(Self {
            value_dtype,
            fill_value,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NullKind {
    Null,
    NaN,
    NaT,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
pub enum Scalar {
    Null(NullKind),
    Bool(bool),
    Int64(i64),
    Float64(f64),
    #[serde(alias = "string", alias = "str")]
    Utf8(String),
    Timedelta64(i64),
    /// Nanoseconds since Unix epoch. Matches pandas `datetime64[ns]`.
    /// Uses `Timestamp::NAT` (i64::MIN) for missing values.
    Datetime64(i64),
    /// Period ordinal. Uses i64::MIN for NaT (missing value).
    Period(i64),
    /// Numeric interval value. Missing values remain `Scalar::Null`.
    Interval(Interval),
}

impl std::fmt::Display for Scalar {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Null(NullKind::NaN) => write!(f, "NaN"),
            Self::Null(NullKind::NaT) => write!(f, "NaT"),
            Self::Null(NullKind::Null) => write!(f, "None"),
            Self::Bool(b) => write!(f, "{}", if *b { "True" } else { "False" }),
            Self::Int64(v) => write!(f, "{v}"),
            Self::Float64(v) => write!(f, "{v}"),
            Self::Utf8(s) => write!(f, "{s}"),
            Self::Timedelta64(nanos) => write!(f, "{}", Timedelta::format(*nanos)),
            Self::Datetime64(nanos) => {
                if *nanos == Timestamp::NAT {
                    write!(f, "NaT")
                } else {
                    write!(f, "Timestamp[{nanos}]")
                }
            }
            Self::Period(ordinal) => {
                if *ordinal == i64::MIN {
                    write!(f, "NaT")
                } else {
                    write!(f, "Period[{ordinal}]")
                }
            }
            Self::Interval(interval) => write!(f, "{interval}"),
        }
    }
}

// Ergonomic From impls (br-frankenpandas-esjjy / fd90.182). Mirrors
// IndexLabel's From<i64>/From<&str>/From<String> so users can write
//   let v: Vec<Scalar> = vec![1i64.into(), 2.0.into(), "three".into()];
// instead of the explicit Scalar::Int64(...)/Scalar::Float64(...) form.
//
// i64 maps to Int64 (more common than Timedelta64 in pandas-style code).
// Users wanting Timedelta64 should construct it explicitly with
// Scalar::Timedelta64(nanos) or via Timedelta::parse / to_timedelta.

impl From<bool> for Scalar {
    fn from(value: bool) -> Self {
        Self::Bool(value)
    }
}

impl From<i64> for Scalar {
    fn from(value: i64) -> Self {
        Self::Int64(value)
    }
}

impl From<f64> for Scalar {
    fn from(value: f64) -> Self {
        Self::Float64(value)
    }
}

impl From<&str> for Scalar {
    fn from(value: &str) -> Self {
        Self::Utf8(value.to_owned())
    }
}

impl From<String> for Scalar {
    fn from(value: String) -> Self {
        Self::Utf8(value)
    }
}

impl Scalar {
    #[must_use]
    pub fn dtype(&self) -> DType {
        match self {
            Self::Null(_) => DType::Null,
            Self::Bool(_) => DType::Bool,
            Self::Int64(_) => DType::Int64,
            Self::Float64(_) => DType::Float64,
            Self::Utf8(_) => DType::Utf8,
            Self::Timedelta64(_) => DType::Timedelta64,
            Self::Datetime64(_) => DType::Datetime64,
            Self::Period(_) => DType::Period,
            Self::Interval(_) => DType::Interval,
        }
    }

    #[must_use]
    pub fn is_missing(&self) -> bool {
        match self {
            Self::Null(_) => true,
            Self::Float64(v) => v.is_nan(),
            Self::Timedelta64(v) => *v == Timedelta::NAT,
            Self::Datetime64(v) => *v == Timestamp::NAT,
            Self::Period(v) => *v == i64::MIN,
            _ => false,
        }
    }

    #[must_use]
    pub fn is_nan(&self) -> bool {
        matches!(self, Self::Null(NullKind::NaN)) || matches!(self, Self::Float64(v) if v.is_nan())
    }

    /// Returns true if this is a Bool scalar.
    #[must_use]
    pub const fn is_bool(&self) -> bool {
        matches!(self, Self::Bool(_))
    }

    /// Returns true if this is an Int64 scalar.
    #[must_use]
    pub const fn is_integer(&self) -> bool {
        matches!(self, Self::Int64(_))
    }

    /// Returns true if this is a Float64 scalar.
    #[must_use]
    pub const fn is_float(&self) -> bool {
        matches!(self, Self::Float64(_))
    }

    /// Returns true if this is a numeric scalar (Int64 or Float64).
    #[must_use]
    pub const fn is_numeric(&self) -> bool {
        matches!(self, Self::Int64(_) | Self::Float64(_))
    }

    /// Returns true if this is a Utf8 (string) scalar.
    #[must_use]
    pub const fn is_string(&self) -> bool {
        matches!(self, Self::Utf8(_))
    }

    /// Returns true if this is a Datetime64 scalar.
    #[must_use]
    pub const fn is_datetime(&self) -> bool {
        matches!(self, Self::Datetime64(_))
    }

    /// Returns true if this is a Timedelta64 scalar.
    #[must_use]
    pub const fn is_timedelta(&self) -> bool {
        matches!(self, Self::Timedelta64(_))
    }

    /// Returns true if this is a Period scalar.
    #[must_use]
    pub const fn is_period(&self) -> bool {
        matches!(self, Self::Period(_))
    }

    /// Returns true if this is an Interval scalar.
    #[must_use]
    pub const fn is_interval(&self) -> bool {
        matches!(self, Self::Interval(_))
    }

    #[must_use]
    pub fn missing_for_dtype(dtype: DType) -> Self {
        match dtype {
            DType::Float64 => Self::Null(NullKind::NaN),
            DType::Timedelta64 => Self::Timedelta64(Timedelta::NAT),
            DType::Datetime64 => Self::Datetime64(Timestamp::NAT),
            DType::Period => Self::Period(i64::MIN),
            DType::Null => Self::Null(NullKind::Null),
            DType::Bool
            | DType::BoolNullable
            | DType::Int64
            | DType::Int64Nullable
            | DType::Utf8
            | DType::Categorical
            | DType::Interval
            | DType::Sparse => Self::Null(NullKind::Null),
        }
    }

    #[must_use]
    pub fn semantic_eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::Float64(a), Self::Float64(b)) => {
                if a.is_nan() && b.is_nan() {
                    return true;
                }
                if *a == *b {
                    return true;
                }
                let diff = (*a - *b).abs();
                let max_abs = a.abs().max(b.abs());
                if max_abs == 0.0 {
                    diff < f64::EPSILON
                } else {
                    diff / max_abs < 1e-14
                }
            }
            (Self::Null(_), Self::Float64(v)) | (Self::Float64(v), Self::Null(_)) => v.is_nan(),
            // All Null kinds (Null / NaN / NaT) mark missingness; they are
            // semantically indistinguishable for oracle-parity checks even
            // though derived PartialEq would reject a cross-kind pair.
            // fp-frame normalizes Float64 column missing cells to
            // Null(NaN) at Column::new time, while fixture oracles encode
            // the canonical missing marker as Null(Null).
            (Self::Null(_), Self::Null(_)) => true,
            _ => self == other,
        }
    }

    #[must_use]
    pub fn semantic_le(&self, other: &Self) -> bool {
        match self.semantic_cmp(other) {
            std::cmp::Ordering::Less | std::cmp::Ordering::Equal => true,
            std::cmp::Ordering::Greater => false,
        }
    }

    #[must_use]
    pub fn semantic_ge(&self, other: &Self) -> bool {
        match self.semantic_cmp(other) {
            std::cmp::Ordering::Greater | std::cmp::Ordering::Equal => true,
            std::cmp::Ordering::Less => false,
        }
    }

    #[must_use]
    pub fn is_null(&self) -> bool {
        matches!(self, Self::Null(_))
    }

    #[must_use]
    pub fn is_na(&self) -> bool {
        self.is_missing()
    }

    #[must_use]
    pub fn coalesce(&self, other: &Self) -> Self {
        if self.is_missing() {
            other.clone()
        } else {
            self.clone()
        }
    }

    #[must_use]
    pub fn semantic_cmp(&self, other: &Self) -> std::cmp::Ordering {
        match (self, other) {
            (Self::Int64(a), Self::Int64(b)) => a.cmp(b),
            (Self::Float64(a), Self::Float64(b)) => {
                a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)
            }
            (Self::Utf8(a), Self::Utf8(b)) => a.cmp(b),
            (Self::Bool(a), Self::Bool(b)) => a.cmp(b),
            (Self::Null(a), Self::Null(b)) => a.cmp(b),
            (Self::Timedelta64(a), Self::Timedelta64(b)) => {
                if *a == Timedelta::NAT || *b == Timedelta::NAT {
                    std::cmp::Ordering::Equal
                } else {
                    a.cmp(b)
                }
            }
            (Self::Datetime64(a), Self::Datetime64(b)) => {
                if *a == Timestamp::NAT || *b == Timestamp::NAT {
                    std::cmp::Ordering::Equal
                } else {
                    a.cmp(b)
                }
            }
            (Self::Period(a), Self::Period(b)) => {
                if *a == i64::MIN || *b == i64::MIN {
                    std::cmp::Ordering::Equal
                } else {
                    a.cmp(b)
                }
            }
            (Self::Interval(a), Self::Interval(b)) => a
                .left
                .partial_cmp(&b.left)
                .unwrap_or(std::cmp::Ordering::Equal)
                .then_with(|| {
                    a.right
                        .partial_cmp(&b.right)
                        .unwrap_or(std::cmp::Ordering::Equal)
                })
                .then_with(|| a.closed.cmp(&b.closed)),
            // Cross-numeric comparison
            (Self::Int64(a), Self::Float64(b)) => (*a as f64)
                .partial_cmp(b)
                .unwrap_or(std::cmp::Ordering::Equal),
            (Self::Float64(a), Self::Int64(b)) => a
                .partial_cmp(&(*b as f64))
                .unwrap_or(std::cmp::Ordering::Equal),
            // Fallback to debug representation for inconsistent types
            (a, b) => format!("{a:?}").cmp(&format!("{b:?}")),
        }
    }

    pub fn to_f64(&self) -> Result<f64, TypeError> {
        match self {
            Self::Bool(v) => Ok(if *v { 1.0 } else { 0.0 }),
            Self::Int64(v) => Ok(*v as f64),
            Self::Float64(v) => Ok(*v),
            Self::Null(kind) => Err(TypeError::ValueIsMissing { kind: *kind }),
            Self::Utf8(v) => Err(TypeError::NonNumericValue {
                value: v.clone(),
                dtype: DType::Utf8,
            }),
            Self::Timedelta64(v) if *v == Timedelta::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Timedelta64(v) => Err(TypeError::NonNumericValue {
                value: Timedelta::format(*v),
                dtype: DType::Timedelta64,
            }),
            Self::Datetime64(v) if *v == Timestamp::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Datetime64(v) => Err(TypeError::NonNumericValue {
                value: format!("Timestamp[{v}]"),
                dtype: DType::Datetime64,
            }),
            Self::Period(v) if *v == i64::MIN => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Period(v) => Err(TypeError::NonNumericValue {
                value: format!("Period[{v}]"),
                dtype: DType::Period,
            }),
            Self::Interval(v) => Err(TypeError::NonNumericValue {
                value: v.to_string(),
                dtype: DType::Interval,
            }),
        }
    }

    /// Try to convert to i64. Returns error for missing or non-numeric values.
    pub fn to_i64(&self) -> Result<i64, TypeError> {
        match self {
            Self::Bool(v) => Ok(if *v { 1 } else { 0 }),
            Self::Int64(v) => Ok(*v),
            Self::Float64(v) => Ok(*v as i64),
            Self::Null(kind) => Err(TypeError::ValueIsMissing { kind: *kind }),
            Self::Utf8(v) => Err(TypeError::NonNumericValue {
                value: v.clone(),
                dtype: DType::Utf8,
            }),
            Self::Timedelta64(v) if *v == Timedelta::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Timedelta64(v) => Ok(*v),
            Self::Datetime64(v) if *v == Timestamp::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Datetime64(v) => Ok(*v),
            Self::Period(v) if *v == i64::MIN => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Period(v) => Ok(*v),
            Self::Interval(v) => Err(TypeError::NonNumericValue {
                value: v.to_string(),
                dtype: DType::Interval,
            }),
        }
    }

    /// Try to convert to bool. Returns error for missing values.
    pub fn to_bool(&self) -> Result<bool, TypeError> {
        match self {
            Self::Bool(v) => Ok(*v),
            Self::Int64(v) => Ok(*v != 0),
            Self::Float64(v) => Ok(*v != 0.0 && !v.is_nan()),
            Self::Null(kind) => Err(TypeError::ValueIsMissing { kind: *kind }),
            Self::Utf8(v) => Ok(!v.is_empty()),
            Self::Timedelta64(v) if *v == Timedelta::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Timedelta64(v) => Ok(*v != 0),
            Self::Datetime64(v) if *v == Timestamp::NAT => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Datetime64(v) => Ok(*v != 0),
            Self::Period(v) if *v == i64::MIN => Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            }),
            Self::Period(v) => Ok(*v != 0),
            Self::Interval(_) => Ok(true),
        }
    }

    /// Try to convert to string representation.
    pub fn to_str(&self) -> String {
        match self {
            Self::Bool(v) => if *v { "True" } else { "False" }.to_string(),
            Self::Int64(v) => v.to_string(),
            Self::Float64(v) => {
                if v.is_nan() {
                    "nan".to_string()
                } else if v.is_infinite() {
                    if *v > 0.0 { "inf" } else { "-inf" }.to_string()
                } else {
                    v.to_string()
                }
            }
            Self::Null(_) => "NaN".to_string(),
            Self::Utf8(v) => v.clone(),
            Self::Timedelta64(v) => Timedelta::format(*v),
            Self::Datetime64(v) if *v == Timestamp::NAT => "NaT".to_string(),
            Self::Datetime64(v) => Timestamp::from_nanos(*v).isoformat(),
            Self::Period(v) if *v == i64::MIN => "NaT".to_string(),
            Self::Period(v) => format!("Period[{}]", v),
            Self::Interval(v) => v.to_string(),
        }
    }
}

#[derive(Debug, Error, Clone, PartialEq)]
pub enum TypeError {
    #[error("dtype coercion from {left:?} to {right:?} has no compatible common type")]
    IncompatibleDtypes { left: DType, right: DType },
    #[error("cannot cast scalar of dtype {from:?} to {to:?}")]
    InvalidCast { from: DType, to: DType },
    #[error("cannot cast float {value} to int64 without loss")]
    LossyFloatToInt { value: f64 },
    #[error("expected 0/1 for bool cast from int64 but found {value}")]
    InvalidBoolInt { value: i64 },
    #[error("expected 0.0/1.0 for bool cast from float64 but found {value}")]
    InvalidBoolFloat { value: f64 },
    #[error("value {value:?} has non-numeric dtype {dtype:?}")]
    NonNumericValue { value: String, dtype: DType },
    #[error("value is missing ({kind:?})")]
    ValueIsMissing { kind: NullKind },
    #[error("sparse value dtype cannot be {dtype:?}")]
    InvalidSparseValueDType { dtype: DType },
    #[error("interval_range step must be finite, positive, and not NaN (got {step})")]
    InvalidIntervalStep { step: f64 },
    #[error("interval_range step {step} does not evenly divide range end-start={span}")]
    IntervalStepDoesNotDivide { step: f64, span: f64 },
    #[error("cannot parse '{value}' as {target}")]
    ValueNotParseable { value: String, target: String },
}

pub fn common_dtype(left: DType, right: DType) -> Result<DType, TypeError> {
    use DType::{
        Bool, BoolNullable, Categorical, Datetime64, Float64, Int64, Int64Nullable, Null, Sparse,
        Timedelta64,
    };

    let out = match (left, right) {
        (a, b) if a == b => a,
        (Null, other) | (other, Null) => other,
        (Categorical, Categorical) => Categorical,

        // Bool promotions (nullable absorbs non-nullable)
        (Bool, Int64) | (Int64, Bool) => Int64,
        (Bool, Int64Nullable) | (Int64Nullable, Bool) => Int64Nullable,
        (BoolNullable, Int64) | (Int64, BoolNullable) => Int64Nullable,
        (BoolNullable, Int64Nullable) | (Int64Nullable, BoolNullable) => Int64Nullable,
        (Bool, BoolNullable) | (BoolNullable, Bool) => BoolNullable,
        (Bool, Float64) | (Float64, Bool) => Float64,
        (BoolNullable, Float64) | (Float64, BoolNullable) => Float64,

        // Int64 promotions (nullable absorbs non-nullable)
        (Int64, Float64) | (Float64, Int64) => Float64,
        (Int64Nullable, Float64) | (Float64, Int64Nullable) => Float64,
        (Int64, Int64Nullable) | (Int64Nullable, Int64) => Int64Nullable,

        // Datetime/Timedelta
        (Timedelta64, Timedelta64) => Timedelta64,
        (Datetime64, Datetime64) => Datetime64,

        (Sparse, _) | (_, Sparse) => return Err(TypeError::IncompatibleDtypes { left, right }),
        _ => return Err(TypeError::IncompatibleDtypes { left, right }),
    };

    Ok(out)
}

pub fn infer_dtype(values: &[Scalar]) -> Result<DType, TypeError> {
    let mut current = DType::Null;
    let mut saw_utf8 = false;
    let mut saw_timedelta = false;
    let mut saw_datetime = false;
    let mut saw_non_utf8_non_null = false;

    for value in values {
        match value.dtype() {
            DType::Null => {}
            DType::Utf8 => saw_utf8 = true,
            DType::Timedelta64 => {
                saw_timedelta = true;
                if current == DType::Null {
                    current = DType::Timedelta64;
                } else if current != DType::Timedelta64 {
                    return Err(TypeError::IncompatibleDtypes {
                        left: current,
                        right: DType::Timedelta64,
                    });
                }
            }
            DType::Datetime64 => {
                saw_datetime = true;
                if current == DType::Null {
                    current = DType::Datetime64;
                } else if current != DType::Datetime64 {
                    return Err(TypeError::IncompatibleDtypes {
                        left: current,
                        right: DType::Datetime64,
                    });
                }
            }
            other => {
                saw_non_utf8_non_null = true;
                current = common_dtype(current, other)?;
            }
        }

        if saw_utf8 && saw_non_utf8_non_null {
            // Constructor inference follows pandas object-dtype behavior for
            // heterogeneous string/scalar payloads while arithmetic coercion
            // remains governed by the stricter common_dtype lattice.
            return Ok(DType::Utf8);
        }
        if saw_timedelta && saw_non_utf8_non_null {
            return Err(TypeError::IncompatibleDtypes {
                left: DType::Timedelta64,
                right: current,
            });
        }
        if saw_datetime && saw_non_utf8_non_null {
            return Err(TypeError::IncompatibleDtypes {
                left: DType::Datetime64,
                right: current,
            });
        }
    }

    if saw_utf8 {
        Ok(DType::Utf8)
    } else {
        Ok(current)
    }
}

/// Cast a scalar to a target dtype, taking ownership to avoid redundant clones
/// when the value already has the correct type (AG-03: identity-cast skip).
pub fn cast_scalar_owned(value: Scalar, target: DType) -> Result<Scalar, TypeError> {
    let from = value.dtype();
    if from == target {
        return Ok(value);
    }
    // Int64 <-> Int64Nullable: same representation, just different dtype tracking
    if (from == DType::Int64 && target == DType::Int64Nullable)
        || (from == DType::Int64Nullable && target == DType::Int64)
    {
        return Ok(value);
    }
    // Bool <-> BoolNullable: same representation
    if (from == DType::Bool && target == DType::BoolNullable)
        || (from == DType::BoolNullable && target == DType::Bool)
    {
        return Ok(value);
    }
    if target == DType::Utf8 {
        return Ok(Scalar::Utf8(scalar_to_string_for_astype(value)));
    }
    // Per br-frankenpandas-cyi4h: pandas astype(bool) (the numpy bool dtype)
    // treats a float NaN as truthy -> True (bool(nan) is True), unlike the
    // nullable 'boolean' dtype which keeps NA. FP's NaN=missing model would
    // otherwise fall through to the missing branch below and yield a null.
    // Verified vs live pandas 2.2.3.
    if target == DType::Bool
        && let Scalar::Float64(v) = &value
        && v.is_nan()
    {
        return Ok(Scalar::Bool(true));
    }
    if value.is_missing() {
        return Ok(Scalar::missing_for_dtype(target));
    }

    // Note: identity casts (from == target) are handled above, so same-type
    // arms are omitted from the match below.
    match target {
        DType::Null => Ok(Scalar::Null(NullKind::Null)),
        DType::Bool => match &value {
            // numpy bool: zero -> False, ANY nonzero -> True (it does not
            // restrict to 0/1). e.g. bool of -3 / 2.5 is True.
            Scalar::Int64(v) => Ok(Scalar::Bool(*v != 0)),
            // 0.0 and -0.0 -> False; every other value, INCLUDING NaN, -> True
            // (NaN != 0.0 is true), matching numpy/pandas truthiness.
            Scalar::Float64(v) => Ok(Scalar::Bool(*v != 0.0)),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::BoolNullable => match &value {
            // pandas nullable Boolean ('boolean') is STRICT: only 0/1 (or
            // True/False) are accepted — any other value raises "Need to pass
            // bool-like values", UNLIKE numpy bool which is nonzero-truthy.
            // (br-frankenpandas-tjomg)
            Scalar::Bool(b) => Ok(Scalar::Bool(*b)),
            Scalar::Int64(0) => Ok(Scalar::Bool(false)),
            Scalar::Int64(1) => Ok(Scalar::Bool(true)),
            Scalar::Int64(v) => Err(TypeError::InvalidBoolInt { value: *v }),
            Scalar::Float64(v) if *v == 0.0 => Ok(Scalar::Bool(false)),
            Scalar::Float64(v) if *v == 1.0 => Ok(Scalar::Bool(true)),
            Scalar::Float64(v) => Err(TypeError::InvalidBoolFloat { value: *v }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Int64 | DType::Int64Nullable => match &value {
            Scalar::Bool(v) => Ok(Scalar::Int64(i64::from(*v))),
            Scalar::Float64(v) => {
                // pandas astype(int64) truncates a finite float toward zero
                // (1.9 -> 1, -1.9 -> -1, 2.5 -> 2); only non-finite (NaN/±inf)
                // or out-of-i64-range values raise. Verified vs pandas 2.2.3
                // (br-frankenpandas-qcutc). NaN is handled as missing above, so
                // here non-finite means ±inf. `as i64` performs the toward-zero
                // truncation for in-range finite values.
                if !v.is_finite() {
                    return Err(TypeError::LossyFloatToInt { value: *v });
                }
                if *v < i64::MIN as f64 || *v >= 9223372036854775808.0 {
                    return Err(TypeError::LossyFloatToInt { value: *v });
                }
                Ok(Scalar::Int64(*v as i64))
            }
            Scalar::Utf8(s) => {
                // Try direct int parse first, then try float parse + truncate
                // (pandas accepts "1.0" as valid int via float intermediate)
                if let Ok(v) = s.parse::<i64>() {
                    return Ok(Scalar::Int64(v));
                }
                if let Ok(f) = s.parse::<f64>()
                    && f.is_finite()
                    && f.fract() == 0.0
                    && f >= i64::MIN as f64
                    && f < 9223372036854775808.0
                {
                    return Ok(Scalar::Int64(f as i64));
                }
                Err(TypeError::InvalidCast { from, to: target })
            }
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Float64 => match &value {
            Scalar::Bool(v) => Ok(Scalar::Float64(if *v { 1.0 } else { 0.0 })),
            Scalar::Int64(v) => Ok(Scalar::Float64(*v as f64)),
            Scalar::Utf8(s) => s
                .parse::<f64>()
                .map(Scalar::Float64)
                .map_err(|_| TypeError::InvalidCast { from, to: target }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Utf8 => Ok(Scalar::Utf8(scalar_to_string_for_astype(value))),
        DType::Categorical => Err(TypeError::InvalidCast { from, to: target }),
        DType::Timedelta64 => match &value {
            Scalar::Int64(v) => Ok(Scalar::Timedelta64(*v)),
            Scalar::Utf8(s) => Timedelta::parse(s)
                .map(Scalar::Timedelta64)
                .map_err(|_| TypeError::InvalidCast { from, to: target }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Datetime64 => match &value {
            Scalar::Int64(v) => Ok(Scalar::Datetime64(*v)),
            Scalar::Utf8(s) => Timestamp::parse(s)
                .map(|timestamp| Scalar::Datetime64(timestamp.nanos))
                .map_err(|_| TypeError::InvalidCast { from, to: target }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Period => match &value {
            Scalar::Int64(v) => Ok(Scalar::Period(*v)),
            Scalar::Utf8(s) => Period::parse(s)
                .map(|period| Scalar::Period(period.ordinal))
                .map_err(|_| TypeError::InvalidCast { from, to: target }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Interval => match &value {
            Scalar::Utf8(s) => Interval::parse(s)
                .map(Scalar::Interval)
                .map_err(|_| TypeError::InvalidCast { from, to: target }),
            _ => Err(TypeError::InvalidCast { from, to: target }),
        },
        DType::Sparse => Err(TypeError::InvalidCast { from, to: target }),
    }
}

fn scalar_to_string_for_astype(value: Scalar) -> String {
    match value {
        Scalar::Null(NullKind::Null) => "None".to_owned(),
        Scalar::Null(NullKind::NaN) => "nan".to_owned(),
        Scalar::Null(NullKind::NaT) => "NaT".to_owned(),
        Scalar::Bool(true) => "True".to_owned(),
        Scalar::Bool(false) => "False".to_owned(),
        Scalar::Int64(v) => v.to_string(),
        Scalar::Float64(v) => float_to_string_for_astype(v),
        Scalar::Utf8(s) => s,
        Scalar::Timedelta64(v) if v == Timedelta::NAT => "NaT".to_owned(),
        Scalar::Timedelta64(v) => Timedelta::format(v),
        Scalar::Datetime64(v) if v == Timestamp::NAT => "NaT".to_owned(),
        Scalar::Datetime64(v) => format!("Timestamp[{v}]"),
        Scalar::Period(v) if v == i64::MIN => "NaT".to_owned(),
        Scalar::Period(v) => format!("Period[{v}]"),
        Scalar::Interval(v) => v.to_string(),
    }
}

fn float_to_string_for_astype(value: f64) -> String {
    if value.is_nan() {
        return "nan".to_owned();
    }
    if value.is_infinite() {
        return value.to_string(); // "inf" / "-inf"
    }
    // pandas astype(str) renders floats via Python str(float): whole numbers
    // keep ".0", decimals use the shortest round-trip, and extreme magnitudes use
    // scientific notation ("1e+16", "1e-05"). Rust's Debug formatter matches this
    // (shortest round-trip, ".0" on whole numbers, scientific at Python's
    // boundaries); only the exponent spelling differs (Rust "1e16"/"1e-5" vs
    // Python "1e+16"/"1e-05"), so normalize that. The old `{:.1}` whole / Display
    // decimal path lost scientific notation (1e16 -> "10000000000000000.0").
    let s = format!("{value:?}");
    match s.split_once('e') {
        None => s,
        Some((mantissa, exp)) => {
            let (sign, digits) = match exp.strip_prefix('-') {
                Some(d) => ('-', d),
                None => ('+', exp.strip_prefix('+').unwrap_or(exp)),
            };
            format!("{mantissa}e{sign}{digits:0>2}")
        }
    }
}

/// Cast a scalar reference to a target dtype (clones only when conversion is needed).
pub fn cast_scalar(value: &Scalar, target: DType) -> Result<Scalar, TypeError> {
    cast_scalar_owned(value.clone(), target)
}

// ── Timedelta support ──────────────────────────────────────────────────

#[derive(Debug, Error, Clone, PartialEq)]
pub enum TimedeltaError {
    #[error("invalid timedelta string: {0}")]
    InvalidFormat(String),
    #[error("overflow in timedelta computation")]
    Overflow,
}

#[derive(Debug, Clone, Copy, Default)]
pub struct TimedeltaComponents {
    pub days: i64,
    pub hours: i64,
    pub minutes: i64,
    pub seconds: i64,
    pub milliseconds: i64,
    pub microseconds: i64,
    pub nanoseconds: i64,
}

pub struct Timedelta;

impl Timedelta {
    pub const NANOS_PER_MICRO: i64 = 1_000;
    pub const NANOS_PER_MILLI: i64 = 1_000_000;
    pub const NANOS_PER_SEC: i64 = 1_000_000_000;
    pub const NANOS_PER_MIN: i64 = 60 * Self::NANOS_PER_SEC;
    pub const NANOS_PER_HOUR: i64 = 60 * Self::NANOS_PER_MIN;
    pub const NANOS_PER_DAY: i64 = 24 * Self::NANOS_PER_HOUR;
    pub const NANOS_PER_WEEK: i64 = 7 * Self::NANOS_PER_DAY;

    pub const NAT: i64 = i64::MIN;

    pub fn parse(s: &str) -> Result<i64, TimedeltaError> {
        let s = s.trim();

        if s.eq_ignore_ascii_case("nat") {
            return Ok(Self::NAT);
        }

        let (negative, s) = if let Some(rest) = s.strip_prefix('-') {
            (true, rest.trim())
        } else {
            (false, s)
        };

        if let Some(nanos) = Self::try_parse_time_format(s) {
            return Ok(if negative { -nanos } else { nanos });
        }

        if let Some(nanos) = Self::try_parse_iso8601_duration(s) {
            return Ok(if negative { -nanos } else { nanos });
        }

        let nanos = Self::parse_compound(s)?;
        Ok(if negative { -nanos } else { nanos })
    }

    /// Parse an ISO-8601 duration the way pandas `Timedelta` accepts it:
    /// a leading uppercase `P`, an optional `T` separator that is otherwise
    /// ignored, and unit letters `W`/`D`/`H`/`M`/`S` in any position. `M` is
    /// always MINUTES (never months), and only seconds may be fractional —
    /// years/months and lowercase units are rejected, matching pandas.
    /// (pandas mis-handles fractional non-second components; those are rejected
    /// here rather than reproducing the buggy value.) (br-frankenpandas-c3p8b)
    fn try_parse_iso8601_duration(s: &str) -> Option<i64> {
        let mut rest = s.strip_prefix('P')?;
        if rest.is_empty() {
            return None;
        }
        let mut total: i64 = 0;
        let mut saw_component = false;
        while !rest.is_empty() {
            if let Some(after_t) = rest.strip_prefix('T') {
                rest = after_t;
                continue;
            }
            let num_end = rest.find(|c: char| !c.is_ascii_digit() && c != '.')?;
            if num_end == 0 {
                return None;
            }
            let num_str = &rest[..num_end];
            let unit = rest.as_bytes()[num_end];
            let is_fractional = num_str.contains('.');
            rest = &rest[num_end + 1..];

            let (multiplier, frac_ok) = match unit {
                b'W' => (Self::NANOS_PER_WEEK, false),
                b'D' => (Self::NANOS_PER_DAY, false),
                b'H' => (Self::NANOS_PER_HOUR, false),
                b'M' => (Self::NANOS_PER_MIN, false),
                b'S' => (Self::NANOS_PER_SEC, true),
                _ => return None,
            };
            if is_fractional {
                if !frac_ok {
                    return None;
                }
                let value: f64 = num_str.parse().ok()?;
                let product = value * multiplier as f64;
                if !product.is_finite() || product.abs() >= 9223372036854775808.0 {
                    return None;
                }
                total = total.checked_add(product.round() as i64)?;
            } else {
                let value: i64 = num_str.parse().ok()?;
                total = total.checked_add(value.checked_mul(multiplier)?)?;
            }
            saw_component = true;
        }
        saw_component.then_some(total)
    }

    fn try_parse_time_format(s: &str) -> Option<i64> {
        let parts: Vec<&str> = s.split(':').collect();
        if parts.len() < 2 || parts.len() > 3 {
            return None;
        }

        let hours: i64 = parts[0].parse().ok()?;
        let minutes: i64 = parts[1].parse().ok()?;

        let (seconds, frac_nanos) = if parts.len() == 3 {
            if let Some((sec_str, frac_str)) = parts[2].split_once('.') {
                let sec: i64 = sec_str.parse().ok()?;
                if !frac_str.bytes().all(|byte| byte.is_ascii_digit()) {
                    return None;
                }
                let mut frac = 0_i64;
                let taken = frac_str.len().min(9);
                for byte in frac_str.bytes().take(9) {
                    frac = frac * 10 + i64::from(byte - b'0');
                }
                for _ in taken..9 {
                    frac *= 10;
                }
                (sec, frac)
            } else {
                let sec: i64 = parts[2].parse().ok()?;
                (sec, 0)
            }
        } else {
            (0, 0)
        };

        hours
            .checked_mul(Self::NANOS_PER_HOUR)?
            .checked_add(minutes.checked_mul(Self::NANOS_PER_MIN)?)?
            .checked_add(seconds.checked_mul(Self::NANOS_PER_SEC)?)?
            .checked_add(frac_nanos)
    }

    fn parse_compound(s: &str) -> Result<i64, TimedeltaError> {
        let mut total: i64 = 0;
        let mut remaining = s;

        while !remaining.is_empty() {
            remaining = remaining.trim_start();
            if remaining.is_empty() {
                break;
            }

            // Per br-frankenpandas-i9bah: check if remaining is a time format
            // (HH:MM:SS) which can appear after "N days " in pandas timedelta strings.
            if remaining.contains(':')
                && let Some(time_nanos) = Self::try_parse_time_format(remaining)
            {
                total = total
                    .checked_add(time_nanos)
                    .ok_or(TimedeltaError::Overflow)?;
                break;
            }

            let num_end = remaining
                .find(|c: char| !c.is_ascii_digit() && c != '.' && c != '-')
                .unwrap_or(remaining.len());

            if num_end == 0 {
                return Err(TimedeltaError::InvalidFormat(s.to_string()));
            }

            let num_str = &remaining[..num_end];
            let num: f64 = num_str
                .parse()
                .map_err(|_| TimedeltaError::InvalidFormat(s.to_string()))?;

            remaining = remaining[num_end..].trim_start();

            let unit_end = remaining
                .find(|c: char| c.is_ascii_digit() || c.is_whitespace())
                .unwrap_or(remaining.len());

            let unit = &remaining[..unit_end];
            remaining = &remaining[unit_end..];

            let multiplier = Self::unit_to_nanos(unit)
                .ok_or_else(|| TimedeltaError::InvalidFormat(s.to_string()))?;

            // Per br-frankenpandas-zw3mg: pandas raises OverflowError on
            // huge scientific-notation Timedeltas like "1e100 days". The
            // raw `as i64` cast silently saturated to i64::MAX, masking
            // overflow before checked_add could catch it.
            let product = num * multiplier as f64;
            if !product.is_finite() || product.abs() >= 9223372036854775808.0 {
                return Err(TimedeltaError::Overflow);
            }
            let nanos = product.round() as i64;
            total = total.checked_add(nanos).ok_or(TimedeltaError::Overflow)?;
        }

        if total == 0 && !s.trim().is_empty() && s.trim() != "0" {
            return Err(TimedeltaError::InvalidFormat(s.to_string()));
        }

        Ok(total)
    }

    /// Map a pandas-style frequency-alias string to a nanosecond-count.
    ///
    /// Recognizes pandas's offset alias core set plus common word forms:
    /// W/week(s), D/day(s), H/hr/hour(s), m/T/min/minute(s), s/sec/second(s),
    /// ms/milli/millisecond(s)/L, us/µs/micro/microsecond(s)/U, ns/nano/
    /// nanosecond(s)/N. Empty string maps to days (matches pandas default).
    /// Returns `None` for unrecognized aliases — callers can choose to map
    /// that to NaT (consistent with the rest of fp-types) or surface as a
    /// typed error.
    ///
    /// Per br-frankenpandas-lbsx (9p0u Phase 2.6): public surface so
    /// downstream crates can consume the same alias map fp-types uses for
    /// `Timedelta::from_unit` / `Timestamp::*_to_unit`.
    #[must_use]
    pub fn unit_to_nanos(unit: &str) -> Option<i64> {
        match unit.to_lowercase().as_str() {
            "w" | "week" | "weeks" => Some(Self::NANOS_PER_WEEK),
            "d" | "day" | "days" => Some(Self::NANOS_PER_DAY),
            "h" | "hr" | "hour" | "hours" => Some(Self::NANOS_PER_HOUR),
            "m" | "min" | "minute" | "minutes" | "t" => Some(Self::NANOS_PER_MIN),
            "s" | "sec" | "second" | "seconds" => Some(Self::NANOS_PER_SEC),
            "ms" | "milli" | "millis" | "millisecond" | "milliseconds" | "l" => {
                Some(Self::NANOS_PER_MILLI)
            }
            "us" | "µs" | "micro" | "micros" | "microsecond" | "microseconds" | "u" => {
                Some(Self::NANOS_PER_MICRO)
            }
            "ns" | "nano" | "nanos" | "nanosecond" | "nanoseconds" | "n" => Some(1),
            "" => Some(Self::NANOS_PER_DAY),
            _ => None,
        }
    }

    pub fn components(nanos: i64) -> TimedeltaComponents {
        if nanos == Self::NAT {
            return TimedeltaComponents::default();
        }

        // pandas/Python normalize via FLOOR division (like format()): the days
        // component can be negative while the time-of-day remainder is always in
        // [0, 1 day). So pd.Timedelta(-1,'s').components == (-1, 23, 59, 59, 0, 0, 0),
        // NOT the abs-based (0, 0, 0, 1, 0, 0, 0).
        let days = nanos.div_euclid(Self::NANOS_PER_DAY);
        let rem = nanos.rem_euclid(Self::NANOS_PER_DAY);

        let hours = rem / Self::NANOS_PER_HOUR;
        let rem = rem % Self::NANOS_PER_HOUR;

        let minutes = rem / Self::NANOS_PER_MIN;
        let rem = rem % Self::NANOS_PER_MIN;

        let seconds = rem / Self::NANOS_PER_SEC;
        let rem = rem % Self::NANOS_PER_SEC;

        let milliseconds = rem / Self::NANOS_PER_MILLI;
        let rem = rem % Self::NANOS_PER_MILLI;

        let microseconds = rem / Self::NANOS_PER_MICRO;
        let nanoseconds = rem % Self::NANOS_PER_MICRO;

        TimedeltaComponents {
            days,
            hours,
            minutes,
            seconds,
            milliseconds,
            microseconds,
            nanoseconds,
        }
    }

    pub fn total_seconds(nanos: i64) -> f64 {
        if nanos == Self::NAT {
            f64::NAN
        } else {
            nanos as f64 / Self::NANOS_PER_SEC as f64
        }
    }

    /// Convert to specified time unit.
    ///
    /// Matches pd.Timedelta.as_unit(). Supported units: ns, us, ms, s, m, h, D.
    #[must_use]
    pub fn as_unit(nanos: i64, unit: &str) -> f64 {
        if nanos == Self::NAT {
            return f64::NAN;
        }
        let nanos_f = nanos as f64;
        match unit {
            "ns" | "nanoseconds" => nanos_f,
            "us" | "microseconds" => nanos_f / Self::NANOS_PER_MICRO as f64,
            "ms" | "milliseconds" => nanos_f / Self::NANOS_PER_MILLI as f64,
            "s" | "seconds" => nanos_f / Self::NANOS_PER_SEC as f64,
            "m" | "minutes" => nanos_f / Self::NANOS_PER_MIN as f64,
            "h" | "hours" => nanos_f / Self::NANOS_PER_HOUR as f64,
            "D" | "days" => nanos_f / Self::NANOS_PER_DAY as f64,
            _ => f64::NAN,
        }
    }

    /// Return the days component. Matches `pd.Timedelta.days`.
    #[must_use]
    pub fn days(nanos: i64) -> i64 {
        if nanos == Self::NAT {
            return 0; // pandas returns 0 for NaT.days (no error)
        }
        // FLOOR division like pandas: pd.Timedelta(-1,'s').days == -1, not 0.
        nanos.div_euclid(Self::NANOS_PER_DAY)
    }

    /// Return the seconds component (0-86399). Matches `pd.Timedelta.seconds`.
    #[must_use]
    pub fn seconds(nanos: i64) -> i64 {
        if nanos == Self::NAT {
            return 0;
        }
        // Floor-normalized time-of-day remainder: pd.Timedelta(-1,'s').seconds == 86399.
        nanos.rem_euclid(Self::NANOS_PER_DAY) / Self::NANOS_PER_SEC
    }

    /// Return the microseconds component (0-999999). Matches `pd.Timedelta.microseconds`.
    #[must_use]
    pub fn microseconds(nanos: i64) -> i64 {
        if nanos == Self::NAT {
            return 0;
        }
        nanos.rem_euclid(Self::NANOS_PER_SEC) / Self::NANOS_PER_MICRO
    }

    /// Return the nanoseconds component (0-999). Matches `pd.Timedelta.nanoseconds`.
    #[must_use]
    pub fn nanoseconds(nanos: i64) -> i64 {
        if nanos == Self::NAT {
            return 0;
        }
        nanos.rem_euclid(Self::NANOS_PER_MICRO)
    }

    pub fn format(nanos: i64) -> String {
        if nanos == Self::NAT {
            return "NaT".to_string();
        }

        // pandas / Python timedelta normalize via FLOOR division: the days
        // component can be negative while the time-of-day remainder is always
        // non-negative, and a negative-days value prints a '+' before the time
        // (e.g. -1s -> "-1 days +23:59:59", not "-0 days 00:00:01"). Compute the
        // components with Euclidean div/rem so the remainder is in [0, 1 day).
        let days = nanos.div_euclid(Self::NANOS_PER_DAY);
        let rem = nanos.rem_euclid(Self::NANOS_PER_DAY);
        let hours = rem / Self::NANOS_PER_HOUR;
        let minutes = (rem % Self::NANOS_PER_HOUR) / Self::NANOS_PER_MIN;
        let seconds = (rem % Self::NANOS_PER_MIN) / Self::NANOS_PER_SEC;
        let frac = rem % Self::NANOS_PER_SEC;

        let time_part = format!("{hours:02}:{minutes:02}:{seconds:02}");
        // '+' joins the negative day count to the positive time remainder.
        let sep = if days < 0 { "+" } else { "" };

        if frac > 0 {
            // pandas renders the sub-second part with microsecond precision
            // (6 digits) unless a sub-microsecond (nanosecond) component is
            // present, in which case it widens to 9 digits.
            if frac % 1_000 == 0 {
                format!("{days} days {sep}{time_part}.{:06}", frac / 1_000)
            } else {
                format!("{days} days {sep}{time_part}.{frac:09}")
            }
        } else {
            format!("{days} days {sep}{time_part}")
        }
    }

    pub fn from_unit(value: f64, unit: &str) -> Result<i64, TimedeltaError> {
        let multiplier = Self::unit_to_nanos(unit)
            .ok_or_else(|| TimedeltaError::InvalidFormat(unit.to_string()))?;
        Ok((value * multiplier as f64).round() as i64)
    }

    // ── Arithmetic (br-frankenpandas-4r56 Phase 1) ──────────────────────
    //
    // NaT propagation: any arithmetic with `NAT` returns `NAT`. Matches
    // pandas `pd.NaT + anything == NaT`, `pd.NaT - anything == NaT`, etc.
    // Saturation: i64 overflow clamps to i64::MAX/MIN (never wraps). Matches
    // pandas's OverflowError surface at the type-system boundary.

    /// Add two Timedelta nanosecond values. NaT propagates; saturates on overflow.
    #[must_use]
    pub fn add(a: i64, b: i64) -> i64 {
        if a == Self::NAT || b == Self::NAT {
            return Self::NAT;
        }
        a.saturating_add(b)
    }

    /// Subtract two Timedelta nanosecond values. NaT propagates; saturates on overflow.
    #[must_use]
    pub fn sub(a: i64, b: i64) -> i64 {
        if a == Self::NAT || b == Self::NAT {
            return Self::NAT;
        }
        a.saturating_sub(b)
    }

    /// Negate a Timedelta value. NaT stays NaT. Saturates on overflow
    /// (pandas: `-pd.Timedelta.min` is NaT since min == -max - 1 cannot be negated).
    #[must_use]
    pub fn neg(a: i64) -> i64 {
        if a == Self::NAT {
            return Self::NAT;
        }
        a.saturating_neg()
    }

    /// Absolute value of a Timedelta. NaT stays NaT. Saturates on overflow.
    #[must_use]
    pub fn abs(a: i64) -> i64 {
        if a == Self::NAT {
            return Self::NAT;
        }
        a.saturating_abs()
    }

    /// Multiply a Timedelta value by an integer factor. NaT propagates;
    /// saturates on overflow.
    ///
    /// Matches pandas `pd.Timedelta(...) * int`.
    #[must_use]
    pub fn mul_scalar(a: i64, factor: i64) -> i64 {
        if a == Self::NAT {
            return Self::NAT;
        }
        a.saturating_mul(factor)
    }

    /// Floor-divide a Timedelta value by an integer divisor. NaT propagates.
    /// Returns NaT on divide-by-zero (matches pandas, which raises, but we
    /// surface as NaT to avoid panics at the type-system boundary).
    ///
    /// Matches pandas / Python `pd.Timedelta(...) // int`: floor division,
    /// not truncation toward zero. `-100 // 3 == -34`, and `100 // -3 ==
    /// -34`. Rust's `/` truncates toward zero and `div_euclid` keeps the
    /// remainder non-negative — neither matches pandas when the divisor is
    /// negative. This helper adjusts trunc-toward-zero into floor.
    #[must_use]
    pub fn div_scalar(a: i64, divisor: i64) -> i64 {
        if a == Self::NAT || divisor == 0 {
            return Self::NAT;
        }
        // NAT == i64::MIN so the classic `i64::MIN / -1` overflow path is
        // already handled by the NAT check above. `(i64::MIN + 1) / -1`
        // equals `i64::MAX` with no overflow, so we never need a
        // saturation branch here.
        let q = a / divisor;
        let r = a % divisor;
        // If remainder is non-zero and has opposite sign from divisor,
        // Rust's trunc-toward-zero `/` is one step above the floor. Adjust
        // down by 1 to match Python/pandas floor division.
        if r != 0 && (r < 0) != (divisor < 0) {
            q - 1
        } else {
            q
        }
    }

    /// Divide two Timedelta values, returning the ratio as f64.
    /// Matches pandas `pd.Timedelta(...) / pd.Timedelta(...)` → float.
    /// NaT in either operand → NaN. Zero divisor → ±Inf (per IEEE 754).
    #[must_use]
    pub fn div_timedelta(a: i64, b: i64) -> f64 {
        if a == Self::NAT || b == Self::NAT {
            return f64::NAN;
        }
        (a as f64) / (b as f64)
    }

    /// Returns ISO 8601 duration format string.
    ///
    /// Matches pandas `pd.Timedelta.isoformat()`. Returns format like
    /// "P1DT2H3M4.567890123S" for 1 day, 2 hours, 3 minutes, 4.567890123 seconds.
    /// NaT returns "NaT".
    #[must_use]
    pub fn isoformat(nanos: i64) -> String {
        if nanos == Self::NAT {
            return "NaT".to_string();
        }

        let negative = nanos < 0;
        let abs_nanos = nanos.saturating_abs();

        let days = abs_nanos / Self::NANOS_PER_DAY;
        let remaining = abs_nanos % Self::NANOS_PER_DAY;

        let hours = remaining / Self::NANOS_PER_HOUR;
        let remaining = remaining % Self::NANOS_PER_HOUR;

        let minutes = remaining / Self::NANOS_PER_MIN;
        let remaining = remaining % Self::NANOS_PER_MIN;

        let seconds = remaining / Self::NANOS_PER_SEC;
        let sub_sec_nanos = remaining % Self::NANOS_PER_SEC;

        let mut result = String::new();
        if negative {
            result.push('-');
        }

        result.push_str(&format!("P{days}DT{hours}H{minutes}M"));

        if sub_sec_nanos == 0 {
            result.push_str(&format!("{seconds}S"));
        } else {
            let frac = format!("{:09}", sub_sec_nanos);
            let trimmed = frac.trim_end_matches('0');
            result.push_str(&format!("{seconds}.{trimmed}S"));
        }

        result
    }

    /// Rounds down to the nearest frequency unit.
    ///
    /// Matches pandas `pd.Timedelta.floor(freq)`. NaT is preserved.
    #[must_use]
    pub fn floor(nanos: i64, freq: &str) -> i64 {
        if nanos == Self::NAT {
            return Self::NAT;
        }
        let Some(unit_nanos) = Self::unit_to_nanos(freq) else {
            return Self::NAT;
        };
        if unit_nanos == 0 {
            return Self::NAT;
        }
        let negative = nanos < 0;
        let abs_nanos = nanos.saturating_abs();
        let floored = (abs_nanos / unit_nanos) * unit_nanos;
        if negative { -floored } else { floored }
    }

    /// Rounds up to the nearest frequency unit.
    ///
    /// Matches pandas `pd.Timedelta.ceil(freq)`. NaT is preserved.
    #[must_use]
    pub fn ceil(nanos: i64, freq: &str) -> i64 {
        if nanos == Self::NAT {
            return Self::NAT;
        }
        let Some(unit_nanos) = Self::unit_to_nanos(freq) else {
            return Self::NAT;
        };
        if unit_nanos == 0 {
            return Self::NAT;
        }
        let negative = nanos < 0;
        let abs_nanos = nanos.saturating_abs();
        let ceiled = ((abs_nanos + unit_nanos - 1) / unit_nanos) * unit_nanos;
        if negative { -ceiled } else { ceiled }
    }

    /// Rounds to the nearest frequency unit.
    ///
    /// Matches pandas `pd.Timedelta.round(freq)`. Uses banker's rounding
    /// (round half to even). NaT is preserved.
    #[must_use]
    pub fn round(nanos: i64, freq: &str) -> i64 {
        if nanos == Self::NAT {
            return Self::NAT;
        }
        let Some(unit_nanos) = Self::unit_to_nanos(freq) else {
            return Self::NAT;
        };
        if unit_nanos == 0 {
            return Self::NAT;
        }
        let negative = nanos < 0;
        let abs_nanos = nanos.saturating_abs();

        let quotient = abs_nanos / unit_nanos;
        let remainder = abs_nanos % unit_nanos;
        let half = unit_nanos / 2;

        let rounded = if remainder > half {
            (quotient + 1) * unit_nanos
        } else if remainder < half {
            quotient * unit_nanos
        } else {
            // Exactly half: round to even
            if quotient % 2 == 0 {
                quotient * unit_nanos
            } else {
                (quotient + 1) * unit_nanos
            }
        };

        if negative { -rounded } else { rounded }
    }
}

// ── Timestamp types (br-frankenpandas-9p0u — 4r56 Phase 2) ─────────────
//
// Nanosecond-precision i64 since Unix epoch + optional IANA tz name.
// TZ-dependent arithmetic (DST transitions, tz conversion) is deferred
// to Phase 3 which pulls chrono_tz into fp-types; Phase 2 stores the
// tz name as opaque metadata and performs arithmetic on the absolute
// nanos axis only.

/// Number of days in a given month (1-12) of a given year.
fn days_in_month(year: i64, month: u32) -> Option<u32> {
    if !(1..=12).contains(&month) {
        return None;
    }
    let is_leap = (year % 4 == 0 && year % 100 != 0) || year % 400 == 0;
    let days: [u32; 12] = [
        31,
        if is_leap { 29 } else { 28 },
        31,
        30,
        31,
        30,
        31,
        31,
        30,
        31,
        30,
        31,
    ];
    Some(days[(month - 1) as usize])
}

/// Number of ISO-8601 weeks in a year (52 or 53).
///
/// A year has 53 ISO weeks iff its first day falls on a Thursday, or it is a
/// leap year whose first day is a Wednesday — captured by the dominical
/// closed form `p(year) == 4 || p(year - 1) == 3`, where
/// `p(y) = (y + ⌊y/4⌋ − ⌊y/100⌋ + ⌊y/400⌋) mod 7` is the weekday of Dec 31.
fn iso_weeks_in_year(year: i64) -> i64 {
    fn p(y: i64) -> i64 {
        (y + y.div_euclid(4) - y.div_euclid(100) + y.div_euclid(400)).rem_euclid(7)
    }
    if p(year) == 4 || p(year - 1) == 3 {
        53
    } else {
        52
    }
}

/// A nanosecond-precision point in time, Unix-epoch anchored.
///
/// Phase 2 scope: construction, arithmetic, equality, ordering, serde.
/// TZ semantics (IANA tz lookup, DST-aware shift) are deferred to Phase
/// 3 — see br-frankenpandas-4r56.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct Timestamp {
    /// Nanoseconds since Unix epoch. `i64::MIN` is NaT.
    pub nanos: i64,
    /// Optional IANA time-zone name (e.g. `"US/Eastern"`). `None` means
    /// naive / UTC-anchored. Phase 2 treats this as opaque metadata;
    /// Phase 3 wires chrono_tz interpretation.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tz: Option<String>,
}

impl Timestamp {
    /// NaT sentinel, parallel to `Timedelta::NAT`.
    pub const NAT: i64 = i64::MIN;

    /// Construct a UTC-anchored (tz=None) Timestamp from nanoseconds
    /// since Unix epoch.
    #[must_use]
    pub const fn from_nanos(nanos: i64) -> Self {
        Self { nanos, tz: None }
    }

    /// Construct a Timestamp tagged with an IANA tz name.
    ///
    /// Phase 2 doesn't interpret the tz — it only carries the name
    /// through arithmetic + serde. Phase 3 wires chrono_tz conversion.
    #[must_use]
    pub fn from_nanos_tz(nanos: i64, tz_name: impl Into<String>) -> Self {
        Self {
            nanos,
            tz: Some(tz_name.into()),
        }
    }

    /// Returns the current UTC timestamp.
    ///
    /// Matches `pd.Timestamp.now()` / `pd.Timestamp.utcnow()`.
    #[must_use]
    pub fn now() -> Self {
        use std::time::{SystemTime, UNIX_EPOCH};
        let duration = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default();
        let nanos = duration.as_nanos() as i64;
        Self { nanos, tz: None }
    }

    /// Alias for `now()`. Matches `pd.Timestamp.utcnow()`.
    #[must_use]
    pub fn utcnow() -> Self {
        Self::now()
    }

    /// Returns today's date at midnight UTC.
    ///
    /// Matches `pd.Timestamp.today()`.
    #[must_use]
    pub fn today() -> Self {
        let now = Self::now();
        now.normalize()
    }

    /// The NaT sentinel value for a Timestamp.
    #[must_use]
    pub const fn nat() -> Self {
        Self {
            nanos: Self::NAT,
            tz: None,
        }
    }

    /// True iff this Timestamp is NaT.
    #[must_use]
    pub const fn is_nat(&self) -> bool {
        self.nanos == Self::NAT
    }

    /// Nanoseconds since Unix epoch, matching `pd.Timestamp.value`.
    #[must_use]
    pub const fn value(&self) -> i64 {
        self.nanos
    }

    /// Stored resolution unit, matching `pd.Timestamp.unit`.
    ///
    /// FrankenPandas `Timestamp` stores nanoseconds internally, so non-NaT
    /// values report `ns`. `NaT` has no unit.
    #[must_use]
    pub const fn unit(&self) -> Option<&'static str> {
        if self.is_nat() { None } else { Some("ns") }
    }

    /// Return the resolution of the timestamp (always "ns" for nanoseconds).
    ///
    /// Matches `pd.Timestamp.resolution`. Returns None for NaT.
    #[must_use]
    pub const fn resolution(&self) -> Option<&'static str> {
        if self.is_nat() { None } else { Some("ns") }
    }

    /// Numpy datetime64 scalar payload, matching `pd.Timestamp.asm8`.
    #[must_use]
    pub const fn asm8(&self) -> i64 {
        self.value()
    }

    /// Convert to a datetime64 payload, matching `pd.Timestamp.to_datetime64()`.
    #[must_use]
    pub const fn to_datetime64(&self) -> i64 {
        self.value()
    }

    /// Convert to a numpy scalar payload, matching `pd.Timestamp.to_numpy()`.
    #[must_use]
    pub const fn to_numpy(&self) -> i64 {
        self.value()
    }

    /// POSIX timestamp in seconds, matching `pd.Timestamp.timestamp()`.
    ///
    /// Pandas exposes this through Python's datetime surface, so sub-microsecond
    /// nanoseconds are rounded to six decimal places. `NaT` raises in pandas;
    /// fp-types surfaces the same condition as a missing-value error.
    pub fn timestamp(&self) -> Result<f64, TypeError> {
        if self.is_nat() {
            return Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            });
        }
        let seconds = self.nanos as f64 / 1_000_000_000.0;
        let rounded = format!("{seconds:.6}").parse().unwrap_or(seconds);
        Ok(rounded)
    }

    /// Add a Timedelta. NaT in either operand → NaT; saturates on overflow.
    /// TZ is preserved from `self`.
    #[must_use]
    pub fn add_timedelta(&self, td_nanos: i64) -> Self {
        if self.is_nat() || td_nanos == Timedelta::NAT {
            return Self::nat();
        }
        Self {
            nanos: self.nanos.saturating_add(td_nanos),
            tz: self.tz.clone(),
        }
    }

    /// Subtract a Timedelta. NaT propagation + saturation; TZ preserved.
    #[must_use]
    pub fn sub_timedelta(&self, td_nanos: i64) -> Self {
        if self.is_nat() || td_nanos == Timedelta::NAT {
            return Self::nat();
        }
        Self {
            nanos: self.nanos.saturating_sub(td_nanos),
            tz: self.tz.clone(),
        }
    }

    /// Subtract another Timestamp. Returns a Timedelta (i64 nanos).
    /// NaT in either → `Timedelta::NAT`; saturates on overflow.
    #[must_use]
    pub fn sub_timestamp(&self, other: &Self) -> i64 {
        if self.is_nat() || other.is_nat() {
            return Timedelta::NAT;
        }
        self.nanos.saturating_sub(other.nanos)
    }

    /// NaT-aware semantic equality: two NaT Timestamps are equal to each
    /// other (matches pandas `pd.NaT == pd.NaT` under `equals()`, though
    /// pandas's `==` operator returns False for NaT==NaT — we follow the
    /// `semantic_eq` convention used elsewhere in fp-types).
    #[must_use]
    pub fn semantic_eq(&self, other: &Self) -> bool {
        if self.is_nat() && other.is_nat() {
            return true;
        }
        if self.is_nat() || other.is_nat() {
            return false;
        }
        self.nanos == other.nanos && self.tz == other.tz
    }

    // ── Rounding to a Timedelta unit (br-frankenpandas-5h6n) ────────────
    //
    // Pure i64 arithmetic on the nanos axis. tz is preserved. Phase 3
    // chrono_tz integration will add a tz-aware variant that handles DST
    // boundaries correctly; these methods operate on the absolute time
    // axis, matching pandas's tz-naive `.floor` / `.ceil` / `.round`
    // semantics for unit values smaller than a day.

    /// Round down to the nearest multiple of `unit_nanos`.
    ///
    /// Matches `pd.Timestamp(...).floor(unit)`. NaT in `self` or a
    /// non-positive `unit_nanos` returns NaT.
    #[must_use]
    pub fn floor_to(&self, unit_nanos: i64) -> Self {
        if self.is_nat() || unit_nanos <= 0 {
            return Self::nat();
        }
        Self {
            nanos: self.nanos.div_euclid(unit_nanos) * unit_nanos,
            tz: self.tz.clone(),
        }
    }

    /// Round up to the nearest multiple of `unit_nanos`.
    ///
    /// Matches `pd.Timestamp(...).ceil(unit)`. NaT or non-positive
    /// `unit_nanos` returns NaT. Already-multiple inputs return self.
    #[must_use]
    pub fn ceil_to(&self, unit_nanos: i64) -> Self {
        if self.is_nat() || unit_nanos <= 0 {
            return Self::nat();
        }
        let rem = self.nanos.rem_euclid(unit_nanos);
        let nanos = if rem == 0 {
            self.nanos
        } else {
            self.nanos.saturating_add(unit_nanos - rem)
        };
        Self {
            nanos,
            tz: self.tz.clone(),
        }
    }

    /// Round to the nearest multiple of `unit_nanos`, banker's rounding
    /// (half-to-even) on ties.
    ///
    /// Matches `pd.Timestamp(...).round(unit)`. NaT or non-positive
    /// `unit_nanos` returns NaT.
    #[must_use]
    pub fn round_to(&self, unit_nanos: i64) -> Self {
        if self.is_nat() || unit_nanos <= 0 {
            return Self::nat();
        }
        let floor = self.nanos.div_euclid(unit_nanos);
        let rem = self.nanos.rem_euclid(unit_nanos);
        let half = unit_nanos / 2;
        let chosen_floor = if rem < half {
            floor
        } else if rem > half {
            floor + 1
        } else if unit_nanos % 2 != 0 {
            // Odd unit can't have a true half; treat as round-up.
            floor + 1
        } else {
            // Tie: pick the even multiple.
            if floor % 2 == 0 { floor } else { floor + 1 }
        };
        Self {
            nanos: chosen_floor.saturating_mul(unit_nanos),
            tz: self.tz.clone(),
        }
    }

    // ── String-unit rounding (br-frankenpandas-lbsx) ────────────────────
    //
    // Pandas convenience: `.floor('H')` / `.ceil('1D')` / `.round('s')`.
    // These delegate to `Timedelta::unit_to_nanos` for unit lookup, then to
    // the nanos-based `floor_to`/`ceil_to`/`round_to`. Unknown unit strings
    // return NaT, matching the rest of fp-types' "missing-input → missing-
    // output" convention.

    /// Round down to the nearest multiple of the named unit.
    ///
    /// Matches `pd.Timestamp(...).floor(unit)`. Unknown unit → NaT.
    #[must_use]
    pub fn floor_to_unit(&self, unit: &str) -> Self {
        match Timedelta::unit_to_nanos(unit) {
            Some(unit_nanos) => self.floor_to(unit_nanos),
            None => Self::nat(),
        }
    }

    /// Round up to the nearest multiple of the named unit.
    ///
    /// Matches `pd.Timestamp(...).ceil(unit)`. Unknown unit → NaT.
    #[must_use]
    pub fn ceil_to_unit(&self, unit: &str) -> Self {
        match Timedelta::unit_to_nanos(unit) {
            Some(unit_nanos) => self.ceil_to(unit_nanos),
            None => Self::nat(),
        }
    }

    /// Round to the nearest multiple of the named unit, banker's rounding.
    ///
    /// Matches `pd.Timestamp(...).round(unit)`. Unknown unit → NaT.
    #[must_use]
    pub fn round_to_unit(&self, unit: &str) -> Self {
        match Timedelta::unit_to_nanos(unit) {
            Some(unit_nanos) => self.round_to(unit_nanos),
            None => Self::nat(),
        }
    }

    /// Pandas-named alias for [`floor_to_unit`](Self::floor_to_unit).
    #[must_use]
    pub fn floor(&self, freq: &str) -> Self {
        self.floor_to_unit(freq)
    }

    /// Pandas-named alias for [`ceil_to_unit`](Self::ceil_to_unit).
    #[must_use]
    pub fn ceil(&self, freq: &str) -> Self {
        self.ceil_to_unit(freq)
    }

    /// Pandas-named alias for [`round_to_unit`](Self::round_to_unit).
    #[must_use]
    pub fn round(&self, freq: &str) -> Self {
        self.round_to_unit(freq)
    }

    /// Extract the year component from the timestamp.
    ///
    /// Matches `pd.Timestamp.year`. Returns None for NaT.
    #[must_use]
    pub fn year(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let days_since_epoch = total_secs / 86400;
        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let y = yoe + era * 400;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        let m = if mp < 10 { mp + 3 } else { mp - 9 };
        Some(if m <= 2 { y + 1 } else { y })
    }

    /// Extract the month component (1-12) from the timestamp.
    ///
    /// Matches `pd.Timestamp.month`. Returns None for NaT.
    #[must_use]
    pub fn month(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let days_since_epoch = total_secs / 86400;
        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        Some(if mp < 10 { mp + 3 } else { mp - 9 })
    }

    /// Extract the day component (1-31) from the timestamp.
    ///
    /// Matches `pd.Timestamp.day`. Returns None for NaT.
    #[must_use]
    pub fn day(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let days_since_epoch = total_secs / 86400;
        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        Some(doy - (153 * mp + 2) / 5 + 1)
    }

    /// Extract the hour component (0-23) from the timestamp.
    ///
    /// Matches `pd.Timestamp.hour`. Returns None for NaT.
    #[must_use]
    pub fn hour(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let secs_of_day = (total_secs % 86400 + 86400) % 86400;
        Some(secs_of_day / 3600)
    }

    /// Extract the minute component (0-59) from the timestamp.
    ///
    /// Matches `pd.Timestamp.minute`. Returns None for NaT.
    #[must_use]
    pub fn minute(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let secs_of_day = (total_secs % 86400 + 86400) % 86400;
        Some((secs_of_day % 3600) / 60)
    }

    /// Extract the second component (0-59) from the timestamp.
    ///
    /// Matches `pd.Timestamp.second`. Returns None for NaT.
    #[must_use]
    pub fn second(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let secs_of_day = (total_secs % 86400 + 86400) % 86400;
        Some(secs_of_day % 60)
    }

    /// Extract the microsecond component (0-999999) from the timestamp.
    ///
    /// Matches `pd.Timestamp.microsecond`. Returns None for NaT.
    #[must_use]
    pub fn microsecond(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let sub_nanos = (self.nanos % Timedelta::NANOS_PER_SEC).unsigned_abs();
        Some((sub_nanos / 1000) as i64)
    }

    /// Extract the nanosecond component (0-999) from the timestamp.
    ///
    /// Matches `pd.Timestamp.nanosecond`. Returns None for NaT.
    #[must_use]
    pub fn nanosecond(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let sub_nanos = (self.nanos % Timedelta::NANOS_PER_SEC).unsigned_abs();
        Some((sub_nanos % 1000) as i64)
    }

    /// Return the day of the week (Monday=0, Sunday=6).
    ///
    /// Matches `pd.Timestamp.dayofweek`. Returns None for NaT.
    #[must_use]
    pub fn dayofweek(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let days_since_epoch = self.nanos / Timedelta::NANOS_PER_DAY;
        let dow = ((days_since_epoch + 3) % 7 + 7) % 7;
        Some(dow)
    }

    /// Alias for dayofweek(). Matches `pd.Timestamp.weekday`.
    #[must_use]
    pub fn weekday(&self) -> Option<i64> {
        self.dayofweek()
    }

    /// Alias for dayofweek(). Matches `pd.Timestamp.day_of_week`.
    #[must_use]
    pub fn day_of_week(&self) -> Option<i64> {
        self.dayofweek()
    }

    /// Return the day of the year (1-366).
    ///
    /// Matches `pd.Timestamp.dayofyear`. Returns None for NaT.
    #[must_use]
    pub fn dayofyear(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let m = self.month()?;
        let d = self.day()?;
        let y = self.year()?;
        let is_leap = (y % 4 == 0 && y % 100 != 0) || y % 400 == 0;
        let days_before: [i64; 12] = [0, 31, 59, 90, 120, 151, 181, 212, 243, 273, 304, 334];
        let base = days_before[(m - 1) as usize] + d;
        if is_leap && m > 2 {
            Some(base + 1)
        } else {
            Some(base)
        }
    }

    /// Alias for dayofyear(). Matches `pd.Timestamp.day_of_year`.
    #[must_use]
    pub fn day_of_year(&self) -> Option<i64> {
        self.dayofyear()
    }

    /// Return the proleptic Gregorian ordinal (number of days since Jan 1, year 1).
    ///
    /// Matches `pd.Timestamp.toordinal()`. Returns None for NaT.
    #[must_use]
    pub fn toordinal(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let y = self.year()?;
        let m = self.month()?;
        let d = self.day()?;
        // Algorithm: count days from year 1 to the start of the given year,
        // add days in the months before the given month, add the day of month.
        // Account for leap years.
        let y_minus_1 = y - 1;
        let mut ordinal = y_minus_1 * 365 + y_minus_1 / 4 - y_minus_1 / 100 + y_minus_1 / 400;
        let is_leap = (y % 4 == 0 && y % 100 != 0) || y % 400 == 0;
        let days_before: [i64; 12] = [0, 31, 59, 90, 120, 151, 181, 212, 243, 273, 304, 334];
        ordinal += days_before[(m - 1) as usize];
        if is_leap && m > 2 {
            ordinal += 1;
        }
        ordinal += d;
        Some(ordinal)
    }

    /// Construct a Timestamp from a proleptic Gregorian ordinal.
    ///
    /// Matches `pd.Timestamp.fromordinal(ordinal)`. Returns NaT for invalid ordinals.
    #[must_use]
    pub fn fromordinal(ordinal: i64) -> Self {
        if ordinal <= 0 {
            return Self {
                nanos: Self::NAT,
                tz: None,
            };
        }
        // Convert y/m/d to days since Unix epoch, then to nanos
        // Unix epoch is 1970-01-01, which is ordinal 719163
        let days_since_epoch = ordinal - 719163;
        let nanos = days_since_epoch * 24 * 60 * 60 * 1_000_000_000_i64;
        Self { nanos, tz: None }
    }

    /// Return the Julian Date (astronomical day number).
    ///
    /// Matches `pd.Timestamp.to_julian_date()`. Returns NaN for NaT.
    /// The Julian Date is the continuous count of days since the beginning
    /// of the Julian Period (January 1, 4713 BC in the proleptic Julian calendar).
    #[must_use]
    pub fn to_julian_date(&self) -> f64 {
        if self.is_nat() {
            return f64::NAN;
        }
        // Gregorian ordinal 1 (Jan 1, year 1) corresponds to Julian Day 1721425.5
        // (at noon, since JD starts at noon)
        // For a timestamp at midnight, we subtract 0.5
        let ordinal = match self.toordinal() {
            Some(o) => o,
            None => return f64::NAN,
        };
        // Fractional day from time components
        let h = self.hour().unwrap_or(0) as f64;
        let m = self.minute().unwrap_or(0) as f64;
        let s = self.second().unwrap_or(0) as f64;
        let us = self.microsecond().unwrap_or(0) as f64;
        let ns = self.nanosecond().unwrap_or(0) as f64;
        let frac_day =
            (h + m / 60.0 + s / 3600.0 + us / 3_600_000_000.0 + ns / 3_600_000_000_000.0) / 24.0;
        // Julian day at midnight of ordinal 1 is 1721424.5
        1721424.5 + ordinal as f64 + frac_day
    }

    /// Return the quarter (1-4) of the year.
    ///
    /// Matches `pd.Timestamp.quarter`. Returns None for NaT.
    #[must_use]
    pub fn quarter(&self) -> Option<i64> {
        self.month().map(|m| (m - 1) / 3 + 1)
    }

    /// Return the ISO week number (1-53).
    ///
    /// Matches `pd.Timestamp.week`. Returns None for NaT.
    #[must_use]
    pub fn weekofyear(&self) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        let doy = self.dayofyear()?;
        let dow = self.dayofweek()?;
        let year = self.year()?;
        let iso_dow = if dow == 6 { 7 } else { dow + 1 };
        let week = (doy - iso_dow + 10) / 7;
        // ISO-8601 has 53-week years, so the clamps must consult the actual
        // week count, not hardcode 52/1: a week<1 belongs to the LAST week of
        // the previous year (52 OR 53), and a week beyond this year's count
        // wraps to week 1 of the next year. pandas isocalendar().week agrees:
        // 2021-01-01 -> 53 (2020 is a 53-week year), 2026-12-31 -> 53.
        if week < 1 {
            Some(iso_weeks_in_year(year - 1))
        } else if week > iso_weeks_in_year(year) {
            Some(1)
        } else {
            Some(week)
        }
    }

    /// Alias for weekofyear(). Matches `pd.Timestamp.week`.
    #[must_use]
    pub fn week(&self) -> Option<i64> {
        self.weekofyear()
    }

    /// Return the timestamp value in the specified unit.
    ///
    /// Matches `pd.Timestamp.value` when unit is nanoseconds.
    /// Supported units: "ns", "us", "ms", "s".
    #[must_use]
    pub fn to_unit(&self, unit: &str) -> Option<i64> {
        if self.is_nat() {
            return None;
        }
        match unit {
            "ns" | "nanosecond" | "nanoseconds" => Some(self.nanos),
            "us" | "microsecond" | "microseconds" => Some(self.nanos / 1_000),
            "ms" | "millisecond" | "milliseconds" => Some(self.nanos / 1_000_000),
            "s" | "second" | "seconds" => Some(self.nanos / 1_000_000_000),
            _ => None,
        }
    }

    /// Whether the year is a leap year.
    ///
    /// Matches `pd.Timestamp.is_leap_year`. Returns None for NaT.
    #[must_use]
    pub fn is_leap_year(&self) -> Option<bool> {
        self.year()
            .map(|y| (y % 4 == 0 && y % 100 != 0) || y % 400 == 0)
    }

    /// Whether the day is the first day of the month.
    ///
    /// Matches `pd.Timestamp.is_month_start`. Returns None for NaT.
    #[must_use]
    pub fn is_month_start(&self) -> Option<bool> {
        self.day().map(|d| d == 1)
    }

    /// Whether the day is the last day of the month.
    ///
    /// Matches `pd.Timestamp.is_month_end`. Returns None for NaT.
    #[must_use]
    pub fn is_month_end(&self) -> Option<bool> {
        let y = self.year()?;
        let m = self.month()?;
        let d = self.day()?;
        let is_leap = (y % 4 == 0 && y % 100 != 0) || y % 400 == 0;
        let days_in_month: [i64; 12] = [
            31,
            if is_leap { 29 } else { 28 },
            31,
            30,
            31,
            30,
            31,
            31,
            30,
            31,
            30,
            31,
        ];
        Some(d == days_in_month[(m - 1) as usize])
    }

    /// Whether the day is the first day of a quarter.
    ///
    /// Matches `pd.Timestamp.is_quarter_start`. Returns None for NaT.
    #[must_use]
    pub fn is_quarter_start(&self) -> Option<bool> {
        let m = self.month()?;
        let d = self.day()?;
        Some(d == 1 && (m == 1 || m == 4 || m == 7 || m == 10))
    }

    /// Whether the day is the last day of a quarter.
    ///
    /// Matches `pd.Timestamp.is_quarter_end`. Returns None for NaT.
    #[must_use]
    pub fn is_quarter_end(&self) -> Option<bool> {
        let m = self.month()?;
        let d = self.day()?;
        Some(
            (m == 3 && d == 31)
                || (m == 6 && d == 30)
                || (m == 9 && d == 30)
                || (m == 12 && d == 31),
        )
    }

    /// Whether the day is the first day of the year (Jan 1).
    ///
    /// Matches `pd.Timestamp.is_year_start`. Returns None for NaT.
    #[must_use]
    pub fn is_year_start(&self) -> Option<bool> {
        let m = self.month()?;
        let d = self.day()?;
        Some(m == 1 && d == 1)
    }

    /// Whether the day is the last day of the year (Dec 31).
    ///
    /// Matches `pd.Timestamp.is_year_end`. Returns None for NaT.
    #[must_use]
    pub fn is_year_end(&self) -> Option<bool> {
        let m = self.month()?;
        let d = self.day()?;
        Some(m == 12 && d == 31)
    }

    /// Return the number of days in the month of this timestamp.
    ///
    /// Matches `pd.Timestamp.days_in_month`. Returns None for NaT.
    #[must_use]
    pub fn days_in_month(&self) -> Option<i64> {
        let y = self.year()?;
        let m = self.month()?;
        let is_leap = (y % 4 == 0 && y % 100 != 0) || y % 400 == 0;
        let days: [i64; 12] = [
            31,
            if is_leap { 29 } else { 28 },
            31,
            30,
            31,
            30,
            31,
            31,
            30,
            31,
            30,
            31,
        ];
        Some(days[(m - 1) as usize])
    }

    /// Alias for days_in_month(). Matches `pd.Timestamp.daysinmonth`.
    #[must_use]
    pub fn daysinmonth(&self) -> Option<i64> {
        self.days_in_month()
    }

    /// Normalize to midnight/day boundary, matching `pd.Timestamp.normalize()`.
    #[must_use]
    pub fn normalize(&self) -> Self {
        self.floor_to_unit("D")
    }

    /// Replace timestamp components with new values.
    ///
    /// Matches pd.Timestamp.replace(). None values keep the existing component.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub fn replace(
        &self,
        year: Option<i64>,
        month: Option<i64>,
        day: Option<i64>,
        hour: Option<i64>,
        minute: Option<i64>,
        second: Option<i64>,
        microsecond: Option<i64>,
        nanosecond: Option<i64>,
    ) -> Self {
        if self.is_nat() {
            return self.clone();
        }
        let cur_year = self.year().unwrap_or(1970);
        let cur_month = self.month().unwrap_or(1);
        let cur_day = self.day().unwrap_or(1);
        let cur_hour = self.hour().unwrap_or(0);
        let cur_minute = self.minute().unwrap_or(0);
        let cur_second = self.second().unwrap_or(0);
        let cur_micro = self.microsecond().unwrap_or(0);
        let cur_nano = self.nanosecond().unwrap_or(0);

        let y = year.unwrap_or(cur_year);
        let mo = month.unwrap_or(cur_month);
        let d = day.unwrap_or(cur_day);
        let h = hour.unwrap_or(cur_hour);
        let mi = minute.unwrap_or(cur_minute);
        let s = second.unwrap_or(cur_second);
        let us = microsecond.unwrap_or(cur_micro);
        let ns = nanosecond.unwrap_or(cur_nano);

        let days_from_epoch = Self::days_from_ymd(y, mo, d);
        let secs = h * 3600 + mi * 60 + s;
        let total_nanos = days_from_epoch * Timedelta::NANOS_PER_DAY
            + secs * Timedelta::NANOS_PER_SEC
            + us * Timedelta::NANOS_PER_MICRO
            + ns;

        Self {
            nanos: total_nanos,
            tz: self.tz.clone(),
        }
    }

    fn days_from_ymd(year: i64, month: i64, day: i64) -> i64 {
        let y = if month <= 2 { year - 1 } else { year };
        let era = if y >= 0 { y } else { y - 399 } / 400;
        let yoe = y - era * 400;
        let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
        let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
        era * 146097 + doe - 719468
    }

    /// Return an ISO 8601 string representation of the timestamp.
    ///
    /// Matches `pd.Timestamp.isoformat()`. NaT returns "NaT".
    #[must_use]
    pub fn isoformat(&self) -> String {
        if self.is_nat() {
            return "NaT".to_string();
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let sub_nanos = (self.nanos % Timedelta::NANOS_PER_SEC).unsigned_abs();
        let days_since_epoch = total_secs / 86400;
        let secs_of_day = (total_secs % 86400 + 86400) % 86400;

        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let y = yoe + era * 400;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        let d = doy - (153 * mp + 2) / 5 + 1;
        let m = if mp < 10 { mp + 3 } else { mp - 9 };
        let year = if m <= 2 { y + 1 } else { y };

        let hour = secs_of_day / 3600;
        let minute = (secs_of_day % 3600) / 60;
        let second = secs_of_day % 60;

        let base = if sub_nanos == 0 {
            format!("{year:04}-{m:02}-{d:02}T{hour:02}:{minute:02}:{second:02}")
        } else {
            let micros = sub_nanos / 1000;
            format!("{year:04}-{m:02}-{d:02}T{hour:02}:{minute:02}:{second:02}.{micros:06}")
        };
        match &self.tz {
            Some(tz) if tz == "UTC" => format!("{base}+00:00"),
            Some(tz) => format!("{base}[{tz}]"),
            None => base,
        }
    }

    /// Alias for isoformat.
    #[must_use]
    pub fn to_iso8601(&self) -> String {
        self.isoformat()
    }

    /// Parse a datetime string into a Timestamp.
    ///
    /// Supports ISO 8601 formats:
    /// - "2024-01-15" (date only, time defaults to 00:00:00)
    /// - "2024-01-15T10:30:00" (datetime)
    /// - "2024-01-15 10:30:00" (space separator)
    /// - "2024-01-15T10:30:00.123456" (with fractional seconds)
    /// - "2024-01-15T10:30:00Z" (UTC timezone)
    /// - "2024-01-15T10:30:00+05:30" (offset timezone)
    /// - "NaT" (Not a Timestamp)
    ///
    /// Matches `pd.Timestamp()` constructor behavior.
    pub fn parse(s: &str) -> Result<Self, TypeError> {
        let s = s.trim();

        if s.eq_ignore_ascii_case("nat") {
            return Ok(Self::nat());
        }

        let (datetime_part, tz) = Self::split_timezone(s);

        let (date_part, time_part) = if datetime_part.contains('T') {
            datetime_part
                .split_once('T')
                .ok_or_else(|| TypeError::ValueNotParseable {
                    value: s.to_string(),
                    target: "Timestamp".to_string(),
                })?
        } else if datetime_part.contains(' ')
            && datetime_part.chars().filter(|&c| c == ' ').count() == 1
        {
            datetime_part
                .split_once(' ')
                .ok_or_else(|| TypeError::ValueNotParseable {
                    value: s.to_string(),
                    target: "Timestamp".to_string(),
                })?
        } else {
            (datetime_part, "00:00:00")
        };

        let (year, month, day) =
            Self::parse_date(date_part).ok_or_else(|| TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Timestamp".to_string(),
            })?;

        let (hour, minute, second, nanos) =
            Self::parse_time(time_part).ok_or_else(|| TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Timestamp".to_string(),
            })?;

        let total_nanos = Self::ymd_hms_to_nanos(year, month, day, hour, minute, second, nanos);

        Ok(if let Some(tz_name) = tz {
            Self::from_nanos_tz(total_nanos, tz_name)
        } else {
            Self::from_nanos(total_nanos)
        })
    }

    fn split_timezone(s: &str) -> (&str, Option<String>) {
        if let Some(stripped) = s.strip_suffix('Z') {
            (stripped, Some("UTC".to_string()))
        } else if let Some(idx) = s.rfind('+') {
            if idx > 10 {
                (&s[..idx], Some(s[idx..].to_string()))
            } else {
                (s, None)
            }
        } else if let Some(idx) = s.rfind('-') {
            if idx > 10 && s[idx..].contains(':') {
                (&s[..idx], Some(s[idx..].to_string()))
            } else {
                (s, None)
            }
        } else {
            (s, None)
        }
    }

    fn parse_date(s: &str) -> Option<(i64, u32, u32)> {
        let parts: Vec<&str> = s.split('-').collect();
        if parts.len() != 3 {
            return None;
        }
        let year: i64 = parts[0].parse().ok()?;
        let month: u32 = parts[1].parse().ok()?;
        let day: u32 = parts[2].parse().ok()?;
        if !(1..=days_in_month(year, month)?).contains(&day) {
            return None;
        }
        Some((year, month, day))
    }

    fn parse_time(s: &str) -> Option<(u32, u32, u32, u64)> {
        let (time_str, frac_str) = s.split_once('.').unwrap_or((s, ""));
        let parts: Vec<&str> = time_str.split(':').collect();
        if parts.is_empty() || parts.len() > 3 {
            return None;
        }
        let hour: u32 = parts.first().and_then(|p| p.parse().ok())?;
        let minute: u32 = parts.get(1).and_then(|p| p.parse().ok()).unwrap_or(0);
        let second: u32 = parts.get(2).and_then(|p| p.parse().ok()).unwrap_or(0);

        if hour > 23 || minute > 59 || second > 59 {
            return None;
        }

        let nanos = if frac_str.is_empty() {
            0
        } else {
            let padded = format!("{:0<9}", &frac_str[..frac_str.len().min(9)]);
            padded.parse::<u64>().unwrap_or(0)
        };

        Some((hour, minute, second, nanos))
    }

    fn ymd_hms_to_nanos(
        year: i64,
        month: u32,
        day: u32,
        hour: u32,
        minute: u32,
        second: u32,
        sub_nanos: u64,
    ) -> i64 {
        let m = month as i64;
        let d = day as i64;

        let y = if m <= 2 { year - 1 } else { year };
        let era = if y >= 0 { y } else { y - 399 } / 400;
        let yoe = y - era * 400;
        let doy = (153 * (if m > 2 { m - 3 } else { m + 9 }) + 2) / 5 + d - 1;
        let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
        let days_since_epoch = era * 146_097 + doe - 719_468;

        let total_seconds = days_since_epoch * 86400
            + (hour as i64) * 3600
            + (minute as i64) * 60
            + (second as i64);
        total_seconds * Timedelta::NANOS_PER_SEC + sub_nanos as i64
    }

    /// Format timestamp using strftime directives.
    ///
    /// Matches `pd.Timestamp.strftime(format)`. Supports: %Y (year), %m (month),
    /// %d (day), %H (hour), %M (minute), %S (second), %f (microsecond).
    /// NaT returns "NaT".
    #[must_use]
    pub fn strftime(&self, format: &str) -> String {
        if self.is_nat() {
            return "NaT".to_string();
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let sub_nanos = (self.nanos % Timedelta::NANOS_PER_SEC).unsigned_abs();

        let days_since_epoch = total_secs / 86400;
        let secs_of_day = (total_secs % 86400 + 86400) % 86400;

        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let y = yoe + era * 400;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        let d = doy - (153 * mp + 2) / 5 + 1;
        let m = if mp < 10 { mp + 3 } else { mp - 9 };
        let year = if m <= 2 { y + 1 } else { y };

        let hour = secs_of_day / 3600;
        let minute = (secs_of_day % 3600) / 60;
        let second = secs_of_day % 60;
        let micros = sub_nanos / 1000;

        format
            .replace("%Y", &format!("{year:04}"))
            .replace("%m", &format!("{m:02}"))
            .replace("%d", &format!("{d:02}"))
            .replace("%H", &format!("{hour:02}"))
            .replace("%M", &format!("{minute:02}"))
            .replace("%S", &format!("{second:02}"))
            .replace("%f", &format!("{micros:06}"))
    }

    /// Return the day of the week as a string (e.g., "Monday").
    ///
    /// Matches `pd.Timestamp.day_name()`. NaT returns "NaT".
    #[must_use]
    pub fn day_name(&self) -> String {
        const NAMES: [&str; 7] = [
            "Thursday",
            "Friday",
            "Saturday",
            "Sunday",
            "Monday",
            "Tuesday",
            "Wednesday",
        ];
        if self.is_nat() {
            return "NaT".to_string();
        }
        let days_since_epoch = self.nanos / Timedelta::NANOS_PER_DAY;
        let dow = ((days_since_epoch % 7) + 7) % 7;
        NAMES[dow as usize].to_string()
    }

    /// Return the month name as a string (e.g., "January").
    ///
    /// Matches `pd.Timestamp.month_name()`. NaT returns "NaT".
    #[must_use]
    pub fn month_name(&self) -> String {
        const NAMES: [&str; 12] = [
            "January",
            "February",
            "March",
            "April",
            "May",
            "June",
            "July",
            "August",
            "September",
            "October",
            "November",
            "December",
        ];
        if self.is_nat() {
            return "NaT".to_string();
        }
        let total_secs = self.nanos / Timedelta::NANOS_PER_SEC;
        let days_since_epoch = total_secs / 86400;
        let days = days_since_epoch + 719_468;
        let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
        let doe = days - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        let m = if mp < 10 { mp + 3 } else { mp - 9 };
        NAMES[(m - 1) as usize].to_string()
    }

    /// Localize a naive timestamp to a timezone.
    ///
    /// Matches `pd.Timestamp.tz_localize(tz)`. If `tz` is None, removes the
    /// timezone (makes timestamp naive). NaT propagates.
    #[must_use]
    pub fn tz_localize(&self, tz: Option<&str>) -> Self {
        if self.is_nat() {
            return Self::nat();
        }
        Self {
            nanos: self.nanos,
            tz: tz.map(String::from),
        }
    }

    /// Convert timezone-aware timestamp to another timezone.
    ///
    /// Matches `pd.Timestamp.tz_convert(tz)`. If timestamp is naive (no tz),
    /// the timezone is simply attached without conversion. NaT propagates.
    /// Note: actual UTC offset conversion requires chrono-tz (Phase 3).
    #[must_use]
    pub fn tz_convert(&self, tz: &str) -> Self {
        if self.is_nat() {
            return Self::nat();
        }
        Self {
            nanos: self.nanos,
            tz: Some(tz.to_string()),
        }
    }

    /// Create a Timestamp from a Unix timestamp (seconds since epoch).
    ///
    /// Matches `pd.Timestamp.fromtimestamp(ts)`. The optional `tz` parameter
    /// specifies the timezone to localize to.
    #[must_use]
    pub fn fromtimestamp(ts: f64, tz: Option<&str>) -> Self {
        if ts.is_nan() || ts.is_infinite() {
            return Self::nat();
        }
        let nanos_f64 = ts * 1_000_000_000.0;
        // Check for overflow before casting - i64 range is roughly ±9.2e18
        const MAX_NANOS: f64 = i64::MAX as f64;
        const MIN_NANOS: f64 = i64::MIN as f64;
        if !(MIN_NANOS..=MAX_NANOS).contains(&nanos_f64) {
            return Self::nat();
        }
        Self {
            nanos: nanos_f64 as i64,
            tz: tz.map(String::from),
        }
    }

    /// Create a Timestamp from milliseconds since epoch.
    ///
    /// Convenience constructor complementing fromtimestamp.
    #[must_use]
    pub fn from_millis(ms: i64, tz: Option<&str>) -> Self {
        Self {
            nanos: ms.saturating_mul(1_000_000),
            tz: tz.map(String::from),
        }
    }

    /// Create a Timestamp from microseconds since epoch.
    ///
    /// Convenience constructor complementing fromtimestamp.
    #[must_use]
    pub fn from_micros(us: i64, tz: Option<&str>) -> Self {
        Self {
            nanos: us.saturating_mul(1_000),
            tz: tz.map(String::from),
        }
    }

    /// Return the timezone string, or None if naive.
    #[must_use]
    pub fn tzinfo(&self) -> Option<&str> {
        self.tz.as_deref()
    }

    /// Return the timezone name, or None if naive.
    ///
    /// Alias for tzinfo() matching pandas Timestamp.tzname().
    #[must_use]
    pub fn tzname(&self) -> Option<&str> {
        self.tzinfo()
    }
}

impl std::fmt::Display for Timestamp {
    /// Phase 2 debug-style format; Phase 3 replaces with pandas ISO-8601
    /// notation once chrono interpretation lands.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.is_nat() {
            return f.write_str("NaT");
        }
        match &self.tz {
            Some(tz) => write!(f, "Timestamp[{}, {}]", self.nanos, tz),
            None => write!(f, "Timestamp[{}, UTC]", self.nanos),
        }
    }
}

impl PartialOrd for Timestamp {
    /// Orders by nanos axis; NaT is incomparable (`None`). Tz difference
    /// does not affect ordering — two Timestamps at the same absolute
    /// nanos compare equal regardless of tz label (Phase 3 will revisit
    /// whether tz affects ordering semantics).
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        if self.is_nat() || other.is_nat() {
            return None;
        }
        Some(self.nanos.cmp(&other.nanos))
    }
}

// ── Missingness utilities ──────────────────────────────────────────────

pub fn isna(values: &[Scalar]) -> Vec<bool> {
    values.iter().map(Scalar::is_missing).collect()
}

pub fn isnull(values: &[Scalar]) -> Vec<bool> {
    isna(values)
}

pub fn notna(values: &[Scalar]) -> Vec<bool> {
    values.iter().map(|v| !v.is_missing()).collect()
}

pub fn notnull(values: &[Scalar]) -> Vec<bool> {
    notna(values)
}

pub fn count_na(values: &[Scalar]) -> usize {
    values.iter().filter(|v| v.is_missing()).count()
}

pub fn fill_na(values: &[Scalar], fill: &Scalar) -> Vec<Scalar> {
    values
        .iter()
        .map(|v| {
            if v.is_missing() {
                fill.clone()
            } else {
                v.clone()
            }
        })
        .collect()
}

pub fn dropna(values: &[Scalar]) -> Vec<Scalar> {
    values.iter().filter(|v| !v.is_missing()).cloned().collect()
}

// ── Nanops: null-skipping numeric reductions ───────────────────────────

fn collect_finite(values: &[Scalar]) -> Vec<f64> {
    values
        .iter()
        .filter(|v| !v.is_missing())
        .filter_map(|v| v.to_f64().ok())
        .collect()
}

/// Per br-frankenpandas-620mj: if a column is uniformly Timedelta64
/// (with optional NAT/Null missing), sum/mean preserve Timedelta dtype
/// matching pandas — instead of silently coercing to Float64(0.0) via
/// the collect_finite path (which drops Timedelta64 because to_f64
/// errors). Returns Some(sum_in_ns, observed_count) when applicable.
fn collect_timedelta_ns(values: &[Scalar]) -> Option<(i128, usize)> {
    let mut sum: i128 = 0;
    let mut count: usize = 0;
    let mut saw_timedelta = false;
    for v in values {
        if v.is_missing() {
            continue;
        }
        match v {
            Scalar::Timedelta64(ns) => {
                saw_timedelta = true;
                sum += i128::from(*ns);
                count += 1;
            }
            // Any non-Timedelta non-missing value bails out to the
            // existing Float64 path, preserving cross-type behavior.
            _ => return None,
        }
    }
    if saw_timedelta {
        Some((sum, count))
    } else {
        None
    }
}

pub fn nansum(values: &[Scalar]) -> Scalar {
    if let Some((sum, _)) = collect_timedelta_ns(values) {
        let clamped = sum.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
        return Scalar::Timedelta64(clamped as i64);
    }
    // Fused single-pass fold: filter missing / non-f64-coercible and accumulate
    // in one scan, avoiding the intermediate `collect_finite` Vec<f64> and its
    // second pass. Bit-identical to `collect_finite(..).iter().sum()`: same
    // finite values in the same order, same left-fold f64 `+` (empty -> 0.0).
    let mut sum = 0.0_f64;
    for v in values {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            sum += x;
        }
    }
    Scalar::Float64(sum)
}

pub fn nanmean(values: &[Scalar]) -> Scalar {
    if let Some((sum, count)) = collect_timedelta_ns(values) {
        if count == 0 {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let mean = sum / count as i128;
        let clamped = mean.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
        return Scalar::Timedelta64(clamped as i64);
    }
    // Fused single-pass fold (see `nansum`): accumulate sum + count of finite
    // values in one scan. Bit-identical to the prior `collect_finite` two-pass:
    // count == nums.len(), sum folds the same values in the same order.
    let mut sum = 0.0_f64;
    let mut count = 0usize;
    for v in values {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            sum += x;
            count += 1;
        }
    }
    if count == 0 {
        return Scalar::Null(NullKind::NaN);
    }
    Scalar::Float64(sum / count as f64)
}

pub fn nanany(values: &[Scalar]) -> Scalar {
    for v in values {
        if v.is_missing() {
            continue;
        }
        match v {
            Scalar::Bool(flag) if *flag => return Scalar::Bool(true),
            Scalar::Int64(val) if *val != 0 => return Scalar::Bool(true),
            Scalar::Float64(val) if !val.is_nan() && *val != 0.0 => return Scalar::Bool(true),
            Scalar::Utf8(val) if !val.is_empty() => return Scalar::Bool(true),
            // pandas Series([td]).any() returns True for any non-zero
            // Timedelta. NaT is already filtered by is_missing() above.
            Scalar::Timedelta64(ns) if *ns != 0 => return Scalar::Bool(true),
            _ => continue,
        }
    }
    Scalar::Bool(false)
}

pub fn nanall(values: &[Scalar]) -> Scalar {
    for v in values {
        if v.is_missing() {
            continue;
        }
        match v {
            Scalar::Bool(flag) if !*flag => return Scalar::Bool(false),
            Scalar::Int64(val) if *val == 0 => return Scalar::Bool(false),
            Scalar::Float64(val) if val.is_nan() || *val == 0.0 => return Scalar::Bool(false),
            Scalar::Utf8(val) if val.is_empty() => return Scalar::Bool(false),
            // pandas Series([td(0)]).all() returns False; any non-zero
            // Timedelta is truthy. NaT is already filtered by is_missing.
            Scalar::Timedelta64(ns) if *ns == 0 => return Scalar::Bool(false),
            _ => continue,
        }
    }
    Scalar::Bool(true)
}

pub fn nancount(values: &[Scalar]) -> Scalar {
    let n = values.iter().filter(|v| !v.is_missing()).count();
    Scalar::Int64(n as i64)
}

pub fn nanmin(values: &[Scalar]) -> Scalar {
    let mut min: Option<&Scalar> = None;
    for v in values {
        if v.is_missing() {
            continue;
        }
        match (min, v) {
            (None, _) => min = Some(v),
            (Some(Scalar::Int64(a)), Scalar::Int64(b)) => {
                if b < a {
                    min = Some(v)
                }
            }
            (Some(Scalar::Float64(a)), Scalar::Float64(b)) => {
                if *b < *a {
                    min = Some(v)
                }
            }
            (Some(Scalar::Utf8(a)), Scalar::Utf8(b)) => {
                if b < a {
                    min = Some(v)
                }
            }
            (Some(Scalar::Bool(a)), Scalar::Bool(b)) => {
                if b < a {
                    min = Some(v)
                }
            }
            // Per br-frankenpandas-yic5m: Timedelta64.to_f64() errors, so
            // the catch-all below would silently return NaN. Compare ns
            // representations directly; NAT is already filtered by
            // is_missing() above.
            (Some(Scalar::Timedelta64(a)), Scalar::Timedelta64(b)) => {
                if b < a {
                    min = Some(v)
                }
            }
            (Some(a), b) => match (a.to_f64(), b.to_f64()) {
                (Ok(af), Ok(bf)) if bf < af => min = Some(v),
                (Ok(_), Ok(_)) => {}
                _ => return Scalar::Null(NullKind::NaN),
            },
        }
    }
    match min {
        Some(v) => v.clone(),
        None => Scalar::Null(NullKind::NaN),
    }
}

pub fn nanmax(values: &[Scalar]) -> Scalar {
    let mut max: Option<&Scalar> = None;
    for v in values {
        if v.is_missing() {
            continue;
        }
        match (max, v) {
            (None, _) => max = Some(v),
            (Some(Scalar::Int64(a)), Scalar::Int64(b)) => {
                if b > a {
                    max = Some(v)
                }
            }
            (Some(Scalar::Float64(a)), Scalar::Float64(b)) => {
                if *b > *a {
                    max = Some(v)
                }
            }
            (Some(Scalar::Utf8(a)), Scalar::Utf8(b)) => {
                if b > a {
                    max = Some(v)
                }
            }
            (Some(Scalar::Bool(a)), Scalar::Bool(b)) => {
                if b > a {
                    max = Some(v)
                }
            }
            // Per br-frankenpandas-yic5m: Timedelta64.to_f64() errors, so
            // the catch-all below would silently return NaN. Compare ns
            // representations directly; NAT is already filtered above.
            (Some(Scalar::Timedelta64(a)), Scalar::Timedelta64(b)) => {
                if b > a {
                    max = Some(v)
                }
            }
            (Some(a), b) => match (a.to_f64(), b.to_f64()) {
                (Ok(af), Ok(bf)) if bf > af => max = Some(v),
                (Ok(_), Ok(_)) => {}
                _ => return Scalar::Null(NullKind::NaN),
            },
        }
    }
    match max {
        Some(v) => v.clone(),
        None => Scalar::Null(NullKind::NaN),
    }
}

/// Per br-frankenpandas-j8ntk: harvest ns values from a uniformly-Timedelta64
/// input as f64 (the f64 representation has 53 bits of mantissa, sufficient
/// for ns spans up to ~104 days exactly; beyond that pandas itself loses
/// precision the same way). Returns None if any non-missing value is not
/// Timedelta64.
fn collect_timedelta_ns_f64(values: &[Scalar]) -> Option<Vec<f64>> {
    let mut out = Vec::with_capacity(values.len());
    let mut saw_td = false;
    for v in values {
        if v.is_missing() {
            continue;
        }
        match v {
            Scalar::Timedelta64(ns) => {
                saw_td = true;
                out.push(*ns as f64);
            }
            _ => return None,
        }
    }
    if saw_td { Some(out) } else { None }
}

/// Clamp an f64 result into i64 range and wrap as Scalar::Timedelta64.
fn float_ns_to_timedelta(value: f64) -> Scalar {
    if !value.is_finite() {
        return Scalar::Timedelta64(Timedelta::NAT);
    }
    let clamped = value.clamp(i64::MIN as f64, i64::MAX as f64);
    Scalar::Timedelta64(clamped as i64)
}

pub fn nanmedian(values: &[Scalar]) -> Scalar {
    // Per br-frankenpandas-j8ntk: Timedelta64 median preserves dtype.
    if let Some(mut td) = collect_timedelta_ns_f64(values) {
        if td.is_empty() {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        // O(n) selection instead of a full sort (see the numeric arm below):
        // collect_timedelta_ns_f64 yields finite ns (NaT excluded), so the
        // comparator is a total order; order statistics depend only on values,
        // so the unstable partition yields the same td[mid-1]/td[mid].
        let n = td.len();
        let mid = n / 2;
        let cmp = |a: &f64, b: &f64| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal);
        let (left, mid_ref, _right) = td.select_nth_unstable_by(mid, cmp);
        let mid_val = *mid_ref;
        let median_ns = if n.is_multiple_of(2) {
            let lower = left.iter().copied().fold(f64::NEG_INFINITY, f64::max);
            (lower + mid_val) / 2.0
        } else {
            mid_val
        };
        return float_ns_to_timedelta(median_ns);
    }
    let mut nums = collect_finite(values);
    if nums.is_empty() {
        return Scalar::Null(NullKind::NaN);
    }
    // O(n) selection instead of an O(n log n) full sort: select_nth_unstable_by
    // places the `mid`-th smallest at index `mid` with all smaller elements
    // (unordered) in the left partition. For even n the (mid-1)-th smallest is
    // the MAX of that left partition. Bit-identical to the sort path: order
    // statistics depend only on VALUES, and ties share a value, so the
    // unstable partition yields the same nums[mid-1]/nums[mid] the sort did.
    let n = nums.len();
    let mid = n / 2;
    let cmp = |a: &f64, b: &f64| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal);
    let (left, mid_ref, _right) = nums.select_nth_unstable_by(mid, cmp);
    let mid_val = *mid_ref;
    if n.is_multiple_of(2) {
        let lower = left.iter().copied().fold(f64::NEG_INFINITY, f64::max);
        Scalar::Float64((lower + mid_val) / 2.0)
    } else {
        Scalar::Float64(mid_val)
    }
}

pub fn nanvar(values: &[Scalar], ddof: usize) -> Scalar {
    // Per br-frankenpandas-j8ntk: Timedelta64 var preserves dtype — pandas
    // returns Timedelta even though variance is ns² conceptually; matching.
    if let Some(td) = collect_timedelta_ns_f64(values) {
        if td.len() <= ddof {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let mean: f64 = td.iter().sum::<f64>() / td.len() as f64;
        let sum_sq: f64 = td.iter().map(|x| (x - mean).powi(2)).sum();
        return float_ns_to_timedelta(sum_sq / (td.len() - ddof) as f64);
    }
    let nums = collect_finite(values);
    if nums.len() <= ddof {
        return Scalar::Null(NullKind::NaN);
    }
    let mean: f64 = nums.iter().sum::<f64>() / nums.len() as f64;
    let sum_sq: f64 = nums.iter().map(|x| (x - mean).powi(2)).sum();
    Scalar::Float64(sum_sq / (nums.len() - ddof) as f64)
}

pub fn nanstd(values: &[Scalar], ddof: usize) -> Scalar {
    // Per br-frankenpandas-j8ntk: Timedelta64 std preserves dtype.
    if let Some(td) = collect_timedelta_ns_f64(values) {
        if td.len() <= ddof {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let mean: f64 = td.iter().sum::<f64>() / td.len() as f64;
        let sum_sq: f64 = td.iter().map(|x| (x - mean).powi(2)).sum();
        let var = sum_sq / (td.len() - ddof) as f64;
        return float_ns_to_timedelta(var.sqrt());
    }
    match nanvar(values, ddof) {
        Scalar::Float64(v) => Scalar::Float64(v.sqrt()),
        other => other,
    }
}

/// Standard error of the mean over non-missing values.
///
/// Matches `pd.Series.sem(ddof=1)` / `scipy.stats.sem`. Computed as
/// `std(values, ddof) / sqrt(n)` where `n` is the non-missing count.
/// Returns `Null(NaN)` when `n <= ddof`.
pub fn nansem(values: &[Scalar], ddof: usize) -> Scalar {
    // Per br-frankenpandas-j8ntk: Timedelta64 sem preserves dtype.
    if let Some(td) = collect_timedelta_ns_f64(values) {
        if td.len() <= ddof {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let mean: f64 = td.iter().sum::<f64>() / td.len() as f64;
        let sum_sq: f64 = td.iter().map(|x| (x - mean).powi(2)).sum();
        let var = sum_sq / (td.len() - ddof) as f64;
        let std = var.sqrt();
        return float_ns_to_timedelta(std / (td.len() as f64).sqrt());
    }
    let nums = collect_finite(values);
    if nums.len() <= ddof {
        return Scalar::Null(NullKind::NaN);
    }
    match nanstd(values, ddof) {
        Scalar::Float64(s) => Scalar::Float64(s / (nums.len() as f64).sqrt()),
        other => other,
    }
}

/// Peak-to-peak range of non-missing values (max − min).
///
/// Matches `np.ptp` behavior on nan-safe inputs. Returns `Null(NaN)`
/// for empty or all-missing inputs.
pub fn nanptp(values: &[Scalar]) -> Scalar {
    // Per br-frankenpandas-u2g0r: Timedelta64 peak-to-peak returns
    // Timedelta64 (max - min in ns). collect_timedelta_ns_f64 is defined
    // in the cumulative-aggregations section below.
    if let Some(td) = collect_timedelta_ns_f64(values) {
        if td.is_empty() {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let (mut lo, mut hi) = (f64::INFINITY, f64::NEG_INFINITY);
        for x in &td {
            if *x < lo {
                lo = *x;
            }
            if *x > hi {
                hi = *x;
            }
        }
        return float_ns_to_timedelta(hi - lo);
    }
    // Fused single-pass min/max (see `nansum`): track lo/hi while filtering, no
    // intermediate Vec<f64>. Bit-identical to the prior collect_finite two-pass:
    // `seen` is true exactly when collect_finite would be non-empty, and the
    // lo/hi comparisons fold the same finite values in the same order.
    let (mut lo, mut hi) = (f64::INFINITY, f64::NEG_INFINITY);
    let mut seen = false;
    for v in values {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            seen = true;
            if x < lo {
                lo = x;
            }
            if x > hi {
                hi = x;
            }
        }
    }
    if !seen {
        return Scalar::Null(NullKind::NaN);
    }
    Scalar::Float64(hi - lo)
}

/// Sample skewness (bias-corrected, Fisher-Pearson) over non-missing values.
///
/// Matches `pd.Series.skew()`. Requires at least 3 non-missing values;
/// returns `Null(NaN)` otherwise, and when the sample standard deviation
/// is zero.
pub fn nanskew(values: &[Scalar]) -> Scalar {
    let nums = collect_finite(values);
    let n = nums.len() as f64;
    if n < 3.0 {
        return Scalar::Null(NullKind::NaN);
    }
    let mean = nums.iter().sum::<f64>() / n;
    let m2: f64 = nums.iter().map(|x| (x - mean).powi(2)).sum();
    let m3: f64 = nums.iter().map(|x| (x - mean).powi(3)).sum();
    let s2 = m2 / (n - 1.0);
    if s2 == 0.0 {
        return Scalar::Float64(0.0);
    }
    let s3 = s2.powf(1.5);
    Scalar::Float64((n / ((n - 1.0) * (n - 2.0))) * (m3 / s3))
}

/// Excess sample kurtosis (Fisher's definition, bias-corrected) over
/// non-missing values.
///
/// Matches `pd.Series.kurt()`. Requires at least 4 non-missing values;
/// returns `Null(NaN)` otherwise, and when the sample standard deviation
/// is zero.
pub fn nankurt(values: &[Scalar]) -> Scalar {
    let nums = collect_finite(values);
    let n = nums.len() as f64;
    if n < 4.0 {
        return Scalar::Null(NullKind::NaN);
    }
    let mean = nums.iter().sum::<f64>() / n;
    let m2: f64 = nums.iter().map(|x| (x - mean).powi(2)).sum();
    let m4: f64 = nums.iter().map(|x| (x - mean).powi(4)).sum();
    let s2 = m2 / (n - 1.0);
    if s2 == 0.0 {
        return Scalar::Float64(0.0);
    }
    let adj = (n * (n + 1.0)) / ((n - 1.0) * (n - 2.0) * (n - 3.0));
    let sub = (3.0 * (n - 1.0).powi(2)) / ((n - 2.0) * (n - 3.0));
    Scalar::Float64(adj * (m4 / (s2 * s2)) - sub)
}

/// Product of non-missing values. Returns 1.0 for empty input (matching pandas).
pub fn nanprod(values: &[Scalar]) -> Scalar {
    // Per br-frankenpandas-szq6a: pandas raises TypeError on
    // td_series.prod() because Timedelta² has no dimension. Returning the
    // misleading Float64(1.0) (empty-iterator default after collect_finite
    // drops every Timedelta64) is worse than surfacing missing. NaT
    // propagates the "type-incompatible" signal in lieu of a Result-level
    // error.
    if is_timedelta_input(values) {
        return Scalar::Null(NullKind::NaN);
    }
    // Fused single-pass fold (see `nansum`): filter missing / non-coercible and
    // multiply in one scan, no intermediate Vec<f64>. Bit-identical to
    // `collect_finite(..).iter().product()`: same finite values, same order,
    // same f64 `*` (Product for f64 == fold(1.0, *)); empty -> 1.0.
    let mut prod = 1.0_f64;
    for v in values {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            prod *= x;
        }
    }
    Scalar::Float64(prod)
}

/// Cumulative sum respecting null propagation.
///
/// Per br-frankenpandas-x0x91: detect uniformly-Timedelta64 input
/// (allowing Null/NAT missing markers). Returns true when at least one
/// non-missing value is Timedelta64 and no other dtype appears.
fn is_timedelta_input(values: &[Scalar]) -> bool {
    let mut saw_td = false;
    for v in values {
        if v.is_missing() {
            continue;
        }
        match v {
            Scalar::Timedelta64(_) => saw_td = true,
            _ => return false,
        }
    }
    saw_td
}

/// Per br-frankenpandas-x0x91: cumulative running aggregation over a
/// uniformly-Timedelta64 input. NaT/Null positions emit NaT and skip
/// the accumulator. Saturating i128 keeps overflow contained at i64
/// bounds when emitting.
fn timedelta_cumulative<F>(values: &[Scalar], init: i128, mut step: F) -> Vec<Scalar>
where
    F: FnMut(i128, i128) -> i128,
{
    let mut out = Vec::with_capacity(values.len());
    let mut running: i128 = init;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaT));
            continue;
        }
        if let Scalar::Timedelta64(ns) = v {
            running = step(running, i128::from(*ns));
            let clamped = running.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
            out.push(Scalar::Timedelta64(clamped as i64));
        } else {
            out.push(Scalar::Null(NullKind::NaT));
        }
    }
    out
}

/// Per br-frankenpandas-x0x91: running extrema (min/max) over a
/// uniformly-Timedelta64 input. `sentinel` is the identity element
/// (i64::MAX for min, i64::MIN for max) used until the first
/// non-missing value initializes the accumulator.
fn timedelta_cumulative_extrema<F>(values: &[Scalar], sentinel: i64, mut step: F) -> Vec<Scalar>
where
    F: FnMut(i64, i64) -> i64,
{
    let mut out = Vec::with_capacity(values.len());
    let mut running: Option<i64> = None;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaT));
            continue;
        }
        if let Scalar::Timedelta64(ns) = v {
            let new_val = match running {
                Some(prev) => step(prev, *ns),
                None => *ns,
            };
            running = Some(new_val);
            out.push(Scalar::Timedelta64(new_val));
        } else {
            out.push(Scalar::Null(NullKind::NaT));
        }
    }
    let _ = sentinel; // silence unused warning if closure ignores it
    out
}

/// Matches `np.nancumsum` / `pd.Series.cumsum()`. Missing input positions
/// pass through as `Null(NaN)` in the output; the running sum ignores
/// those positions when accumulating.
pub fn nancumsum(values: &[Scalar]) -> Vec<Scalar> {
    // Per br-frankenpandas-x0x91: when input is uniformly Timedelta64 (with
    // optional NaT/Null missing markers), preserve Timedelta dtype to match
    // pandas td_series.cumsum() returning Timedelta64.
    if is_timedelta_input(values) {
        return timedelta_cumulative(values, 0_i128, |acc, x| acc.saturating_add(x));
    }
    let mut out = Vec::with_capacity(values.len());
    let mut running = 0.0_f64;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaN));
            continue;
        }
        match v.to_f64() {
            Ok(x) if !x.is_nan() => {
                running += x;
                out.push(Scalar::Float64(running));
            }
            _ => out.push(Scalar::Null(NullKind::NaN)),
        }
    }
    out
}

/// Cumulative product respecting null propagation.
///
/// Matches `np.nancumprod` / `pd.Series.cumprod()`. Missing positions
/// pass through as `Null(NaN)` without advancing the running product.
pub fn nancumprod(values: &[Scalar]) -> Vec<Scalar> {
    let mut out = Vec::with_capacity(values.len());
    let mut running = 1.0_f64;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaN));
            continue;
        }
        match v.to_f64() {
            Ok(x) if !x.is_nan() => {
                running *= x;
                out.push(Scalar::Float64(running));
            }
            _ => out.push(Scalar::Null(NullKind::NaN)),
        }
    }
    out
}

/// Cumulative maximum respecting null propagation.
///
/// Matches `pd.Series.cummax()`. Missing positions pass through as
/// `Null(NaN)` without updating the running maximum. The first
/// non-missing value initializes the running maximum.
pub fn nancummax(values: &[Scalar]) -> Vec<Scalar> {
    // Per br-frankenpandas-x0x91: Timedelta64 preserves dtype.
    if is_timedelta_input(values) {
        return timedelta_cumulative_extrema(values, i64::MAX, |acc, x| acc.max(x));
    }
    let mut out = Vec::with_capacity(values.len());
    let mut running: Option<f64> = None;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaN));
            continue;
        }
        match v.to_f64() {
            Ok(x) if !x.is_nan() => {
                let new_val = match running {
                    Some(prev) => prev.max(x),
                    None => x,
                };
                running = Some(new_val);
                out.push(Scalar::Float64(new_val));
            }
            _ => out.push(Scalar::Null(NullKind::NaN)),
        }
    }
    out
}

/// Cumulative minimum respecting null propagation.
///
/// Matches `pd.Series.cummin()`. Symmetric to `nancummax`.
pub fn nancummin(values: &[Scalar]) -> Vec<Scalar> {
    // Per br-frankenpandas-x0x91: Timedelta64 preserves dtype.
    if is_timedelta_input(values) {
        return timedelta_cumulative_extrema(values, i64::MIN, |acc, x| acc.min(x));
    }
    let mut out = Vec::with_capacity(values.len());
    let mut running: Option<f64> = None;
    for v in values {
        if v.is_missing() {
            out.push(Scalar::Null(NullKind::NaN));
            continue;
        }
        match v.to_f64() {
            Ok(x) if !x.is_nan() => {
                let new_val = match running {
                    Some(prev) => prev.min(x),
                    None => x,
                };
                running = Some(new_val);
                out.push(Scalar::Float64(new_val));
            }
            _ => out.push(Scalar::Null(NullKind::NaN)),
        }
    }
    out
}

/// Linear-interpolation quantile over non-missing numeric values.
///
/// Matches `np.nanquantile(values, q)` with `interpolation='linear'`.
/// Returns `Null(NaN)` for empty inputs or when `q` is outside
/// `[0.0, 1.0]`.
pub fn nanquantile(values: &[Scalar], q: f64) -> Scalar {
    if !(0.0..=1.0).contains(&q) {
        return Scalar::Null(NullKind::NaN);
    }
    // Per br-frankenpandas-5djk7: pandas td_series.quantile(q) returns
    // Timedelta64 with linear-interpolated ns. Was silently NaN before.
    if let Some(mut td) = collect_timedelta_ns_f64(values) {
        if td.is_empty() {
            return Scalar::Timedelta64(Timedelta::NAT);
        }
        let n = td.len();
        if n == 1 {
            return float_ns_to_timedelta(td[0]);
        }
        let pos = q * (n - 1) as f64;
        let lo = pos.floor() as usize;
        let hi = pos.ceil() as usize;
        // O(n) selection instead of a full sort (see the numeric arm below):
        // select the lo-th order statistic; the (lo+1)-th is the MIN of the
        // right partition. Bit-identical (finite ns, values-only order stats).
        let cmp = |a: &f64, b: &f64| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal);
        let (_left, lo_ref, right) = td.select_nth_unstable_by(lo, cmp);
        let lo_val = *lo_ref;
        let ns = if lo == hi {
            lo_val
        } else {
            let hi_val = right.iter().copied().fold(f64::INFINITY, f64::min);
            let weight = pos - lo as f64;
            lo_val + (hi_val - lo_val) * weight
        };
        return float_ns_to_timedelta(ns);
    }
    let mut nums = collect_finite(values);
    if nums.is_empty() {
        return Scalar::Null(NullKind::NaN);
    }
    let n = nums.len();
    if n == 1 {
        return Scalar::Float64(nums[0]);
    }
    let pos = q * (n - 1) as f64;
    let lo = pos.floor() as usize;
    let hi = pos.ceil() as usize;
    // O(n) selection instead of a full sort: select the `lo`-th order statistic;
    // when interpolation is needed (hi == lo+1) the (lo+1)-th smallest is the
    // MIN of the right partition. Bit-identical to the sort path (same
    // nums[lo]/nums[hi] values, since order statistics depend only on values).
    let cmp = |a: &f64, b: &f64| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal);
    let (_left, lo_ref, right) = nums.select_nth_unstable_by(lo, cmp);
    let lo_val = *lo_ref;
    if lo == hi {
        return Scalar::Float64(lo_val);
    }
    let hi_val = right.iter().copied().fold(f64::INFINITY, f64::min);
    let weight = pos - lo as f64;
    Scalar::Float64(lo_val + (hi_val - lo_val) * weight)
}

/// Position (in the original slice) of the non-missing maximum.
///
/// Matches `np.nanargmax`. Returns `None` if every value is missing.
/// Ties resolve to the first position seen (matching numpy).
pub fn nanargmax(values: &[Scalar]) -> Option<usize> {
    // Per br-frankenpandas-ql1t5: Timedelta64.to_f64() errors, so the
    // generic path would silently skip every Timedelta64 value and
    // return None. Pandas td_series.argmax() returns the position of
    // the largest Timedelta — compare i64 ns directly.
    if is_timedelta_input(values) {
        let mut best: Option<(usize, i64)> = None;
        for (i, v) in values.iter().enumerate() {
            if v.is_missing() {
                continue;
            }
            if let Scalar::Timedelta64(ns) = v {
                match best {
                    None => best = Some((i, *ns)),
                    Some((_, cur)) if *ns > cur => best = Some((i, *ns)),
                    _ => {}
                }
            }
        }
        return best.map(|(i, _)| i);
    }
    let mut best: Option<(usize, f64)> = None;
    for (i, v) in values.iter().enumerate() {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            if x.is_nan() {
                continue;
            }
            match best {
                None => best = Some((i, x)),
                Some((_, cur)) if x > cur => best = Some((i, x)),
                _ => {}
            }
        }
    }
    best.map(|(i, _)| i)
}

/// Position (in the original slice) of the non-missing minimum.
///
/// Matches `np.nanargmin`. Returns `None` if every value is missing.
pub fn nanargmin(values: &[Scalar]) -> Option<usize> {
    // Per br-frankenpandas-ql1t5: Timedelta64 argmin via i64 ns compare.
    if is_timedelta_input(values) {
        let mut best: Option<(usize, i64)> = None;
        for (i, v) in values.iter().enumerate() {
            if v.is_missing() {
                continue;
            }
            if let Scalar::Timedelta64(ns) = v {
                match best {
                    None => best = Some((i, *ns)),
                    Some((_, cur)) if *ns < cur => best = Some((i, *ns)),
                    _ => {}
                }
            }
        }
        return best.map(|(i, _)| i);
    }
    let mut best: Option<(usize, f64)> = None;
    for (i, v) in values.iter().enumerate() {
        if v.is_missing() {
            continue;
        }
        if let Ok(x) = v.to_f64() {
            if x.is_nan() {
                continue;
            }
            match best {
                None => best = Some((i, x)),
                Some((_, cur)) if x < cur => best = Some((i, x)),
                _ => {}
            }
        }
    }
    best.map(|(i, _)| i)
}

/// Count of unique non-missing values.
pub fn nannunique(values: &[Scalar]) -> Scalar {
    use rustc_hash::FxHashSet;
    #[derive(Hash, PartialEq, Eq)]
    enum ScalarKey<'a> {
        Bool(bool),
        Int64(i64),
        FloatBits(u64),
        Utf8(&'a str),
        Timedelta64(i64),
        Datetime64(i64),
        Period(i64),
        Interval(u64, u64, IntervalClosed),
    }

    let mut seen = FxHashSet::default();
    for val in values {
        if val.is_missing() {
            continue;
        }
        let key = match val {
            Scalar::Bool(v) => ScalarKey::Bool(*v),
            Scalar::Int64(v) => ScalarKey::Int64(*v),
            Scalar::Float64(v) => {
                let normalized = if *v == 0.0 { 0.0 } else { *v };
                ScalarKey::FloatBits(normalized.to_bits())
            }
            Scalar::Utf8(v) => ScalarKey::Utf8(v.as_str()),
            Scalar::Timedelta64(v) => ScalarKey::Timedelta64(*v),
            Scalar::Datetime64(v) => ScalarKey::Datetime64(*v),
            Scalar::Period(v) => ScalarKey::Period(*v),
            Scalar::Interval(v) => ScalarKey::Interval(
                normalized_float_bits(v.left),
                normalized_float_bits(v.right),
                v.closed,
            ),
            Scalar::Null(_) => continue,
        };
        seen.insert(key);
    }
    Scalar::Int64(seen.len() as i64)
}

fn normalized_float_bits(value: f64) -> u64 {
    let normalized = if value == 0.0 { 0.0 } else { value };
    normalized.to_bits()
}

// ── Interval types (br-frankenpandas-j8k4 Phase 1) ──────────────────────
//
// Scaffolding for pandas `pd.Interval` / `pd.IntervalIndex` / `pd.IntervalDtype`.
//
// Phase 1 ships float-valued intervals only (matches `cut`/`qcut` output on
// numeric bins — the dominant pandas use case). Generic-subtype intervals
// over Int64 / Timestamp are deferred to Phase 2 alongside the DType::Interval
// enum-variant wiring. See br-j8k4 for the phased roadmap.
//
// Semantics mirror pandas: closed tells which endpoints are included.
//   Left    → [left, right)
//   Right   → (left, right]       ← pandas default
//   Both    → [left, right]
//   Neither → (left, right)

/// Endpoint-inclusion policy for an `Interval`.
///
/// Matches pandas `pd.Interval.closed` / `pd.IntervalDtype.closed` string
/// values ("left" / "right" / "both" / "neither").
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize,
)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum IntervalClosed {
    /// `[left, right)` — left-inclusive, right-exclusive.
    Left,
    /// `(left, right]` — left-exclusive, right-inclusive. Pandas default.
    #[default]
    Right,
    /// `[left, right]` — both endpoints included.
    Both,
    /// `(left, right)` — neither endpoint included.
    Neither,
}

impl IntervalClosed {
    /// Left endpoint included?
    #[must_use]
    pub fn left_closed(self) -> bool {
        matches!(self, Self::Left | Self::Both)
    }

    /// Right endpoint included?
    #[must_use]
    pub fn right_closed(self) -> bool {
        matches!(self, Self::Right | Self::Both)
    }
}

impl std::fmt::Display for IntervalClosed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Left => write!(f, "left"),
            Self::Right => write!(f, "right"),
            Self::Both => write!(f, "both"),
            Self::Neither => write!(f, "neither"),
        }
    }
}

/// A bounded numeric interval between two `f64` endpoints.
///
/// Matches `pd.Interval(left, right, closed)` on the numeric-subtype path.
/// Accessors match pandas: `.left`, `.right`, `.closed`, `.length`, `.mid`,
/// `.contains`, `.is_empty`, `.overlaps`.
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub struct Interval {
    pub left: f64,
    pub right: f64,
    #[serde(default)]
    pub closed: IntervalClosed,
}

impl Interval {
    /// Construct an interval. No validation on `left <= right` — pandas also
    /// accepts reversed intervals (they're non-empty only if empty-by-design).
    #[must_use]
    pub const fn new(left: f64, right: f64, closed: IntervalClosed) -> Self {
        Self {
            left,
            right,
            closed,
        }
    }

    /// `right - left` (pandas `.length`). Negative for reversed intervals.
    #[must_use]
    pub fn length(&self) -> f64 {
        self.right - self.left
    }

    /// Midpoint `(left + right) / 2` (pandas `.mid`).
    #[must_use]
    pub fn mid(&self) -> f64 {
        (self.left + self.right) / 2.0
    }

    /// Empty iff endpoints coincide AND at least one side is open.
    /// Pandas semantics: `pd.Interval(3, 3, 'right').is_empty → True`.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.left == self.right && !matches!(self.closed, IntervalClosed::Both)
    }

    /// Does `value` fall inside this interval?
    ///
    /// NaN always returns false, matching pandas `pd.Interval.__contains__`
    /// behavior (NaN doesn't compare equal to anything).
    #[must_use]
    pub fn contains(&self, value: f64) -> bool {
        if value.is_nan() {
            return false;
        }
        let left_ok = if self.closed.left_closed() {
            value >= self.left
        } else {
            value > self.left
        };
        let right_ok = if self.closed.right_closed() {
            value <= self.right
        } else {
            value < self.right
        };
        left_ok && right_ok
    }

    /// Do `self` and `other` share any point?
    ///
    /// Matches `pd.Interval.overlaps(other)`. Two intervals overlap iff the
    /// max of their lefts is less than the min of their rights, with
    /// endpoint-inclusion determining the strictness of the comparison when
    /// they touch exactly.
    #[must_use]
    pub fn overlaps(&self, other: &Self) -> bool {
        if self.left > other.right || other.left > self.right {
            return false;
        }
        // Touching-at-a-point: overlap iff both sides at that touchpoint are closed.
        if self.right == other.left {
            return self.closed.right_closed() && other.closed.left_closed();
        }
        if other.right == self.left {
            return other.closed.right_closed() && self.closed.left_closed();
        }
        true
    }

    /// Parse an interval string. Supports bracket notation:
    /// - '[0, 1]' -> closed on both ends
    /// - '(0, 1)' -> open on both ends
    /// - '[0, 1)' -> closed left, open right (pandas default)
    /// - '(0, 1]' -> open left, closed right
    pub fn parse(s: &str) -> Result<Self, TypeError> {
        let s = s.trim();
        if s.len() < 5 {
            return Err(TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Interval".to_string(),
            });
        }

        let first_char = s.chars().next().unwrap();
        let last_char = s.chars().last().unwrap();

        let left_closed = match first_char {
            '[' => true,
            '(' => false,
            _ => {
                return Err(TypeError::ValueNotParseable {
                    value: s.to_string(),
                    target: "Interval".to_string(),
                });
            }
        };

        let right_closed = match last_char {
            ']' => true,
            ')' => false,
            _ => {
                return Err(TypeError::ValueNotParseable {
                    value: s.to_string(),
                    target: "Interval".to_string(),
                });
            }
        };

        let closed = match (left_closed, right_closed) {
            (true, true) => IntervalClosed::Both,
            (true, false) => IntervalClosed::Left,
            (false, true) => IntervalClosed::Right,
            (false, false) => IntervalClosed::Neither,
        };

        let inner = &s[1..s.len() - 1];
        let parts: Vec<&str> = inner.split(',').collect();
        if parts.len() != 2 {
            return Err(TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Interval".to_string(),
            });
        }

        let left: f64 = parts[0]
            .trim()
            .parse()
            .map_err(|_| TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Interval".to_string(),
            })?;

        let right: f64 = parts[1]
            .trim()
            .parse()
            .map_err(|_| TypeError::ValueNotParseable {
                value: s.to_string(),
                target: "Interval".to_string(),
            })?;

        Ok(Self::new(left, right, closed))
    }
}

impl std::fmt::Display for Interval {
    /// Matches `str(pd.Interval(0, 5, 'right'))` → `"(0, 5]"`.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let left_bracket = if self.closed.left_closed() { '[' } else { '(' };
        let right_bracket = if self.closed.right_closed() { ']' } else { ')' };
        write!(
            f,
            "{left_bracket}{}, {}{right_bracket}",
            self.left, self.right
        )
    }
}

// ── interval_range builders (br-frankenpandas-xaom — Phase 2 of j8k4) ────

/// Build `periods` equal-width intervals spanning `[start, end]`.
///
/// Matches `pd.interval_range(start, end, periods=N, closed=...)` for the
/// numeric-subtype case. Returns exactly `periods` intervals; when
/// `periods == 0` or `start >= end`, returns an empty vector (matches
/// pandas's empty IntervalIndex).
///
/// ```
/// use fp_types::{interval_range_by_periods, IntervalClosed};
/// let bins = interval_range_by_periods(0.0, 10.0, 5, IntervalClosed::Right);
/// assert_eq!(bins.len(), 5);
/// assert_eq!(bins[0].left, 0.0);
/// assert_eq!(bins[0].right, 2.0);
/// assert_eq!(bins[4].right, 10.0);
/// ```
#[must_use]
pub fn interval_range_by_periods(
    start: f64,
    end: f64,
    periods: usize,
    closed: IntervalClosed,
) -> Vec<Interval> {
    if periods == 0 || !start.is_finite() || !end.is_finite() || start >= end {
        return Vec::new();
    }
    let step = (end - start) / (periods as f64);
    let mut out = Vec::with_capacity(periods);
    for i in 0..periods {
        let left = start + step * (i as f64);
        // Use end exactly for the final right edge to avoid float drift.
        let right = if i + 1 == periods {
            end
        } else {
            start + step * ((i + 1) as f64)
        };
        out.push(Interval::new(left, right, closed));
    }
    out
}

/// Build equal-`step`-width intervals spanning `[start, end]`.
///
/// Matches `pd.interval_range(start, end, freq=step, closed=...)` for the
/// numeric-subtype case. `step` must be finite and positive; `(end - start)`
/// must be an integer multiple of `step` (within float tolerance) — pandas
/// raises `ValueError` otherwise; this fn returns `Err(TypeError::IntervalStepDoesNotDivide)`.
///
/// Returns an empty vector when `start == end` (matches pandas' zero-bin
/// IntervalIndex); returns an empty vector when `start > end` (pandas also
/// returns empty rather than erroring in this case).
pub fn interval_range_by_step(
    start: f64,
    end: f64,
    step: f64,
    closed: IntervalClosed,
) -> Result<Vec<Interval>, TypeError> {
    if !step.is_finite() || !step.is_sign_positive() || step == 0.0 {
        return Err(TypeError::InvalidIntervalStep { step });
    }
    if !start.is_finite() || !end.is_finite() || start >= end {
        return Ok(Vec::new());
    }
    let span = end - start;
    let periods_f = span / step;
    let periods = periods_f.round() as i64;
    if periods <= 0 {
        return Ok(Vec::new());
    }
    let reconstructed = step * (periods as f64);
    // Relative tolerance: allow float-rounding noise proportional to span.
    if (span - reconstructed).abs() > span.abs() * 1e-9 + 1e-12 {
        return Err(TypeError::IntervalStepDoesNotDivide { step, span });
    }
    let periods = periods as usize;
    let mut out = Vec::with_capacity(periods);
    for i in 0..periods {
        let left = start + step * (i as f64);
        let right = if i + 1 == periods {
            end
        } else {
            start + step * ((i + 1) as f64)
        };
        out.push(Interval::new(left, right, closed));
    }
    Ok(out)
}

// ── Period types (br-frankenpandas-epoj Phase 1) ────────────────────────
//
// Scaffolding for pandas `pd.Period` / `pd.PeriodIndex` / `pd.PeriodDtype`.
//
// A Period is a calendar *span* (Q1 2024, Jan 2024, 2024-03-15), distinct
// from a Timestamp (an instant). Phase 1 ships the PeriodFreq enum +
// Period struct with ordinal-based arithmetic (Period + n, Period - Period),
// Display in pandas notation, and parse from standard strings. Calendar-
// conversion (ordinal ↔ ymd) and DType::Period wiring land in Phase 2.

/// Period frequency code. Matches pandas offset alias core set.
///
/// The ordinal axis is frequency-specific: for Monthly, ordinal 0 is a
/// fixed anchor (pandas uses months since 1970-01). Phase 1 doesn't
/// commit to a specific epoch yet — the ordinal scheme is opaque until
/// Phase 2 wires calendar arithmetic. What Phase 1 DOES nail down is:
/// same-freq Periods compare + subtract; Period + i64 shifts by `n`
/// periods of the declared frequency.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING-KEBAB-CASE")]
#[non_exhaustive]
pub enum PeriodFreq {
    /// `Y-DEC` / `A` / `Y` — annual periods.
    Annual,
    /// `Q-DEC` / `Q` — quarterly periods.
    Quarterly,
    /// `M` — monthly periods.
    Monthly,
    /// `W-SUN` / `W` — weekly periods.
    Weekly,
    /// `D` — daily periods.
    Daily,
    /// `B` — business-day periods.
    Business,
    /// `h` / `H` — hourly periods.
    Hourly,
    /// `min` / `T` — minutely periods.
    Minutely,
    /// `s` / `S` — secondly periods.
    Secondly,
}

impl PeriodFreq {
    /// Parse a pandas-style frequency alias. Recognizes the common subset
    /// (Y-DEC/A/Y, Q-DEC/Q, M, W-SUN/W, D, B, h/H, min/T, s/S).
    /// Case-insensitive.
    pub fn parse(alias: &str) -> Option<Self> {
        match alias.to_ascii_uppercase().as_str() {
            "A" | "Y" | "A-DEC" | "Y-DEC" | "ANNUAL" | "YEARLY" => Some(Self::Annual),
            "Q" | "Q-DEC" | "QUARTERLY" => Some(Self::Quarterly),
            "M" | "MONTHLY" => Some(Self::Monthly),
            "W" | "W-SUN" | "WEEKLY" => Some(Self::Weekly),
            "D" | "DAILY" => Some(Self::Daily),
            "B" | "BUSINESS" => Some(Self::Business),
            "H" | "HOURLY" => Some(Self::Hourly),
            "T" | "MIN" | "MINUTELY" => Some(Self::Minutely),
            "S" | "SECONDLY" => Some(Self::Secondly),
            _ => None,
        }
    }

    /// Canonical pandas alias string.
    #[must_use]
    pub const fn alias(self) -> &'static str {
        match self {
            Self::Annual => "Y-DEC",
            Self::Quarterly => "Q-DEC",
            Self::Monthly => "M",
            Self::Weekly => "W-SUN",
            Self::Daily => "D",
            Self::Business => "B",
            Self::Hourly => "h",
            Self::Minutely => "min",
            Self::Secondly => "s",
        }
    }

    /// Per br-frankenpandas-qigpe: resolution string for PeriodIndex.resolution.
    #[must_use]
    pub const fn resolution(self) -> &'static str {
        match self {
            Self::Annual => "A-DEC",
            Self::Quarterly => "Q-DEC",
            Self::Monthly => "M",
            Self::Weekly => "W-SUN",
            Self::Daily => "D",
            Self::Business => "B",
            Self::Hourly => "H",
            Self::Minutely => "T",
            Self::Secondly => "S",
        }
    }
}

impl std::fmt::Display for PeriodFreq {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.alias())
    }
}

/// A single pandas-style Period value.
///
/// Stored as an integer ordinal on a frequency-specific axis plus the
/// frequency code. Two Periods with different `freq` are incompatible —
/// arithmetic and comparison require same-freq operands.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct Period {
    pub ordinal: i64,
    pub freq: PeriodFreq,
}

impl Period {
    #[must_use]
    pub const fn new(ordinal: i64, freq: PeriodFreq) -> Self {
        Self { ordinal, freq }
    }

    /// Integer position on this period's frequency axis, matching
    /// `pd.Period.ordinal`.
    #[must_use]
    pub const fn ordinal(&self) -> i64 {
        self.ordinal
    }

    /// Frequency code for this period, matching `pd.Period.freq`.
    #[must_use]
    pub const fn freq(&self) -> PeriodFreq {
        self.freq
    }

    /// Canonical pandas frequency alias, matching `pd.Period.freqstr`.
    #[must_use]
    pub const fn freqstr(&self) -> &'static str {
        self.freq.alias()
    }

    /// Same-freq ordinal comparison. Returns `None` if `freq` differs —
    /// caller decides whether that's an error or a panic site.
    #[must_use]
    pub fn cmp_same_freq(&self, other: &Self) -> Option<std::cmp::Ordering> {
        if self.freq != other.freq {
            return None;
        }
        Some(self.ordinal.cmp(&other.ordinal))
    }

    /// Shift by `n` periods of the current frequency.
    /// Matches `pd.Period + n` and `pd.Period - n`.
    #[must_use]
    pub fn shift(&self, n: i64) -> Self {
        Self {
            ordinal: self.ordinal.saturating_add(n),
            freq: self.freq,
        }
    }

    /// Period-difference in units of the shared frequency.
    /// Returns `None` if `freq` differs (pandas raises IncompatibleFrequency).
    #[must_use]
    pub fn diff(&self, other: &Self) -> Option<i64> {
        if self.freq != other.freq {
            return None;
        }
        Some(self.ordinal.saturating_sub(other.ordinal))
    }

    /// Parse common pandas `Period(...)` strings and infer the frequency.
    ///
    /// Supported forms mirror pandas' unambiguous scalar constructor cases:
    /// annual (`"2024"`), quarterly (`"2024Q1"`), monthly (`"2024-01"`),
    /// and daily (`"2024-01-15"`). The ordinal axes match pandas:
    /// 1970, 1970Q1, 1970-01, and 1970-01-01 all have ordinal 0.
    pub fn parse(s: &str) -> Result<Self, TypeError> {
        let trimmed = s.trim();
        if trimmed.eq_ignore_ascii_case("nat") {
            return Ok(Self::new(i64::MIN, PeriodFreq::Daily));
        }

        if let Some((year, quarter)) = parse_quarter_period(trimmed) {
            let ordinal = year
                .checked_sub(1970)
                .and_then(|offset| offset.checked_mul(4))
                .and_then(|base| base.checked_add(i64::from(quarter) - 1))
                .ok_or_else(|| TypeError::ValueNotParseable {
                    value: s.to_owned(),
                    target: "Period".to_owned(),
                })?;
            return Ok(Self::new(ordinal, PeriodFreq::Quarterly));
        }

        if let Some((year, month, day)) = parse_ymd_period(trimmed) {
            let ordinal = Timestamp::days_from_ymd(year, i64::from(month), i64::from(day));
            return Ok(Self::new(ordinal, PeriodFreq::Daily));
        }

        if let Some((year, month)) = parse_year_month_period(trimmed) {
            let ordinal = year
                .checked_sub(1970)
                .and_then(|offset| offset.checked_mul(12))
                .and_then(|base| base.checked_add(i64::from(month) - 1))
                .ok_or_else(|| TypeError::ValueNotParseable {
                    value: s.to_owned(),
                    target: "Period".to_owned(),
                })?;
            return Ok(Self::new(ordinal, PeriodFreq::Monthly));
        }

        if let Some(year) = parse_annual_period(trimmed) {
            let ordinal = year
                .checked_sub(1970)
                .ok_or_else(|| TypeError::ValueNotParseable {
                    value: s.to_owned(),
                    target: "Period".to_owned(),
                })?;
            return Ok(Self::new(ordinal, PeriodFreq::Annual));
        }

        Err(TypeError::ValueNotParseable {
            value: s.to_owned(),
            target: "Period".to_owned(),
        })
    }
}

fn parse_annual_period(value: &str) -> Option<i64> {
    (value.len() == 4 && value.chars().all(|ch| ch.is_ascii_digit()))
        .then(|| value.parse::<i64>().ok())
        .flatten()
}

fn parse_year_month_period(value: &str) -> Option<(i64, u32)> {
    let (year, month) = value.split_once('-')?;
    if year.len() != 4 || month.len() != 2 {
        return None;
    }
    let year = year.parse::<i64>().ok()?;
    let month = month.parse::<u32>().ok()?;
    (1..=12).contains(&month).then_some((year, month))
}

fn parse_ymd_period(value: &str) -> Option<(i64, u32, u32)> {
    let mut parts = value.split('-');
    let year = parts.next()?;
    let month = parts.next()?;
    let day = parts.next()?;
    if parts.next().is_some() || year.len() != 4 || month.len() != 2 || day.len() != 2 {
        return None;
    }
    let year = year.parse::<i64>().ok()?;
    let month = month.parse::<u32>().ok()?;
    let day = day.parse::<u32>().ok()?;
    (1..=days_in_month(year, month)?)
        .contains(&day)
        .then_some((year, month, day))
}

fn parse_quarter_period(value: &str) -> Option<(i64, u32)> {
    let (year, quarter) = value.split_once('Q').or_else(|| value.split_once('q'))?;
    if year.len() != 4 || quarter.len() != 1 {
        return None;
    }
    let year = year.parse::<i64>().ok()?;
    let quarter = quarter.parse::<u32>().ok()?;
    (1..=4).contains(&quarter).then_some((year, quarter))
}

impl std::fmt::Display for Period {
    /// Phase 1: ordinal+freq form, e.g. `Period[Q-DEC, 216]`. Calendar-
    /// formatted display (`2024Q1`, `2024-03`) lands in Phase 2 once the
    /// ordinal-to-ymd arithmetic is wired.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Period[{}, {}]", self.freq, self.ordinal)
    }
}

/// Build `periods` consecutive Periods starting at `start`.
///
/// Matches `pd.period_range(start, periods=N, freq=start.freq)` for the
/// count-based form. The frequency is taken from `start` — pandas requires
/// `freq` to match when both are passed; mismatches are an error in
/// pandas, but here we sidestep ambiguity by deriving from `start.freq`.
///
/// Per br-frankenpandas-2jef (epoj Phase 2). Pure ordinal arithmetic — no
/// calendar conversion (Phase 3 wires chrono). `periods=0` returns empty.
///
/// ```
/// use fp_types::{period_range, Period, PeriodFreq};
/// let q1 = Period::new(216, PeriodFreq::Quarterly);
/// let year = period_range(q1, 4);
/// assert_eq!(year.len(), 4);
/// assert_eq!(year[0].ordinal, 216);
/// assert_eq!(year[3].ordinal, 219);
/// ```
#[must_use]
pub fn period_range(start: Period, periods: usize) -> Vec<Period> {
    (0..periods).map(|i| start.shift(i as i64)).collect()
}

#[cfg(test)]
mod tests {
    use super::{
        DType, Interval, IntervalClosed, NullKind, Period, PeriodFreq, Scalar, SparseDType,
        cast_scalar, common_dtype, infer_dtype,
    };

    /// br-frankenpandas-esjjy / fd90.182: ergonomic From impls for Scalar.
    #[test]
    fn scalar_from_primitive_types() {
        // Each primitive maps to its canonical Scalar variant.
        assert_eq!(Scalar::from(true), Scalar::Bool(true));
        assert_eq!(Scalar::from(42i64), Scalar::Int64(42));
        assert_eq!(Scalar::from(1.5f64), Scalar::Float64(1.5));
        assert_eq!(Scalar::from("hi"), Scalar::Utf8("hi".to_owned()));
        assert_eq!(
            Scalar::from(String::from("world")),
            Scalar::Utf8("world".to_owned())
        );

        // .into() works in mixed-type Vec<Scalar> contexts (the README's
        // case_when example pattern, and what fd90.181 needed for apply_row
        // closures).
        let mixed: Vec<Scalar> = vec![1i64.into(), 2.0f64.into(), "three".into()];
        assert_eq!(mixed.len(), 3);
        assert_eq!(mixed[0], Scalar::Int64(1));
        assert_eq!(mixed[1], Scalar::Float64(2.0));
        assert_eq!(mixed[2], Scalar::Utf8("three".to_owned()));
    }

    #[test]
    fn dtype_inference_coerces_numeric_values() {
        let values = vec![Scalar::Bool(true), Scalar::Int64(7), Scalar::Float64(3.5)];
        assert_eq!(
            infer_dtype(&values).expect("dtype should infer"),
            DType::Float64
        );
    }

    #[test]
    fn interval_scalar_has_dtype_storage_and_unique_semantics_5g5uj() {
        let left = Scalar::Interval(Interval::new(0.0, 1.0, IntervalClosed::Right));
        let right = Scalar::Interval(Interval::new(1.0, 2.0, IntervalClosed::Right));
        assert_eq!(left.dtype(), DType::Interval);
        assert!(!left.is_missing());
        assert_eq!(
            infer_dtype(&[left.clone(), right.clone()]).expect("interval dtype"),
            DType::Interval
        );
        assert_eq!(
            common_dtype(DType::Interval, DType::Interval).expect("same interval dtype"),
            DType::Interval
        );
        assert_eq!(
            cast_scalar(&Scalar::Null(NullKind::Null), DType::Interval).expect("missing casts"),
            Scalar::Null(NullKind::Null)
        );
        assert_eq!(
            cast_scalar(&left, DType::Utf8).expect("interval string cast"),
            Scalar::Utf8("(0, 1]".to_owned())
        );
        assert_eq!(
            super::nannunique(&[left.clone(), right, left, Scalar::Null(NullKind::Null)]),
            Scalar::Int64(2)
        );
    }

    #[test]
    fn cast_scalar_parses_temporal_extension_strings_avm08() {
        let expected_nanos = super::Timestamp::parse("2024-01-15T10:30:45")
            .expect("timestamp parse")
            .nanos;
        assert_eq!(
            cast_scalar(
                &Scalar::Utf8("2024-01-15T10:30:45".to_owned()),
                DType::Datetime64
            )
            .expect("datetime cast"),
            Scalar::Datetime64(expected_nanos)
        );
        assert_eq!(
            cast_scalar(&Scalar::Utf8("2024Q1".to_owned()), DType::Period).expect("period cast"),
            Scalar::Period(216)
        );
        assert_eq!(
            cast_scalar(&Scalar::Utf8("(0, 1]".to_owned()), DType::Interval)
                .expect("interval cast"),
            Scalar::Interval(Interval::new(0.0, 1.0, IntervalClosed::Right))
        );
    }

    #[test]
    fn missing_values_get_target_missing_marker() {
        let missing = Scalar::Null(NullKind::Null);
        let cast = cast_scalar(&missing, DType::Float64).expect("missing casts");
        assert_eq!(cast, Scalar::Null(NullKind::NaN));
    }

    #[test]
    fn cast_scalar_to_utf8_uses_pandas_string_spellings() {
        let cases = [
            (Scalar::Bool(true), "True"),
            (Scalar::Bool(false), "False"),
            (Scalar::Int64(-7), "-7"),
            (Scalar::Float64(1.0), "1.0"),
            (Scalar::Float64(1.5), "1.5"),
            (Scalar::Float64(f64::NAN), "nan"),
            (Scalar::Null(NullKind::Null), "None"),
            (Scalar::Null(NullKind::NaN), "nan"),
            (Scalar::Null(NullKind::NaT), "NaT"),
        ];

        for (value, expected) in cases {
            assert_eq!(
                cast_scalar(&value, DType::Utf8).expect("cast"),
                Scalar::Utf8(expected.to_owned())
            );
        }
    }

    #[test]
    fn semantic_eq_treats_nan_as_equal() {
        let left = Scalar::Float64(f64::NAN);
        let right = Scalar::Null(NullKind::NaN);
        assert!(left.semantic_eq(&right));
    }

    #[test]
    fn semantic_eq_treats_nan_as_missing_null() {
        let left = Scalar::Float64(f64::NAN);
        let right = Scalar::Null(NullKind::Null);
        assert!(left.semantic_eq(&right));
    }

    #[test]
    fn common_dtype_rejects_string_numeric_mix() {
        let err = common_dtype(DType::Utf8, DType::Int64).expect_err("must fail");
        assert_eq!(
            err.to_string(),
            "dtype coercion from Utf8 to Int64 has no compatible common type"
        );
        let err = common_dtype(DType::Float64, DType::Utf8).expect_err("must fail");
        assert_eq!(
            err.to_string(),
            "dtype coercion from Float64 to Utf8 has no compatible common type"
        );
    }

    #[test]
    fn sparse_dtype_normalizes_fill_value_to_value_dtype() {
        let dtype = SparseDType::new(DType::Float64, Scalar::Int64(0)).expect("fill should cast");

        assert_eq!(dtype.value_dtype, DType::Float64);
        assert_eq!(dtype.fill_value, Scalar::Float64(0.0));
    }

    #[test]
    fn sparse_dtype_rejects_sparse_value_dtype() {
        let err = SparseDType::new(DType::Sparse, Scalar::Int64(0)).expect_err("must reject");

        assert_eq!(err.to_string(), "sparse value dtype cannot be Sparse");
    }

    #[test]
    fn common_dtype_rejects_sparse_dense_mix() {
        let err = common_dtype(DType::Sparse, DType::Int64).expect_err("must fail");

        assert_eq!(
            err.to_string(),
            "dtype coercion from Sparse to Int64 has no compatible common type"
        );
    }

    // ── Nullable Int64/Bool dtype tests (br-frankenpandas-rg8ys.6.4) ────

    #[test]
    fn nullable_int64_promotion_matrix() {
        // Int64 + Int64Nullable -> Int64Nullable (nullable absorbs)
        assert_eq!(
            common_dtype(DType::Int64, DType::Int64Nullable).unwrap(),
            DType::Int64Nullable
        );
        assert_eq!(
            common_dtype(DType::Int64Nullable, DType::Int64).unwrap(),
            DType::Int64Nullable
        );

        // Int64Nullable + Float64 -> Float64 (float always wins)
        assert_eq!(
            common_dtype(DType::Int64Nullable, DType::Float64).unwrap(),
            DType::Float64
        );
        assert_eq!(
            common_dtype(DType::Float64, DType::Int64Nullable).unwrap(),
            DType::Float64
        );

        // Int64Nullable + Int64Nullable -> Int64Nullable
        assert_eq!(
            common_dtype(DType::Int64Nullable, DType::Int64Nullable).unwrap(),
            DType::Int64Nullable
        );

        // Bool + Int64Nullable -> Int64Nullable
        assert_eq!(
            common_dtype(DType::Bool, DType::Int64Nullable).unwrap(),
            DType::Int64Nullable
        );

        // BoolNullable + Int64 -> Int64Nullable
        assert_eq!(
            common_dtype(DType::BoolNullable, DType::Int64).unwrap(),
            DType::Int64Nullable
        );
    }

    #[test]
    fn nullable_bool_promotion_matrix() {
        // Bool + BoolNullable -> BoolNullable
        assert_eq!(
            common_dtype(DType::Bool, DType::BoolNullable).unwrap(),
            DType::BoolNullable
        );
        assert_eq!(
            common_dtype(DType::BoolNullable, DType::Bool).unwrap(),
            DType::BoolNullable
        );

        // BoolNullable + Float64 -> Float64
        assert_eq!(
            common_dtype(DType::BoolNullable, DType::Float64).unwrap(),
            DType::Float64
        );
    }

    #[test]
    fn dtype_is_nullable_helper() {
        assert!(DType::Int64Nullable.is_nullable());
        assert!(DType::BoolNullable.is_nullable());
        assert!(!DType::Int64.is_nullable());
        assert!(!DType::Bool.is_nullable());
        assert!(!DType::Float64.is_nullable());
        assert!(!DType::Utf8.is_nullable());
    }

    #[test]
    fn dtype_to_nullable_conversions() {
        assert_eq!(DType::Int64.to_nullable(), DType::Int64Nullable);
        assert_eq!(DType::Bool.to_nullable(), DType::BoolNullable);
        assert_eq!(DType::Float64.to_nullable(), DType::Float64); // unchanged
        assert_eq!(DType::Int64Nullable.to_nullable(), DType::Int64Nullable);
    }

    #[test]
    fn dtype_to_non_nullable_conversions() {
        assert_eq!(DType::Int64Nullable.to_non_nullable(), DType::Int64);
        assert_eq!(DType::BoolNullable.to_non_nullable(), DType::Bool);
        assert_eq!(DType::Int64.to_non_nullable(), DType::Int64); // unchanged
        assert_eq!(DType::Float64.to_non_nullable(), DType::Float64);
    }

    #[test]
    fn nullable_dtype_name_reports_pandas_style() {
        assert_eq!(DType::Int64.name(), "int64");
        assert_eq!(DType::Int64Nullable.name(), "Int64"); // capital I
        assert_eq!(DType::Bool.name(), "bool");
        assert_eq!(DType::BoolNullable.name(), "boolean");
    }

    #[test]
    fn cast_scalar_int64_nullable_identity() {
        let val = Scalar::Int64(42);
        // Int64 -> Int64Nullable is identity (no actual conversion)
        let result = cast_scalar(&val, DType::Int64Nullable).unwrap();
        assert_eq!(result, Scalar::Int64(42));

        // Int64Nullable -> Int64 is also identity
        let result2 = cast_scalar(&val, DType::Int64).unwrap();
        assert_eq!(result2, Scalar::Int64(42));
    }

    #[test]
    fn cast_float_to_utf8_uses_pandas_str_float_with_scientific() {
        // pandas astype(str) of floats == Python str(float): whole -> ".0",
        // shortest round-trip decimals, scientific (e+NN/e-NN) for large/small,
        // inf -> "inf", NaN -> "nan". Verified vs live pandas 2.2.3. (Previously
        // large/small lost scientific notation, e.g. 1e16 -> "10000000000000000.0".)
        let cases: &[(f64, &str)] = &[
            (1.0, "1.0"),
            (2.5, "2.5"),
            (100.0, "100.0"),
            (0.1, "0.1"),
            (0.0001, "0.0001"),
            (1e16, "1e+16"),
            (1e20, "1e+20"),
            (1e-5, "1e-05"),
            (1e-7, "1e-07"),
            (f64::INFINITY, "inf"),
            (f64::NEG_INFINITY, "-inf"),
        ];
        for (v, expected) in cases {
            assert_eq!(
                cast_scalar(&Scalar::Float64(*v), DType::Utf8).unwrap(),
                Scalar::Utf8((*expected).to_owned()),
                "float {v} -> str"
            );
        }
    }

    #[test]
    fn cast_to_bool_uses_pandas_nonzero_truthiness() {
        // pandas astype(bool): zero -> False, any nonzero -> True (not just 0/1),
        // -0.0 -> False, and NaN -> True (numpy bool(nan), br-cyi4h). Verified vs
        // live pandas 2.2.3.
        let cases_int: &[(i64, bool)] = &[(0, false), (1, true), (-3, true), (2, true)];
        for (v, expected) in cases_int {
            assert_eq!(
                cast_scalar(&Scalar::Int64(*v), DType::Bool).unwrap(),
                Scalar::Bool(*expected),
                "int {v} -> bool"
            );
        }
        let cases_float: &[(f64, bool)] = &[
            (0.0, false),
            (-0.0, false),
            (0.1, true),
            (2.5, true),
            (1.0, true),
            // pandas astype(bool): NaN is truthy -> True (numpy bool). br-cyi4h.
            (f64::NAN, true),
        ];
        for (v, expected) in cases_float {
            assert_eq!(
                cast_scalar(&Scalar::Float64(*v), DType::Bool).unwrap(),
                Scalar::Bool(*expected),
                "float {v} -> bool"
            );
        }
    }

    #[test]
    fn nullable_dtype_is_extension() {
        assert!(DType::Int64Nullable.is_extension());
        assert!(DType::BoolNullable.is_extension());
        assert!(!DType::Int64.is_extension());
        assert!(!DType::Bool.is_extension());
    }

    #[test]
    fn infer_dtype_preserves_string_numeric_mix_as_utf8_bucket() {
        let values = vec![Scalar::Utf8("x".into()), Scalar::Int64(7)];
        assert_eq!(
            infer_dtype(&values).expect("dtype should infer"),
            DType::Utf8
        );
    }

    // ── Scalar missingness methods ─────────────────────────────────────

    #[test]
    fn is_null_detects_explicit_nulls() {
        assert!(Scalar::Null(NullKind::Null).is_null());
        assert!(Scalar::Null(NullKind::NaN).is_null());
        assert!(!Scalar::Int64(42).is_null());
        assert!(!Scalar::Float64(f64::NAN).is_null());
    }

    #[test]
    fn is_na_matches_is_missing() {
        let vals = vec![
            Scalar::Null(NullKind::Null),
            Scalar::Float64(f64::NAN),
            Scalar::Int64(0),
            Scalar::Bool(false),
        ];
        for v in &vals {
            assert_eq!(v.is_na(), v.is_missing());
        }
    }

    #[test]
    fn coalesce_picks_first_non_missing() {
        let null = Scalar::Null(NullKind::Null);
        let fill = Scalar::Int64(99);
        assert_eq!(null.coalesce(&fill), fill);
        assert_eq!(fill.coalesce(&null), fill);
    }

    // ── Missingness utilities ──────────────────────────────────────────

    #[test]
    fn isna_notna_complement() {
        let vals = vec![
            Scalar::Int64(1),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(f64::NAN),
            Scalar::Float64(3.0),
        ];
        let na = super::isna(&vals);
        let not = super::notna(&vals);
        assert_eq!(na, vec![false, true, true, false]);
        for (a, b) in na.iter().zip(not.iter()) {
            assert_ne!(a, b);
        }
    }

    #[test]
    fn count_na_counts_missing() {
        let vals = vec![
            Scalar::Int64(1),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(f64::NAN),
        ];
        assert_eq!(super::count_na(&vals), 2);
    }

    #[test]
    fn fill_na_replaces_missing() {
        let vals = vec![
            Scalar::Int64(1),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(f64::NAN),
            Scalar::Int64(4),
        ];
        let filled = super::fill_na(&vals, &Scalar::Int64(0));
        assert_eq!(filled[0], Scalar::Int64(1));
        assert_eq!(filled[1], Scalar::Int64(0));
        assert_eq!(filled[2], Scalar::Int64(0));
        assert_eq!(filled[3], Scalar::Int64(4));
    }

    #[test]
    fn dropna_removes_missing() {
        let vals = vec![
            Scalar::Int64(1),
            Scalar::Null(NullKind::Null),
            Scalar::Int64(3),
            Scalar::Float64(f64::NAN),
        ];
        let kept = super::dropna(&vals);
        assert_eq!(kept.len(), 2);
        assert_eq!(kept[0], Scalar::Int64(1));
        assert_eq!(kept[1], Scalar::Int64(3));
    }

    // ── Nanops ─────────────────────────────────────────────────────────

    #[test]
    fn nansum_skips_nulls() {
        let vals = vec![
            Scalar::Float64(1.0),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(2.0),
            Scalar::Float64(f64::NAN),
            Scalar::Int64(7),
        ];
        assert_eq!(super::nansum(&vals), Scalar::Float64(10.0));
    }

    #[test]
    fn nansum_empty_returns_zero() {
        assert_eq!(super::nansum(&[]), Scalar::Float64(0.0));
    }

    #[test]
    fn nannunique_merges_negative_zero_and_zero() {
        let vals = vec![
            Scalar::Float64(-0.0),
            Scalar::Float64(0.0),
            Scalar::Float64(1.0),
        ];
        assert_eq!(super::nannunique(&vals), Scalar::Int64(2));
    }

    #[test]
    fn nanmean_basic() {
        let vals = vec![
            Scalar::Float64(2.0),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(4.0),
        ];
        assert_eq!(super::nanmean(&vals), Scalar::Float64(3.0));
    }

    #[test]
    fn nanmean_all_null_returns_nan() {
        let vals = vec![Scalar::Null(NullKind::Null), Scalar::Float64(f64::NAN)];
        assert!(super::nanmean(&vals).is_missing());
    }

    #[test]
    fn nansum_nanmean_timedelta64_preserves_dtype_620mj() {
        // Per br-frankenpandas-620mj: pandas td_series.sum()/mean() return
        // Timedelta64, not Float64(0.0). Was silently zero before because
        // collect_finite drops Timedelta64 (to_f64 errors).
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        assert_eq!(super::nansum(&vals), Scalar::Timedelta64(6 * one_hour));
        assert_eq!(super::nanmean(&vals), Scalar::Timedelta64(2 * one_hour));
    }

    #[test]
    fn nansum_nanmean_timedelta64_skips_nat_620mj() {
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![
            Scalar::Timedelta64(Timedelta::NAT),
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(3 * one_hour),
            Scalar::Timedelta64(Timedelta::NAT),
        ];
        // NAT is missing → skipped. Sum: 1h+3h=4h; mean: 2h.
        assert_eq!(super::nansum(&vals), Scalar::Timedelta64(4 * one_hour));
        assert_eq!(super::nanmean(&vals), Scalar::Timedelta64(2 * one_hour));
    }

    #[test]
    fn nansum_nanmean_mixed_timedelta_other_falls_back_620mj() {
        // Mixed Timedelta64 + other type bails out of the Timedelta path
        // and uses Float64 collect_finite (which drops Timedelta).
        // Preserves existing cross-type behavior (effectively ignores TD).
        let vals = vec![Scalar::Timedelta64(3600 * 1_000_000_000), Scalar::Int64(5)];
        // Int64(5) makes it through to_f64 → 5.0; Timedelta is dropped.
        assert_eq!(super::nansum(&vals), Scalar::Float64(5.0));
    }

    #[test]
    fn nancount_counts_non_missing() {
        let vals = vec![
            Scalar::Int64(1),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(3.0),
        ];
        assert_eq!(super::nancount(&vals), Scalar::Int64(2));
    }

    #[test]
    fn nanmin_basic() {
        let vals = vec![
            Scalar::Float64(5.0),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(2.0),
            Scalar::Float64(8.0),
        ];
        assert_eq!(super::nanmin(&vals), Scalar::Float64(2.0));
    }

    #[test]
    fn nanmax_basic() {
        let vals = vec![
            Scalar::Float64(5.0),
            Scalar::Float64(f64::NAN),
            Scalar::Float64(8.0),
        ];
        assert_eq!(super::nanmax(&vals), Scalar::Float64(8.0));
    }

    #[test]
    fn nanmin_nanmax_empty_returns_nan() {
        assert!(super::nanmin(&[]).is_missing());
        assert!(super::nanmax(&[]).is_missing());
    }

    #[test]
    fn nanmin_nanmax_mixed_incompatible_types_returns_nan() {
        let vals = vec![Scalar::Int64(5), Scalar::Utf8("hello".into())];
        assert!(super::nanmin(&vals).is_missing());
        assert!(super::nanmax(&vals).is_missing());

        let vals2 = vec![Scalar::Utf8("a".into()), Scalar::Float64(3.0)];
        assert!(super::nanmin(&vals2).is_missing());
        assert!(super::nanmax(&vals2).is_missing());
    }

    #[test]
    fn nanmin_nanmax_compatible_numeric_types_ok() {
        let vals = vec![Scalar::Int64(5), Scalar::Float64(3.0), Scalar::Bool(true)];
        assert_eq!(super::nanmin(&vals), Scalar::Bool(true));
        assert_eq!(super::nanmax(&vals), Scalar::Int64(5));
    }

    #[test]
    fn nanmin_nanmax_timedelta64_returns_timedelta_yic5m() {
        // Per br-frankenpandas-yic5m: nanmin/nanmax on Timedelta64 returns
        // the smallest/largest Timedelta64 — was silently NaN before
        // because Timedelta64.to_f64() errors and the catch-all swallowed it.
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![
            Scalar::Timedelta64(3 * one_hour),
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
        ];
        assert_eq!(super::nanmin(&vals), Scalar::Timedelta64(one_hour));
        assert_eq!(super::nanmax(&vals), Scalar::Timedelta64(3 * one_hour));
    }

    #[test]
    fn nanmin_nanmax_timedelta64_skips_nat_yic5m() {
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![
            Scalar::Timedelta64(Timedelta::NAT),
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(Timedelta::NAT),
        ];
        assert_eq!(super::nanmin(&vals), Scalar::Timedelta64(one_hour));
        assert_eq!(super::nanmax(&vals), Scalar::Timedelta64(2 * one_hour));
    }

    #[test]
    fn nanmedian_odd_count() {
        let vals = vec![
            Scalar::Float64(3.0),
            Scalar::Null(NullKind::Null),
            Scalar::Float64(1.0),
            Scalar::Float64(2.0),
        ];
        assert_eq!(super::nanmedian(&vals), Scalar::Float64(2.0));
    }

    #[test]
    fn nanmedian_even_count() {
        let vals = vec![
            Scalar::Float64(1.0),
            Scalar::Float64(3.0),
            Scalar::Float64(2.0),
            Scalar::Float64(4.0),
        ];
        assert_eq!(super::nanmedian(&vals), Scalar::Float64(2.5));
    }

    #[test]
    fn nanvar_population() {
        let vals = vec![
            Scalar::Float64(2.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
            Scalar::Float64(7.0),
            Scalar::Float64(9.0),
        ];
        let var = super::nanvar(&vals, 0);
        assert!(matches!(var, Scalar::Float64(_)), "expected Float64");
        if let Scalar::Float64(v) = var {
            assert!((v - 4.0).abs() < 1e-10);
        }
    }

    #[test]
    fn nanvar_sample_ddof1() {
        let vals = vec![
            Scalar::Float64(2.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
            Scalar::Float64(7.0),
            Scalar::Float64(9.0),
        ];
        let var = super::nanvar(&vals, 1);
        assert!(matches!(var, Scalar::Float64(_)), "expected Float64");
        if let Scalar::Float64(v) = var {
            assert!((v - 32.0 / 7.0).abs() < 1e-10);
        }
    }

    #[test]
    fn nanvar_insufficient_values_returns_nan() {
        let vals = vec![Scalar::Float64(5.0)];
        assert!(super::nanvar(&vals, 1).is_missing());
    }

    #[test]
    fn nanstd_is_sqrt_of_var() {
        let vals = vec![
            Scalar::Float64(2.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
            Scalar::Float64(7.0),
            Scalar::Float64(9.0),
        ];
        let std = super::nanstd(&vals, 0);
        assert!(matches!(std, Scalar::Float64(_)), "expected Float64");
        if let Scalar::Float64(v) = std {
            assert!((v - 2.0).abs() < 1e-10);
        }
    }

    #[test]
    fn nanmedian_timedelta64_preserves_dtype_j8ntk() {
        // Per br-frankenpandas-j8ntk: pandas td_series.median() returns
        // Timedelta64; was silently NaN before via collect_finite.
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        assert_eq!(super::nanmedian(&vals), Scalar::Timedelta64(2 * one_hour));
    }

    #[test]
    fn nanstd_timedelta64_preserves_dtype_j8ntk() {
        // Per br-frankenpandas-j8ntk: pandas td_series.std() returns
        // Timedelta64. Check Timedelta64 output and reasonable magnitude
        // for population std of [1h, 2h, 3h] = sqrt(2/3) * 1h.
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let vals = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        let std = super::nanstd(&vals, 0);
        match std {
            Scalar::Timedelta64(ns) => {
                let expected = (2.0_f64 / 3.0).sqrt() * one_hour as f64;
                assert!(
                    (ns as f64 - expected).abs() < 1e6,
                    "expected ~{expected} ns, got {ns}"
                );
            }
            other => panic!("expected Timedelta64, got {other:?}"),
        }
    }

    #[test]
    fn nanstd_nansem_timedelta64_insufficient_returns_nat_j8ntk() {
        let one_hour = 3_600 * 1_000_000_000_i64;
        let vals = vec![Scalar::Timedelta64(one_hour)];
        // ddof=1 with n=1 → underflow, returns NaT
        match super::nanstd(&vals, 1) {
            Scalar::Timedelta64(v) => assert_eq!(v, Timedelta::NAT),
            other => panic!("expected Timedelta64 NAT, got {other:?}"),
        }
        match super::nansem(&vals, 1) {
            Scalar::Timedelta64(v) => assert_eq!(v, Timedelta::NAT),
            other => panic!("expected Timedelta64 NAT, got {other:?}"),
        }
    }

    #[test]
    fn nanops_with_mixed_types() {
        let vals = vec![
            Scalar::Bool(true),
            Scalar::Int64(3),
            Scalar::Float64(6.0),
            Scalar::Null(NullKind::Null),
        ];
        assert_eq!(super::nansum(&vals), Scalar::Float64(10.0));
        assert_eq!(super::nancount(&vals), Scalar::Int64(3));
    }

    #[test]
    fn nanops_all_missing_returns_identity() {
        let vals = vec![Scalar::Null(NullKind::Null), Scalar::Float64(f64::NAN)];
        assert_eq!(super::nansum(&vals), Scalar::Float64(0.0));
        assert!(super::nanmean(&vals).is_missing());
        assert!(super::nanmedian(&vals).is_missing());
        assert!(super::nanvar(&vals, 0).is_missing());
        assert!(super::nanstd(&vals, 0).is_missing());
    }

    // ── Timedelta tests ────────────────────────────────────────────────

    #[test]
    fn timedelta_parse_simple_units() {
        use super::Timedelta;
        assert_eq!(Timedelta::parse("1d").unwrap(), Timedelta::NANOS_PER_DAY);
        assert_eq!(
            Timedelta::parse("2h").unwrap(),
            2 * Timedelta::NANOS_PER_HOUR
        );
        assert_eq!(
            Timedelta::parse("30m").unwrap(),
            30 * Timedelta::NANOS_PER_MIN
        );
        assert_eq!(
            Timedelta::parse("45s").unwrap(),
            45 * Timedelta::NANOS_PER_SEC
        );
        assert_eq!(
            Timedelta::parse("100ms").unwrap(),
            100 * Timedelta::NANOS_PER_MILLI
        );
        assert_eq!(
            Timedelta::parse("500us").unwrap(),
            500 * Timedelta::NANOS_PER_MICRO
        );
        assert_eq!(Timedelta::parse("1000ns").unwrap(), 1000);
    }

    #[test]
    fn timedelta_parse_compound() {
        use super::Timedelta;
        let expected = Timedelta::NANOS_PER_DAY
            + 2 * Timedelta::NANOS_PER_HOUR
            + 30 * Timedelta::NANOS_PER_MIN;
        assert_eq!(Timedelta::parse("1d 2h 30m").unwrap(), expected);
        assert_eq!(Timedelta::parse("1d2h30m").unwrap(), expected);
    }

    #[test]
    fn timedelta_parse_iso8601_matches_pandas_tdiso() {
        use super::Timedelta;
        // Verified vs pandas 2.2.3 Timedelta(...).value.
        assert_eq!(Timedelta::parse("P1DT2H3M4S").unwrap(), 93_784_000_000_000);
        assert_eq!(Timedelta::parse("PT1H").unwrap(), 3_600_000_000_000);
        assert_eq!(Timedelta::parse("PT1H30M").unwrap(), 5_400_000_000_000);
        assert_eq!(Timedelta::parse("P1D").unwrap(), 86_400_000_000_000);
        assert_eq!(Timedelta::parse("P2W").unwrap(), 1_209_600_000_000_000);
        assert_eq!(Timedelta::parse("PT0.5S").unwrap(), 500_000_000);
        // pandas quirks: T ignored, M is minutes everywhere, units in any order.
        assert_eq!(Timedelta::parse("P1M").unwrap(), 60_000_000_000);
        assert_eq!(Timedelta::parse("P1H").unwrap(), 3_600_000_000_000);
        assert_eq!(Timedelta::parse("PT1D").unwrap(), 86_400_000_000_000);
        assert_eq!(Timedelta::parse("P1D1H").unwrap(), 90_000_000_000_000);
        assert_eq!(Timedelta::parse("-P1DT2H").unwrap(), -93_600_000_000_000);
        // Rejected like pandas: years, lowercase, bare P/PT.
        assert!(Timedelta::parse("P1Y").is_err());
        assert!(Timedelta::parse("p1d").is_err());
        assert!(Timedelta::parse("P").is_err());
        assert!(Timedelta::parse("PT").is_err());
    }

    #[test]
    fn timedelta_parse_time_format() {
        use super::Timedelta;
        let expected = Timedelta::NANOS_PER_HOUR
            + 30 * Timedelta::NANOS_PER_MIN
            + 45 * Timedelta::NANOS_PER_SEC;
        assert_eq!(Timedelta::parse("01:30:45").unwrap(), expected);
    }

    #[test]
    fn timedelta_parse_time_fraction_rejects_unicode_without_panic() {
        use super::{Timedelta, TimedeltaError};
        let err = Timedelta::parse("00:00:00.\u{00e9}\u{00e9}\u{00e9}\u{00e9}\u{00e9}")
            .expect_err("non-ASCII fractional seconds must reject");
        assert!(matches!(err, TimedeltaError::InvalidFormat(_)));
    }

    #[test]
    fn timedelta_parse_time_format_rejects_overflow_without_panic() {
        use super::{Timedelta, TimedeltaError};
        let err = Timedelta::parse("9223372036854775807:00")
            .expect_err("oversized hour component must reject");
        assert!(matches!(err, TimedeltaError::InvalidFormat(_)));
    }

    #[test]
    fn timedelta_parse_rejects_huge_value_overflow_zw3mg() {
        // Per br-frankenpandas-zw3mg: the compound parser used a raw
        // `as i64` cast that silently saturated to i64::MAX when the
        // product of (decimal-digit f64) × unit multiplier overflows.
        // Use a large literal (no scientific notation — the lexer only
        // accepts digits, '.', '-'). 1e18 days × NANOS_PER_DAY (~8.64e13)
        // overflows i64.
        use super::{Timedelta, TimedeltaError};
        let huge = format!("{} days", "9".repeat(18));
        assert!(matches!(
            Timedelta::parse(&huge).expect_err("9...(18 9s) days must overflow"),
            TimedeltaError::Overflow
        ));
    }

    #[test]
    fn timedelta_parse_nat() {
        use super::Timedelta;
        assert_eq!(Timedelta::parse("NaT").unwrap(), Timedelta::NAT);
        assert_eq!(Timedelta::parse("nat").unwrap(), Timedelta::NAT);
    }

    #[test]
    fn timedelta_parse_negative() {
        use super::Timedelta;
        assert_eq!(Timedelta::parse("-1d").unwrap(), -Timedelta::NANOS_PER_DAY);
    }

    #[test]
    fn timedelta_components() {
        use super::Timedelta;
        let nanos = Timedelta::NANOS_PER_DAY
            + Timedelta::NANOS_PER_HOUR
            + Timedelta::NANOS_PER_MIN
            + Timedelta::NANOS_PER_SEC
            + Timedelta::NANOS_PER_MILLI
            + 2 * Timedelta::NANOS_PER_MICRO
            + 3;
        let comp = Timedelta::components(nanos);
        assert_eq!(comp.days, 1);
        assert_eq!(comp.hours, 1);
        assert_eq!(comp.minutes, 1);
        assert_eq!(comp.seconds, 1);
        assert_eq!(comp.milliseconds, 1);
        assert_eq!(comp.microseconds, 2);
        assert_eq!(comp.nanoseconds, 3);
    }

    #[test]
    fn timedelta_negative_components_floor_div() {
        use super::Timedelta;
        // pandas floor-normalizes negative timedeltas: pd.Timedelta(-1,'s') has
        // days=-1, seconds=86399, components=(-1, 23, 59, 59, 0, 0, 0).
        let neg_1s = -Timedelta::NANOS_PER_SEC;
        assert_eq!(Timedelta::days(neg_1s), -1);
        assert_eq!(Timedelta::seconds(neg_1s), 86399);
        assert_eq!(Timedelta::microseconds(neg_1s), 0);
        assert_eq!(Timedelta::nanoseconds(neg_1s), 0);
        let comp = Timedelta::components(neg_1s);
        assert_eq!(
            (
                comp.days,
                comp.hours,
                comp.minutes,
                comp.seconds,
                comp.milliseconds,
                comp.microseconds,
                comp.nanoseconds
            ),
            (-1, 23, 59, 59, 0, 0, 0)
        );

        // pd.Timedelta(-86401,'s'): days=-2, seconds=86399.
        let neg = -86_401 * Timedelta::NANOS_PER_SEC;
        assert_eq!(Timedelta::days(neg), -2);
        assert_eq!(Timedelta::seconds(neg), 86399);
    }

    #[test]
    fn timedelta_total_seconds() {
        use super::Timedelta;
        let nanos = 90_000_000_000i64; // 90 seconds
        assert!((Timedelta::total_seconds(nanos) - 90.0).abs() < 1e-9);
        assert!(Timedelta::total_seconds(Timedelta::NAT).is_nan());
    }

    #[test]
    fn timedelta_format_basic() {
        use super::Timedelta;
        assert_eq!(Timedelta::format(Timedelta::NAT), "NaT");
        assert_eq!(
            Timedelta::format(Timedelta::NANOS_PER_DAY),
            "1 days 00:00:00"
        );
        assert_eq!(
            Timedelta::format(Timedelta::NANOS_PER_DAY + 2 * Timedelta::NANOS_PER_HOUR),
            "1 days 02:00:00"
        );
    }

    #[test]
    fn timedelta_format_subsecond_matches_pandas() {
        use super::Timedelta;
        // pandas str(Timedelta) uses 6 fractional digits (microseconds) unless a
        // sub-microsecond (nanosecond) component is present, then 9 digits.
        // Verified vs live pandas 2.2.3.
        assert_eq!(
            Timedelta::format(1_500_000_000), // 1.5s
            "0 days 00:00:01.500000"
        );
        assert_eq!(
            Timedelta::format(1_000_000), // 1ms
            "0 days 00:00:00.001000"
        );
        assert_eq!(
            Timedelta::format(123_456_000), // 123456us
            "0 days 00:00:00.123456"
        );
        // Nanosecond component -> 9 digits.
        assert_eq!(
            Timedelta::format(500), // 500ns
            "0 days 00:00:00.000000500"
        );
        assert_eq!(Timedelta::format(123_456_789), "0 days 00:00:00.123456789");
    }

    #[test]
    fn timedelta_format_negative_uses_python_borrow_form() {
        use super::Timedelta;
        // pandas/Python normalize negatives via floor division: the days count
        // goes negative, the time remainder stays non-negative, and a '+' joins
        // them. Verified vs live pandas 2.2.3.
        assert_eq!(Timedelta::format(-1_000_000_000), "-1 days +23:59:59");
        assert_eq!(
            Timedelta::format(-Timedelta::NANOS_PER_DAY),
            "-1 days +00:00:00"
        );
        assert_eq!(
            Timedelta::format(-25 * Timedelta::NANOS_PER_HOUR),
            "-2 days +23:00:00"
        );
        assert_eq!(
            Timedelta::format(-1_500_000_000),
            "-1 days +23:59:58.500000"
        );
        assert_eq!(Timedelta::format(-500), "-1 days +23:59:59.999999500");
        assert_eq!(Timedelta::format(-1), "-1 days +23:59:59.999999999");
    }

    #[test]
    fn timedelta_isoformat_basic() {
        use super::Timedelta;
        assert_eq!(Timedelta::isoformat(Timedelta::NAT), "NaT");
        assert_eq!(Timedelta::isoformat(0), "P0DT0H0M0S");
        assert_eq!(Timedelta::isoformat(Timedelta::NANOS_PER_DAY), "P1DT0H0M0S");
        assert_eq!(
            Timedelta::isoformat(
                Timedelta::NANOS_PER_DAY
                    + 2 * Timedelta::NANOS_PER_HOUR
                    + 30 * Timedelta::NANOS_PER_MIN
                    + 45 * Timedelta::NANOS_PER_SEC
            ),
            "P1DT2H30M45S"
        );
        assert_eq!(
            Timedelta::isoformat(Timedelta::NANOS_PER_SEC + 500_000_000),
            "P0DT0H0M1.5S"
        );
        assert_eq!(
            Timedelta::isoformat(-(Timedelta::NANOS_PER_DAY + Timedelta::NANOS_PER_HOUR)),
            "-P1DT1H0M0S"
        );
    }

    #[test]
    fn timedelta_floor_ceil_round() {
        use super::Timedelta;
        let nanos = Timedelta::NANOS_PER_HOUR + 30 * Timedelta::NANOS_PER_MIN;

        // floor: rounds down
        assert_eq!(Timedelta::floor(nanos, "h"), Timedelta::NANOS_PER_HOUR);
        assert_eq!(Timedelta::floor(nanos, "d"), 0);

        // ceil: rounds up
        assert_eq!(Timedelta::ceil(nanos, "h"), 2 * Timedelta::NANOS_PER_HOUR);
        assert_eq!(Timedelta::ceil(nanos, "d"), Timedelta::NANOS_PER_DAY);

        // round: rounds to nearest (banker's rounding on tie)
        assert_eq!(Timedelta::round(nanos, "h"), 2 * Timedelta::NANOS_PER_HOUR);

        // NaT preserved
        assert_eq!(Timedelta::floor(Timedelta::NAT, "h"), Timedelta::NAT);
        assert_eq!(Timedelta::ceil(Timedelta::NAT, "h"), Timedelta::NAT);
        assert_eq!(Timedelta::round(Timedelta::NAT, "h"), Timedelta::NAT);

        // Invalid freq returns NAT
        assert_eq!(Timedelta::floor(nanos, "invalid"), Timedelta::NAT);
    }

    #[test]
    fn timedelta_scalar_dtype() {
        let td = Scalar::Timedelta64(86_400_000_000_000);
        assert_eq!(td.dtype(), DType::Timedelta64);
    }

    #[test]
    fn timedelta_scalar_is_missing() {
        use super::Timedelta;
        let valid = Scalar::Timedelta64(1000);
        let nat = Scalar::Timedelta64(Timedelta::NAT);
        assert!(!valid.is_missing());
        assert!(nat.is_missing());
    }

    #[test]
    fn dtype_utf8_deserializes_legacy_aliases() {
        let dtype: DType = serde_json::from_str("\"str\"").unwrap();
        assert_eq!(dtype, DType::Utf8);

        let dtype: DType = serde_json::from_str("\"string\"").unwrap();
        assert_eq!(dtype, DType::Utf8);
    }

    #[test]
    fn scalar_utf8_deserializes_legacy_aliases() {
        let scalar: Scalar = serde_json::from_str(r#"{"kind":"str","value":"x"}"#).unwrap();
        assert_eq!(scalar, Scalar::Utf8("x".to_owned()));

        let scalar: Scalar = serde_json::from_str(r#"{"kind":"string","value":"y"}"#).unwrap();
        assert_eq!(scalar, Scalar::Utf8("y".to_owned()));
    }

    #[test]
    fn nancumsum_skips_nulls_and_accumulates() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(2.0),
            Scalar::Float64(3.0),
        ];
        let out = super::nancumsum(&values);
        assert!(matches!(out[0], Scalar::Float64(v) if (v - 1.0).abs() < 1e-9));
        assert!(out[1].is_missing());
        assert!(matches!(out[2], Scalar::Float64(v) if (v - 3.0).abs() < 1e-9));
        assert!(matches!(out[3], Scalar::Float64(v) if (v - 6.0).abs() < 1e-9));
    }

    #[test]
    fn nancumprod_skips_nulls_and_multiplies() {
        let values = vec![
            Scalar::Float64(2.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(3.0),
            Scalar::Float64(4.0),
        ];
        let out = super::nancumprod(&values);
        assert!(matches!(out[0], Scalar::Float64(v) if (v - 2.0).abs() < 1e-9));
        assert!(out[1].is_missing());
        assert!(matches!(out[2], Scalar::Float64(v) if (v - 6.0).abs() < 1e-9));
        assert!(matches!(out[3], Scalar::Float64(v) if (v - 24.0).abs() < 1e-9));
    }

    #[test]
    fn nancummax_tracks_running_max() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Float64(3.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(2.0),
            Scalar::Float64(5.0),
        ];
        let out = super::nancummax(&values);
        assert_eq!(out[0], Scalar::Float64(1.0));
        assert_eq!(out[1], Scalar::Float64(3.0));
        assert!(out[2].is_missing());
        assert_eq!(out[3], Scalar::Float64(3.0));
        assert_eq!(out[4], Scalar::Float64(5.0));
    }

    #[test]
    fn nancummin_tracks_running_min() {
        let values = vec![
            Scalar::Float64(5.0),
            Scalar::Float64(3.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(4.0),
            Scalar::Float64(1.0),
        ];
        let out = super::nancummin(&values);
        assert_eq!(out[0], Scalar::Float64(5.0));
        assert_eq!(out[1], Scalar::Float64(3.0));
        assert!(out[2].is_missing());
        assert_eq!(out[3], Scalar::Float64(3.0));
        assert_eq!(out[4], Scalar::Float64(1.0));
    }

    #[test]
    fn nancumsum_timedelta64_preserves_dtype_x0x91() {
        // Per br-frankenpandas-x0x91: pandas td_series.cumsum() returns
        // Timedelta64 running sums. Was silently NaN before.
        let one_hour = 3_600 * 1_000_000_000_i64;
        let values = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        let out = super::nancumsum(&values);
        assert_eq!(out[0], Scalar::Timedelta64(one_hour));
        assert_eq!(out[1], Scalar::Timedelta64(3 * one_hour));
        assert_eq!(out[2], Scalar::Timedelta64(6 * one_hour));
    }

    #[test]
    fn nancummax_nancummin_timedelta64_preserves_dtype_x0x91() {
        let one_hour = 3_600 * 1_000_000_000_i64;
        let values = vec![
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(5 * one_hour),
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        let mx = super::nancummax(&values);
        assert_eq!(mx[0], Scalar::Timedelta64(2 * one_hour));
        assert_eq!(mx[1], Scalar::Timedelta64(5 * one_hour));
        assert_eq!(mx[2], Scalar::Timedelta64(5 * one_hour));
        assert_eq!(mx[3], Scalar::Timedelta64(5 * one_hour));

        let mn = super::nancummin(&values);
        assert_eq!(mn[0], Scalar::Timedelta64(2 * one_hour));
        assert_eq!(mn[1], Scalar::Timedelta64(2 * one_hour));
        assert_eq!(mn[2], Scalar::Timedelta64(one_hour));
        assert_eq!(mn[3], Scalar::Timedelta64(one_hour));
    }

    #[test]
    fn nancumulative_timedelta64_skips_nat_x0x91() {
        let one_hour = 3_600 * 1_000_000_000_i64;
        let values = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(Timedelta::NAT),
            Scalar::Timedelta64(2 * one_hour),
        ];
        let cs = super::nancumsum(&values);
        assert_eq!(cs[0], Scalar::Timedelta64(one_hour));
        assert!(cs[1].is_missing());
        assert_eq!(cs[2], Scalar::Timedelta64(3 * one_hour));
    }

    #[test]
    fn nanquantile_linear_interpolation_matches_numpy() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Float64(2.0),
            Scalar::Float64(3.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
        ];
        // median
        let q = super::nanquantile(&values, 0.5);
        assert!(matches!(q, Scalar::Float64(v) if (v - 3.0).abs() < 1e-9));
        // 25th percentile: interpolate between 2.0 and 3.0 at pos 1.0 → 2.0
        let q25 = super::nanquantile(&values, 0.25);
        assert!(matches!(q25, Scalar::Float64(v) if (v - 2.0).abs() < 1e-9));
    }

    #[test]
    fn nanquantile_ignores_nulls() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(3.0),
        ];
        let q = super::nanquantile(&values, 0.5);
        assert!(matches!(q, Scalar::Float64(v) if (v - 2.0).abs() < 1e-9));
    }

    #[test]
    fn nanquantile_empty_and_out_of_range_yield_null() {
        assert!(super::nanquantile(&[], 0.5).is_missing());
        assert!(super::nanquantile(&[Scalar::Float64(1.0)], 1.5).is_missing());
        assert!(super::nanquantile(&[Scalar::Float64(1.0)], -0.1).is_missing());
    }

    #[test]
    fn nanquantile_timedelta64_preserves_dtype_5djk7() {
        // Per br-frankenpandas-5djk7: pandas td_series.quantile(q) returns
        // Timedelta64 — was silently NaN before via collect_finite.
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let vals = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
            Scalar::Timedelta64(4 * one_hour),
            Scalar::Timedelta64(5 * one_hour),
        ];
        assert_eq!(
            super::nanquantile(&vals, 0.5),
            Scalar::Timedelta64(3 * one_hour)
        );
        assert_eq!(
            super::nanquantile(&vals, 0.0),
            Scalar::Timedelta64(one_hour)
        );
        assert_eq!(
            super::nanquantile(&vals, 1.0),
            Scalar::Timedelta64(5 * one_hour)
        );
    }

    #[test]
    fn nanquantile_timedelta64_linear_interpolation_5djk7() {
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let vals = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        // Linear interpolation: at q=0.5, midpoint = 2h
        assert_eq!(
            super::nanquantile(&vals, 0.5),
            Scalar::Timedelta64(2 * one_hour)
        );
    }

    #[test]
    fn nanargmax_returns_first_position() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(2.0),
        ];
        assert_eq!(super::nanargmax(&values), Some(2));
    }

    #[test]
    fn nanargmin_returns_first_position() {
        let values = vec![
            Scalar::Float64(3.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(1.0),
            Scalar::Float64(1.0),
        ];
        assert_eq!(super::nanargmin(&values), Some(2));
    }

    #[test]
    fn nanargmax_all_missing_returns_none() {
        let values = vec![Scalar::Null(NullKind::NaN), Scalar::Null(NullKind::Null)];
        assert_eq!(super::nanargmax(&values), None);
        assert_eq!(super::nanargmin(&values), None);
    }

    #[test]
    fn nansem_matches_std_over_sqrt_n() {
        let values = vec![
            Scalar::Float64(2.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
            Scalar::Float64(7.0),
            Scalar::Float64(9.0),
        ];
        // numpy/scipy: std(ddof=1) = 2.138089935299395; sem = std/sqrt(8) = 0.7559
        let sem = super::nansem(&values, 1);
        assert!(matches!(sem, Scalar::Float64(_)));
        let Scalar::Float64(v) = sem else {
            return;
        };
        assert!((v - 0.7559289460184544).abs() < 1e-9);
    }

    #[test]
    fn nansem_empty_returns_null() {
        assert!(super::nansem(&[], 1).is_missing());
        assert!(super::nansem(&[Scalar::Float64(1.0)], 1).is_missing());
    }

    #[test]
    fn nanptp_returns_max_minus_min() {
        let values = vec![
            Scalar::Float64(3.0),
            Scalar::Null(NullKind::NaN),
            Scalar::Float64(7.0),
            Scalar::Float64(1.0),
        ];
        assert_eq!(super::nanptp(&values), Scalar::Float64(6.0));
    }

    #[test]
    fn nanptp_empty_returns_null() {
        assert!(super::nanptp(&[]).is_missing());
        assert!(super::nanptp(&[Scalar::Null(NullKind::NaN)]).is_missing());
    }

    #[test]
    fn nanptp_timedelta64_preserves_dtype_u2g0r() {
        // Per br-frankenpandas-u2g0r: ptp on Timedelta64 returns Timedelta64.
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let values = vec![
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(5 * one_hour),
            Scalar::Timedelta64(2 * one_hour),
        ];
        assert_eq!(super::nanptp(&values), Scalar::Timedelta64(4 * one_hour));
    }

    #[test]
    fn nanargmax_nanargmin_timedelta64_compare_by_ns_ql1t5() {
        // Per br-frankenpandas-ql1t5: argmax/argmin on Timedelta64 compare
        // i64 ns directly instead of silently skipping via to_f64.
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let values = vec![
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(5 * one_hour),
            Scalar::Timedelta64(one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        assert_eq!(super::nanargmax(&values), Some(1));
        assert_eq!(super::nanargmin(&values), Some(2));
    }

    #[test]
    fn nanprod_timedelta64_returns_null_szq6a() {
        // Per br-frankenpandas-szq6a: pandas raises on Timedelta prod
        // (dimensionally undefined). We surface Null instead of the
        // misleading Float64(1.0) the old empty-iterator default emitted.
        let one_hour: i64 = 3_600 * 1_000_000_000;
        let values = vec![
            Scalar::Timedelta64(2 * one_hour),
            Scalar::Timedelta64(3 * one_hour),
        ];
        assert!(super::nanprod(&values).is_missing());
    }

    #[test]
    fn nanskew_symmetric_distribution_near_zero() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Float64(2.0),
            Scalar::Float64(3.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
        ];
        // Perfectly symmetric -> skew = 0
        let skew = super::nanskew(&values);
        assert!(matches!(skew, Scalar::Float64(_)));
        let Scalar::Float64(v) = skew else {
            return;
        };
        assert!(v.abs() < 1e-9);
    }

    #[test]
    fn nanskew_too_few_values_returns_null() {
        assert!(super::nanskew(&[]).is_missing());
        assert!(super::nanskew(&[Scalar::Float64(1.0), Scalar::Float64(2.0)]).is_missing());
    }

    #[test]
    fn nankurt_symmetric_uniform_distribution() {
        let values = vec![
            Scalar::Float64(1.0),
            Scalar::Float64(2.0),
            Scalar::Float64(3.0),
            Scalar::Float64(4.0),
            Scalar::Float64(5.0),
        ];
        // pandas kurt([1,2,3,4,5]) = -1.2
        let kurt = super::nankurt(&values);
        assert!(matches!(kurt, Scalar::Float64(_)));
        let Scalar::Float64(v) = kurt else {
            return;
        };
        assert!((v + 1.2).abs() < 1e-9);
    }

    #[test]
    fn nankurt_too_few_values_returns_null() {
        let vals: Vec<Scalar> = (0..3).map(|i| Scalar::Float64(i as f64)).collect();
        assert!(super::nankurt(&vals).is_missing());
    }

    #[test]
    fn nanskew_constant_series_returns_zero() {
        let values = vec![
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
            Scalar::Float64(5.0),
        ];
        assert_eq!(super::nanskew(&values), Scalar::Float64(0.0));
        assert_eq!(
            super::nankurt(&[
                Scalar::Float64(5.0),
                Scalar::Float64(5.0),
                Scalar::Float64(5.0),
                Scalar::Float64(5.0),
            ]),
            Scalar::Float64(0.0)
        );
    }

    // ── Interval tests (br-frankenpandas-j8k4) ──────────────────────────

    #[test]
    fn interval_default_closed_is_right() {
        assert_eq!(IntervalClosed::default(), IntervalClosed::Right);
    }

    #[test]
    fn interval_left_and_right_closed_helpers() {
        assert!(IntervalClosed::Left.left_closed());
        assert!(!IntervalClosed::Left.right_closed());
        assert!(!IntervalClosed::Right.left_closed());
        assert!(IntervalClosed::Right.right_closed());
        assert!(IntervalClosed::Both.left_closed());
        assert!(IntervalClosed::Both.right_closed());
        assert!(!IntervalClosed::Neither.left_closed());
        assert!(!IntervalClosed::Neither.right_closed());
    }

    #[test]
    fn interval_display_matches_pandas_notation() {
        assert_eq!(
            Interval::new(0.0, 5.0, IntervalClosed::Right).to_string(),
            "(0, 5]"
        );
        assert_eq!(
            Interval::new(0.0, 5.0, IntervalClosed::Left).to_string(),
            "[0, 5)"
        );
        assert_eq!(
            Interval::new(0.0, 5.0, IntervalClosed::Both).to_string(),
            "[0, 5]"
        );
        assert_eq!(
            Interval::new(0.0, 5.0, IntervalClosed::Neither).to_string(),
            "(0, 5)"
        );
    }

    #[test]
    fn interval_length_and_mid() {
        let i = Interval::new(2.0, 10.0, IntervalClosed::Right);
        assert_eq!(i.length(), 8.0);
        assert_eq!(i.mid(), 6.0);
    }

    #[test]
    fn interval_contains_matches_closed_policy() {
        let right = Interval::new(0.0, 5.0, IntervalClosed::Right);
        assert!(!right.contains(0.0));
        assert!(right.contains(2.5));
        assert!(right.contains(5.0));

        let left = Interval::new(0.0, 5.0, IntervalClosed::Left);
        assert!(left.contains(0.0));
        assert!(left.contains(2.5));
        assert!(!left.contains(5.0));

        let both = Interval::new(0.0, 5.0, IntervalClosed::Both);
        assert!(both.contains(0.0));
        assert!(both.contains(5.0));

        let neither = Interval::new(0.0, 5.0, IntervalClosed::Neither);
        assert!(!neither.contains(0.0));
        assert!(!neither.contains(5.0));
        assert!(neither.contains(2.5));
    }

    #[test]
    fn interval_contains_nan_returns_false() {
        let i = Interval::new(0.0, 10.0, IntervalClosed::Both);
        assert!(!i.contains(f64::NAN));
    }

    #[test]
    fn interval_is_empty_matches_pandas() {
        // pd.Interval(3, 3, 'right').is_empty → True
        assert!(Interval::new(3.0, 3.0, IntervalClosed::Right).is_empty());
        assert!(Interval::new(3.0, 3.0, IntervalClosed::Left).is_empty());
        assert!(Interval::new(3.0, 3.0, IntervalClosed::Neither).is_empty());
        // pd.Interval(3, 3, 'both').is_empty → False (single point)
        assert!(!Interval::new(3.0, 3.0, IntervalClosed::Both).is_empty());
        // Non-degenerate intervals are never empty.
        assert!(!Interval::new(0.0, 5.0, IntervalClosed::Right).is_empty());
    }

    #[test]
    fn interval_overlaps_disjoint_returns_false() {
        let a = Interval::new(0.0, 1.0, IntervalClosed::Right);
        let b = Interval::new(2.0, 3.0, IntervalClosed::Right);
        assert!(!a.overlaps(&b));
        assert!(!b.overlaps(&a));
    }

    #[test]
    fn interval_overlaps_nested_returns_true() {
        let outer = Interval::new(0.0, 10.0, IntervalClosed::Right);
        let inner = Interval::new(3.0, 7.0, IntervalClosed::Right);
        assert!(outer.overlaps(&inner));
        assert!(inner.overlaps(&outer));
    }

    #[test]
    fn interval_overlaps_touching_respects_closed_policy() {
        // (0, 1] touching (1, 2] at point 1.
        let right_right = (
            Interval::new(0.0, 1.0, IntervalClosed::Right),
            Interval::new(1.0, 2.0, IntervalClosed::Right),
        );
        // right_right.0 is closed at 1; right_right.1 is open at 1 → no overlap.
        assert!(!right_right.0.overlaps(&right_right.1));

        // [0, 1] touching [1, 2] — both closed at 1 → overlap.
        let both_both = (
            Interval::new(0.0, 1.0, IntervalClosed::Both),
            Interval::new(1.0, 2.0, IntervalClosed::Both),
        );
        assert!(both_both.0.overlaps(&both_both.1));
    }

    #[test]
    fn interval_roundtrips_through_serde_json() {
        let i = Interval::new(1.5, 3.25, IntervalClosed::Both);
        let json = serde_json::to_string(&i).expect("serialize");
        let back: Interval = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(i, back);
    }

    #[test]
    fn interval_serde_default_closed_is_right_when_missing() {
        // JSON payloads that omit `closed` deserialize with the pandas default.
        let back: Interval =
            serde_json::from_str(r#"{"left":0.0,"right":5.0}"#).expect("deserialize");
        assert_eq!(back.closed, IntervalClosed::Right);
    }

    // ── Period tests (br-frankenpandas-epoj) ────────────────────────────

    #[test]
    fn period_freq_parses_canonical_aliases() {
        assert_eq!(PeriodFreq::parse("A"), Some(PeriodFreq::Annual));
        assert_eq!(PeriodFreq::parse("Y"), Some(PeriodFreq::Annual));
        assert_eq!(PeriodFreq::parse("Q"), Some(PeriodFreq::Quarterly));
        assert_eq!(PeriodFreq::parse("M"), Some(PeriodFreq::Monthly));
        assert_eq!(PeriodFreq::parse("W"), Some(PeriodFreq::Weekly));
        assert_eq!(PeriodFreq::parse("D"), Some(PeriodFreq::Daily));
        assert_eq!(PeriodFreq::parse("B"), Some(PeriodFreq::Business));
        assert_eq!(PeriodFreq::parse("H"), Some(PeriodFreq::Hourly));
        assert_eq!(PeriodFreq::parse("T"), Some(PeriodFreq::Minutely));
        assert_eq!(PeriodFreq::parse("min"), Some(PeriodFreq::Minutely));
        assert_eq!(PeriodFreq::parse("S"), Some(PeriodFreq::Secondly));
    }

    #[test]
    fn period_freq_parse_is_case_insensitive() {
        assert_eq!(PeriodFreq::parse("quarterly"), Some(PeriodFreq::Quarterly));
        assert_eq!(PeriodFreq::parse("MONTHLY"), Some(PeriodFreq::Monthly));
    }

    #[test]
    fn period_freq_rejects_unknown_aliases() {
        assert_eq!(PeriodFreq::parse("nanosec"), None);
        assert_eq!(PeriodFreq::parse(""), None);
        assert_eq!(PeriodFreq::parse("xyz"), None);
    }

    #[test]
    fn period_freq_alias_roundtrip() {
        for freq in [
            PeriodFreq::Annual,
            PeriodFreq::Quarterly,
            PeriodFreq::Monthly,
            PeriodFreq::Weekly,
            PeriodFreq::Daily,
            PeriodFreq::Business,
            PeriodFreq::Hourly,
            PeriodFreq::Minutely,
            PeriodFreq::Secondly,
        ] {
            assert_eq!(PeriodFreq::parse(freq.alias()), Some(freq));
        }
    }

    #[test]
    fn period_freq_anchored_aliases_are_pandas_canonical_h2wiv() {
        assert_eq!(PeriodFreq::Annual.alias(), "Y-DEC");
        assert_eq!(PeriodFreq::Quarterly.alias(), "Q-DEC");
        assert_eq!(PeriodFreq::Weekly.alias(), "W-SUN");

        assert_eq!(PeriodFreq::parse("A"), Some(PeriodFreq::Annual));
        assert_eq!(PeriodFreq::parse("Y"), Some(PeriodFreq::Annual));
        assert_eq!(PeriodFreq::parse("Y-DEC"), Some(PeriodFreq::Annual));
        assert_eq!(PeriodFreq::parse("Q"), Some(PeriodFreq::Quarterly));
        assert_eq!(PeriodFreq::parse("Q-DEC"), Some(PeriodFreq::Quarterly));
        assert_eq!(PeriodFreq::parse("W"), Some(PeriodFreq::Weekly));
        assert_eq!(PeriodFreq::parse("W-SUN"), Some(PeriodFreq::Weekly));
    }

    #[test]
    fn period_freq_intraday_aliases_are_pandas_canonical_8kfdo() {
        assert_eq!(PeriodFreq::Hourly.alias(), "h");
        assert_eq!(PeriodFreq::Minutely.alias(), "min");
        assert_eq!(PeriodFreq::Secondly.alias(), "s");

        assert_eq!(PeriodFreq::parse("H"), Some(PeriodFreq::Hourly));
        assert_eq!(PeriodFreq::parse("T"), Some(PeriodFreq::Minutely));
        assert_eq!(PeriodFreq::parse("S"), Some(PeriodFreq::Secondly));
    }

    #[test]
    fn period_scalar_accessors_match_pandas_star8() {
        let period = Period::new(600, PeriodFreq::Monthly);

        assert_eq!(period.ordinal(), 600);
        assert_eq!(period.freq(), PeriodFreq::Monthly);
        assert_eq!(period.freqstr(), "M");
    }

    #[test]
    fn period_parse_common_pandas_ordinals_avm08() {
        assert_eq!(
            Period::parse("2024").unwrap(),
            Period::new(54, PeriodFreq::Annual)
        );
        assert_eq!(
            Period::parse("2024Q1").unwrap(),
            Period::new(216, PeriodFreq::Quarterly)
        );
        assert_eq!(
            Period::parse("2024-01").unwrap(),
            Period::new(648, PeriodFreq::Monthly)
        );
        assert_eq!(
            Period::parse("2024-01-15").unwrap(),
            Period::new(19_737, PeriodFreq::Daily)
        );
        assert!(Period::parse("216").is_err());
    }

    #[test]
    fn period_shift_advances_ordinal() {
        let q1 = Period::new(216, PeriodFreq::Quarterly);
        let q2 = q1.shift(1);
        assert_eq!(q2.ordinal, 217);
        assert_eq!(q2.freq, PeriodFreq::Quarterly);
        let q0 = q1.shift(-1);
        assert_eq!(q0.ordinal, 215);
    }

    #[test]
    fn period_shift_saturates_on_overflow() {
        let p = Period::new(i64::MAX - 2, PeriodFreq::Daily);
        assert_eq!(p.shift(100).ordinal, i64::MAX);
        let p = Period::new(i64::MIN + 2, PeriodFreq::Daily);
        assert_eq!(p.shift(-100).ordinal, i64::MIN);
    }

    #[test]
    fn period_diff_returns_period_count() {
        let a = Period::new(216, PeriodFreq::Quarterly);
        let b = Period::new(220, PeriodFreq::Quarterly);
        assert_eq!(b.diff(&a), Some(4));
        assert_eq!(a.diff(&b), Some(-4));
    }

    #[test]
    fn period_diff_rejects_mismatched_freq() {
        let monthly = Period::new(100, PeriodFreq::Monthly);
        let quarterly = Period::new(100, PeriodFreq::Quarterly);
        assert_eq!(monthly.diff(&quarterly), None);
        assert_eq!(quarterly.diff(&monthly), None);
    }

    #[test]
    fn period_cmp_same_freq_respects_ordinal_order() {
        use std::cmp::Ordering;
        let a = Period::new(10, PeriodFreq::Monthly);
        let b = Period::new(20, PeriodFreq::Monthly);
        assert_eq!(a.cmp_same_freq(&b), Some(Ordering::Less));
        assert_eq!(b.cmp_same_freq(&a), Some(Ordering::Greater));
        assert_eq!(a.cmp_same_freq(&a), Some(Ordering::Equal));
    }

    #[test]
    fn period_cmp_cross_freq_returns_none() {
        let m = Period::new(1, PeriodFreq::Monthly);
        let q = Period::new(1, PeriodFreq::Quarterly);
        assert_eq!(m.cmp_same_freq(&q), None);
    }

    #[test]
    fn period_display_carries_freq_and_ordinal() {
        let p = Period::new(216, PeriodFreq::Quarterly);
        assert_eq!(p.to_string(), "Period[Q-DEC, 216]");
    }

    #[test]
    fn period_roundtrips_through_serde_json() {
        let p = Period::new(42, PeriodFreq::Weekly);
        let json = serde_json::to_string(&p).expect("serialize");
        let back: Period = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(p, back);
    }

    // ── period_range tests (br-frankenpandas-2jef — epoj Phase 2) ───────

    use super::period_range;

    #[test]
    fn period_range_zero_periods_is_empty() {
        let start = Period::new(216, PeriodFreq::Quarterly);
        assert!(period_range(start, 0).is_empty());
    }

    #[test]
    fn period_range_single_period_returns_start_only() {
        let start = Period::new(216, PeriodFreq::Quarterly);
        let r = period_range(start, 1);
        assert_eq!(r.len(), 1);
        assert_eq!(r[0], start);
    }

    #[test]
    fn period_range_increments_ordinal_by_one_per_step() {
        let start = Period::new(216, PeriodFreq::Quarterly);
        let r = period_range(start, 4);
        assert_eq!(r.len(), 4);
        assert_eq!(r[0].ordinal, 216);
        assert_eq!(r[1].ordinal, 217);
        assert_eq!(r[2].ordinal, 218);
        assert_eq!(r[3].ordinal, 219);
    }

    #[test]
    fn period_range_preserves_frequency() {
        let start = Period::new(0, PeriodFreq::Monthly);
        let r = period_range(start, 12);
        assert!(r.iter().all(|p| p.freq == PeriodFreq::Monthly));
    }

    #[test]
    fn period_range_negative_starting_ordinal_works() {
        // Ordinal axis is signed — pre-epoch periods are valid.
        let start = Period::new(-3, PeriodFreq::Annual);
        let r = period_range(start, 5);
        assert_eq!(
            r.iter().map(|p| p.ordinal).collect::<Vec<_>>(),
            vec![-3, -2, -1, 0, 1]
        );
    }

    #[test]
    fn period_range_large_n_does_not_panic() {
        // 1024 monthly periods — large enough to catch any allocation bug.
        let start = Period::new(0, PeriodFreq::Monthly);
        let r = period_range(start, 1024);
        assert_eq!(r.len(), 1024);
        assert_eq!(r[1023].ordinal, 1023);
    }

    // ── interval_range tests (br-frankenpandas-xaom) ────────────────────

    use super::{TypeError, interval_range_by_periods, interval_range_by_step};

    #[test]
    fn interval_range_by_periods_matches_pandas_default_case() {
        // pd.interval_range(0, 10, periods=5) → [(0,2],(2,4],(4,6],(6,8],(8,10]]
        let bins = interval_range_by_periods(0.0, 10.0, 5, IntervalClosed::Right);
        assert_eq!(bins.len(), 5);
        for (i, bin) in bins.iter().enumerate() {
            assert_eq!(bin.left, (i as f64) * 2.0);
            assert_eq!(bin.right, ((i + 1) as f64) * 2.0);
            assert_eq!(bin.closed, IntervalClosed::Right);
        }
    }

    #[test]
    fn interval_range_by_periods_final_edge_is_exact_end() {
        // Guards against accumulated float drift on the last right edge.
        let bins = interval_range_by_periods(0.0, 1.0, 7, IntervalClosed::Right);
        assert_eq!(bins.last().unwrap().right, 1.0);
    }

    #[test]
    fn interval_range_by_periods_zero_periods_is_empty() {
        assert!(interval_range_by_periods(0.0, 10.0, 0, IntervalClosed::Right).is_empty());
    }

    #[test]
    fn interval_range_by_periods_reversed_range_is_empty() {
        // pandas: pd.interval_range(10, 0, periods=5) → IntervalIndex([]).
        assert!(interval_range_by_periods(10.0, 0.0, 5, IntervalClosed::Right).is_empty());
    }

    #[test]
    fn interval_range_by_periods_preserves_closed_policy() {
        for closed in [
            IntervalClosed::Left,
            IntervalClosed::Right,
            IntervalClosed::Both,
            IntervalClosed::Neither,
        ] {
            let bins = interval_range_by_periods(0.0, 4.0, 2, closed);
            assert!(bins.iter().all(|b| b.closed == closed));
        }
    }

    #[test]
    fn interval_range_by_step_matches_pandas_default_case() {
        // pd.interval_range(0, 10, freq=2) → [(0,2],(2,4],(4,6],(6,8],(8,10]]
        let bins = interval_range_by_step(0.0, 10.0, 2.0, IntervalClosed::Right).expect("ok");
        assert_eq!(bins.len(), 5);
        assert_eq!(bins[0].left, 0.0);
        assert_eq!(bins[4].right, 10.0);
    }

    #[test]
    fn interval_range_by_step_rejects_non_positive_step() {
        assert!(matches!(
            interval_range_by_step(0.0, 10.0, 0.0, IntervalClosed::Right),
            Err(TypeError::InvalidIntervalStep { .. })
        ));
        assert!(matches!(
            interval_range_by_step(0.0, 10.0, -2.0, IntervalClosed::Right),
            Err(TypeError::InvalidIntervalStep { .. })
        ));
        assert!(matches!(
            interval_range_by_step(0.0, 10.0, f64::NAN, IntervalClosed::Right),
            Err(TypeError::InvalidIntervalStep { .. })
        ));
        assert!(matches!(
            interval_range_by_step(0.0, 10.0, f64::INFINITY, IntervalClosed::Right),
            Err(TypeError::InvalidIntervalStep { .. })
        ));
    }

    #[test]
    fn interval_range_by_step_rejects_non_dividing_step() {
        // pandas: pd.interval_range(0, 10, freq=3) → ValueError
        // (span=10 not divisible by step=3). Reject with IntervalStepDoesNotDivide.
        assert!(matches!(
            interval_range_by_step(0.0, 10.0, 3.0, IntervalClosed::Right),
            Err(TypeError::IntervalStepDoesNotDivide { .. })
        ));
    }

    #[test]
    fn interval_range_by_step_reversed_range_is_empty() {
        let bins = interval_range_by_step(10.0, 0.0, 2.0, IntervalClosed::Right).expect("ok");
        assert!(bins.is_empty());
    }

    #[test]
    fn interval_range_by_step_degenerate_zero_span_is_empty() {
        let bins = interval_range_by_step(5.0, 5.0, 1.0, IntervalClosed::Right).expect("ok");
        assert!(bins.is_empty());
    }

    #[test]
    fn interval_range_by_step_accepts_float_step_within_tolerance() {
        // step=0.1 ten times == 1.0 but float arithmetic produces 0.9999...
        let bins = interval_range_by_step(0.0, 1.0, 0.1, IntervalClosed::Right).expect("ok");
        assert_eq!(bins.len(), 10);
        assert_eq!(bins.last().unwrap().right, 1.0);
    }

    // ── Timedelta arithmetic tests (br-frankenpandas-4r56 Phase 1) ──────

    use super::Timedelta;

    #[test]
    fn timedelta_add_sums_non_nat() {
        let one_hour = Timedelta::NANOS_PER_HOUR;
        let one_day = Timedelta::NANOS_PER_DAY;
        assert_eq!(Timedelta::add(one_hour, one_day), one_hour + one_day);
    }

    #[test]
    fn timedelta_add_propagates_nat() {
        assert_eq!(Timedelta::add(Timedelta::NAT, 100), Timedelta::NAT);
        assert_eq!(Timedelta::add(100, Timedelta::NAT), Timedelta::NAT);
        assert_eq!(
            Timedelta::add(Timedelta::NAT, Timedelta::NAT),
            Timedelta::NAT
        );
    }

    #[test]
    fn timedelta_add_saturates_on_overflow() {
        assert_eq!(Timedelta::add(i64::MAX - 10, 100), i64::MAX);
        // Note: i64::MIN is NaT; use MIN+1 to test saturation on the negative side.
        assert_eq!(Timedelta::add(i64::MIN + 10, -100), i64::MIN);
    }

    #[test]
    fn timedelta_sub_subtracts_non_nat() {
        let one_hour = Timedelta::NANOS_PER_HOUR;
        assert_eq!(
            Timedelta::sub(one_hour, Timedelta::NANOS_PER_MIN),
            one_hour - Timedelta::NANOS_PER_MIN
        );
    }

    #[test]
    fn timedelta_sub_propagates_nat() {
        assert_eq!(Timedelta::sub(Timedelta::NAT, 100), Timedelta::NAT);
        assert_eq!(Timedelta::sub(100, Timedelta::NAT), Timedelta::NAT);
    }

    #[test]
    fn timedelta_neg_flips_sign_non_nat() {
        assert_eq!(Timedelta::neg(5), -5);
        assert_eq!(Timedelta::neg(-5), 5);
        assert_eq!(Timedelta::neg(0), 0);
    }

    #[test]
    fn timedelta_neg_preserves_nat() {
        assert_eq!(Timedelta::neg(Timedelta::NAT), Timedelta::NAT);
    }

    #[test]
    fn timedelta_abs_returns_magnitude() {
        assert_eq!(Timedelta::abs(-5), 5);
        assert_eq!(Timedelta::abs(5), 5);
        assert_eq!(Timedelta::abs(0), 0);
        assert_eq!(Timedelta::abs(Timedelta::NAT), Timedelta::NAT);
    }

    #[test]
    fn timedelta_mul_scalar_scales() {
        let three_hours = Timedelta::NANOS_PER_HOUR * 3;
        assert_eq!(
            Timedelta::mul_scalar(Timedelta::NANOS_PER_HOUR, 3),
            three_hours
        );
        assert_eq!(Timedelta::mul_scalar(100, 0), 0);
        assert_eq!(Timedelta::mul_scalar(100, -2), -200);
    }

    #[test]
    fn timedelta_mul_scalar_saturates() {
        assert_eq!(Timedelta::mul_scalar(i64::MAX, 2), i64::MAX);
        // (i64::MIN + 1) * 2 saturates to i64::MIN (magnitude too large).
        assert_eq!(Timedelta::mul_scalar(i64::MIN + 1, 2), i64::MIN);
    }

    #[test]
    fn timedelta_mul_scalar_propagates_nat() {
        assert_eq!(Timedelta::mul_scalar(Timedelta::NAT, 5), Timedelta::NAT);
    }

    #[test]
    fn timedelta_div_scalar_floor_divides() {
        // Floor division (matches Python / pandas): -100 // 3 == -34, not -33.
        assert_eq!(Timedelta::div_scalar(100, 3), 33);
        assert_eq!(Timedelta::div_scalar(-100, 3), -34);
        assert_eq!(Timedelta::div_scalar(100, -3), -34);
        assert_eq!(Timedelta::div_scalar(-100, -3), 33);
    }

    #[test]
    fn timedelta_div_scalar_zero_divisor_returns_nat() {
        assert_eq!(Timedelta::div_scalar(100, 0), Timedelta::NAT);
    }

    #[test]
    fn timedelta_div_scalar_min_neg_one_propagates_nat() {
        // i64::MIN aliases NaT, so `div_scalar(i64::MIN, _)` propagates NaT
        // — the `i64::MIN / -1` arithmetic-overflow case is subsumed.
        assert_eq!(Timedelta::div_scalar(i64::MIN, -1), Timedelta::NAT);
        // (i64::MIN + 1) is a real timedelta; `/ -1` does not overflow.
        assert_eq!(Timedelta::div_scalar(i64::MIN + 1, -1), i64::MAX);
    }

    #[test]
    fn timedelta_div_scalar_propagates_nat() {
        assert_eq!(Timedelta::div_scalar(Timedelta::NAT, 10), Timedelta::NAT);
    }

    #[test]
    fn timedelta_div_timedelta_returns_float_ratio() {
        let two_hours = Timedelta::NANOS_PER_HOUR * 2;
        let one_hour = Timedelta::NANOS_PER_HOUR;
        assert!((Timedelta::div_timedelta(two_hours, one_hour) - 2.0).abs() < 1e-12);
        assert!((Timedelta::div_timedelta(one_hour, two_hours) - 0.5).abs() < 1e-12);
    }

    #[test]
    fn timedelta_div_timedelta_nat_returns_nan() {
        assert!(Timedelta::div_timedelta(Timedelta::NAT, 100).is_nan());
        assert!(Timedelta::div_timedelta(100, Timedelta::NAT).is_nan());
    }

    // ── Timestamp tests (br-frankenpandas-9p0u — 4r56 Phase 2) ──────────

    use super::Timestamp;

    #[test]
    fn timestamp_from_nanos_is_naive_utc() {
        let ts = Timestamp::from_nanos(1_700_000_000_000_000_000);
        assert_eq!(ts.nanos, 1_700_000_000_000_000_000);
        assert_eq!(ts.tz, None);
        assert!(!ts.is_nat());
    }

    #[test]
    fn timestamp_from_nanos_tz_carries_tz_name() {
        let ts = Timestamp::from_nanos_tz(1_700_000_000_000_000_000, "US/Eastern");
        assert_eq!(ts.tz.as_deref(), Some("US/Eastern"));
    }

    #[test]
    fn timestamp_now_returns_current_time() {
        let before = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as i64;
        let ts = Timestamp::now();
        let after = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as i64;
        assert!(ts.nanos >= before);
        assert!(ts.nanos <= after);
        assert!(!ts.is_nat());
    }

    #[test]
    fn timestamp_today_returns_midnight() {
        let ts = Timestamp::today();
        assert!(!ts.is_nat());
        // Today should be normalized (midnight), so hour/min/sec should be 0
        assert_eq!(ts.hour(), Some(0));
        assert_eq!(ts.minute(), Some(0));
        assert_eq!(ts.second(), Some(0));
    }

    #[test]
    fn timestamp_add_timedelta_shifts_nanos_and_preserves_tz() {
        let ts = Timestamp::from_nanos_tz(0, "US/Eastern");
        let one_day = Timedelta::NANOS_PER_DAY;
        let shifted = ts.add_timedelta(one_day);
        assert_eq!(shifted.nanos, one_day);
        assert_eq!(shifted.tz.as_deref(), Some("US/Eastern"));
    }

    #[test]
    fn timestamp_add_timedelta_saturates_on_overflow() {
        let ts = Timestamp::from_nanos(i64::MAX - 10);
        let shifted = ts.add_timedelta(100);
        assert_eq!(shifted.nanos, i64::MAX);
    }

    #[test]
    fn timestamp_add_timedelta_propagates_nat() {
        // NaT Timestamp + anything = NaT.
        assert!(Timestamp::nat().add_timedelta(100).is_nat());
        // Timestamp + NaT Timedelta = NaT.
        assert!(
            Timestamp::from_nanos(0)
                .add_timedelta(Timedelta::NAT)
                .is_nat()
        );
    }

    #[test]
    fn timestamp_sub_timedelta_shifts_backward() {
        let ts = Timestamp::from_nanos(1_000);
        let shifted = ts.sub_timedelta(Timedelta::NANOS_PER_MICRO);
        assert_eq!(shifted.nanos, 0);
    }

    #[test]
    fn timestamp_sub_timestamp_returns_timedelta_nanos() {
        let t0 = Timestamp::from_nanos(0);
        let t1 = Timestamp::from_nanos(Timedelta::NANOS_PER_HOUR);
        assert_eq!(t1.sub_timestamp(&t0), Timedelta::NANOS_PER_HOUR);
        assert_eq!(t0.sub_timestamp(&t1), -Timedelta::NANOS_PER_HOUR);
    }

    #[test]
    fn timestamp_sub_timestamp_nat_propagates() {
        let ts = Timestamp::from_nanos(1_000);
        assert_eq!(Timestamp::nat().sub_timestamp(&ts), Timedelta::NAT);
        assert_eq!(ts.sub_timestamp(&Timestamp::nat()), Timedelta::NAT);
    }

    #[test]
    fn timestamp_semantic_eq_treats_two_nat_as_equal() {
        assert!(Timestamp::nat().semantic_eq(&Timestamp::nat()));
        assert!(!Timestamp::nat().semantic_eq(&Timestamp::from_nanos(0)));
        assert!(!Timestamp::from_nanos(0).semantic_eq(&Timestamp::nat()));
    }

    #[test]
    fn timestamp_partial_cmp_orders_by_nanos_nat_is_incomparable() {
        use std::cmp::Ordering;
        let a = Timestamp::from_nanos(0);
        let b = Timestamp::from_nanos(100);
        assert_eq!(a.partial_cmp(&b), Some(Ordering::Less));
        assert_eq!(b.partial_cmp(&a), Some(Ordering::Greater));
        assert_eq!(a.partial_cmp(&a), Some(Ordering::Equal));
        assert_eq!(a.partial_cmp(&Timestamp::nat()), None);
        assert_eq!(Timestamp::nat().partial_cmp(&Timestamp::nat()), None);
    }

    #[test]
    fn timestamp_display_matches_phase2_debug_format() {
        assert_eq!(Timestamp::from_nanos(42).to_string(), "Timestamp[42, UTC]");
        assert_eq!(
            Timestamp::from_nanos_tz(42, "US/Eastern").to_string(),
            "Timestamp[42, US/Eastern]"
        );
        assert_eq!(Timestamp::nat().to_string(), "NaT");
    }

    #[test]
    fn timestamp_value_and_unit_match_pandas_l0edr() {
        let ts = Timestamp::from_nanos(1_000_000_123);
        assert_eq!(ts.value(), 1_000_000_123);
        assert_eq!(ts.unit(), Some("ns"));

        let nat = Timestamp::nat();
        assert_eq!(nat.value(), Timestamp::NAT);
        assert_eq!(nat.unit(), None);
    }

    #[test]
    fn timestamp_numpy_datetime64_materializers_match_value_twksi() {
        let ts = Timestamp::from_nanos(1_000_000_123);
        assert_eq!(ts.asm8(), ts.value());
        assert_eq!(ts.to_datetime64(), ts.value());
        assert_eq!(ts.to_numpy(), ts.value());

        let nat = Timestamp::nat();
        assert_eq!(nat.asm8(), Timestamp::NAT);
        assert_eq!(nat.to_datetime64(), Timestamp::NAT);
        assert_eq!(nat.to_numpy(), Timestamp::NAT);
    }

    #[test]
    fn timestamp_timestamp_accessor_matches_pandas_microsecond_rounding_py0h3() {
        assert_eq!(Timestamp::from_nanos(0).timestamp(), Ok(0.0));
        assert_eq!(Timestamp::from_nanos(1_500_000_000).timestamp(), Ok(1.5));
        assert_eq!(Timestamp::from_nanos(500).timestamp(), Ok(0.0));
        assert_eq!(Timestamp::from_nanos(501).timestamp(), Ok(0.000001));
        assert_eq!(Timestamp::from_nanos(2_500).timestamp(), Ok(0.000003));

        assert!(matches!(
            Timestamp::from_nanos(-500).timestamp(),
            Ok(value) if value == -0.0 && value.is_sign_negative()
        ));
        assert_eq!(Timestamp::from_nanos(-2_500).timestamp(), Ok(-0.000003));
        assert_eq!(
            Timestamp::nat().timestamp(),
            Err(TypeError::ValueIsMissing {
                kind: NullKind::NaT,
            })
        );
    }

    #[test]
    fn timestamp_roundtrips_through_serde_json() {
        let naive = Timestamp::from_nanos(1_700_000_000_000_000_000);
        let json = serde_json::to_string(&naive).expect("serialize");
        let back: Timestamp = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(naive, back);

        let tz_aware = Timestamp::from_nanos_tz(1_700_000_000_000_000_000, "US/Eastern");
        let json = serde_json::to_string(&tz_aware).expect("serialize");
        let back: Timestamp = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(tz_aware, back);
    }

    #[test]
    fn timestamp_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<Timestamp>();
    }

    // ── Timestamp rounding tests (br-frankenpandas-5h6n) ────────────────

    #[test]
    fn timestamp_floor_to_rounds_down() {
        // 12:34:56 → floor by 1H → 12:00:00
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let twelve_thirty_four =
            twelve_h + Timedelta::NANOS_PER_MIN * 34 + Timedelta::NANOS_PER_SEC * 56;
        let ts = Timestamp::from_nanos(twelve_thirty_four);
        let floored = ts.floor_to(h);
        assert_eq!(floored.nanos, twelve_h);
    }

    #[test]
    fn timestamp_floor_to_handles_already_aligned() {
        // 12:00:00 floored by 1H → 12:00:00 (no change).
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let ts = Timestamp::from_nanos(twelve_h);
        assert_eq!(ts.floor_to(h).nanos, twelve_h);
    }

    #[test]
    fn timestamp_floor_to_handles_negative_nanos() {
        // -100 ns floored by 60 ns:
        //   div_euclid(-100, 60) = -2 (since -2*60=-120, rem=20 ≥ 0).
        //   result = -2 * 60 = -120.
        let ts = Timestamp::from_nanos(-100);
        assert_eq!(ts.floor_to(60).nanos, -120);
    }

    #[test]
    fn timestamp_ceil_to_rounds_up() {
        // 12:34:56 → ceil by 1H → 13:00:00.
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let thirteen_h = h * 13;
        let twelve_thirty_four =
            twelve_h + Timedelta::NANOS_PER_MIN * 34 + Timedelta::NANOS_PER_SEC * 56;
        let ts = Timestamp::from_nanos(twelve_thirty_four);
        assert_eq!(ts.ceil_to(h).nanos, thirteen_h);
    }

    #[test]
    fn timestamp_ceil_to_no_op_on_aligned() {
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let ts = Timestamp::from_nanos(twelve_h);
        assert_eq!(ts.ceil_to(h).nanos, twelve_h);
    }

    #[test]
    fn timestamp_round_to_rounds_to_nearest() {
        // 12:30:01 (one second past the half-hour): round to 1H → 13:00:00.
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let twelve_thirty_one_sec =
            twelve_h + Timedelta::NANOS_PER_MIN * 30 + Timedelta::NANOS_PER_SEC;
        let ts = Timestamp::from_nanos(twelve_thirty_one_sec);
        assert_eq!(ts.round_to(h).nanos, h * 13);

        // 12:29:59 (one second before half): round to 1H → 12:00:00.
        let twelve_twenty_nine_sec =
            twelve_h + Timedelta::NANOS_PER_MIN * 29 + Timedelta::NANOS_PER_SEC * 59;
        let ts = Timestamp::from_nanos(twelve_twenty_nine_sec);
        assert_eq!(ts.round_to(h).nanos, twelve_h);
    }

    #[test]
    fn timestamp_round_to_bankers_tie_to_even() {
        // Tie cases: rem == unit/2 exactly. Pick even-multiple floor.
        // unit=10, so half=5. nanos=5: floor=0 (even), so → 0.
        // nanos=15: floor=1 (odd), so → 20.
        // nanos=25: floor=2 (even), so → 20.
        // nanos=35: floor=3 (odd), so → 40.
        assert_eq!(Timestamp::from_nanos(5).round_to(10).nanos, 0);
        assert_eq!(Timestamp::from_nanos(15).round_to(10).nanos, 20);
        assert_eq!(Timestamp::from_nanos(25).round_to(10).nanos, 20);
        assert_eq!(Timestamp::from_nanos(35).round_to(10).nanos, 40);
    }

    #[test]
    fn timestamp_round_to_zero_unit_returns_nat() {
        let ts = Timestamp::from_nanos(100);
        assert!(ts.round_to(0).is_nat());
        assert!(ts.floor_to(0).is_nat());
        assert!(ts.ceil_to(0).is_nat());
    }

    #[test]
    fn timestamp_round_to_negative_unit_returns_nat() {
        let ts = Timestamp::from_nanos(100);
        assert!(ts.round_to(-10).is_nat());
        assert!(ts.floor_to(-10).is_nat());
        assert!(ts.ceil_to(-10).is_nat());
    }

    #[test]
    fn timestamp_rounding_propagates_nat() {
        let nat = Timestamp::nat();
        assert!(nat.floor_to(60).is_nat());
        assert!(nat.ceil_to(60).is_nat());
        assert!(nat.round_to(60).is_nat());
    }

    #[test]
    fn timestamp_rounding_preserves_tz() {
        let ts = Timestamp::from_nanos_tz(100, "US/Eastern");
        assert_eq!(ts.floor_to(60).tz.as_deref(), Some("US/Eastern"));
        assert_eq!(ts.ceil_to(60).tz.as_deref(), Some("US/Eastern"));
        assert_eq!(ts.round_to(60).tz.as_deref(), Some("US/Eastern"));
    }

    // ── Timestamp string-unit rounding tests (br-frankenpandas-lbsx) ────

    #[test]
    fn timestamp_floor_to_unit_h_rounds_to_hour() {
        let h = Timedelta::NANOS_PER_HOUR;
        let twelve_h = h * 12;
        let twelve_thirty_four =
            twelve_h + Timedelta::NANOS_PER_MIN * 34 + Timedelta::NANOS_PER_SEC * 56;
        let ts = Timestamp::from_nanos(twelve_thirty_four);
        assert_eq!(ts.floor_to_unit("H").nanos, twelve_h);
        assert_eq!(ts.floor_to_unit("h").nanos, twelve_h);
        assert_eq!(ts.floor_to_unit("hour").nanos, twelve_h);
        assert_eq!(ts.floor_to_unit("hours").nanos, twelve_h);
        assert_eq!(ts.floor_to_unit("hr").nanos, twelve_h);
    }

    #[test]
    fn timestamp_ceil_to_unit_d_rounds_to_day() {
        // 12:34:56 → ceil to 1 day → 24:00:00 (next day).
        let h = Timedelta::NANOS_PER_HOUR;
        let d = Timedelta::NANOS_PER_DAY;
        let twelve_thirty_four = h * 12 + Timedelta::NANOS_PER_MIN * 34;
        let ts = Timestamp::from_nanos(twelve_thirty_four);
        assert_eq!(ts.ceil_to_unit("D").nanos, d);
        assert_eq!(ts.ceil_to_unit("day").nanos, d);
        assert_eq!(ts.ceil_to_unit("days").nanos, d);
    }

    #[test]
    fn timestamp_round_to_unit_min_rounds_to_minute() {
        // 12:34:31 → round to 1 minute → 12:35:00.
        let m = Timedelta::NANOS_PER_MIN;
        let twelve_thirty_four_thirty_one =
            Timedelta::NANOS_PER_HOUR * 12 + m * 34 + Timedelta::NANOS_PER_SEC * 31;
        let ts = Timestamp::from_nanos(twelve_thirty_four_thirty_one);
        let expected = Timedelta::NANOS_PER_HOUR * 12 + m * 35;
        assert_eq!(ts.round_to_unit("min").nanos, expected);
        assert_eq!(ts.round_to_unit("T").nanos, expected); // pandas pre-2.2 alias
        assert_eq!(ts.round_to_unit("minute").nanos, expected);
    }

    #[test]
    fn timestamp_floor_ceil_round_aliases_match_unit_methods_li897() {
        let ts = Timestamp::from_nanos(
            Timedelta::NANOS_PER_HOUR * 12
                + Timedelta::NANOS_PER_MIN * 34
                + Timedelta::NANOS_PER_SEC * 31,
        );

        assert_eq!(ts.floor("H"), ts.floor_to_unit("H"));
        assert_eq!(ts.ceil("D"), ts.ceil_to_unit("D"));
        assert_eq!(ts.round("min"), ts.round_to_unit("min"));
    }

    #[test]
    fn timestamp_normalize_floors_to_day_and_preserves_tz_455op() {
        let ts = Timestamp::from_nanos_tz(
            Timedelta::NANOS_PER_DAY * 3
                + Timedelta::NANOS_PER_HOUR * 12
                + Timedelta::NANOS_PER_MIN * 34,
            "US/Eastern",
        );
        let normalized = ts.normalize();

        assert_eq!(normalized.nanos, Timedelta::NANOS_PER_DAY * 3);
        assert_eq!(normalized.tz.as_deref(), Some("US/Eastern"));
        assert!(Timestamp::nat().normalize().is_nat());
    }

    #[test]
    fn timestamp_unit_rounding_unknown_unit_returns_nat() {
        let ts = Timestamp::from_nanos(100);
        assert!(ts.floor_to_unit("fortnight").is_nat());
        assert!(ts.ceil_to_unit("century").is_nat());
        assert!(ts.round_to_unit("xyz").is_nat());
    }

    #[test]
    fn timestamp_unit_rounding_propagates_nat() {
        let nat = Timestamp::nat();
        assert!(nat.floor_to_unit("H").is_nat());
        assert!(nat.ceil_to_unit("H").is_nat());
        assert!(nat.round_to_unit("H").is_nat());
    }

    #[test]
    fn timestamp_unit_rounding_preserves_tz() {
        let ts = Timestamp::from_nanos_tz(Timedelta::NANOS_PER_HOUR * 12 + 100, "US/Eastern");
        assert_eq!(ts.floor_to_unit("H").tz.as_deref(), Some("US/Eastern"));
        assert_eq!(ts.ceil_to_unit("H").tz.as_deref(), Some("US/Eastern"));
        assert_eq!(ts.round_to_unit("H").tz.as_deref(), Some("US/Eastern"));
    }

    #[test]
    fn timedelta_unit_to_nanos_is_now_public_and_matches_pandas_aliases() {
        // Public surface check: pandas alias core set.
        assert_eq!(
            Timedelta::unit_to_nanos("W"),
            Some(Timedelta::NANOS_PER_WEEK)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("D"),
            Some(Timedelta::NANOS_PER_DAY)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("H"),
            Some(Timedelta::NANOS_PER_HOUR)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("min"),
            Some(Timedelta::NANOS_PER_MIN)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("s"),
            Some(Timedelta::NANOS_PER_SEC)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("ms"),
            Some(Timedelta::NANOS_PER_MILLI)
        );
        assert_eq!(
            Timedelta::unit_to_nanos("us"),
            Some(Timedelta::NANOS_PER_MICRO)
        );
        assert_eq!(Timedelta::unit_to_nanos("ns"), Some(1));
        // Empty string → days (pandas default).
        assert_eq!(Timedelta::unit_to_nanos(""), Some(Timedelta::NANOS_PER_DAY));
        // Unknown alias → None.
        assert_eq!(Timedelta::unit_to_nanos("century"), None);
    }

    #[test]
    fn timestamp_isoformat_basic() {
        let ts = Timestamp::from_nanos(0);
        assert_eq!(ts.isoformat(), "1970-01-01T00:00:00");

        let ts_utc = Timestamp::from_nanos_tz(0, "UTC");
        assert_eq!(ts_utc.isoformat(), "1970-01-01T00:00:00+00:00");

        let ts_tz = Timestamp::from_nanos_tz(
            Timedelta::NANOS_PER_DAY
                + Timedelta::NANOS_PER_HOUR * 14
                + Timedelta::NANOS_PER_MIN * 30,
            "America/New_York",
        );
        assert!(ts_tz.isoformat().contains("1970-01-02T14:30:00"));
        assert!(ts_tz.isoformat().contains("[America/New_York]"));

        assert_eq!(Timestamp::nat().isoformat(), "NaT");
    }

    #[test]
    fn timestamp_strftime_basic() {
        let ts = Timestamp::from_nanos(
            Timedelta::NANOS_PER_DAY * 365
                + Timedelta::NANOS_PER_HOUR * 9
                + Timedelta::NANOS_PER_MIN * 15,
        );
        assert_eq!(ts.strftime("%Y-%m-%d"), "1971-01-01");
        assert_eq!(ts.strftime("%H:%M:%S"), "09:15:00");
        assert_eq!(ts.strftime("%Y/%m/%d %H:%M"), "1971/01/01 09:15");
        assert_eq!(Timestamp::nat().strftime("%Y-%m-%d"), "NaT");
    }

    #[test]
    fn timestamp_day_name_and_month_name() {
        let ts = Timestamp::from_nanos(0);
        assert_eq!(ts.day_name(), "Thursday");
        assert_eq!(ts.month_name(), "January");

        let ts2 = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 365);
        assert_eq!(ts2.day_name(), "Friday");
        assert_eq!(ts2.month_name(), "January");

        assert_eq!(Timestamp::nat().day_name(), "NaT");
        assert_eq!(Timestamp::nat().month_name(), "NaT");
    }

    #[test]
    fn timestamp_component_accessors() {
        let ts = Timestamp::from_nanos(0);
        assert_eq!(ts.year(), Some(1970));
        assert_eq!(ts.month(), Some(1));
        assert_eq!(ts.day(), Some(1));
        assert_eq!(ts.hour(), Some(0));
        assert_eq!(ts.minute(), Some(0));
        assert_eq!(ts.second(), Some(0));
        assert_eq!(ts.microsecond(), Some(0));
        assert_eq!(ts.nanosecond(), Some(0));

        let ts2 = Timestamp::from_nanos(
            Timedelta::NANOS_PER_DAY * 365
                + Timedelta::NANOS_PER_HOUR * 14
                + Timedelta::NANOS_PER_MIN * 30
                + Timedelta::NANOS_PER_SEC * 45
                + 123_456_789,
        );
        assert_eq!(ts2.year(), Some(1971));
        assert_eq!(ts2.month(), Some(1));
        assert_eq!(ts2.day(), Some(1));
        assert_eq!(ts2.hour(), Some(14));
        assert_eq!(ts2.minute(), Some(30));
        assert_eq!(ts2.second(), Some(45));
        assert_eq!(ts2.microsecond(), Some(123456));
        assert_eq!(ts2.nanosecond(), Some(789));

        assert_eq!(Timestamp::nat().year(), None);
        assert_eq!(Timestamp::nat().month(), None);
        assert_eq!(Timestamp::nat().day(), None);
    }

    #[test]
    fn timestamp_dayofweek_dayofyear_quarter() {
        let ts = Timestamp::from_nanos(0);
        assert_eq!(ts.dayofweek(), Some(3));
        assert_eq!(ts.weekday(), Some(3));
        assert_eq!(ts.dayofyear(), Some(1));
        assert_eq!(ts.quarter(), Some(1));

        let ts2 = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 90);
        assert_eq!(ts2.quarter(), Some(2));

        let ts3 = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 365);
        assert_eq!(ts3.dayofyear(), Some(1));
        assert_eq!(ts3.dayofweek(), Some(4));

        assert_eq!(Timestamp::nat().dayofweek(), None);
        assert_eq!(Timestamp::nat().dayofyear(), None);
        assert_eq!(Timestamp::nat().quarter(), None);
    }

    #[test]
    fn timestamp_is_boundary_methods() {
        let jan1 = Timestamp::from_nanos(0);
        assert_eq!(jan1.is_leap_year(), Some(false));
        assert_eq!(jan1.is_month_start(), Some(true));
        assert_eq!(jan1.is_month_end(), Some(false));
        assert_eq!(jan1.is_quarter_start(), Some(true));
        assert_eq!(jan1.is_quarter_end(), Some(false));
        assert_eq!(jan1.is_year_start(), Some(true));
        assert_eq!(jan1.is_year_end(), Some(false));

        let dec31 = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 364);
        assert_eq!(dec31.is_month_start(), Some(false));
        assert_eq!(dec31.is_month_end(), Some(true));
        assert_eq!(dec31.is_quarter_end(), Some(true));
        assert_eq!(dec31.is_year_end(), Some(true));

        assert_eq!(Timestamp::nat().is_leap_year(), None);
        assert_eq!(Timestamp::nat().is_month_start(), None);
    }

    #[test]
    fn timestamp_days_in_month() {
        let jan = Timestamp::from_nanos(0);
        assert_eq!(jan.days_in_month(), Some(31));
        assert_eq!(jan.daysinmonth(), Some(31));

        let feb_non_leap = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 31);
        assert_eq!(feb_non_leap.days_in_month(), Some(28));

        assert_eq!(Timestamp::nat().days_in_month(), None);
    }

    #[test]
    fn timestamp_weekofyear() {
        let jan1 = Timestamp::from_nanos(0);
        assert_eq!(jan1.weekofyear(), Some(1));
        assert_eq!(jan1.week(), Some(1));

        let jan8 = Timestamp::from_nanos(Timedelta::NANOS_PER_DAY * 7);
        assert_eq!(jan8.weekofyear(), Some(2));

        assert_eq!(Timestamp::nat().weekofyear(), None);
        assert_eq!(Timestamp::nat().week(), None);
    }

    #[test]
    fn timestamp_weekofyear_iso_53_week_boundaries() {
        // ISO-8601 53-week-year boundaries vs pandas 2.2.3 isocalendar().week.
        // (br-frankenpandas-xmfmd) Date -> nanos via days since 1970-01-01.
        fn week_of(date_days: i64) -> Option<i64> {
            Timestamp::from_nanos(date_days * Timedelta::NANOS_PER_DAY).weekofyear()
        }
        // Days from 1970-01-01 for each date (UTC, no tz).
        // 2021-01-01 -> week 53 (2020 is a 53-week year); FP used to give 52.
        assert_eq!(week_of(18_628), Some(53)); // 2021-01-01
        // 2016-01-01 -> week 53 (2015 is a 53-week year).
        assert_eq!(week_of(16_801), Some(53)); // 2016-01-01
        // 2026-12-31 -> week 53; FP used to give 1.
        assert_eq!(week_of(20_818), Some(53)); // 2026-12-31
        // 2020-12-31 -> week 53.
        assert_eq!(week_of(18_627), Some(53)); // 2020-12-31
        // Cases that must stay correct (non-53 boundaries):
        assert_eq!(week_of(19_358), Some(52)); // 2023-01-01 -> week 52
        assert_eq!(week_of(20_087), Some(1)); // 2024-12-30 -> week 1
        assert_eq!(week_of(18_260), Some(1)); // 2019-12-30 -> week 1
    }

    #[test]
    fn iso_weeks_in_year_53_week_years() {
        use super::iso_weeks_in_year;
        // Known 53-week years; everything else is 52.
        for y in [2004, 2009, 2015, 2020, 2026] {
            assert_eq!(iso_weeks_in_year(y), 53, "{y} should have 53 ISO weeks");
        }
        for y in [2018, 2019, 2021, 2022, 2023, 2024] {
            assert_eq!(iso_weeks_in_year(y), 52, "{y} should have 52 ISO weeks");
        }
    }

    #[test]
    fn timestamp_to_unit() {
        let ts = Timestamp::from_nanos(1_000_000_000);
        assert_eq!(ts.to_unit("ns"), Some(1_000_000_000));
        assert_eq!(ts.to_unit("us"), Some(1_000_000));
        assert_eq!(ts.to_unit("ms"), Some(1_000));
        assert_eq!(ts.to_unit("s"), Some(1));
        assert_eq!(ts.to_unit("invalid"), None);

        assert_eq!(Timestamp::nat().to_unit("ns"), None);
    }

    #[test]
    fn timestamp_toordinal() {
        // 2026-01-01 is ordinal 738886 (days since Jan 1, year 1)
        // Days from Unix epoch: 738886 - 719163 = 19723
        let nanos_2026_01_01 = 19723_i64 * 24 * 60 * 60 * 1_000_000_000;
        let ts = Timestamp::from_nanos(nanos_2026_01_01);
        assert_eq!(ts.toordinal(), Some(738886));

        // NaT returns None
        assert_eq!(Timestamp::nat().toordinal(), None);
    }

    #[test]
    fn timestamp_fromordinal() {
        // Round-trip test: create a timestamp from ordinal derived from toordinal
        // First create a known timestamp
        let nanos_2026_01_01 = 19723_i64 * 24 * 60 * 60 * 1_000_000_000;
        let ts_orig = Timestamp::from_nanos(nanos_2026_01_01);
        let ordinal = ts_orig.toordinal().unwrap();

        // Now convert back using fromordinal
        let ts = Timestamp::fromordinal(ordinal);
        assert_eq!(ts.year(), ts_orig.year());
        assert_eq!(ts.month(), ts_orig.month());
        assert_eq!(ts.day(), ts_orig.day());

        // Invalid ordinal returns NaT
        let nat = Timestamp::fromordinal(0);
        assert!(nat.is_nat());
    }

    #[test]
    fn timestamp_parse_iso8601_date_only() {
        let ts = Timestamp::parse("2024-01-15").unwrap();
        assert_eq!(ts.year(), Some(2024));
        assert_eq!(ts.month(), Some(1));
        assert_eq!(ts.day(), Some(15));
        assert_eq!(ts.hour(), Some(0));
        assert_eq!(ts.minute(), Some(0));
        assert_eq!(ts.second(), Some(0));
    }

    #[test]
    fn timestamp_parse_iso8601_datetime() {
        let ts = Timestamp::parse("2024-01-15T10:30:45").unwrap();
        assert_eq!(ts.year(), Some(2024));
        assert_eq!(ts.month(), Some(1));
        assert_eq!(ts.day(), Some(15));
        assert_eq!(ts.hour(), Some(10));
        assert_eq!(ts.minute(), Some(30));
        assert_eq!(ts.second(), Some(45));
    }

    #[test]
    fn timestamp_parse_space_separator() {
        let ts = Timestamp::parse("2024-01-15 10:30:45").unwrap();
        assert_eq!(ts.year(), Some(2024));
        assert_eq!(ts.hour(), Some(10));
    }

    #[test]
    fn timestamp_parse_with_fractional_seconds() {
        let ts = Timestamp::parse("2024-01-15T10:30:45.123456789").unwrap();
        assert_eq!(ts.second(), Some(45));
        assert_eq!(ts.microsecond(), Some(123456));
        assert_eq!(ts.nanosecond(), Some(789));
    }

    #[test]
    fn timestamp_parse_utc_timezone() {
        let ts = Timestamp::parse("2024-01-15T10:30:45Z").unwrap();
        assert_eq!(ts.tz, Some("UTC".to_string()));
    }

    #[test]
    fn timestamp_parse_offset_timezone() {
        let ts = Timestamp::parse("2024-01-15T10:30:45+05:30").unwrap();
        assert_eq!(ts.tz, Some("+05:30".to_string()));
    }

    #[test]
    fn timestamp_parse_nat() {
        let ts = Timestamp::parse("NaT").unwrap();
        assert!(ts.is_nat());
        let ts2 = Timestamp::parse("nat").unwrap();
        assert!(ts2.is_nat());
    }

    #[test]
    fn timestamp_parse_invalid() {
        assert!(Timestamp::parse("not a date").is_err());
        assert!(Timestamp::parse("2024-13-01").is_err()); // invalid month
        assert!(Timestamp::parse("2024-01-32").is_err()); // invalid day
    }

    #[test]
    fn period_parse_annual() {
        let p = Period::parse("2024").unwrap();
        assert_eq!(p.freq(), PeriodFreq::Annual);
        assert_eq!(p.ordinal(), 2024 - 1970);
    }

    #[test]
    fn period_parse_quarterly() {
        let p = Period::parse("2024Q1").unwrap();
        assert_eq!(p.freq(), PeriodFreq::Quarterly);
        assert_eq!(p.ordinal(), (2024 - 1970) * 4);

        let p2 = Period::parse("2024q3").unwrap();
        assert_eq!(p2.freq(), PeriodFreq::Quarterly);
        assert_eq!(p2.ordinal(), (2024 - 1970) * 4 + 2);
    }

    #[test]
    fn period_parse_monthly() {
        let p = Period::parse("2024-01").unwrap();
        assert_eq!(p.freq(), PeriodFreq::Monthly);
        assert_eq!(p.ordinal(), (2024 - 1970) * 12);

        let p2 = Period::parse("2024-12").unwrap();
        assert_eq!(p2.freq(), PeriodFreq::Monthly);
        assert_eq!(p2.ordinal(), (2024 - 1970) * 12 + 11);
    }

    #[test]
    fn period_parse_nat() {
        let p = Period::parse("NaT").unwrap();
        assert_eq!(p.ordinal(), i64::MIN);
    }

    #[test]
    fn period_parse_invalid() {
        assert!(Period::parse("not a period").is_err());
        assert!(Period::parse("2024Q5").is_err()); // invalid quarter
        assert!(Period::parse("2024-13").is_err()); // invalid month
    }

    #[test]
    fn interval_parse_basic() {
        let i = Interval::parse("[0, 1]").unwrap();
        assert_eq!(i.left, 0.0);
        assert_eq!(i.right, 1.0);
        assert_eq!(i.closed, IntervalClosed::Both);

        let i2 = Interval::parse("(0, 1)").unwrap();
        assert_eq!(i2.left, 0.0);
        assert_eq!(i2.right, 1.0);
        assert_eq!(i2.closed, IntervalClosed::Neither);

        let i3 = Interval::parse("[0, 1)").unwrap();
        assert_eq!(i3.closed, IntervalClosed::Left);

        let i4 = Interval::parse("(0, 1]").unwrap();
        assert_eq!(i4.closed, IntervalClosed::Right);
    }

    #[test]
    fn interval_parse_floats() {
        let i = Interval::parse("[-1.5, 2.5)").unwrap();
        assert_eq!(i.left, -1.5);
        assert_eq!(i.right, 2.5);
        assert_eq!(i.closed, IntervalClosed::Left);
    }

    #[test]
    fn interval_parse_invalid() {
        assert!(Interval::parse("invalid").is_err());
        assert!(Interval::parse("[0]").is_err());
        assert!(Interval::parse("0, 1").is_err()); // missing brackets
    }
}