af_backtest/

lib.rs

//! `af-backtest` — deterministic historical replay for product adapters.
//!
//! This crate owns the reusable mechanics only: ordered historical ingress,
//! state transitions, decision traces, value curves, metrics, and bounded
//! report projection. Domain event types, accounting, execution semantics,
//! data providers, and concrete strategies stay in product/extension crates.

#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]

use serde::{Deserialize, Serialize};

/// Durable persistence port for definitions, snapshots, runs and bounded reports.
#[async_trait::async_trait]
pub trait BacktestStore: Send + Sync {
    /// Save or replace a tenant-owned definition.
    async fn save_definition(
        &self,
        context: &af_context::RequestContext,
        definition: BacktestDefinition,
    ) -> Result<(), BacktestStoreError>;
    /// Freeze definition and input under an owner-scoped key; retries return the original Run.
    async fn start_run(
        &self,
        context: &af_context::RequestContext,
        definition_id: &af_context::BacktestDefinitionId,
        input_snapshot: serde_json::Value,
        idempotency_key: &str,
    ) -> Result<af_context::BacktestRunId, BacktestStoreError>;
    /// Read a Run and its immutable snapshots, enforcing tenant and subject ownership.
    async fn get_run(
        &self,
        context: &af_context::RequestContext,
        run_id: &af_context::BacktestRunId,
    ) -> Result<BacktestRun, BacktestStoreError>;
    /// Atomically cancel a running Run; repeating cancellation is safe.
    async fn cancel_run(
        &self,
        context: &af_context::RequestContext,
        run_id: &af_context::BacktestRunId,
    ) -> Result<(), BacktestStoreError>;
    /// Complete a run with a report below `max_bytes`.
    async fn complete_run(
        &self,
        context: &af_context::RequestContext,
        run_id: &af_context::BacktestRunId,
        report: serde_json::Value,
        max_bytes: usize,
    ) -> Result<(), BacktestStoreError>;
    /// Preserve a terminal failure for recovery diagnostics.
    async fn fail_run(
        &self,
        context: &af_context::RequestContext,
        run_id: &af_context::BacktestRunId,
        message: &str,
    ) -> Result<(), BacktestStoreError>;
}

/// Tenant-owned deterministic replay definition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BacktestDefinition {
    /// Stable definition identity.
    pub id: af_context::BacktestDefinitionId,
    /// Product-neutral display name.
    pub name: String,
    /// Immutable model implementation version.
    pub model_version: String,
    /// Bounded model configuration.
    pub config: serde_json::Value,
}

/// Durable replay record; computation resumes from these snapshots, never a mutable definition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BacktestRun {
    /// Stable Run identity.
    pub id: af_context::BacktestRunId,
    /// Frozen definition, absent only for legacy runs whose original version is unknowable.
    pub definition: Option<BacktestDefinition>,
    /// Immutable replay inputs.
    pub input_snapshot: serde_json::Value,
    /// Current terminal or running state.
    pub status: BacktestRunStatus,
    /// Bounded result when completed.
    pub report: Option<serde_json::Value>,
    /// Failure diagnostic when failed.
    pub error_message: Option<String>,
}

/// Durable replay lifecycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BacktestRunStatus {
    /// Computation may run or be recovered.
    Running,
    /// Immutable result was committed.
    Completed,
    /// Computation failed.
    Failed,
    /// Owner cancelled computation; no later result may overwrite it.
    Cancelled,
}

/// Failure owned by the durable replay store.
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum BacktestStoreError {
    /// Store dependency is unavailable.
    #[error("backtest store unavailable: {0}")]
    Unavailable(String),
    /// Requested record does not exist in the tenant.
    #[error("backtest record not found")]
    NotFound,
    /// An owned Run exists but rejects this state transition or result.
    #[error("backtest run state conflicts with requested mutation")]
    Conflict,
    /// Input is invalid.
    #[error("invalid backtest state: {0}")]
    Invalid(String),
    /// Report exceeds its configured serialized size cap.
    #[error("report size {actual} exceeds maximum {maximum}")]
    ReportTooLarge {
        /// Serialized report bytes.
        actual: usize,
        /// Configured maximum bytes.
        maximum: usize,
    },
}

/// Product-supplied deterministic replay behavior.
pub trait BacktestModel {
    /// Typed product error returned by decisions and transitions.
    type Error: std::error::Error + Send + Sync + 'static;
    /// Historical input event type.
    type Event;
    /// Replay state carried between events.
    type State: Clone;
    /// Decision produced by the model.
    type Action: Clone;

