feat(backtest): pre-soak EthFlow replay harness (COW-1078)

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "crates/nexum-engine",
+    "crates/shepherd-backtest",
     "crates/shepherd-sdk",
     "crates/shepherd-sdk-test",
     "modules/ethflow-watcher",

crates/shepherd-backtest/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "shepherd-backtest"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+description = "Offline replay harness for Shepherd modules — drives strategy code against a fixtures dump of real Sepolia events via MockHost. COW-1078."
+
+[[bin]]
+name = "shepherd-backtest"
+path = "src/main.rs"
+
+[dependencies]
+# Strategy code under test. ethflow-watcher exposes a native rlib
+# (alongside its wasm cdylib) specifically so this crate can drive
+# `strategy::on_logs` directly without an embedded runtime.
+ethflow-watcher = { path = "../../modules/ethflow-watcher" }
+shepherd-sdk = { path = "../shepherd-sdk" }
+shepherd-sdk-test = { path = "../shepherd-sdk-test" }
+
+clap = { version = "4", features = ["derive"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+hex = "0.4"
+thiserror = "2"

crates/shepherd-backtest/src/fixtures.rs

@@ -0,0 +1,113 @@
+//! JSON deserialization for the Python collector's
+//! `tools/backtest-collect/fixtures-YYYY-MM-DD.json` output.
+//!
+//! Mirrors `tools/backtest-collect/backtest_collect.py` exactly:
+//! every field present in the JSON must round-trip into a
+//! [`Fixtures`] without information loss, since the replay
+//! harness relies on raw `eth_getLogs` topics + data to reconstruct
+//! a faithful `LogView`. TWAP fields are deserialised but not yet
+//! consumed by the replay (Phase 2B); keep them on the struct so
+//! the fixture file is the canonical schema.
+
+#![allow(dead_code)]
+
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+pub struct Fixtures {
+    pub metadata: Metadata,
+    pub ethflow_orders: Vec<EthFlowFixture>,
+    pub twap_conditionals: Vec<TwapFixture>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Metadata {
+    pub collected_at: String,
+    pub chain_id: u64,
+    pub chain_name: String,
+    pub window_days: u32,
+    pub from_block: u64,
+    pub to_block: u64,
+    pub rpc_url: String,
+    pub cow_api: String,
+    pub ethflow_owner: String,
+    pub composable_cow: String,
+    #[serde(default)]
+    pub notes: Vec<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct EthFlowFixture {
+    pub uid: String,
+    pub block_number: u64,
+    pub block_timestamp: u64,
+    pub tx_hash: Option<String>,
+    pub log_index: u64,
+    pub contract: String,
+    pub sender: Option<String>,
+    pub app_data_hash: String,
+    /// Resolved app_data document fetched from
+    /// `GET /api/v1/app_data/{hash}` at collection time. `None` if
+    /// the hash 404'd (no mirror in the orderbook's app_data store).
+    pub app_data_resolved: Option<serde_json::Value>,
+    pub raw_log: RawLog,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct TwapFixture {
+    pub owner: Option<String>,
+    pub block_number: u64,
+    pub block_timestamp: u64,
+    pub tx_hash: Option<String>,
+    pub log_index: u64,
+    pub params: TwapParams,
+    pub raw_log: RawLog,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct TwapParams {
+    pub handler: String,
+    pub salt: String,
+    pub static_input: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RawLog {
+    /// Each topic is a 32-byte hex string with `0x` prefix. The
+    /// `OrderPlacement` and `ConditionalOrderCreated` events both
+    /// carry exactly 2 topics: `topic0` (the signature hash) and
+    /// `topic1` (the indexed `sender` / `owner` address).
+    pub topics: Vec<String>,
+    /// ABI-encoded payload, hex-prefixed.
+    pub data: String,
+}
+
+impl RawLog {
+    /// Decode each `0x...` topic into a 32-byte vector. The strategy
+    /// layer reads topics as `&[u8]` (right-padded address in topic1
+    /// for indexed parameters), so we preserve the byte order.
+    pub fn topics_bytes(&self) -> Result<Vec<Vec<u8>>, hex::FromHexError> {
+        self.topics
+            .iter()
+            .map(|t| hex::decode(t.strip_prefix("0x").unwrap_or(t.as_str())))
+            .collect()
+    }
+
+    /// Decode the `data` hex string.
+    pub fn data_bytes(&self) -> Result<Vec<u8>, hex::FromHexError> {
+        hex::decode(self.data.strip_prefix("0x").unwrap_or(self.data.as_str()))
+    }
+}
+
+/// Decode a `0x...` address string into the 20-byte representation
+/// the strategy uses.
+pub fn parse_address(s: &str) -> Result<[u8; 20], String> {
+    let raw = s.strip_prefix("0x").unwrap_or(s);
+    let bytes = hex::decode(raw).map_err(|e| format!("hex decode: {e}"))?;
+    if bytes.len() != 20 {
+        return Err(format!("expected 20-byte address, got {}", bytes.len()));
+    }
+    let mut out = [0u8; 20];
+    out.copy_from_slice(&bytes);
+    Ok(out)
+}

crates/shepherd-backtest/src/main.rs

@@ -0,0 +1,132 @@
+//! # shepherd-backtest
+//!
+//! Offline replay harness for Shepherd modules. Loads a fixtures
+//! JSON produced by `tools/backtest-collect/backtest_collect.py`,
+//! drives each on-chain event through the production strategy code
+//! via `shepherd_sdk_test::MockHost`, classifies the result, and
+//! emits a Markdown report at
+//! `docs/operations/backtest-reports/backtest-7d-YYYY-MM-DD.md`.
+//!
+//! ## Scope vs. the COW-1078 issue
+//!
+//! v1 covers the EthFlow lane end-to-end. The TWAP lane requires
+//! per-part eth_call walking against an archive RPC which the
+//! current public-tier endpoints refuse (see the
+//! `tools/baseline-latency` finding, COW-1031). TWAP fixtures are
+//! still loaded and counted in the report so the gap is visible,
+//! but the replay is gated on a paid endpoint (Phase 2B).
+
+use std::path::PathBuf;
+
+use clap::Parser;
+
+mod fixtures;
+mod replay;
+mod report;
+
+use fixtures::Fixtures;
+use replay::{Classification, replay_ethflow};
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "shepherd-backtest",
+    about = "Replay collected Sepolia events through production strategies (COW-1078)"
+)]
+struct Args {
+    /// Fixtures JSON produced by `tools/backtest-collect/backtest_collect.py`.
+    #[arg(long)]
+    fixtures: PathBuf,
+
+    /// Markdown report output. The default path follows the
+    /// `backtest-{window}d-{date}.md` convention the
+    /// `docs/operations/backtest-reports/` directory expects.
+    #[arg(long)]
+    out: Option<PathBuf>,
+
+    /// Acceptance threshold for the report's sign-off line. The
+    /// COW-1078 acceptance criterion is ≥ 95% of replayed events
+    /// land in `Submitted` or `RejectedExpected`; the threshold is
+    /// surfaced as a CLI flag so a soak-team override is possible
+    /// without re-editing the binary.
+    #[arg(long, default_value_t = 0.95)]
+    accept_threshold: f64,
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = Args::parse();
+    eprintln!(
+        "=== shepherd-backtest — loading {} ===",
+        args.fixtures.display()
+    );
+    let raw = std::fs::read_to_string(&args.fixtures)?;
+    let fx: Fixtures = serde_json::from_str(&raw)?;
+    eprintln!(
+        "  chain: {} (id={})  window: {}d  blocks {}..{}",
+        fx.metadata.chain_name,
+        fx.metadata.chain_id,
+        fx.metadata.window_days,
+        fx.metadata.from_block,
+        fx.metadata.to_block,
+    );
+    eprintln!("  ethflow fixtures: {}", fx.ethflow_orders.len());
+    eprintln!("  twap fixtures: {}", fx.twap_conditionals.len());
+
+    // ---- replay EthFlow ----
+    let mut outcomes = Vec::with_capacity(fx.ethflow_orders.len());
+    for (idx, order) in fx.ethflow_orders.iter().enumerate() {
+        let outcome = replay_ethflow(order, fx.metadata.chain_id);
+        if idx < 3 || idx == fx.ethflow_orders.len() - 1 {
+            eprintln!(
+                "  [{}/{}] {} {}",
+                idx + 1,
+                fx.ethflow_orders.len(),
+                outcome.class.label(),
+                outcome.uid,
+            );
+        }
+        outcomes.push(outcome);
+    }
+
+    let report_md = report::render(&fx, &outcomes, args.accept_threshold);
+    let out_path = args.out.unwrap_or_else(|| {
+        let date = fx.metadata.collected_at.split('T').next().unwrap_or("unknown");
+        PathBuf::from(format!(
+            "docs/operations/backtest-reports/backtest-{}d-{}.md",
+            fx.metadata.window_days, date
+        ))
+    });
+    if let Some(parent) = out_path.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+    std::fs::write(&out_path, &report_md)?;
+    eprintln!("\nreport written: {}", out_path.display());
+
+    // ---- summary + exit code ----
+    let total = outcomes.len();
+    let accepted = outcomes
+        .iter()
+        .filter(|o| {
+            matches!(
+                o.class,
+                Classification::Submitted | Classification::RejectedExpected(_)
+            )
+        })
+        .count();
+    let ratio = if total == 0 {
+        0.0
+    } else {
+        accepted as f64 / total as f64
+    };
+    eprintln!(
+        "summary: {}/{} ({:.1}%) Accepted+RejectedExpected (threshold {:.1}%)",
+        accepted,
+        total,
+        ratio * 100.0,
+        args.accept_threshold * 100.0,
+    );
+    if total > 0 && ratio < args.accept_threshold {
+        eprintln!("FAIL: below threshold");
+        std::process::exit(1);
+    }
+    Ok(())
+}