//! Confirmed manual fills are external observations, not simulated broker fills.
//! The producer must bind these records to the runtime's durable order/audit facts.

use std::collections::{BTreeMap, BTreeSet};

use chrono::{DateTime, FixedOffset, NaiveDate, Timelike, Utc};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::events::OrderSide;
use crate::{DataSet, FixedMoney, PortfolioState};
use rust_decimal::prelude::ToPrimitive;

pub const MANUAL_REPLAY_SCHEMA: &str = "fidc.observed-manual-executions/v3";

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualExecutionReplay {
    pub schema: String,
    pub runtime_id: String,
    pub account_id: String,
    pub source_contract_sha256: String,
    pub content_sha256: String,
    pub observation_cutoff: DateTime<Utc>,
    pub actions: Vec<ManualExecutionAction>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub position_exposure_events: Vec<crate::position_exposure::PositionExposureEvent>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub legacy_position_exposure_bps: BTreeMap<NaiveDate, i32>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualExecutionAction {
    pub action_id: String,
    pub source: ManualExecutionSource,
    pub audit_event_ids: Vec<String>,
    pub confirmed_at: DateTime<Utc>,
    pub confirmation_observed_at: DateTime<Utc>,
    pub outcome: ManualActionOutcome,
    pub orders: Vec<ManualExecutionOrder>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ManualActionOutcome {
    NoOrdersNeeded,
    NotExecuted,
    OrdersTerminal,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ManualExecutionSource {
    ManualSecurityTrade,
    ManualPositionAction,
    ManualRebalance,
    StockPoolAllocation,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualExecutionOrder {
    pub order_id: String,
    pub broker_order_id: Option<String>,
    pub source_adapter: Option<String>,
    pub symbol: String,
    pub side: OrderSide,
    pub quantity: u32,
    pub order_created_at: DateTime<Utc>,
    pub terminal_observed_at: DateTime<Utc>,
    pub terminal_status: ManualOrderTerminalStatus,
    pub fills: Vec<ManualExecutionFill>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ManualOrderTerminalStatus {
    Filled,
    Cancelled,
    Rejected,
    Expired,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualExecutionFill {
    pub trade_id: String,
    pub observation_event_id: String,
    pub observation_sequence: u64,
    pub fee_observation_event_id: String,
    pub fee_observation_sequence: u64,
    pub fee_observed_at: DateTime<Utc>,
    pub trade_date: NaiveDate,
    pub executed_at: DateTime<Utc>,
    pub observed_at: DateTime<Utc>,
    pub timestamp_precision: ManualTimestampPrecision,
    pub quantity: u32,
    #[serde(with = "rust_decimal::serde::str")]
    pub price: Decimal,
    #[serde(default, with = "rust_decimal::serde::str_option")]
    pub commission: Option<Decimal>,
    #[serde(default, with = "rust_decimal::serde::str_option")]
    pub stamp_tax: Option<Decimal>,
    #[serde(default, with = "rust_decimal::serde::str_option")]
    pub transfer_fee: Option<Decimal>,
    /// Full observed charge, including any venue fees not itemized above.
    #[serde(with = "rust_decimal::serde::str")]
    pub total_fee: Decimal,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ManualTimestampPrecision {
    Second,
    Millisecond,
    Microsecond,
    Nanosecond,
}

impl ManualTimestampPrecision {
    fn nanoseconds(self) -> i64 {
        match self {
            Self::Second => 1_000_000_000,
            Self::Millisecond => 1_000_000,
            Self::Microsecond => 1_000,
            Self::Nanosecond => 1,
        }
    }
}

impl ManualExecutionFill {
    pub fn gross_amount(&self) -> Result<Decimal, String> {
        self.price
            .checked_mul(Decimal::from(self.quantity))
            .ok_or_else(|| "manual fill gross amount overflow".into())
    }

    pub fn total_fees(&self) -> Result<Decimal, String> {
        let known = [self.commission, self.stamp_tax, self.transfer_fee]
            .into_iter()
            .flatten()
            .try_fold(Decimal::ZERO, |sum, fee| {
                if fee < Decimal::ZERO {
                    return Err("manual fill fee component is negative");
                }
                sum.checked_add(fee).ok_or("manual fill fees overflow")
            })?;
        if self.total_fee < known {
            return Err("manual total fee is below its known components".into());
        }
        Ok(self.total_fee)
    }
}

fn identifier(value: &str) -> Result<(), String> {
    if value.is_empty()
        || value.trim() != value
        || value.len() > 256
        || value.chars().any(char::is_control)
    {
        return Err("manual execution identity is empty, untrimmed or invalid".into());
    }
    Ok(())
}

impl ManualExecutionReplay {
    /// Market/indicator data is needed for securities whose observed fills
    /// change the portfolio. A rejected, never-filled order is not data demand.
    pub fn required_data_symbols(&self) -> Result<BTreeSet<String>, String> {
        self.validate()?;
        Ok(self
            .actions
            .iter()
            .flat_map(|action| &action.orders)
            .filter(|order| !order.fills.is_empty())
            .map(|order| order.symbol.clone())
            .collect())
    }

    pub fn observations(&self) -> Result<Vec<ManualFillObservation<'_>>, String> {
        self.validate()?;
        let mut observations = Vec::new();
        for action in &self.actions {
            for order in &action.orders {
                for fill in &order.fills {
                    observations.push(ManualFillObservation {
                        action,
                        order,
                        fill,
                    });
                }
            }
        }
        observations.sort_by_key(|entry| (entry.fill.observed_at, entry.fill.observation_sequence));
        Ok(observations)
    }
    pub fn content_digest(&self) -> Result<String, String> {
        let mut value = serde_json::to_value(self).map_err(|error| error.to_string())?;
        value
            .as_object_mut()
            .ok_or("manual replay is not an object")?
            .remove("contentSha256");
        let bytes = serde_json::to_vec(&value).map_err(|error| error.to_string())?;
        Ok(format!("{:x}", Sha256::digest(bytes)))
    }

    pub fn validate(&self) -> Result<(), String> {
        if self.schema != MANUAL_REPLAY_SCHEMA
            && self.schema != "fidc.observed-manual-executions/v2"
        {
            return Err("unsupported manual replay schema".into());
        }
        if self.schema == "fidc.observed-manual-executions/v2"
            && (!self.position_exposure_events.is_empty()
                || !self.legacy_position_exposure_bps.is_empty())
        {
            return Err("runtime configuration requires manual replay v3".into());
        }
        crate::position_exposure::PositionExposureTimeline::from_events(
            &self.position_exposure_events,
        )?;
        if self.position_exposure_events.iter().any(|event| event.effective_at > self.observation_cutoff) {
            return Err("observed runtime position event is after the evidence cutoff".into());
        }
        if self
            .legacy_position_exposure_bps
            .values()
            .any(|value| !(0..=10000).contains(value))
        {
            return Err("legacy manual exposure is outside 0..10000 bps".into());
        }
        identifier(&self.runtime_id)?;
        identifier(&self.account_id)?;
        if self.source_contract_sha256.len() != 64
            || !self
                .source_contract_sha256
                .bytes()
                .all(|v| v.is_ascii_hexdigit())
        {
            return Err("manual replay source contract hash is invalid".into());
        }
        if self.content_digest()? != self.content_sha256 {
            return Err("manual replay content digest mismatch".into());
        }
        if self.actions.len() > 100_000 {
            return Err("manual replay action limit exceeded; trace was not truncated".into());
        }
        let shanghai = FixedOffset::east_opt(8 * 3600).unwrap();
        let mut actions = BTreeSet::new();
        let mut audits = BTreeSet::new();
        let mut orders = BTreeSet::new();
        let mut broker_orders = BTreeSet::new();
        let mut trades = BTreeSet::new();
        let mut observation_events = BTreeSet::new();
        let mut observation_sequences = BTreeSet::new();
        let mut fee_observations = BTreeSet::new();
        let mut receipt_ids = BTreeMap::new();
        let mut receipt_sequences = BTreeMap::new();
        for action in &self.actions {
            identifier(&action.action_id)?;
            if !actions.insert(action.action_id.as_str())
                || action.confirmed_at > self.observation_cutoff
                || action.confirmation_observed_at < action.confirmed_at
                || action.confirmation_observed_at > self.observation_cutoff
            {
                return Err("duplicate manual action or confirmation after cutoff".into());
            }
            if action.audit_event_ids.is_empty() {
                return Err("manual action has no immutable audit binding".into());
            }
            if (action.outcome != ManualActionOutcome::OrdersTerminal) != action.orders.is_empty() {
                return Err("manual action outcome does not prove its order coverage".into());
            }
            for id in &action.audit_event_ids {
                identifier(id)?;
                if !audits.insert(id.as_str()) {
                    return Err("manual audit event is bound more than once".into());
                }
            }
            for order in &action.orders {
                identifier(&order.order_id)?;
                if let Some(adapter) = &order.source_adapter {
                    identifier(adapter)?;
                }
                identifier(&order.symbol)?;
                if let Some(id) = &order.broker_order_id {
                    identifier(id)?;
                    if !broker_orders.insert((
                        order
                            .source_adapter
                            .as_deref()
                            .ok_or("broker identity requires its source adapter")?,
                        order.order_created_at.with_timezone(&shanghai).date_naive(),
                        id.as_str(),
                    )) {
                        return Err("manual local orders share one broker order identity".into());
                    }
                }
                if !order.fills.is_empty() && order.source_adapter.is_none() {
                    return Err("manual fills require a known source adapter".into());
                }
                if !order.fills.is_empty()
                    && order.source_adapter.as_deref() != Some("paper")
                    && order.broker_order_id.is_none()
                {
                    return Err(
                        "manual broker fills require their original broker order identity".into(),
                    );
                }
                if !orders.insert(order.order_id.as_str())
                    || order.quantity == 0
                    || order.quantity > i32::MAX as u32
                {
                    return Err("duplicate manual order or invalid quantity".into());
                }
                if order.order_created_at < action.confirmed_at
                    || order.terminal_observed_at < order.order_created_at
                    || order.terminal_observed_at > self.observation_cutoff
                {
                    return Err(
                        "manual order confirmation/submission/terminal time is inconsistent".into(),
                    );
                }
                let mut filled = 0_u32;
                for fill in &order.fills {
                    identifier(&fill.trade_id)?;
                    identifier(&fill.observation_event_id)?;
                    identifier(&fill.fee_observation_event_id)?;
                    if fill.observation_sequence == 0
                        || fill.observation_sequence > i64::MAX as u64
                        || !observation_events.insert(fill.observation_event_id.as_str())
                        || !observation_sequences.insert(fill.observation_sequence)
                    {
                        return Err(
                            "manual fill requires a unique durable observation event and sequence"
                                .into(),
                        );
                    }
                    if fill.fee_observation_sequence == 0
                        || fill.fee_observation_sequence > i64::MAX as u64
                        || fill.fee_observed_at < fill.observed_at
                        || fill.fee_observed_at > self.observation_cutoff
                        || !fee_observations.insert((
                            fill.fee_observation_event_id.as_str(),
                            fill.fee_observation_sequence,
                        ))
                    {
                        return Err("manual finalized fees require their own unique observation within the cutoff".into());
                    }
                    if (fill.fee_observation_event_id == fill.observation_event_id)
                        != (fill.fee_observation_sequence == fill.observation_sequence)
                        || (fill.fee_observation_event_id == fill.observation_event_id
                            && fill.fee_observed_at != fill.observed_at)
                    {
                        return Err("manual fill and fee observation identities disagree".into());
                    }
                    if !trades.insert((fill.trade_date, fill.trade_id.as_str()))
                        || fill.quantity == 0
                    {
                        return Err("duplicate manual trade or zero fill quantity".into());
                    }
                    for (id, sequence) in [
                        (&fill.observation_event_id, fill.observation_sequence),
                        (
                            &fill.fee_observation_event_id,
                            fill.fee_observation_sequence,
                        ),
                    ] {
                        if receipt_ids
                            .insert(id, (&fill.trade_id, sequence))
                            .is_some_and(|owner| owner != (&fill.trade_id, sequence))
                            || receipt_sequences
                                .insert(sequence, (&fill.trade_id, id))
                                .is_some_and(|owner| owner != (&fill.trade_id, id))
                        {
                            return Err("manual observation identity is reused by a different trade or sequence".into());
                        }
                    }
                    if fill.executed_at.with_timezone(&shanghai).date_naive() != fill.trade_date
                        || fill.observed_at > self.observation_cutoff
                        || fill.observed_at < order.order_created_at
                        || fill.observed_at < action.confirmation_observed_at
                        || fill.observed_at < fill.executed_at
                        || fill.executed_at > order.terminal_observed_at
                    {
                        return Err("manual fill execution/observation time is inconsistent".into());
                    }
                    if i64::from(fill.executed_at.nanosecond())
                        % fill.timestamp_precision.nanoseconds()
                        != 0
                    {
                        return Err(
                            "broker timestamp contains digits finer than its declared precision"
                                .into(),
                        );
                    }
                    let upper = fill
                        .executed_at
                        .checked_add_signed(chrono::Duration::nanoseconds(
                            fill.timestamp_precision.nanoseconds(),
                        ))
                        .ok_or("manual execution timestamp overflow")?;
                    let earliest = order.order_created_at.max(action.confirmation_observed_at);
                    if fill.executed_at < earliest && earliest >= upper {
                        return Err("manual fill predates its order or durable confirmation".into());
                    }
                    if fill.price <= Decimal::ZERO {
                        return Err("manual fill requires a positive price".into());
                    }
                    fill.gross_amount()?
                        .checked_add(fill.total_fees()?)
                        .ok_or("manual fill cash amount overflow")?;
                    filled = filled
                        .checked_add(fill.quantity)
                        .ok_or("manual cumulative fill quantity overflow")?;
                }
                if filled > order.quantity
                    || (order.terminal_status == ManualOrderTerminalStatus::Filled
                        && filled != order.quantity)
                    || (order.terminal_status == ManualOrderTerminalStatus::Rejected && filled != 0)
                    || (matches!(
                        order.terminal_status,
                        ManualOrderTerminalStatus::Cancelled | ManualOrderTerminalStatus::Expired
                    ) && filled == order.quantity)
                {
                    return Err("manual terminal status disagrees with cumulative fills".into());
                }
            }
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy)]
pub struct ManualFillObservation<'a> {
    pub action: &'a ManualExecutionAction,
    pub order: &'a ManualExecutionOrder,
    pub fill: &'a ManualExecutionFill,
}

#[derive(Debug, Clone, PartialEq)]
pub struct AppliedManualFill {
    pub gross: FixedMoney,
    pub fees: FixedMoney,
    pub cash_delta: FixedMoney,
    pub quantity_after: u32,
}

/// One replay owns its immutable trace and progress. Advancing is atomic even
/// if a later receipt in the same step disagrees with the shadow account.
pub struct ManualReplayCursor {
    replay: std::sync::Arc<ManualExecutionReplay>,
    indices: Vec<(usize, usize, usize)>,
    cursor: usize,
    clock: Option<DateTime<Utc>>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct ManualReplayApplication {
    pub action_id: String,
    pub order_id: String,
    pub trade_id: String,
    pub observation_event_id: String,
    pub observation_sequence: u64,
    pub observed_at: DateTime<Utc>,
    pub fee_observation_event_id: String,
    pub fee_observed_at: DateTime<Utc>,
    pub executed_at: DateTime<Utc>,
    pub symbol: String,
    pub side: OrderSide,
    pub quantity: u32,
    pub quantity_after: u32,
    pub price: String,
    pub commission: Option<String>,
    pub stamp_tax: Option<String>,
    pub transfer_fee: Option<String>,
    pub source_total_fee: String,
    pub source_gross_amount: String,
    pub ledger_gross_amount: String,
    pub ledger_fees: String,
    pub cash_delta: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub corporate_adjustment: Option<ManualCorporateAdjustment>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporateAdjustment {
    pub schema: String,
    pub observed_at: DateTime<Utc>,
    pub cash_dividends_enabled: bool,
    pub dividend_cost_basis_adjustment: bool,
    #[serde(default, skip_serializing_if = "disabled_flag")]
    pub dividend_reinvestment: bool,
    pub actions: Vec<ManualCorporateActionReference>,
    pub cash_before: String,
    pub cash_after: String,
    pub corporate_cash_delta: String,
    pub positions: BTreeMap<String, ManualCorporatePositionChange>,
    pub reference_sha256: String,
    pub replayed_sha256: String,
}

fn disabled_flag(value: &bool) -> bool { !value }

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporateActionReference {
    pub date: NaiveDate,
    pub symbol: String,
    pub successor_symbol: Option<String>,
    pub share_cash: String,
    pub split_ratio: String,
    pub successor_ratio: Option<String>,
    pub successor_cash: Option<String>,
    pub sha256: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporatePositionChange {
    pub quantity_before: u32,
    pub quantity_after: u32,
    pub cost_basis_before: String,
    pub cost_basis_after: String,
}

impl ManualReplayCursor {
    pub(crate) fn frozen_source(&self) -> std::sync::Arc<ManualExecutionReplay> {
        self.replay.clone()
    }

    pub(crate) fn next_observation(&self) -> Option<ManualFillObservation<'_>> {
        self.indices.get(self.cursor).map(|&(a, o, f)| ManualFillObservation {
            action: &self.replay.actions[a], order: &self.replay.actions[a].orders[o],
            fill: &self.replay.actions[a].orders[o].fills[f],
        })
    }

    pub(crate) fn advance_next_projected<F>(
        &mut self, portfolio: &mut PortfolioState, project: F,
    ) -> Result<Option<ManualReplayApplication>, String>
    where F: FnOnce(ManualFillObservation<'_>, &mut PortfolioState) -> Result<(AppliedManualFill, ManualCorporateAdjustment), String> {
        let Some(observation) = self.next_observation() else { return Ok(None); };
        let at = observation.fill.observed_at;
        if at > self.replay.observation_cutoff || self.clock.is_some_and(|clock| at < clock) {
            return Err("manual projected observation clock violates the frozen trace".into());
        }
        let mut next = portfolio.clone();
        let (applied, adjustment) = project(observation, &mut next)?;
        let mut application = observation.application(applied)?;
        application.corporate_adjustment = Some(adjustment);
        crate::finite_serialization::validate(&application).map_err(|error| error.to_string())?;
        *portfolio = next;
        self.cursor += 1;
        self.clock = Some(at);
        Ok(Some(application))
    }

    pub fn new(replay: ManualExecutionReplay) -> Result<Self, String> {
        Self::from_shared(std::sync::Arc::new(replay))
    }

    pub fn from_shared(replay: std::sync::Arc<ManualExecutionReplay>) -> Result<Self, String> {
        replay.validate()?;
        let mut indices = Vec::new();
        for (a, action) in replay.actions.iter().enumerate() {
            for (o, order) in action.orders.iter().enumerate() {
                for f in 0..order.fills.len() {
                    indices.push((a, o, f));
                }
            }
        }
        indices.sort_by_key(|&(a, o, f)| {
            let fill = &replay.actions[a].orders[o].fills[f];
            (fill.observed_at, fill.observation_sequence)
        });
        Ok(Self {
            replay,
            indices,
            cursor: 0,
            clock: None,
        })
    }

    pub fn next_observation_at(&self) -> Option<DateTime<Utc>> {
        self.indices
            .get(self.cursor)
            .map(|&(a, o, f)| self.replay.actions[a].orders[o].fills[f].observed_at)
    }

    pub fn applied_count(&self) -> usize {
        self.cursor
    }

    pub fn advance(
        &mut self,
        at: DateTime<Utc>,
        portfolio: &mut PortfolioState,
        data: &DataSet,
        has_pending_orders: bool,
    ) -> Result<Vec<ManualReplayApplication>, String> {
        let end = self.cursor
            + self.indices[self.cursor..].iter().take_while(|&&(a, o, f)| {
                self.replay.actions[a].orders[o].fills[f].observed_at <= at
            }).count();
        self.advance_through(at, end, portfolio, data, has_pending_orders)
    }

    /// One receipt at a time lets callbacks observe the intermediate state
    /// when multiple fills share a timestamp but have distinct durable sequences.
    pub fn advance_next(
        &mut self, portfolio: &mut PortfolioState, data: &DataSet, has_pending_orders: bool,
    ) -> Result<Option<ManualReplayApplication>, String> {
        let Some(at) = self.next_observation_at() else { return Ok(None); };
        let mut applications = self.advance_through(at, self.cursor + 1, portfolio, data, has_pending_orders)?;
        Ok(applications.pop())
    }

    fn advance_through(
        &mut self, at: DateTime<Utc>, end: usize, portfolio: &mut PortfolioState,
        data: &DataSet, has_pending_orders: bool,
    ) -> Result<Vec<ManualReplayApplication>, String> {
        if at > self.replay.observation_cutoff {
            return Err("manual observation clock exceeds the frozen evidence cutoff".into());
        }
        if self.clock.is_some_and(|clock| at < clock) {
            return Err("manual observation clock moved backwards".into());
        }
        if end == self.cursor {
            self.clock = Some(at);
            return Ok(vec![]);
        }
        let mut next = portfolio.clone();
        let mut applications = Vec::with_capacity(end - self.cursor);
        for &(a, o, f) in &self.indices[self.cursor..end] {
            let action = &self.replay.actions[a];
            let order = &action.orders[o];
            let fill = &order.fills[f];
            let applied = ManualFillObservation {
                action,
                order,
                fill,
            }
            .apply(&mut next, data, has_pending_orders)?;
            applications.push(ManualReplayApplication {
                action_id: action.action_id.clone(),
                order_id: order.order_id.clone(),
                trade_id: fill.trade_id.clone(),
                observation_event_id: fill.observation_event_id.clone(),
                observation_sequence: fill.observation_sequence,
                observed_at: fill.observed_at,
                fee_observation_event_id: fill.fee_observation_event_id.clone(),
                fee_observed_at: fill.fee_observed_at,
                executed_at: fill.executed_at,
                symbol: order.symbol.clone(),
                side: order.side,
                quantity: fill.quantity,
                quantity_after: applied.quantity_after,
                price: fill.price.to_string(),
                commission: fill.commission.map(|fee| fee.to_string()),
                stamp_tax: fill.stamp_tax.map(|fee| fee.to_string()),
                transfer_fee: fill.transfer_fee.map(|fee| fee.to_string()),
                source_total_fee: fill.total_fee.to_string(),
                source_gross_amount: fill.gross_amount()?.to_string(),
                ledger_gross_amount: applied.gross.to_decimal_string(),
                ledger_fees: applied.fees.to_decimal_string(),
                cash_delta: applied.cash_delta.to_decimal_string(),
                corporate_adjustment: None,
            });
        }
        *portfolio = next;
        self.cursor = end;
        self.clock = Some(at);
        Ok(applications)
    }
}

impl ManualFillObservation<'_> {
    fn application(&self, applied: AppliedManualFill) -> Result<ManualReplayApplication, String> {
        Ok(ManualReplayApplication {
            action_id: self.action.action_id.clone(), order_id: self.order.order_id.clone(),
            trade_id: self.fill.trade_id.clone(), observation_event_id: self.fill.observation_event_id.clone(),
            observation_sequence: self.fill.observation_sequence, observed_at: self.fill.observed_at,
            fee_observation_event_id: self.fill.fee_observation_event_id.clone(), fee_observed_at: self.fill.fee_observed_at,
            executed_at: self.fill.executed_at, symbol: self.order.symbol.clone(), side: self.order.side,
            quantity: self.fill.quantity, quantity_after: applied.quantity_after, price: self.fill.price.to_string(),
            commission: self.fill.commission.map(|fee| fee.to_string()), stamp_tax: self.fill.stamp_tax.map(|fee| fee.to_string()),
            transfer_fee: self.fill.transfer_fee.map(|fee| fee.to_string()), source_total_fee: self.fill.total_fee.to_string(),
            source_gross_amount: self.fill.gross_amount()?.to_string(), ledger_gross_amount: applied.gross.to_decimal_string(),
            ledger_fees: applied.fees.to_decimal_string(), cash_delta: applied.cash_delta.to_decimal_string(), corporate_adjustment: None,
        })
    }

    pub(crate) fn apply(
        &self,
        portfolio: &mut PortfolioState,
        data: &DataSet,
        has_pending_orders: bool,
    ) -> Result<AppliedManualFill, String> {
        if has_pending_orders {
            return Err("manual observation conflicts with pending shadow orders".into());
        }
        let instrument = data
            .instrument(&self.order.symbol)
            .ok_or("manual observation instrument is absent from frozen source data")?;
        if instrument
            .dated_market_absence_reason(self.fill.trade_date)
            .is_some()
        {
            return Err("manual execution contradicts the frozen instrument lifecycle".into());
        }
        let gross = FixedMoney::from_decimal_str(&self.fill.gross_amount()?.to_string())?;
        let fees = FixedMoney::from_decimal_str(&self.fill.total_fees()?.to_string())?;
        let price = self
            .fill
            .price
            .to_f64()
            .filter(|price| price.is_finite() && *price > 0.)
            .ok_or("manual execution price cannot be represented for valuation")?;
        // This is the real observed trade price, not a fabricated quote. The
        // normal market clock remains responsible for subsequent marks.
        let cash_delta = portfolio.apply_observed_manual_fill(
            self.fill.trade_date,
            &self.order.symbol,
            self.order.side,
            self.fill.quantity,
            price,
            price,
            gross,
            fees,
        )?;
        Ok(AppliedManualFill {
            gross,
            fees,
            cash_delta,
            quantity_after: portfolio
                .position(&self.order.symbol)
                .map_or(0, |position| position.quantity),
        })
    }
}

#[cfg(test)]
mod tests;