nautilus_model/data/

bar.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Bar aggregate structures, data types, and functionality.

use std::{
    collections::HashMap,
    fmt::{Debug, Display},
    hash::Hash,
    num::{NonZero, NonZeroUsize},
    str::FromStr,
};

use derive_builder::Builder;
use indexmap::IndexMap;
use jiff::{
    SignedDuration, Span, Timestamp,
    civil::Date,
    tz::{Offset, TimeZone},
};
use nautilus_core::{
    DurationNanos, UnixNanos,
    correctness::{FAILED, check_predicate_true},
    serialization::Serializable,
};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use super::{ARROW_TIMESTAMP_NANOSECOND, HasTsInit};
use crate::{
    deserialization::StringInput,
    enums::{AggregationSource, BarAggregation, PriceType},
    identifiers::InstrumentId,
    types::{Price, Quantity, fixed::FIXED_DECIMAL},
};

pub const BAR_SPEC_1_SECOND_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Second,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_1_MINUTE_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_3_MINUTE_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(3).unwrap(),
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_5_MINUTE_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(5).unwrap(),
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_15_MINUTE_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(15).unwrap(),
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_30_MINUTE_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(30).unwrap(),
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_1_HOUR_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_2_HOUR_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(2).unwrap(),
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_4_HOUR_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(4).unwrap(),
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_6_HOUR_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(6).unwrap(),
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_12_HOUR_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(12).unwrap(),
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_1_DAY_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Day,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_2_DAY_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(2).unwrap(),
    aggregation: BarAggregation::Day,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_3_DAY_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(3).unwrap(),
    aggregation: BarAggregation::Day,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_5_DAY_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(5).unwrap(),
    aggregation: BarAggregation::Day,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_1_WEEK_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Week,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_1_MONTH_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(1).unwrap(),
    aggregation: BarAggregation::Month,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_3_MONTH_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(3).unwrap(),
    aggregation: BarAggregation::Month,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_6_MONTH_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(6).unwrap(),
    aggregation: BarAggregation::Month,
    price_type: PriceType::Last,
};

pub const BAR_SPEC_12_MONTH_LAST: BarSpecification = BarSpecification {
    step: NonZero::new(12).unwrap(),
    aggregation: BarAggregation::Month,
    price_type: PriceType::Last,
};

/// Returns the bar interval as a [`SignedDuration`].
///
/// # Panics
///
/// Panics if the aggregation method of the given `bar_type` is not time based,
/// or if `step` is too large for the interval arithmetic.
#[must_use]
pub fn get_bar_interval(bar_type: &BarType) -> SignedDuration {
    let spec = bar_type.spec();
    let step = step_to_i64(spec.step);

    match spec.aggregation {
        BarAggregation::Millisecond => SignedDuration::from_millis(step),
        BarAggregation::Second => SignedDuration::from_secs(step),
        BarAggregation::Minute => SignedDuration::from_mins(step),
        BarAggregation::Hour => SignedDuration::from_hours(step),
        BarAggregation::Day => duration_days(step),
        BarAggregation::Week => {
            duration_days(step.checked_mul(7).expect("`step` overflows i64 days"))
        }
        BarAggregation::Month => {
            // Proxy for comparing bar lengths
            duration_days(step.checked_mul(30).expect("`step` overflows i64 days"))
        }
        BarAggregation::Year => {
            // Proxy for comparing bar lengths
            duration_days(step.checked_mul(365).expect("`step` overflows i64 days"))
        }
        _ => panic!("Aggregation not time based"),
    }
}

/// Returns the bar interval as `UnixNanos`.
///
/// # Panics
///
/// Panics if the aggregation method of the given `bar_type` is not time based.
#[must_use]
pub fn get_bar_interval_ns(bar_type: &BarType) -> DurationNanos {
    DurationNanos::try_from(get_bar_interval(bar_type)).expect("Invalid bar interval")
}

/// Returns the time bar start as a timezone-aware `Timestamp`.
///
/// # Panics
///
/// Panics if computing the base civil date or datetime from `now` fails,
/// if `step` cannot be represented for the calendar arithmetic,
/// or if the aggregation type is unsupported.
#[must_use]
pub fn get_time_bar_start(
    now: Timestamp,
    bar_type: &BarType,
    time_bars_origin: Option<SignedDuration>,
) -> Timestamp {
    get_time_bar_bounds(now, bar_type, time_bars_origin, &TimeZone::UTC, None).1
}

/// Returns the previous, current, and next time bar boundaries in UTC.
///
/// Days and longer periods follow local calendar boundaries in `time_zone`. The origin offset
/// shifts the nominal civil boundary before resolving DST gaps and folds. Shorter periods retain
/// UTC alignment. Gaps shift forward; folds use the first occurrence.
/// `calendar_origin` keeps the recurrence grid anchored to a timer's starting local date;
/// `None` uses the current local date.
///
/// # Panics
///
/// Panics if the aggregation is unsupported or calendar arithmetic exceeds the supported range.
#[must_use]
pub fn get_time_bar_bounds(
    now: Timestamp,
    bar_type: &BarType,
    time_bars_origin: Option<SignedDuration>,
    time_zone: &TimeZone,
    calendar_origin: Option<Date>,
) -> (Timestamp, Timestamp, Timestamp) {
    let spec = bar_type.spec();
    let step = step_to_i64(spec.step);
    let origin = time_bars_origin.unwrap_or(SignedDuration::ZERO);
    let local_date = time_zone.to_datetime(now).date();
    let reference_date = calendar_origin.unwrap_or(local_date);

    let start = |date: Date| {
        let civil = date
            .at(0, 0, 0, 0)
            .checked_add(origin)
            .expect("valid bar origin");
        time_zone
            .to_timestamp(civil)
            .expect("valid time bar boundary")
    };

    match spec.aggregation {
        BarAggregation::Day | BarAggregation::Week => {
            let (span, days_per_period) = if spec.aggregation == BarAggregation::Day {
                (Span::new().try_days(step).expect("valid day step"), step)
            } else {
                (
                    Span::new().try_weeks(step).expect("valid week step"),
                    step * 7,
                )
            };

            let date = if spec.aggregation == BarAggregation::Week {
                reference_date
                    .checked_sub(
                        Span::new()
                            .days(i64::from(reference_date.weekday().to_monday_zero_offset())),
                    )
                    .expect("valid week start")
            } else {
                reference_date
            };

            let days = date
                .until(local_date)
                .expect("valid date difference")
                .get_days();
            let index = i64::from(days).div_euclid(days_per_period);
            find_calendar_bounds(now, index, 1, |index| {
                start(
                    date.checked_add(span.checked_mul(index).expect("valid day offset"))
                        .expect("valid bar date"),
                )
            })
        }
        BarAggregation::Month => {
            let step = i64::from(
                u32::try_from(step).expect("`step` exceeds u32 range for month arithmetic"),
            );
            let origin_year = reference_date.year();
            let base = Date::new(origin_year, 1, 1)
                .expect("valid year start")
                .at(0, 0, 0, 0)
                .checked_add(origin)
                .expect("valid month origin");

            let boundary = |months: i64| {
                let civil = base
                    .checked_add(Span::new().try_months(months).expect("valid month step"))
                    .expect("valid month boundary");
                time_zone.to_timestamp(civil).expect("valid month boundary")
            };

            let index = (i64::from(local_date.year()) - i64::from(origin_year)) * 12;
            find_calendar_bounds(now, index, step, boundary)
        }
        BarAggregation::Year => {
            let step = i32::try_from(step).expect("`step` exceeds i32 range for year arithmetic");

            let boundary = |year: i64| {
                start(
                    Date::new(
                        i16::try_from(year).expect("year exceeds Jiff supported range"),
                        1,
                        1,
                    )
                    .expect("valid year start date"),
                )
            };

            let step = i64::from(step);
            let origin_year = i64::from(reference_date.year());
            let years = i64::from(local_date.year()) - origin_year;
            let index = origin_year + years.div_euclid(step) * step;
            find_calendar_bounds(now, index, step, boundary)
        }
        _ => {
            let origin_offset = origin;

            let start = match spec.aggregation {
                BarAggregation::Millisecond => {
                    find_closest_smaller_time(now, origin_offset, SignedDuration::from_millis(step))
                }
                BarAggregation::Second => {
                    find_closest_smaller_time(now, origin_offset, SignedDuration::from_secs(step))
                }
                BarAggregation::Minute => {
                    find_closest_smaller_time(now, origin_offset, SignedDuration::from_mins(step))
                }
                BarAggregation::Hour => {
                    find_closest_smaller_time(now, origin_offset, SignedDuration::from_hours(step))
                }
                _ => panic!(
                    "Aggregation type {} not supported for time bars",
                    spec.aggregation
                ),
            };

            let interval = get_bar_interval(bar_type);
            (start - interval, start, start + interval)
        }
    }
}

fn find_calendar_bounds(
    now: Timestamp,
    mut index: i64,
    step: i64,
    boundary: impl Fn(i64) -> Timestamp,
) -> (Timestamp, Timestamp, Timestamp) {
    while boundary(index) > now {
        index = index.checked_sub(step).expect("calendar index underflow");
    }

    loop {
        let next = index.checked_add(step).expect("calendar index overflow");
        let close = boundary(next);
        if close > now {
            let previous = index.checked_sub(step).expect("calendar index underflow");
            return (boundary(previous), boundary(index), close);
        }

        index = next;
    }
}