    /// State before the first event.
    fn initial_state(&self) -> Self::State;
    /// Starting value used for return and drawdown.
    fn initial_value(&self, state: &Self::State) -> f64;
    /// Monotonic timestamp of an event.
    fn timestamp(&self, event: &Self::Event) -> i64;

    /// Produce decisions from the current event and pre-transition state.
    fn decide(
        &mut self,
        event: &Self::Event,
        state: &Self::State,
    ) -> Result<Vec<Self::Action>, Self::Error>;

    /// Apply one decision. The model owns domain-specific transition rules.
    fn apply(
        &mut self,
        event: &Self::Event,
        action: &Self::Action,
        state: &mut Self::State,
    ) -> Result<(), Self::Error>;

    /// Mark the state after all actions for the event have been applied.
    fn value(&self, event: &Self::Event, state: &Self::State) -> f64;
}

/// Replay limits.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct BacktestConfig {
    /// Hard resource bound; inputs above it fail before executing any model code.
    pub max_events: usize,
    /// Capture one detailed decision step every N events. Curve points remain full.
    pub capture_every: usize,
}

impl Default for BacktestConfig {
    fn default() -> Self {
        Self {
            max_events: 1_000_000,
            capture_every: 1,
        }
    }
}

/// Value after one event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ValuePoint {
    /// Event timestamp.
    pub timestamp: i64,
    /// Value at this point.
    pub value: f64,
}

/// Captured decision trace for one event.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ReplayStep<A> {
    /// Index of the event in the input.
    pub event_index: usize,
    /// Event timestamp.
    pub timestamp: i64,
    /// Decisions applied for this event.
    pub actions: Vec<A>,
    /// Value at this point.
    pub value: f64,
}

/// Aggregate metrics for a replay.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BacktestSummary {
    /// Events replayed.
    pub events_processed: usize,
    /// Decisions applied.
    pub actions_applied: usize,
    /// Value before the first event.
    pub initial_value: f64,
    /// Value after the last event.
    pub final_value: f64,
    /// `final_value / initial_value - 1`.
    pub total_return: f64,
    /// Largest peak-to-trough decline as a fraction of the peak.
    pub max_drawdown: f64,
}

/// Full replay output.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BacktestReport<S, A> {
    /// Short human-readable summary.
    pub summary: BacktestSummary,
    /// State after the last event.
    pub final_state: S,
    /// Value after every event.
    pub value_curve: Vec<ValuePoint>,
    /// Captured decision traces.
    pub steps: Vec<ReplayStep<A>>,
}

/// Caps for projecting a report to chat or API consumers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ProjectionPolicy {
    /// Maximum curve points kept; evenly down-sampled beyond this.
    pub max_curve_points: usize,
    /// Upper bound on model steps per Run.
    pub max_steps: usize,
}

impl Default for ProjectionPolicy {
    fn default() -> Self {
        Self {
            max_curve_points: 80,
            max_steps: 24,
        }
    }
}

/// A transport/chat-safe projection of a potentially large full report.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BacktestProjection<A> {
    /// Short human-readable summary.
    pub summary: BacktestSummary,
    /// Down-sampled value curve.
    pub value_curve: Vec<ValuePoint>,
    /// Capped decision traces.
    pub steps: Vec<ReplayStep<A>>,
    /// Curve points before capping.
    pub total_curve_points: usize,
    /// Steps before capping.
    pub total_steps: usize,
}

impl<S, A: Clone> BacktestReport<S, A> {
    /// Project a report under `policy` without losing the summary.
    pub fn project(
        &self,
        policy: ProjectionPolicy,
    ) -> Result<BacktestProjection<A>, BacktestError> {
        if policy.max_curve_points < 2 {
            return Err(BacktestError::InvalidProjection(
                "max_curve_points must be at least 2".into(),
            ));
        }
        if policy.max_steps == 0 {
            return Err(BacktestError::InvalidProjection(
                "max_steps must be greater than 0".into(),
            ));
        }
        Ok(BacktestProjection {
            summary: self.summary.clone(),
            value_curve: sample_evenly(&self.value_curve, policy.max_curve_points),
            steps: sample_evenly(&self.steps, policy.max_steps),
            total_curve_points: self.value_curve.len(),
            total_steps: self.steps.len(),
        })
    }
}

