feat: add initial execution framework

crates/exh-kit/examples/basket_parent_allocator.rs

@@ -0,0 +1,122 @@
+use exh::{ExecutionId, ExecutionIntent, ExecutionProgress, ExecutionSnapshot};
+use exh_kit::strategy_prelude::{
+    MultiAssetCoordinator, MultiAssetLeg, MultiAssetPlan, ParentAuditSchema, ParentExecutionRunner,
+    ParentRunLifecycleEffects, ParentRunLifecycleEffectsSchemaExt, ParentRunState,
+    ParentStateSchema,
+};
+use exh_kit::testing::decimal;
+use mkt::types::{Decimal, OrderSide, Symbol};
+use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+struct BasketAllocationState {
+    rebalance_count: u64,
+    requested_parent_budget: Decimal,
+}
+
+impl ParentStateSchema for BasketAllocationState {
+    const SCHEMA: &'static str = "basket_allocator.state";
+    const VERSION: u32 = 1;
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+struct BasketAllocationAudit {
+    parent_completion: Decimal,
+    first_leg_budget: Decimal,
+    second_leg_budget: Decimal,
+}
+
+impl ParentAuditSchema for BasketAllocationAudit {
+    const EVENT_TYPE: &'static str = "basket_allocator.allocation";
+    const VERSION: u32 = 1;
+}
+
+fn intent(id: &str, symbol: &str, target: &str) -> ExecutionIntent {
+    ExecutionIntent::builder()
+        .execution_id(ExecutionId::new(id))
+        .symbol(Symbol::spot(symbol))
+        .side(OrderSide::Sell)
+        .target_quantity(decimal(target))
+        .build()
+        .expect("example intent must build")
+}
+
+fn snapshot(intent: ExecutionIntent, filled: &str) -> ExecutionSnapshot {
+    let mut snapshot = ExecutionSnapshot::from_intent(intent);
+    snapshot.progress = ExecutionProgress::zero();
+    snapshot.progress.filled_base_quantity = decimal(filled);
+    snapshot.progress.cumulative_quote_quantity = Decimal::ZERO;
+    snapshot
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let first = intent("basket-leg-btc", "BTCUSDT", "10");
+    let second = intent("basket-leg-eth", "ETHUSDT", "20");
+    let runner = ParentExecutionRunner::new(MultiAssetCoordinator::new(MultiAssetPlan::new(
+        "basket-shortfall-demo",
+        vec![
+            MultiAssetLeg::new(first.clone(), decimal("60")),
+            MultiAssetLeg::new(second.clone(), decimal("40")),
+        ],
+    )?));
+    let snapshots = vec![snapshot(first, "4"), snapshot(second, "14")];
+    let recorded_at = OffsetDateTime::UNIX_EPOCH;
+    let run = runner
+        .allocate_and_advance_with_parent_lifecycle(
+            &snapshots,
+            decimal("25"),
+            ParentRunState::new(),
+            |snapshot, allocation| {
+                std::future::ready(Ok::<_, exh::Error>(format!(
+                    "{}:{}",
+                    snapshot.intent.symbol.venue_symbol,
+                    allocation.parent_budget.normalize()
+                )))
+            },
+            |context| {
+                let [first_allocation, second_allocation] = context.allocations else {
+                    return Err(exh::Error::PolicyViolation {
+                        message: "basket example expects two leg allocations".to_owned(),
+                    });
+                };
+                let state = BasketAllocationState {
+                    rebalance_count: 1,
+                    requested_parent_budget: context.requested_parent_budget,
+                };
+                let audit = BasketAllocationAudit {
+                    parent_completion: context.parent_snapshot.completion_ratio(),
+                    first_leg_budget: first_allocation.parent_budget,
+                    second_leg_budget: second_allocation.parent_budget,
+                };
+                ParentRunLifecycleEffects::new(recorded_at)
+                    .with_typed_state(&state)?
+                    .with_typed_audit(&audit)
+            },
+        )
+        .await?;
+
+    let parent_run_state = run
+        .parent_run_state
+        .as_ref()
+        .expect("parent lifecycle run must return parent state");
+    let loaded_state = BasketAllocationState::load_parent_state(&parent_run_state.state)?
+        .expect("parent state must decode");
+    let decoded_audit =
+        BasketAllocationAudit::collect_parent_events(&parent_run_state.audit_events)?
+            .into_iter()
+            .next()
+            .expect("parent audit must decode");
+
+    println!(
+        "basket parent_completion={} requested={} first_leg_budget={} second_leg_budget={} state_rebalances={}",
+        decoded_audit.parent_completion.normalize(),
+        loaded_state.requested_parent_budget.normalize(),
+        decoded_audit.first_leg_budget.normalize(),
+        decoded_audit.second_leg_budget.normalize(),
+        loaded_state.rebalance_count
+    );
+    println!("leg_outputs={:?}", run.leg_outputs);
+    Ok(())
+}

crates/exh-kit/examples/deadline_catchup.rs

@@ -0,0 +1,141 @@
+use async_trait::async_trait;
+use exh::{AdvanceInput, AdvanceOutcome, Engine, ExecutionId, ExecutionIntent, MemoryJournal};
+use exh_kit::strategy_prelude::{
+    Algorithm, AlgorithmDecision, ChildBudgetPolicy, ChildId, ChildOrderFactory, ChildOrderStyle,
+    EvaluateContext, ParentSchedule, SignalFrame, SpotChildOrderSpec, TargetProgressView,
+    TerminalState, TimeWindow,
+};
+use exh_kit::testing::{SimulatedVenue, spot_market_fixture};
+use mkt::prelude::MarketInfo;
+use mkt::types::{Decimal, OrderSide, Symbol};
+use time::{Duration, OffsetDateTime};
+
+#[derive(Debug, Clone)]
+struct DeadlineCatchupAlgo {
+    market: MarketInfo,
+    schedule: ParentSchedule,
+}
+
+#[async_trait]
+impl Algorithm<SignalFrame> for DeadlineCatchupAlgo {
+    async fn evaluate(
+        &self,
+        context: &EvaluateContext<SignalFrame>,
+    ) -> Result<AlgorithmDecision, exh::Error> {
+        let progress = TargetProgressView::from_snapshot(&context.snapshot);
+        if progress.is_complete() {
+            return Ok(AlgorithmDecision::finishing(TerminalState::Completed));
+        }
+
+        let price = context
+            .signals
+            .last_price
+            .as_ref()
+            .ok_or_else(|| exh::Error::PolicyViolation {
+                message: "deadline strategy requires last_price".to_owned(),
+            })?
+            .price;
+        let schedule = self
+            .schedule
+            .evaluate(progress.clone(), context.observed_at)?;
+
+        let (style, price_offset, target_value, tag) =
+            if schedule.scheduled_completion < Decimal::new(8, 1) {
+                let target_value = ChildBudgetPolicy::new()
+                    .with_min_child_value(Decimal::ONE)
+                    .target_child_value(&progress, &schedule, None)?
+                    .min(Decimal::ONE);
+                (
+                    ChildOrderStyle::Passive,
+                    -Decimal::ONE,
+                    target_value,
+                    "deadline-passive",
+                )
+            } else {
+                (
+                    ChildOrderStyle::Aggressive,
+                    Decimal::ONE,
+                    progress.remaining_value,
+                    "deadline-ioc",
+                )
+            };
+
+        if target_value <= Decimal::ZERO {
+            return Ok(
+                AlgorithmDecision::paused().wake_at(context.observed_at + Duration::seconds(5))
+            );
+        }
+
+        let id = ChildId::from_sequence(&context.intent.execution_id, tag, 0)?;
+        let child = ChildOrderFactory::new(&self.market).spot_child_from_context(
+            context,
+            SpotChildOrderSpec::new(id, style, target_value, price).with_price_offset(price_offset),
+        )?;
+
+        Ok(AlgorithmDecision::running(vec![child]))
+    }
+}
+
+fn deadline_market(symbol: Symbol) -> MarketInfo {
+    spot_market_fixture(symbol)
+}
+
+fn frame(symbol: &Symbol, price: i64) -> SignalFrame {
+    SignalFrame::builder()
+        .last_price(Some(mkt::types::LastPrice::new(
+            symbol.clone(),
+            Decimal::new(price, 0),
+        )))
+        .build()
+        .expect("signal frame builder cannot fail")
+}
+
+fn next_snapshot(outcome: AdvanceOutcome) -> exh::ExecutionSnapshot {
+    match outcome {
+        AdvanceOutcome::Progressed { snapshot, .. }
+        | AdvanceOutcome::Quiescent { snapshot, .. }
+        | AdvanceOutcome::Completed { snapshot } => snapshot,
+        _ => unreachable!("example handles all current advance outcomes"),
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let symbol = Symbol::spot("SOLUSDT");
+    let start_at = OffsetDateTime::now_utc();
+    let venue = SimulatedVenue::default();
+    let journal = MemoryJournal::default();
+    let engine = Engine::new(
+        venue,
+        journal,
+        DeadlineCatchupAlgo {
+            market: deadline_market(symbol.clone()),
+            schedule: ParentSchedule::linear(TimeWindow::new(
+                start_at,
+                start_at + Duration::minutes(10),
+            )),
+        },
+    );
+
+    let intent = ExecutionIntent::builder()
+        .execution_id(ExecutionId::new("deadline-demo"))
+        .symbol(symbol.clone())
+        .side(OrderSide::Buy)
+        .target_quantity(Decimal::new(2, 0))
+        .build()?;
+    let snapshot = engine.start(intent, start_at).await?;
+    let snapshot = next_snapshot(
+        engine
+            .advance(
+                &snapshot,
+                AdvanceInput::new(start_at + Duration::minutes(9), frame(&symbol, 150), vec![]),
+            )
+            .await?,
+    );
+    println!(
+        "deadline active_children={}, remaining={}",
+        snapshot.active_children.len(),
+        snapshot.remaining_base_quantity().unwrap_or(Decimal::ZERO)
+    );
+    Ok(())
+}