/// Finds the closest smaller time based on a daily time origin and period.
///
/// This function calculates the most recent time that is aligned with the given period
/// and is less than or equal to the current time.
fn find_closest_smaller_time(
    now: Timestamp,
    daily_time_origin: SignedDuration,
    period: SignedDuration,
) -> Timestamp {
    // Floor to start of day
    let day_start = Offset::UTC
        .to_timestamp(Offset::UTC.to_datetime(now).date().at(0, 0, 0, 0))
        .expect("valid UTC day start");
    let base_time = day_start + daily_time_origin;

    let time_difference = base_time.duration_until(now);
    let period_ns = period.as_nanos();
    debug_assert_ne!(period_ns, 0, "bar period must be non-zero");

    // Use div_euclid for floor division (rounds toward -inf, not zero)
    // so negative deltas (now before origin) yield the previous period boundary
    let num_periods = time_difference.as_nanos().div_euclid(period_ns);

    base_time + SignedDuration::from_nanos_i128(num_periods * period_ns)
}

fn duration_days(days: i64) -> SignedDuration {
    try_duration_days(days).unwrap_or_else(|e| panic!("{e}"))
}

fn try_duration_days(days: i64) -> anyhow::Result<SignedDuration> {
    let hours = days
        .checked_mul(24)
        .ok_or_else(|| anyhow::anyhow!("days overflow i64 hours"))?;
    SignedDuration::try_from_hours(hours)
        .ok_or_else(|| anyhow::anyhow!("days exceed signed duration range"))
}

fn try_time_interval(step: usize, aggregation: BarAggregation) -> anyhow::Result<SignedDuration> {
    let step_i64 = i64::try_from(step)
        .map_err(|_| invalid_interval_step(step, aggregation, "step exceeds i64 range"))?;

    let duration = match aggregation {
        BarAggregation::Millisecond => SignedDuration::from_millis(step_i64),
        BarAggregation::Second => SignedDuration::from_secs(step_i64),
        BarAggregation::Minute => SignedDuration::try_from_mins(step_i64).ok_or_else(|| {
            invalid_interval_step(step, aggregation, "step exceeds signed duration range")
        })?,
        BarAggregation::Hour => SignedDuration::try_from_hours(step_i64).ok_or_else(|| {
            invalid_interval_step(step, aggregation, "step exceeds signed duration range")
        })?,
        BarAggregation::Day => try_scaled_days(step, aggregation, step_i64, 1)?,
        BarAggregation::Week => try_scaled_days(step, aggregation, step_i64, 7)?,
        BarAggregation::Month => try_scaled_days(step, aggregation, step_i64, 30)?,
        BarAggregation::Year => try_scaled_days(step, aggregation, step_i64, 365)?,
        _ => anyhow::bail!("Timedelta not supported for aggregation type: {aggregation:?}"),
    };

    u64::try_from(duration.as_nanos())
        .map_err(|_| invalid_interval_step(step, aggregation, "interval overflows nanoseconds"))?;

    Ok(duration)
}

fn try_scaled_days(
    step: usize,
    aggregation: BarAggregation,
    step_i64: i64,
    multiplier: i64,
) -> anyhow::Result<SignedDuration> {
    let days = step_i64
        .checked_mul(multiplier)
        .ok_or_else(|| invalid_interval_step(step, aggregation, "step overflows i64 days"))?;
    try_duration_days(days).map_err(|e| invalid_interval_step(step, aggregation, &e.to_string()))
}

fn invalid_interval_step(step: usize, aggregation: BarAggregation, reason: &str) -> anyhow::Error {
    anyhow::anyhow!(
        "Invalid step in bar_type.spec.step: {step} for aggregation={aggregation}. {reason}"
    )
}

/// Converts a bar specification step to `i64` for time arithmetic.
///
/// # Panics
///
/// Panics if `step` exceeds the `i64` range.
fn step_to_i64(step: NonZeroUsize) -> i64 {
    i64::try_from(step.get()).expect("`step` exceeds i64 range")
}

/// Represents a bar aggregation specification including a step, aggregation
/// method/rule and price type.
#[repr(C)]
#[derive(
    Clone, Copy, Hash, PartialEq, Eq, PartialOrd, Ord, Debug, Serialize, Deserialize, Builder,
)]
#[builder(build_fn(validate = "Self::validate"))]
#[serde(try_from = "BarSpecificationFields")]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
)]
pub struct BarSpecification {
    /// The step for binning samples for bar aggregation.
    pub step: NonZeroUsize,
    /// The type of bar aggregation.
    pub aggregation: BarAggregation,
    /// The price type to use for aggregation.
    pub price_type: PriceType,
}

impl BarSpecificationBuilder {
    fn validate(&self) -> Result<(), String> {
        if let (Some(step), Some(aggregation)) = (self.step, self.aggregation) {
            BarSpecification::validate_step(step.get(), aggregation).map_err(|e| e.to_string())?;
        }

        Ok(())
    }
}

// Deserialization mirror routing through `new_checked` so serde inputs
// cannot bypass step validation
#[derive(Deserialize)]
struct BarSpecificationFields {
    step: NonZeroUsize,
    aggregation: BarAggregation,
    price_type: PriceType,
}

impl TryFrom<BarSpecificationFields> for BarSpecification {
    type Error = anyhow::Error;

    fn try_from(fields: BarSpecificationFields) -> Result<Self, Self::Error> {
        Self::new_checked(fields.step.get(), fields.aggregation, fields.price_type)
    }
}

impl BarSpecification {
    /// Creates a new [`BarSpecification`] instance with correctness checking.
    ///
    /// # Errors
    ///
    /// Returns an error if `step` is not positive (> 0), if `step` is not
    /// valid for a fixed-subunit time aggregation, or if a time-aggregated
    /// `step` overflows the representable duration or nanosecond interval.
    ///
    /// # Notes
    ///
    /// PyO3 requires a `Result` type for proper error handling and stacktrace printing in Python.
    pub fn new_checked(
        step: usize,
        aggregation: BarAggregation,
        price_type: PriceType,
    ) -> anyhow::Result<Self> {
        let step = NonZeroUsize::new(step)
            .ok_or_else(|| anyhow::anyhow!("Invalid step: {step} (must be non-zero)"))?;
        Self::validate_step(step.get(), aggregation)?;

        Ok(Self {
            step,
            aggregation,
            price_type,
        })
    }

    fn validate_step(step: usize, aggregation: BarAggregation) -> anyhow::Result<()> {
        match aggregation {
            BarAggregation::Millisecond => {
                Self::validate_periodic_step(step, aggregation, 1000, false)?;
            }
            BarAggregation::Second | BarAggregation::Minute => {
                Self::validate_periodic_step(step, aggregation, 60, false)?;
            }
            BarAggregation::Hour => Self::validate_periodic_step(step, aggregation, 24, false)?,
            // 12-MONTH is allowed (unlike other full-subunit steps) because the shipped
            // BAR_SPEC_12_MONTH_LAST constant and OKX yearly candles depend on it
            BarAggregation::Month => Self::validate_periodic_step(step, aggregation, 12, true)?,
            BarAggregation::Day | BarAggregation::Week | BarAggregation::Year => {}
            _ => return Ok(()),
        }

        try_time_interval(step, aggregation).map(|_| ())
    }

    fn validate_periodic_step(
        step: usize,
        aggregation: BarAggregation,
        subunits: usize,
        allow_equal: bool,
    ) -> anyhow::Result<()> {
        if !subunits.is_multiple_of(step) {
            anyhow::bail!(
                "Invalid step in bar_type.spec.step: {step} for aggregation={aggregation}. \
                 step must evenly divide {subunits} (so it is periodic).",
            );
        }

        if !allow_equal && subunits == step {
            anyhow::bail!(
                "Invalid step in bar_type.spec.step: {step} for aggregation={aggregation}. \
                 step must not be {subunits}. Use higher aggregation unit instead.",
            );
        }

        Ok(())
    }

    /// Creates a new [`BarSpecification`] instance.
    ///
    /// # Panics
    ///
    /// Panics if `step` is not positive (> 0), if `step` is not valid for
    /// a fixed-subunit time aggregation, or if a time-aggregated `step`
    /// overflows the representable duration or nanosecond interval.
    #[must_use]
    pub fn new(step: usize, aggregation: BarAggregation, price_type: PriceType) -> Self {
        Self::new_checked(step, aggregation, price_type).expect(FAILED)
    }

    /// Returns the [`SignedDuration`] interval for this bar specification.
    ///
    /// # Notes
    ///
    /// For [`BarAggregation::Month`] and [`BarAggregation::Year`], proxy values are used
    /// (30 days for months, 365 days for years) to estimate their respective durations,
    /// since months and years have variable lengths.
    ///
    /// # Panics
    ///
    /// Panics if the aggregation method is not time-based, or if `step` is too
    /// large for the interval arithmetic.
    #[must_use]
    pub fn timedelta(&self) -> SignedDuration {
        let step = step_to_i64(self.step);

        match self.aggregation {
            BarAggregation::Millisecond => SignedDuration::from_millis(step),
            BarAggregation::Second => SignedDuration::from_secs(step),
            BarAggregation::Minute => SignedDuration::from_mins(step),
            BarAggregation::Hour => SignedDuration::from_hours(step),
            BarAggregation::Day => duration_days(step),
            BarAggregation::Week => {
                duration_days(step.checked_mul(7).expect("`step` overflows i64 days"))
            }
            BarAggregation::Month => {
                // Proxy for comparing bar lengths
                duration_days(step.checked_mul(30).expect("`step` overflows i64 days"))
            }
            BarAggregation::Year => {
                // Proxy for comparing bar lengths
                duration_days(step.checked_mul(365).expect("`step` overflows i64 days"))
            }
            _ => panic!(
                "Timedelta not supported for aggregation type: {:?}",
                self.aggregation
            ),
        }
    }