/// Failure while validating input or replaying.
#[derive(Debug, thiserror::Error, PartialEq)]
pub enum BacktestError {
    /// Backtest requires at least one historical event.
    #[error("backtest requires at least one historical event")]
    EmptyEvents,
    /// Max_events must be greater than 0.
    #[error("max_events must be greater than 0")]
    InvalidEventLimit,
    /// Capture_every must be greater than 0.
    #[error("capture_every must be greater than 0")]
    InvalidCaptureRate,
    /// More events than `max_events`.
    #[error("event count {actual} exceeds configured maximum {maximum}")]
    EventLimitExceeded {
        /// Events supplied.
        actual: usize,
        /// Configured `max_events`.
        maximum: usize,
    },
    /// Initial value must be finite and greater than zero, got.
    #[error("initial value must be finite and greater than zero, got {0}")]
    InvalidInitialValue(f64),
    /// Event timestamps are out of order at index `index`: `previous` > `current`.
    #[error("event timestamps are out of order at index {index}: {previous} > {current}")]
    NonMonotonicTimestamp {
        /// Index of the event being applied.
        index: usize,
        /// Timestamp of the previous event.
        previous: i64,
        /// Timestamp of the offending event.
        current: i64,
    },
    /// The model failed to decide on an event.
    #[error("decision failed at event {index}: {message}")]
    Decision {
        /// Index of the event being decided.
        index: usize,
        /// Model-reported failure.
        message: String,
    },
    /// State transition failed at event `index`, action `action_index`: `message`.
    #[error("state transition failed at event {index}, action {action_index}: {message}")]
    Transition {
        /// Index of the event being applied.
        index: usize,
        /// Index of the action within that event.
        action_index: usize,
        /// Model-reported failure.
        message: String,
    },
    /// The model produced NaN or infinity.
    #[error("model returned a non-finite value at event {index}: {value}")]
    NonFiniteValue {
        /// Index of the offending event.
        index: usize,
        /// The non-finite value.
        value: f64,
    },
    /// Invalid projection policy.
    #[error("invalid projection policy: {0}")]
    InvalidProjection(String),
    /// Replay observed cancellation between events.
    #[error("backtest cancelled")]
    Cancelled,
    /// Replay exceeded its caller-provided deadline.
    #[error("backtest deadline exceeded")]
    DeadlineExceeded,
}

/// Execute a deterministic replay. Events with the same timestamp are allowed
/// and preserve input order; decreasing timestamps fail closed.
pub fn run_backtest<M: BacktestModel>(
    model: &mut M,
    events: &[M::Event],
    config: BacktestConfig,
) -> Result<BacktestReport<M::State, M::Action>, BacktestError> {
    if events.is_empty() {
        return Err(BacktestError::EmptyEvents);
    }
    if config.max_events == 0 {
        return Err(BacktestError::InvalidEventLimit);
    }
    if config.capture_every == 0 {
        return Err(BacktestError::InvalidCaptureRate);
    }
    if events.len() > config.max_events {
        return Err(BacktestError::EventLimitExceeded {
            actual: events.len(),
            maximum: config.max_events,
        });
    }

    let timestamps = events
        .iter()
        .map(|event| model.timestamp(event))
        .collect::<Vec<_>>();
    for (index, pair) in timestamps.windows(2).enumerate() {
        if pair[1] < pair[0] {
            return Err(BacktestError::NonMonotonicTimestamp {
                index: index + 1,
                previous: pair[0],
                current: pair[1],
            });
        }
    }

    let mut state = model.initial_state();
    let initial_value = model.initial_value(&state);
    if !initial_value.is_finite() || initial_value <= 0.0 {
        return Err(BacktestError::InvalidInitialValue(initial_value));
    }

    let mut value_curve = Vec::with_capacity(events.len());
    let mut steps = Vec::with_capacity(events.len().div_ceil(config.capture_every));
    let mut actions_applied = 0;

    for (index, event) in events.iter().enumerate() {
        let timestamp = timestamps[index];

        let actions = model
            .decide(event, &state)
            .map_err(|error| BacktestError::Decision {
                index,
                message: error.to_string(),
            })?;
        for (action_index, action) in actions.iter().enumerate() {
            model
                .apply(event, action, &mut state)
                .map_err(|error| BacktestError::Transition {
                    index,
                    action_index,
                    message: error.to_string(),
                })?;
            actions_applied += 1;
        }

        let value = model.value(event, &state);
        if !value.is_finite() {
            return Err(BacktestError::NonFiniteValue { index, value });
        }
        value_curve.push(ValuePoint { timestamp, value });
        if index % config.capture_every == 0 || index + 1 == events.len() {
            steps.push(ReplayStep {
                event_index: index,
                timestamp,
                actions,
                value,
            });
        }
    }

    let final_value = value_curve
        .last()
        .map_or(initial_value, |point| point.value);
    let summary = BacktestSummary {
        events_processed: events.len(),
        actions_applied,
        initial_value,
        final_value,
        total_return: (final_value / initial_value) - 1.0,
        max_drawdown: max_drawdown(initial_value, &value_curve),
    };
    Ok(BacktestReport {
        summary,
        final_state: state,
        value_curve,
        steps,
    })
}