    /// Return a value indicating whether the aggregation method is time-driven:
    ///  - [`BarAggregation::Millisecond`]
    ///  - [`BarAggregation::Second`]
    ///  - [`BarAggregation::Minute`]
    ///  - [`BarAggregation::Hour`]
    ///  - [`BarAggregation::Day`]
    ///  - [`BarAggregation::Week`]
    ///  - [`BarAggregation::Month`]
    ///  - [`BarAggregation::Year`]
    #[must_use]
    pub fn is_time_aggregated(&self) -> bool {
        matches!(
            self.aggregation,
            BarAggregation::Millisecond
                | BarAggregation::Second
                | BarAggregation::Minute
                | BarAggregation::Hour
                | BarAggregation::Day
                | BarAggregation::Week
                | BarAggregation::Month
                | BarAggregation::Year
        )
    }

    /// Return a value indicating whether the aggregation method is threshold-driven:
    ///  - [`BarAggregation::Tick`]
    ///  - [`BarAggregation::TickImbalance`]
    ///  - [`BarAggregation::Volume`]
    ///  - [`BarAggregation::VolumeImbalance`]
    ///  - [`BarAggregation::Value`]
    ///  - [`BarAggregation::ValueImbalance`]
    #[must_use]
    pub fn is_threshold_aggregated(&self) -> bool {
        matches!(
            self.aggregation,
            BarAggregation::Tick
                | BarAggregation::TickImbalance
                | BarAggregation::Volume
                | BarAggregation::VolumeImbalance
                | BarAggregation::Value
                | BarAggregation::ValueImbalance
        )
    }

    /// Return a value indicating whether the aggregation method is information-driven:
    ///  - [`BarAggregation::TickRuns`]
    ///  - [`BarAggregation::VolumeRuns`]
    ///  - [`BarAggregation::ValueRuns`]
    #[must_use]
    pub fn is_information_aggregated(&self) -> bool {
        matches!(
            self.aggregation,
            BarAggregation::TickRuns | BarAggregation::VolumeRuns | BarAggregation::ValueRuns
        )
    }
}

impl Display for BarSpecification {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}-{}-{}", self.step, self.aggregation, self.price_type)
    }
}

/// Represents a bar type including the instrument ID, bar specification and
/// aggregation source.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass_enum(module = "nautilus_trader.model")
)]
pub enum BarType {
    Standard {
        /// The bar type's instrument ID.
        instrument_id: InstrumentId,
        /// The bar type's specification.
        spec: BarSpecification,
        /// The bar type's aggregation source.
        aggregation_source: AggregationSource,
    },
    Composite {
        /// The bar type's instrument ID.
        instrument_id: InstrumentId,
        /// The bar type's specification.
        spec: BarSpecification,
        /// The bar type's aggregation source.
        aggregation_source: AggregationSource,

        /// The composite step for binning samples for bar aggregation.
        composite_step: usize,
        /// The composite type of bar aggregation.
        composite_aggregation: BarAggregation,
        /// The composite bar type's aggregation source.
        composite_aggregation_source: AggregationSource,
    },
}

impl BarType {
    /// Creates a new [`BarType`] instance.
    #[must_use]
    pub fn new(
        instrument_id: InstrumentId,
        spec: BarSpecification,
        aggregation_source: AggregationSource,
    ) -> Self {
        Self::Standard {
            instrument_id,
            spec,
            aggregation_source,
        }
    }

    /// Creates a new composite [`BarType`] instance with correctness checking.
    ///
    /// # Errors
    ///
    /// Returns an error if the composite specification is invalid, i.e. `composite_step`
    /// is not positive (> 0) or is not valid for a fixed-subunit time aggregation.
    pub fn new_composite_checked(
        instrument_id: InstrumentId,
        spec: BarSpecification,
        aggregation_source: AggregationSource,

        composite_step: usize,
        composite_aggregation: BarAggregation,
        composite_aggregation_source: AggregationSource,
    ) -> anyhow::Result<Self> {
        // Validate eagerly so `composite()` cannot panic later
        BarSpecification::new_checked(composite_step, composite_aggregation, spec.price_type)?;

        Ok(Self::Composite {
            instrument_id,
            spec,
            aggregation_source,

            composite_step,
            composite_aggregation,
            composite_aggregation_source,
        })
    }

    /// Creates a new composite [`BarType`] instance.
    ///
    /// # Panics
    ///
    /// Panics if the composite specification is invalid, i.e. `composite_step` is not
    /// positive (> 0) or is not valid for a fixed-subunit time aggregation.
    #[must_use]
    pub fn new_composite(
        instrument_id: InstrumentId,
        spec: BarSpecification,
        aggregation_source: AggregationSource,

        composite_step: usize,
        composite_aggregation: BarAggregation,
        composite_aggregation_source: AggregationSource,
    ) -> Self {
        Self::new_composite_checked(
            instrument_id,
            spec,
            aggregation_source,
            composite_step,
            composite_aggregation,
            composite_aggregation_source,
        )
        .expect(FAILED)
    }

    /// Returns whether this instance is a standard bar type.
    #[must_use]
    pub fn is_standard(&self) -> bool {
        matches!(self, Self::Standard { .. })
    }

    /// Returns whether this instance is a composite bar type.
    #[must_use]
    pub fn is_composite(&self) -> bool {
        matches!(self, Self::Composite { .. })
    }

    /// Returns whether the bar aggregation source is `EXTERNAL`.
    #[must_use]
    pub fn is_externally_aggregated(&self) -> bool {
        self.aggregation_source() == AggregationSource::External
    }

    /// Returns whether the bar aggregation source is `INTERNAL`.
    #[must_use]
    pub fn is_internally_aggregated(&self) -> bool {
        self.aggregation_source() == AggregationSource::Internal
    }

    /// Returns the standard bar type component.
    #[must_use]
    pub fn standard(&self) -> Self {
        match self {
            &b @ Self::Standard { .. } => b,
            Self::Composite {
                instrument_id,
                spec,
                aggregation_source,
                ..
            } => Self::new(*instrument_id, *spec, *aggregation_source),
        }
    }

    /// Returns any composite bar type component.
    #[must_use]
    pub fn composite(&self) -> Self {
        match self {
            &b @ Self::Standard { .. } => b, // case shouldn't be used if is_composite is called before
            Self::Composite {
                instrument_id,
                spec,
                aggregation_source: _,

                composite_step,
                composite_aggregation,
                composite_aggregation_source,
            } => Self::new(
                *instrument_id,
                BarSpecification::new(*composite_step, *composite_aggregation, spec.price_type),
                *composite_aggregation_source,
            ),
        }
    }

    /// Returns the [`InstrumentId`] for this bar type.
    #[must_use]
    pub fn instrument_id(&self) -> InstrumentId {
        match &self {
            Self::Standard { instrument_id, .. } | Self::Composite { instrument_id, .. } => {
                *instrument_id
            }
        }
    }

    /// Returns the [`BarSpecification`] for this bar type.
    #[must_use]
    pub fn spec(&self) -> BarSpecification {
        match &self {
            Self::Standard { spec, .. } | Self::Composite { spec, .. } => *spec,
        }
    }

    /// Returns the [`AggregationSource`] for this bar type.
    #[must_use]
    pub fn aggregation_source(&self) -> AggregationSource {
        match &self {
            Self::Standard {
                aggregation_source, ..
            }
            | Self::Composite {
                aggregation_source, ..
            } => *aggregation_source,
        }
    }

    /// Returns the instrument ID and bar specification as a tuple key.
    ///
    /// Useful as a hashmap key when aggregation source should be ignored,
    /// such as for indicator registration where INTERNAL and EXTERNAL bars
    /// should trigger the same indicators.
    #[must_use]
    pub fn id_spec_key(&self) -> (InstrumentId, BarSpecification) {
        (self.instrument_id(), self.spec())
    }
}

#[derive(thiserror::Error, Debug)]
#[error("Error parsing `BarType` from '{input}', invalid token: '{token}' at position {position}")]
pub struct BarTypeParseError {
    input: String,
    token: String,
    position: usize,
}

impl FromStr for BarType {
    type Err = BarTypeParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let mut parts = s.split('@');
        let standard = parts.next().expect("split always yields a first field");
        let composite_str = parts.next();
        if let Some(extra) = parts.next() {
            return Err(BarTypeParseError {
                input: s.to_string(),
                token: extra.to_string(),
                position: 5,
            });
        }

        let mut pieces = standard.rsplitn(5, '-');
        let mut rev_pieces = [""; 5];
        for piece in rev_pieces.iter_mut().rev() {
            *piece = pieces.next().ok_or_else(|| BarTypeParseError {
                input: s.to_string(),
                token: String::new(),
                position: 0,
            })?;
        }

        let instrument_id =
            InstrumentId::from_str(rev_pieces[0]).map_err(|_| BarTypeParseError {
                input: s.to_string(),
                token: rev_pieces[0].to_string(),
                position: 0,
            })?;

        let step = rev_pieces[1].parse().map_err(|_| BarTypeParseError {
            input: s.to_string(),
            token: rev_pieces[1].to_string(),
            position: 1,
        })?;
        let aggregation =
            BarAggregation::from_str(rev_pieces[2]).map_err(|_| BarTypeParseError {
                input: s.to_string(),
                token: rev_pieces[2].to_string(),
                position: 2,
            })?;
        let price_type = PriceType::from_str(rev_pieces[3]).map_err(|_| BarTypeParseError {
            input: s.to_string(),
            token: rev_pieces[3].to_string(),
            position: 3,
        })?;
        let aggregation_source =
            AggregationSource::from_str(rev_pieces[4]).map_err(|_| BarTypeParseError {
                input: s.to_string(),
                token: rev_pieces[4].to_string(),
                position: 4,
            })?;
        let spec = BarSpecification::new_checked(step, aggregation, price_type).map_err(|_| {
            BarTypeParseError {
                input: s.to_string(),
                token: rev_pieces[1].to_string(),
                position: 1,
            }
        })?;

        if let Some(composite_str) = composite_str {
            let mut composite_pieces = composite_str.rsplitn(3, '-');
            let mut rev_composite_pieces = [""; 3];
            for piece in rev_composite_pieces.iter_mut().rev() {
                *piece = composite_pieces.next().ok_or_else(|| BarTypeParseError {
                    input: s.to_string(),
                    token: String::new(),
                    position: 5,
                })?;
            }

            let composite_step =
                rev_composite_pieces[0]
                    .parse()
                    .map_err(|_| BarTypeParseError {
                        input: s.to_string(),
                        token: rev_composite_pieces[0].to_string(),
                        position: 5,
                    })?;
            let composite_aggregation =
                BarAggregation::from_str(rev_composite_pieces[1]).map_err(|_| {
                    BarTypeParseError {
                        input: s.to_string(),
                        token: rev_composite_pieces[1].to_string(),
                        position: 6,
                    }
                })?;
            let composite_aggregation_source = AggregationSource::from_str(rev_composite_pieces[2])
                .map_err(|_| BarTypeParseError {
                    input: s.to_string(),
                    token: rev_composite_pieces[2].to_string(),
                    position: 7,
                })?;
            BarSpecification::new_checked(composite_step, composite_aggregation, price_type)
                .map_err(|_| BarTypeParseError {
                    input: s.to_string(),
                    token: rev_composite_pieces[0].to_string(),
                    position: 5,
                })?;

            Ok(Self::new_composite(
                instrument_id,
                spec,
                aggregation_source,
                composite_step,
                composite_aggregation,
                composite_aggregation_source,
            ))
        } else {
            Ok(Self::Standard {
                instrument_id,
                spec,
                aggregation_source,
            })
        }
    }
}

impl<T: AsRef<str>> From<T> for BarType {
    fn from(value: T) -> Self {
        Self::from_str(value.as_ref()).expect(FAILED)
    }
}

impl Display for BarType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self {
            Self::Standard {
                instrument_id,
                spec,
                aggregation_source,
            } => {
                write!(f, "{instrument_id}-{spec}-{aggregation_source}")
            }
            Self::Composite {
                instrument_id,
                spec,
                aggregation_source,

                composite_step,
                composite_aggregation,
                composite_aggregation_source,
            } => {
                write!(
                    f,
                    "{}-{}-{}@{}-{}-{}",
                    instrument_id,
                    spec,
                    aggregation_source,
                    *composite_step,
                    *composite_aggregation,
                    *composite_aggregation_source
                )
            }
        }
    }
}

impl Serialize for BarType {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_string())
    }
}

impl<'de> Deserialize<'de> for BarType {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let StringInput(s) = Deserialize::deserialize(deserializer)?;
        Self::from_str(s.as_ref()).map_err(serde::de::Error::custom)
    }
}

/// Represents an aggregated bar.
#[repr(C)]
#[derive(Clone, Copy, Hash, PartialEq, Eq, Debug, Serialize, Deserialize)]
#[serde(tag = "type", try_from = "BarFields")]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
)]
pub struct Bar {
    /// The bar type for this bar.
    pub bar_type: BarType,
    /// The bars open price.
    pub open: Price,
    /// The bars high price.
    pub high: Price,
    /// The bars low price.
    pub low: Price,
    /// The bars close price.
    pub close: Price,
    /// The bars volume.
    pub volume: Quantity,
    /// UNIX timestamp (nanoseconds) when the data event occurred.
    pub ts_event: UnixNanos,
    /// UNIX timestamp (nanoseconds) when the instance was created.
    pub ts_init: UnixNanos,
}

// Deserialization mirror routing through `new_checked` so serde inputs
// cannot bypass OHLC validation
#[derive(Deserialize)]
struct BarFields {
    bar_type: BarType,
    open: Price,
    high: Price,
    low: Price,
    close: Price,
    volume: Quantity,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
}

impl TryFrom<BarFields> for Bar {
    type Error = anyhow::Error;

    fn try_from(fields: BarFields) -> Result<Self, Self::Error> {
        Self::new_checked(
            fields.bar_type,
            fields.open,
            fields.high,
            fields.low,
            fields.close,
            fields.volume,
            fields.ts_event,
            fields.ts_init,
        )
    }
}

impl Bar {
    /// Creates a new [`Bar`] instance with correctness checking.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - `high` is not >= `open`.
    /// - `high` is not >= `low`.
    /// - `high` is not >= `close`.
    /// - `low` is not <= `open`.
    /// - `low` is not <= `close`.
    ///
    /// # Notes
    ///
    /// PyO3 requires a `Result` type for proper error handling and stacktrace printing in Python.
    #[expect(clippy::too_many_arguments)]
    pub fn new_checked(
        bar_type: BarType,
        open: Price,
        high: Price,
        low: Price,
        close: Price,
        volume: Quantity,
        ts_event: UnixNanos,
        ts_init: UnixNanos,
    ) -> anyhow::Result<Self> {
        check_predicate_true(high >= open, "high >= open")?;
        check_predicate_true(high >= low, "high >= low")?;
        check_predicate_true(high >= close, "high >= close")?;
        check_predicate_true(low <= close, "low <= close")?;
        check_predicate_true(low <= open, "low <= open")?;

        debug_assert!(
            open.precision == high.precision
                && open.precision == low.precision
                && open.precision == close.precision,
            "Bar prices must share a uniform precision (Arrow encoding assumes it)"
        );

        Ok(Self {
            bar_type,
            open,
            high,
            low,
            close,
            volume,
            ts_event,
            ts_init,
        })
    }

    /// Creates a new [`Bar`] instance.
    ///
    /// # Panics
    ///
    /// Panics if:
    /// - `high` is not >= `open`.
    /// - `high` is not >= `low`.
    /// - `high` is not >= `close`.
    /// - `low` is not <= `open`.
    /// - `low` is not <= `close`.
    #[expect(clippy::too_many_arguments)]
    #[must_use]
    pub fn new(
        bar_type: BarType,
        open: Price,
        high: Price,
        low: Price,
        close: Price,
        volume: Quantity,
        ts_event: UnixNanos,
        ts_init: UnixNanos,
    ) -> Self {
        Self::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
            .expect(FAILED)
    }

    #[must_use]
    pub fn instrument_id(&self) -> InstrumentId {
        self.bar_type.instrument_id()
    }

    /// Returns the metadata for the type, for use with serialization formats.
    #[must_use]
    pub fn get_metadata(
        bar_type: &BarType,
        price_precision: u8,
        size_precision: u8,
    ) -> HashMap<String, String> {
        let mut metadata = HashMap::new();
        let instrument_id = bar_type.instrument_id();
        metadata.insert("bar_type".to_string(), bar_type.to_string());
        metadata.insert("instrument_id".to_string(), instrument_id.to_string());
        metadata.insert("price_precision".to_string(), price_precision.to_string());
        metadata.insert("size_precision".to_string(), size_precision.to_string());
        metadata
    }

    /// Returns the field map for the type, for use with Arrow schemas.
    #[must_use]
    pub fn get_fields() -> IndexMap<String, String> {
        let mut metadata = IndexMap::new();
        metadata.insert("open".to_string(), FIXED_DECIMAL.to_string());
        metadata.insert("high".to_string(), FIXED_DECIMAL.to_string());
        metadata.insert("low".to_string(), FIXED_DECIMAL.to_string());
        metadata.insert("close".to_string(), FIXED_DECIMAL.to_string());
        metadata.insert("volume".to_string(), FIXED_DECIMAL.to_string());
        metadata.insert(
            "ts_event".to_string(),
            ARROW_TIMESTAMP_NANOSECOND.to_string(),
        );
        metadata.insert(
            "ts_init".to_string(),
            ARROW_TIMESTAMP_NANOSECOND.to_string(),
        );
        metadata
    }
}

impl Display for Bar {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{},{},{},{},{},{},{}",
            self.bar_type, self.open, self.high, self.low, self.close, self.volume, self.ts_event
        )
    }
}

impl Serializable for Bar {}

impl HasTsInit for Bar {
    fn ts_init(&self) -> UnixNanos {
        self.ts_init
    }
}

#[cfg(test)]
mod tests {
    use std::str::FromStr;

    use nautilus_core::serialization::msgpack::{FromMsgPack, ToMsgPack};
    use rstest::rstest;

    use super::*;
    use crate::identifiers::{Symbol, Venue};

    fn timestamp(value: &str) -> Timestamp {
        value.parse().unwrap()
    }