/// Execute a deterministic replay while observing cancellation and deadline
/// between input events.
pub fn run_backtest_with_control<M: BacktestModel>(
    model: &mut M,
    events: &[M::Event],
    config: BacktestConfig,
    cancellation: &tokio_util::sync::CancellationToken,
    deadline: std::time::Instant,
) -> Result<BacktestReport<M::State, M::Action>, BacktestError> {
    if cancellation.is_cancelled() {
        return Err(BacktestError::Cancelled);
    }
    if std::time::Instant::now() >= deadline {
        return Err(BacktestError::DeadlineExceeded);
    }
    let mut checked = CheckedModel {
        inner: model,
        cancellation,
        deadline,
    };
    run_backtest(&mut checked, events, config).map_err(|error| match &error {
        BacktestError::Decision { message, .. } | BacktestError::Transition { message, .. }
            if message == "backtest cancelled" =>
        {
            BacktestError::Cancelled
        }
        BacktestError::Decision { message, .. } | BacktestError::Transition { message, .. }
            if message == "backtest deadline exceeded" =>
        {
            BacktestError::DeadlineExceeded
        }
        _ => error,
    })
}

struct CheckedModel<'a, M> {
    inner: &'a mut M,
    cancellation: &'a tokio_util::sync::CancellationToken,
    deadline: std::time::Instant,
}

impl<M: BacktestModel> BacktestModel for CheckedModel<'_, M> {
    type Error = ControlledModelError<M::Error>;
    type Event = M::Event;
    type State = M::State;
    type Action = M::Action;

    fn initial_state(&self) -> Self::State {
        self.inner.initial_state()
    }
    fn initial_value(&self, state: &Self::State) -> f64 {
        self.inner.initial_value(state)
    }
    fn timestamp(&self, event: &Self::Event) -> i64 {
        self.inner.timestamp(event)
    }
    fn decide(
        &mut self,
        event: &Self::Event,
        state: &Self::State,
    ) -> Result<Vec<Self::Action>, Self::Error> {
        if self.cancellation.is_cancelled() {
            return Err(ControlledModelError::Cancelled);
        }
        if std::time::Instant::now() >= self.deadline {
            return Err(ControlledModelError::DeadlineExceeded);
        }
        self.inner
            .decide(event, state)
            .map_err(ControlledModelError::Model)
    }
    fn apply(
        &mut self,
        event: &Self::Event,
        action: &Self::Action,
        state: &mut Self::State,
    ) -> Result<(), Self::Error> {
        self.inner
            .apply(event, action, state)
            .map_err(ControlledModelError::Model)
    }
    fn value(&self, event: &Self::Event, state: &Self::State) -> f64 {
        self.inner.value(event, state)
    }
}

#[derive(Debug, thiserror::Error)]
enum ControlledModelError<E: std::error::Error> {
    #[error("backtest cancelled")]
    Cancelled,
    #[error("backtest deadline exceeded")]
    DeadlineExceeded,
    #[error(transparent)]
    Model(E),
}

fn max_drawdown(initial_value: f64, curve: &[ValuePoint]) -> f64 {
    let mut peak = initial_value;
    let mut maximum = 0.0_f64;
    for point in curve {
        peak = peak.max(point.value);
        if peak > 0.0 {
            maximum = maximum.max((peak - point.value) / peak);
        }
    }
    maximum
}

fn sample_evenly<T: Clone>(items: &[T], maximum: usize) -> Vec<T> {
    if items.len() <= maximum {
        return items.to_vec();
    }
    if maximum == 1 {
        return vec![items[items.len() - 1].clone()];
    }
    (0..maximum)
        .map(|slot| {
            let index = slot * (items.len() - 1) / (maximum - 1);
            items[index].clone()
        })
        .collect()
}