    #[rstest]
    #[case::spring_gap(
        BarAggregation::Day,
        "America/New_York",
        "2026-03-08T12:00:00Z",
        9_000,
        "2026-03-07T07:30:00Z",
        "2026-03-08T07:30:00Z",
        "2026-03-09T06:30:00Z"
    )]
    #[case::fall_fold(
        BarAggregation::Day,
        "America/New_York",
        "2026-11-01T12:00:00Z",
        5_400,
        "2026-10-31T05:30:00Z",
        "2026-11-01T05:30:00Z",
        "2026-11-02T06:30:00Z"
    )]
    #[case::clamped_month(
        BarAggregation::Month,
        "America/New_York",
        "2026-03-01T12:00:00Z",
        2_592_000,
        "2026-01-31T05:00:00Z",
        "2026-02-28T05:00:00Z",
        "2026-03-31T04:00:00Z"
    )]
    #[case::leap_origin(
        BarAggregation::Year,
        "America/New_York",
        "2024-07-01T12:00:00Z",
        5_097_600,
        "2023-03-01T05:00:00Z",
        "2024-02-29T05:00:00Z",
        "2025-03-01T05:00:00Z"
    )]
    #[case::week(
        BarAggregation::Week,
        "America/New_York",
        "2026-11-01T12:00:00Z",
        0,
        "2026-10-19T04:00:00Z",
        "2026-10-26T04:00:00Z",
        "2026-11-02T05:00:00Z"
    )]
    #[case::month(
        BarAggregation::Month,
        "America/New_York",
        "2026-11-15T12:00:00Z",
        0,
        "2026-10-01T04:00:00Z",
        "2026-11-01T04:00:00Z",
        "2026-12-01T05:00:00Z"
    )]
    #[case::year(
        BarAggregation::Year,
        "Australia/Sydney",
        "2026-06-01T12:00:00Z",
        0,
        "2024-12-31T13:00:00Z",
        "2025-12-31T13:00:00Z",
        "2026-12-31T13:00:00Z"
    )]
    fn test_time_bar_bounds_preserve_local_calendar_origin(
        #[case] aggregation: BarAggregation,
        #[case] zone: &str,
        #[case] now: &str,
        #[case] origin_secs: i64,
        #[case] previous: &str,
        #[case] open: &str,
        #[case] close: &str,
    ) {
        let bar_type = BarType::from(&format!("AAPL.XNAS-1-{aggregation}-LAST-INTERNAL"));
        let zone = nautilus_core::datetime::get_timezone(zone).unwrap();
        let bounds = get_time_bar_bounds(
            now.parse().unwrap(),
            &bar_type,
            Some(SignedDuration::from_secs(origin_secs)),
            &zone,
            None,
        );
        assert_eq!(
            bounds,
            (
                previous.parse::<Timestamp>().unwrap(),
                open.parse::<Timestamp>().unwrap(),
                close.parse::<Timestamp>().unwrap()
            )
        );
    }

    #[rstest]
    fn test_bar_specification_new_invalid() {
        let result = BarSpecification::new_checked(0, BarAggregation::Tick, PriceType::Last);
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Invalid step: 0 (must be non-zero)")
        );
    }

    #[rstest]
    #[should_panic(expected = "Invalid step: 0 (must be non-zero)")]
    fn test_bar_specification_new_checked_with_invalid_step_panics() {
        let aggregation = BarAggregation::Tick;
        let price_type = PriceType::Last;

        let _ = BarSpecification::new(0, aggregation, price_type);
    }

    #[rstest]
    #[should_panic(expected = "Invalid step in bar_type.spec.step: 7")]
    fn test_bar_specification_new_with_invalid_periodic_step_panics() {
        let _ = BarSpecification::new(7, BarAggregation::Minute, PriceType::Last);
    }

    #[rstest]
    #[case(
        BarAggregation::Millisecond,
        12,
        "Invalid step in bar_type.spec.step: 12 for aggregation=MILLISECOND. step must evenly divide 1000"
    )]
    #[case(
        BarAggregation::Millisecond,
        1000,
        "Invalid step in bar_type.spec.step: 1000 for aggregation=MILLISECOND. step must not be 1000"
    )]
    #[case(
        BarAggregation::Second,
        50,
        "Invalid step in bar_type.spec.step: 50 for aggregation=SECOND. step must evenly divide 60"
    )]
    #[case(
        BarAggregation::Second,
        60,
        "Invalid step in bar_type.spec.step: 60 for aggregation=SECOND. step must not be 60"
    )]
    #[case(
        BarAggregation::Minute,
        40,
        "Invalid step in bar_type.spec.step: 40 for aggregation=MINUTE. step must evenly divide 60"
    )]
    #[case(
        BarAggregation::Minute,
        60,
        "Invalid step in bar_type.spec.step: 60 for aggregation=MINUTE. step must not be 60"
    )]
    #[case(
        BarAggregation::Hour,
        5,
        "Invalid step in bar_type.spec.step: 5 for aggregation=HOUR. step must evenly divide 24"
    )]
    #[case(
        BarAggregation::Hour,
        13,
        "Invalid step in bar_type.spec.step: 13 for aggregation=HOUR. step must evenly divide 24"
    )]
    #[case(
        BarAggregation::Hour,
        24,
        "Invalid step in bar_type.spec.step: 24 for aggregation=HOUR. step must not be 24"
    )]
    #[case(
        BarAggregation::Month,
        5,
        "Invalid step in bar_type.spec.step: 5 for aggregation=MONTH. step must evenly divide 12"
    )]
    fn test_bar_specification_new_checked_invalid_periodic_step(
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
        #[case] expected: &str,
    ) {
        let result = BarSpecification::new_checked(step, aggregation, PriceType::Last);

        assert!(result.unwrap_err().to_string().starts_with(expected));
    }

    #[rstest]
    #[case(BarAggregation::Day)]
    #[case(BarAggregation::Week)]
    #[case(BarAggregation::Year)]
    #[case(BarAggregation::Tick)]
    #[case(BarAggregation::TickImbalance)]
    #[case(BarAggregation::TickRuns)]
    #[case(BarAggregation::Volume)]
    #[case(BarAggregation::VolumeImbalance)]
    #[case(BarAggregation::VolumeRuns)]
    #[case(BarAggregation::Value)]
    #[case(BarAggregation::ValueImbalance)]
    #[case(BarAggregation::ValueRuns)]
    #[case(BarAggregation::Renko)]
    fn test_bar_specification_new_checked_allows_non_periodic_steps(
        #[case] aggregation: BarAggregation,
    ) {
        let result = BarSpecification::new_checked(7, aggregation, PriceType::Last);

        assert!(result.is_ok());
    }

    #[rstest]
    #[case(BarAggregation::Day, 213_503)]
    #[case(BarAggregation::Week, 30_500)]
    #[case(BarAggregation::Year, 584)]
    fn test_bar_specification_new_checked_accepts_max_interval_step(
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
    ) {
        let spec = BarSpecification::new_checked(step, aggregation, PriceType::Last).unwrap();
        let interval = spec.timedelta();
        let interval_ns = u64::try_from(interval.as_nanos()).unwrap();

        assert_eq!(spec.step.get(), step);
        assert_eq!(spec.aggregation, aggregation);
        assert_eq!(
            get_bar_interval_ns(&BarType::new(
                InstrumentId::from("BTCUSDT-PERP.BINANCE"),
                spec,
                AggregationSource::Internal,
            ))
            .as_u64(),
            interval_ns
        );
    }

    #[rstest]
    #[case(BarAggregation::Day, 213_504)]
    #[case(BarAggregation::Week, 30_501)]
    #[case(BarAggregation::Year, 585)]
    fn test_bar_specification_new_checked_rejects_unrepresentable_interval(
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
    ) {
        let result = BarSpecification::new_checked(step, aggregation, PriceType::Last);

        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("interval overflows nanoseconds")
        );
    }

    #[rstest]
    #[should_panic(expected = "interval overflows nanoseconds")]
    fn test_bar_specification_new_unrepresentable_interval_panics() {
        let _ = BarSpecification::new(213_504, BarAggregation::Day, PriceType::Last);
    }

    #[rstest]
    fn test_bar_specification_new_checked_accepts_12_month_interval() {
        let spec =
            BarSpecification::new_checked(12, BarAggregation::Month, PriceType::Last).unwrap();

        assert_eq!(spec, BAR_SPEC_12_MONTH_LAST);
        assert_eq!(spec.timedelta(), duration_days(360));
        assert_eq!(
            u64::try_from(spec.timedelta().as_nanos()).unwrap(),
            31_104_000_000_000_000
        );
    }

    #[rstest]
    fn test_try_time_interval_covers_derived_multipliers() {
        let i64_max = usize::try_from(i64::MAX).unwrap();

        assert!(
            BarSpecification::new_checked(i64_max, BarAggregation::Week, PriceType::Last)
                .unwrap_err()
                .to_string()
                .contains("step overflows i64 days")
        );
        assert!(
            try_time_interval(usize::MAX, BarAggregation::Day)
                .unwrap_err()
                .to_string()
                .contains("step exceeds i64 range")
        );
        assert!(
            try_time_interval(i64_max, BarAggregation::Week)
                .unwrap_err()
                .to_string()
                .contains("step overflows i64 days")
        );
        assert!(
            try_time_interval(i64_max, BarAggregation::Month)
                .unwrap_err()
                .to_string()
                .contains("step overflows i64 days")
        );
        assert!(
            try_time_interval(i64_max, BarAggregation::Year)
                .unwrap_err()
                .to_string()
                .contains("step overflows i64 days")
        );
        assert!(
            try_duration_days(i64::MAX)
                .unwrap_err()
                .to_string()
                .contains("days overflow i64 hours")
        );
        assert!(
            try_duration_days(i64::MAX / 24)
                .unwrap_err()
                .to_string()
                .contains("days exceed signed duration range")
        );
    }

    #[rstest]
    fn test_bar_specification_parse_and_builder_reject_unrepresentable_interval() {
        let step = 30_501;
        let json = format!(r#"{{"step":{step},"aggregation":"WEEK","price_type":"LAST"}}"#);

        assert!(serde_json::from_str::<BarSpecification>(&json).is_err());
        assert!(
            BarSpecificationBuilder::default()
                .step(NonZeroUsize::new(step).unwrap())
                .aggregation(BarAggregation::Week)
                .price_type(PriceType::Last)
                .build()
                .is_err()
        );
        assert!(
            BarType::from_str(&format!("BTCUSDT-PERP.BINANCE-{step}-WEEK-LAST-INTERNAL")).is_err()
        );
        assert_eq!(
            BarType::from_str("BTCUSDT-PERP.BINANCE-30500-WEEK-LAST-INTERNAL")
                .unwrap()
                .spec()
                .timedelta(),
            duration_days(213_500)
        );
    }

    #[rstest]
    #[case(BarAggregation::Millisecond, 1, SignedDuration::from_millis(1))]
    #[case(BarAggregation::Millisecond, 10, SignedDuration::from_millis(10))]
    #[case(BarAggregation::Second, 1, SignedDuration::from_secs(1))]
    #[case(BarAggregation::Second, 15, SignedDuration::from_secs(15))]
    #[case(BarAggregation::Minute, 1, SignedDuration::from_mins(1))]
    #[case(BarAggregation::Minute, 30, SignedDuration::from_mins(30))]
    #[case(BarAggregation::Hour, 1, SignedDuration::from_hours(1))]
    #[case(BarAggregation::Hour, 4, SignedDuration::from_hours(4))]
    #[case(BarAggregation::Day, 1, duration_days(1))]
    #[case(BarAggregation::Day, 2, duration_days(2))]
    #[case(BarAggregation::Week, 1, duration_days(7))]
    #[case(BarAggregation::Week, 2, duration_days(14))]
    #[case(BarAggregation::Month, 1, duration_days(30))]
    #[case(BarAggregation::Month, 3, duration_days(90))]
    #[case(BarAggregation::Year, 1, duration_days(365))]
    #[case(BarAggregation::Year, 2, duration_days(730))]
    #[should_panic(expected = "Aggregation not time based")]
    #[case(BarAggregation::Tick, 1, SignedDuration::ZERO)]
    fn test_get_bar_interval(
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
        #[case] expected: SignedDuration,
    ) {
        let bar_type = BarType::Standard {
            instrument_id: InstrumentId::from("BTCUSDT-PERP.BINANCE"),
            spec: BarSpecification::new(step, aggregation, PriceType::Last),
            aggregation_source: AggregationSource::Internal,
        };

        let interval = get_bar_interval(&bar_type);
        assert_eq!(interval, expected);
    }

    #[rstest]
    #[case(BarAggregation::Millisecond, 1, DurationNanos::new(1_000_000))]
    #[case(BarAggregation::Millisecond, 10, DurationNanos::new(10_000_000))]
    #[case(BarAggregation::Second, 1, DurationNanos::new(1_000_000_000))]
    #[case(BarAggregation::Second, 10, DurationNanos::new(10_000_000_000))]
    #[case(BarAggregation::Minute, 1, DurationNanos::new(60_000_000_000))]
    #[case(BarAggregation::Minute, 30, DurationNanos::new(1_800_000_000_000))]
    #[case(BarAggregation::Hour, 1, DurationNanos::new(3_600_000_000_000))]
    #[case(BarAggregation::Hour, 4, DurationNanos::new(14_400_000_000_000))]
    #[case(BarAggregation::Day, 1, DurationNanos::new(86_400_000_000_000))]
    #[case(BarAggregation::Day, 2, DurationNanos::new(172_800_000_000_000))]
    #[case(BarAggregation::Week, 1, DurationNanos::new(604_800_000_000_000))]
    #[case(BarAggregation::Week, 2, DurationNanos::new(1_209_600_000_000_000))]
    #[case(BarAggregation::Month, 1, DurationNanos::new(2_592_000_000_000_000))]
    #[case(BarAggregation::Month, 3, DurationNanos::new(7_776_000_000_000_000))]
    #[case(BarAggregation::Year, 1, DurationNanos::new(31_536_000_000_000_000))]
    #[case(BarAggregation::Year, 2, DurationNanos::new(63_072_000_000_000_000))]
    #[should_panic(expected = "Aggregation not time based")]
    #[case(BarAggregation::Tick, 1, DurationNanos::new(0))]
    fn test_get_bar_interval_ns(
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
        #[case] expected: DurationNanos,
    ) {
        let bar_type = BarType::Standard {
            instrument_id: InstrumentId::from("BTCUSDT-PERP.BINANCE"),
            spec: BarSpecification::new(step, aggregation, PriceType::Last),
            aggregation_source: AggregationSource::Internal,
        };

        let interval_ns = get_bar_interval_ns(&bar_type);
        assert_eq!(interval_ns, expected);
    }

    fn bar_type_with_raw_step(step: usize, aggregation: BarAggregation) -> BarType {
        // Bypasses `BarSpecification::new_checked` to exercise the conversion guards
        let spec = BarSpecification {
            step: NonZeroUsize::new(step).unwrap(),
            aggregation,
            price_type: PriceType::Last,
        };
        BarType::new(
            InstrumentId::from("BTCUSDT-PERP.BINANCE"),
            spec,
            AggregationSource::Internal,
        )
    }

    #[rstest]
    #[should_panic(expected = "`step` exceeds i64 range")]
    fn test_get_bar_interval_step_exceeds_i64_panics() {
        let bar_type = bar_type_with_raw_step(usize::MAX, BarAggregation::Second);
        let _ = get_bar_interval(&bar_type);
    }

    #[rstest]
    #[should_panic(expected = "`step` overflows i64 days")]
    fn test_get_bar_interval_week_step_overflow_panics() {
        let step = usize::try_from(i64::MAX).unwrap();
        let bar_type = bar_type_with_raw_step(step, BarAggregation::Week);
        let _ = get_bar_interval(&bar_type);
    }

    #[rstest]
    #[should_panic(expected = "`step` overflows i64 days")]
    fn test_timedelta_year_step_overflow_panics() {
        let step = usize::try_from(i64::MAX).unwrap();
        let bar_type = bar_type_with_raw_step(step, BarAggregation::Year);
        let _ = bar_type.spec().timedelta();
    }

    #[rstest]
    #[should_panic(expected = "`step` exceeds u32 range for month arithmetic")]
    fn test_get_time_bar_start_month_step_exceeds_u32_panics() {
        let bar_type = bar_type_with_raw_step(1_usize << 40, BarAggregation::Month);
        let now = timestamp("2024-07-21T12:00:00Z");
        let _ = get_time_bar_start(now, &bar_type, None);
    }

    #[rstest]
    #[should_panic(expected = "`step` exceeds i32 range for year arithmetic")]
    fn test_get_time_bar_start_year_step_exceeds_i32_panics() {
        let bar_type = bar_type_with_raw_step(1_usize << 40, BarAggregation::Year);
        let now = timestamp("2024-07-21T12:00:00Z");
        let _ = get_time_bar_start(now, &bar_type, None);
    }

    #[rstest]
    #[should_panic(expected = "year exceeds Jiff supported range")]
    fn test_get_time_bar_start_year_step_exceeds_jiff_range_panics() {
        let bar_type = bar_type_with_raw_step(32_000, BarAggregation::Year);
        let now = timestamp("2024-07-21T12:00:00Z");
        let _ = get_time_bar_start(now, &bar_type, None);
    }

    #[rstest]
    #[case::millisecond(
    Timestamp::new(1_658_349_296, 123_000_000).unwrap(), // 2022-07-20 20:34:56.123 UTC
    BarAggregation::Millisecond,
    1,
    Timestamp::new(1_658_349_296, 123_000_000).unwrap(), // 2022-07-20 20:34:56.123 UTC
    )]
    #[rstest]
    #[case::millisecond(
    Timestamp::new(1_658_349_296, 123_000_000).unwrap(), // 2022-07-20 20:34:56.123 UTC
    BarAggregation::Millisecond,
    10,
    Timestamp::new(1_658_349_296, 120_000_000).unwrap(), // 2022-07-20 20:34:56.120 UTC
    )]
    #[case::second(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Second,
        1,
        timestamp("2024-07-21T12:34:56Z")
    )]
    #[case::second(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Second,
        5,
        timestamp("2024-07-21T12:34:55Z")
    )]
    #[case::minute(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Minute,
        1,
        timestamp("2024-07-21T12:34:00Z")
    )]
    #[case::minute(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Minute,
        5,
        timestamp("2024-07-21T12:30:00Z")
    )]
    #[case::hour(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Hour,
        1,
        timestamp("2024-07-21T12:00:00Z")
    )]
    #[case::hour(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Hour,
        2,
        timestamp("2024-07-21T12:00:00Z")
    )]
    #[case::day(
        timestamp("2024-07-21T12:34:56Z"),
        BarAggregation::Day,
        1,
        timestamp("2024-07-21T00:00:00Z")
    )]
    fn test_get_time_bar_start(
        #[case] now: Timestamp,
        #[case] aggregation: BarAggregation,
        #[case] step: usize,
        #[case] expected: Timestamp,
    ) {
        let bar_type = BarType::Standard {
            instrument_id: InstrumentId::from("BTCUSDT-PERP.BINANCE"),
            spec: BarSpecification::new(step, aggregation, PriceType::Last),
            aggregation_source: AggregationSource::Internal,
        };

        let start_time = get_time_bar_start(now, &bar_type, None);
        assert_eq!(start_time, expected);
    }

    #[rstest]
    fn test_bar_spec_string_reprs() {
        let bar_spec = BarSpecification::new(1, BarAggregation::Minute, PriceType::Bid);
        assert_eq!(bar_spec.to_string(), "1-MINUTE-BID");
        assert_eq!(format!("{bar_spec}"), "1-MINUTE-BID");
    }

    #[rstest]
    fn test_bar_type_parse_valid() {
        let input = "BTCUSDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL";
        let bar_type = BarType::from(input);

        assert_eq!(
            bar_type.instrument_id(),
            InstrumentId::from("BTCUSDT-PERP.BINANCE")
        );
        assert_eq!(
            bar_type.spec(),
            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last)
        );
        assert_eq!(bar_type.aggregation_source(), AggregationSource::External);
        assert_eq!(bar_type, BarType::from(input));
    }

    #[rstest]
    #[case("BTCUSDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL", true, false)]
    #[case("BTCUSDT-PERP.BINANCE-1-MINUTE-LAST-INTERNAL", false, true)]
    #[case(
        "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-MINUTE-EXTERNAL",
        false,
        true
    )]
    #[case(
        "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-EXTERNAL@1-MINUTE-INTERNAL",
        true,
        false
    )]
    fn test_bar_type_aggregation_source_predicates(
        #[case] input: &str,
        #[case] expected_external: bool,
        #[case] expected_internal: bool,
    ) {
        let bar_type = BarType::from(input);
        assert_eq!(bar_type.is_externally_aggregated(), expected_external);
        assert_eq!(bar_type.is_internally_aggregated(), expected_internal);
    }

    #[rstest]
    fn test_bar_type_composite_aggregation_source_predicates_track_inner() {
        let bar_type =
            BarType::from("BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-MINUTE-EXTERNAL");

        assert!(bar_type.is_internally_aggregated());
        assert!(!bar_type.is_externally_aggregated());

        let composite = bar_type.composite();
        assert!(composite.is_externally_aggregated());
        assert!(!composite.is_internally_aggregated());
    }

    #[rstest]
    fn test_bar_type_from_str_with_utf8_symbol() {
        let non_ascii_instrument = "TËST-PÉRP.BINANCE";
        let non_ascii_bar_type = "TËST-PÉRP.BINANCE-1-MINUTE-LAST-EXTERNAL";

        let bar_type = BarType::from_str(non_ascii_bar_type).unwrap();

        assert_eq!(
            bar_type.instrument_id(),
            InstrumentId::from_str(non_ascii_instrument).unwrap()
        );
        assert_eq!(
            bar_type.spec(),
            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last)
        );
        assert_eq!(bar_type.aggregation_source(), AggregationSource::External);
        assert_eq!(bar_type.to_string(), non_ascii_bar_type);
    }

    #[rstest]
    fn test_bar_type_composite_parse_valid() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-MINUTE-EXTERNAL";
        let bar_type = BarType::from(input);
        let standard = bar_type.standard();

        assert_eq!(
            bar_type.instrument_id(),
            InstrumentId::from("BTCUSDT-PERP.BINANCE")
        );
        assert_eq!(
            bar_type.spec(),
            BarSpecification::new(2, BarAggregation::Minute, PriceType::Last,)
        );
        assert_eq!(bar_type.aggregation_source(), AggregationSource::Internal);
        assert_eq!(bar_type, BarType::from(input));
        assert!(bar_type.is_composite());

        assert_eq!(
            standard.instrument_id(),
            InstrumentId::from("BTCUSDT-PERP.BINANCE")
        );
        assert_eq!(
            standard.spec(),
            BarSpecification::new(2, BarAggregation::Minute, PriceType::Last,)
        );
        assert_eq!(standard.aggregation_source(), AggregationSource::Internal);
        assert!(standard.is_standard());

        let composite = bar_type.composite();
        let composite_input = "BTCUSDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL";

        assert_eq!(
            composite.instrument_id(),
            InstrumentId::from("BTCUSDT-PERP.BINANCE")
        );
        assert_eq!(
            composite.spec(),
            BarSpecification::new(1, BarAggregation::Minute, PriceType::Last,)
        );
        assert_eq!(composite.aggregation_source(), AggregationSource::External);
        assert_eq!(composite, BarType::from(composite_input));
        assert!(composite.is_standard());
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_0() {
        let input = "BTCUSDT-PERP-1-MINUTE-LAST-INTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'BTCUSDT-PERP' at position 0"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_1() {
        let input = "BTCUSDT-PERP.BINANCE-INVALID-MINUTE-LAST-INTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 1"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_spec_step() {
        let input = "BTCUSDT-PERP.BINANCE-60-MINUTE-LAST-INTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!("Error parsing `BarType` from '{input}', invalid token: '60' at position 1")
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_2() {
        let input = "BTCUSDT-PERP.BINANCE-1-INVALID-LAST-INTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 2"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_3() {
        let input = "BTCUSDT-PERP.BINANCE-1-MINUTE-INVALID-INTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 3"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_4() {
        let input = "BTCUSDT-PERP.BINANCE-1-MINUTE-BID-INVALID";
        let result = BarType::from_str(input);

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 4"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_5() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@INVALID-MINUTE-EXTERNAL";
        let result = BarType::from_str(input);

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 5"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_composite_spec_step() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@60-MINUTE-EXTERNAL";
        let result = BarType::from_str(input);

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!("Error parsing `BarType` from '{input}', invalid token: '60' at position 5")
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_6() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-INVALID-EXTERNAL";
        let result = BarType::from_str(input);

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 6"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_invalid_token_pos_7() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-MINUTE-INVALID";
        let result = BarType::from_str(input);

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: 'INVALID' at position 7"
            )
        );
    }

    #[rstest]
    fn test_bar_type_parse_rejects_extra_composite_segment() {
        let input = "BTCUSDT-PERP.BINANCE-2-MINUTE-LAST-INTERNAL@1-MINUTE-EXTERNAL@1-HOUR-EXTERNAL";
        let result = BarType::from_str(input);

        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "Error parsing `BarType` from '{input}', invalid token: '1-HOUR-EXTERNAL' at position 5"
            )
        );
    }

    #[rstest]
    fn test_bar_type_equality() {
        let instrument_id1 = InstrumentId {
            symbol: Symbol::new("AUD/USD"),
            venue: Venue::new("SIM"),
        };
        let instrument_id2 = InstrumentId {
            symbol: Symbol::new("GBP/USD"),
            venue: Venue::new("SIM"),
        };
        let bar_spec = BarSpecification::new(1, BarAggregation::Minute, PriceType::Bid);
        let bar_type1 = BarType::Standard {
            instrument_id: instrument_id1,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar_type2 = BarType::Standard {
            instrument_id: instrument_id1,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar_type3 = BarType::Standard {
            instrument_id: instrument_id2,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        assert_eq!(bar_type1, bar_type1);
        assert_eq!(bar_type1, bar_type2);
        assert_ne!(bar_type1, bar_type3);
    }

    #[rstest]
    fn test_bar_type_id_spec_key_ignores_aggregation_source() {
        let bar_type_external = BarType::from_str("ESM4.XCME-1-MINUTE-LAST-EXTERNAL").unwrap();
        let bar_type_internal = BarType::from_str("ESM4.XCME-1-MINUTE-LAST-INTERNAL").unwrap();

        // Full equality should differ
        assert_ne!(bar_type_external, bar_type_internal);

        // id_spec_key should be the same
        assert_eq!(
            bar_type_external.id_spec_key(),
            bar_type_internal.id_spec_key()
        );

        // Verify key components
        let (instrument_id, spec) = bar_type_external.id_spec_key();
        assert_eq!(instrument_id, bar_type_external.instrument_id());
        assert_eq!(spec, bar_type_external.spec());
    }

    #[rstest]
    fn test_bar_type_comparison() {
        let instrument_id1 = InstrumentId {
            symbol: Symbol::new("AUD/USD"),
            venue: Venue::new("SIM"),
        };

        let instrument_id2 = InstrumentId {
            symbol: Symbol::new("GBP/USD"),
            venue: Venue::new("SIM"),
        };
        let bar_spec = BarSpecification::new(1, BarAggregation::Minute, PriceType::Bid);
        let bar_spec2 = BarSpecification::new(2, BarAggregation::Minute, PriceType::Bid);
        let bar_type1 = BarType::Standard {
            instrument_id: instrument_id1,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar_type2 = BarType::Standard {
            instrument_id: instrument_id1,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar_type3 = BarType::Standard {
            instrument_id: instrument_id2,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar_type4 = BarType::Composite {
            instrument_id: instrument_id2,
            spec: bar_spec2,
            aggregation_source: AggregationSource::Internal,

            composite_step: 1,
            composite_aggregation: BarAggregation::Minute,
            composite_aggregation_source: AggregationSource::External,
        };

        assert!(bar_type1 <= bar_type2);
        assert!(bar_type1 < bar_type3);
        assert!(bar_type3 > bar_type1);
        assert!(bar_type3 >= bar_type1);
        assert!(bar_type4 >= bar_type1);
    }

    #[rstest]
    fn test_bar_new() {
        let bar_type = BarType::from("AAPL.XNAS-1-MINUTE-LAST-INTERNAL");
        let open = Price::from("100.0");
        let high = Price::from("105.0");
        let low = Price::from("95.0");
        let close = Price::from("102.0");
        let volume = Quantity::from("1000");
        let ts_event = UnixNanos::from(1_000_000);
        let ts_init = UnixNanos::from(2_000_000);

        let bar = Bar::new(bar_type, open, high, low, close, volume, ts_event, ts_init);

        assert_eq!(bar.bar_type, bar_type);
        assert_eq!(bar.open, open);
        assert_eq!(bar.high, high);
        assert_eq!(bar.low, low);
        assert_eq!(bar.close, close);
        assert_eq!(bar.volume, volume);
        assert_eq!(bar.ts_event, ts_event);
        assert_eq!(bar.ts_init, ts_init);
    }

    #[rstest]
    #[case("100.0", "90.0", "95.0", "92.0", "high >= open")]
    #[case("100.0", "105.0", "110.0", "102.0", "high >= low")]
    #[case("100.0", "105.0", "95.0", "110.0", "high >= close")]
    #[case("100.0", "105.0", "95.0", "90.0", "low <= close")]
    #[case("100.0", "110.0", "105.0", "108.0", "low <= open")]
    #[case("100.0", "90.0", "110.0", "120.0", "high >= open")] // First failing predicate reported
    fn test_bar_new_checked_conditions(
        #[case] open: &str,
        #[case] high: &str,
        #[case] low: &str,
        #[case] close: &str,
        #[case] expected: &str,
    ) {
        let bar_type = BarType::from("AAPL.XNAS-1-MINUTE-LAST-INTERNAL");
        let open = Price::from(open);
        let high = Price::from(high);
        let low = Price::from(low);
        let close = Price::from(close);
        let volume = Quantity::from("1000");
        let ts_event = UnixNanos::from(1_000_000);
        let ts_init = UnixNanos::from(2_000_000);

        let result = Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init);

        let error = result.unwrap_err();
        assert!(
            error.to_string().contains(expected),
            "unexpected message: {error}"
        );
    }

    #[rstest]
    fn test_bar_equality() {
        let instrument_id = InstrumentId {
            symbol: Symbol::new("AUDUSD"),
            venue: Venue::new("SIM"),
        };
        let bar_spec = BarSpecification::new(1, BarAggregation::Minute, PriceType::Bid);
        let bar_type = BarType::Standard {
            instrument_id,
            spec: bar_spec,
            aggregation_source: AggregationSource::External,
        };
        let bar1 = Bar {
            bar_type,
            open: Price::from("1.00001"),
            high: Price::from("1.00004"),
            low: Price::from("1.00002"),
            close: Price::from("1.00003"),
            volume: Quantity::from("100000"),
            ts_event: UnixNanos::default(),
            ts_init: UnixNanos::from(1),
        };

        let bar2 = Bar {
            bar_type,
            open: Price::from("1.00000"),
            high: Price::from("1.00004"),
            low: Price::from("1.00002"),
            close: Price::from("1.00003"),
            volume: Quantity::from("100000"),
            ts_event: UnixNanos::default(),
            ts_init: UnixNanos::from(1),
        };
        assert_eq!(bar1, bar1);
        assert_ne!(bar1, bar2);
    }

    #[rstest]
    fn test_json_serialization() {
        let bar = Bar::default();
        let serialized = bar.to_json_bytes().unwrap();
        let deserialized = Bar::from_json_bytes(serialized.as_ref()).unwrap();
        assert_eq!(deserialized, bar);
    }

    #[rstest]
    fn test_msgpack_serialization() {
        let bar = Bar::default();
        let serialized = bar.to_msgpack_bytes().unwrap();
        let deserialized = Bar::from_msgpack_bytes(serialized.as_ref()).unwrap();
        assert_eq!(deserialized, bar);
    }

    #[rstest]
    fn test_bar_deserialization_rejects_invalid_ohlc() {
        let json = r#"{
            "type": "Bar",
            "bar_type": "AUD/USD.SIM-1-MINUTE-BID-EXTERNAL",
            "open": "1.00010",
            "high": "1.00000",
            "low": "1.00020",
            "close": "1.00010",
            "volume": "100000",
            "ts_event": 0,
            "ts_init": 0
        }"#;

        let result = Bar::from_json_bytes(json.as_bytes());
        assert!(
            result.is_err(),
            "high < low must fail deserialization, was {result:?}"
        );
    }

    #[rstest]
    fn test_bar_specification_deserialization_rejects_invalid_step() {
        let json = r#"{"step":7,"aggregation":"MINUTE","price_type":"LAST"}"#;

        let result = serde_json::from_str::<BarSpecification>(json);
        assert!(
            result.is_err(),
            "non-periodic step must fail deserialization, was {result:?}"
        );
    }

    #[rstest]
    fn test_bar_specification_builder_rejects_invalid_step() {
        let result = BarSpecificationBuilder::default()
            .step(NonZeroUsize::new(7).unwrap())
            .aggregation(BarAggregation::Minute)
            .price_type(PriceType::Last)
            .build();

        assert!(
            result.is_err(),
            "non-periodic step must fail builder validation, was {result:?}"
        );
    }

    #[rstest]
    fn test_bar_spec_12_month_round_trips() {
        // BAR_SPEC_12_MONTH_LAST is shipped (OKX yearly candles), so the string
        // form must parse back
        let bar_type = BarType::new(
            InstrumentId::from("BTC-USDT.OKX"),
            BAR_SPEC_12_MONTH_LAST,
            AggregationSource::External,
        );

        let parsed = BarType::from_str(&bar_type.to_string()).unwrap();
        assert_eq!(parsed, bar_type);
        assert_eq!(
            BarSpecification::new_checked(12, BarAggregation::Month, PriceType::Last).unwrap(),
            BAR_SPEC_12_MONTH_LAST,
        );
    }

    #[rstest]
    fn test_bar_type_new_composite_checked_invalid_step() {
        let instrument_id = InstrumentId::from("AUD/USD.SIM");
        let spec = BarSpecification::new(5, BarAggregation::Minute, PriceType::Bid);

        let result = BarType::new_composite_checked(
            instrument_id,
            spec,
            AggregationSource::Internal,
            0,
            BarAggregation::Minute,
            AggregationSource::External,
        );

        assert!(
            result.is_err(),
            "zero composite step must fail, was {result:?}"
        );
    }
}

#[cfg(test)]
mod property_tests {
    use std::str::FromStr;

    use proptest::prelude::*;
    use rstest::rstest;

    use super::*;
    use crate::identifiers::{Symbol, Venue};

    fn symbol_strategy() -> impl Strategy<Value = &'static str> {
        prop::sample::select(vec![
            "AAPL",
            "BTC-PERP",
            "EUR/USD",
            "ES-MINI-4",
            "MSFT.OQ",
            "6E",
        ])
    }

    fn venue_strategy() -> impl Strategy<Value = &'static str> {
        prop::sample::select(vec!["SIM", "XNAS", "GLBX", "BINANCE"])
    }

    fn time_spec_strategy() -> impl Strategy<Value = (BarAggregation, usize)> {
        prop_oneof![
            (
                Just(BarAggregation::Millisecond),
                prop::sample::select(vec![1usize, 2, 5, 10, 25, 50, 100, 250, 500]),
            ),
            (
                Just(BarAggregation::Second),
                prop::sample::select(vec![1usize, 2, 3, 5, 10, 15, 30]),
            ),
            (
                Just(BarAggregation::Minute),
                prop::sample::select(vec![1usize, 2, 5, 15, 30]),
            ),
            (
                Just(BarAggregation::Hour),
                prop::sample::select(vec![1usize, 2, 4, 12]),
            ),
            (
                Just(BarAggregation::Day),
                prop::sample::select(vec![1usize, 2, 3]),
            ),
            (Just(BarAggregation::Week), Just(1usize)),
        ]
    }

    fn spec_strategy() -> impl Strategy<Value = (BarAggregation, usize)> {
        prop_oneof![
            time_spec_strategy(),
            // Month stays out of time_spec_strategy: the alignment proptest uses the
            // 30-day proxy interval, which is unsound for calendar months
            (
                Just(BarAggregation::Month),
                prop::sample::select(vec![1usize, 2, 3, 4, 6, 12]),
            ),
            (Just(BarAggregation::Tick), 1usize..=10_000),
            (Just(BarAggregation::Volume), 1usize..=10_000),
            (Just(BarAggregation::Value), 1usize..=10_000),
        ]
    }

    fn price_type_strategy() -> impl Strategy<Value = PriceType> {
        prop::sample::select(vec![
            PriceType::Bid,
            PriceType::Ask,
            PriceType::Mid,
            PriceType::Last,
        ])
    }

    fn source_strategy() -> impl Strategy<Value = AggregationSource> {
        prop_oneof![
            Just(AggregationSource::Internal),
            Just(AggregationSource::External),
        ]
    }

    proptest! {
        #[rstest]
        fn prop_bar_type_string_round_trip(
            symbol in symbol_strategy(),
            venue in venue_strategy(),
            (aggregation, step) in spec_strategy(),
            price_type in price_type_strategy(),
            source in source_strategy(),
            composite in prop::option::of((time_spec_strategy(), source_strategy())),
        ) {
            let instrument_id = InstrumentId::new(Symbol::from(symbol), Venue::from(venue));
            let spec = BarSpecification::new(step, aggregation, price_type);

            let bar_type = match composite {
                None => BarType::new(instrument_id, spec, source),
                Some(((composite_aggregation, composite_step), composite_source)) => {
                    BarType::new_composite(
                        instrument_id,
                        spec,
                        source,
                        composite_step,
                        composite_aggregation,
                        composite_source,
                    )
                }
            };

            let parsed = BarType::from_str(&bar_type.to_string());
            prop_assert!(parsed.is_ok(), "failed to parse '{bar_type}': {parsed:?}");
            prop_assert_eq!(parsed.unwrap(), bar_type);
        }

        #[rstest]
        fn prop_get_time_bar_start_alignment(
            (aggregation, step) in time_spec_strategy(),
            epoch_secs in 946_684_800i64..2_524_608_000i64,
            subsec_nanos in 0u32..1_000_000_000u32,
        ) {
            let instrument_id = InstrumentId::from("AAPL.XNAS");
            let spec = BarSpecification::new(step, aggregation, PriceType::Last);
            let bar_type = BarType::new(instrument_id, spec, AggregationSource::Internal);

            let now = Timestamp::new(epoch_secs, subsec_nanos.cast_signed()).unwrap();
            let start = get_time_bar_start(now, &bar_type, None);
            let interval = get_bar_interval(&bar_type);

            prop_assert!(start <= now, "start {start} must not be after now {now}");
            prop_assert!(
                now.duration_since(start) < interval,
                "now {now} must fall within one interval of start {start}"
            );
        }
    }
}