feat(taker): add dual Kipseli Fermi bundles

README.md

@@ -8,7 +8,7 @@ Three scripts, same protocol family:
 | Script | What it does |
 |---|---|
 | `quoter.py` | Subscribe to Titan's pAMM state-diff WebSocket and run pre-trade quote sims against the current state override. |
-| `taker.py` / `src/main.rs` | Wrap ETH, set approvals, sign + submit swap bundles to Titan. Python and Rust ports are byte-equivalent. |
+| `taker.py` / `src/main.rs` | Wrap ETH, set approvals, sign + submit swap bundles to Titan. The Python taker also supports dual Kipseli+Fermi bundles. |
 | `contracts/KipseliGuard.sol` | Optional permissionless slippage-checking wrapper around the Kipseli pool. |
 
 All three scripts accept `--eth-rpc-url <URL>`. You need a mainnet RPC that
@@ -17,7 +17,9 @@ Geth/Reth). Public free RPCs typically don't.
 
 `taker.py` and the Rust taker additionally need a mainnet-funded signer EOA —
 its private key is read from the `PROP_AMM_TAKER_PRIVATE_KEY` environment
-variable (hex, with or without `0x`). The quoter does not need it.
+variable (hex, with or without `0x`). The quoter does not need it. Never commit
+real private keys, RPC API keys, auth tokens, or funded addresses' secret
+material; examples in this repository use placeholders only.
 
 ```sh
 export PROP_AMM_TAKER_PRIVATE_KEY=0x...
@@ -51,6 +53,13 @@ One-time setup (wrap ETH, grant ERC20 approvals to the target contract):
 python taker.py --eth-rpc-url <your-rpc-url> --contract fermi --setup-only
 ```
 
+For dual-bundle mode, setup grants approvals to both FermiSwapper and
+KipseliGuard:
+
+```sh
+python taker.py --eth-rpc-url <your-rpc-url> --dual-bundle --setup-only
+```
+
 Dry-run a single trade (no `--send`):
 
 ```sh
@@ -68,6 +77,17 @@ python taker.py --eth-rpc-url <your-rpc-url> \
     --min-priority-gwei 5 --interval-secs 3
 ```
 
+Submit one Titan bundle containing both Kipseli and Fermi swaps. Live
+`--dual-bundle --send` requires `--stream` so both swaps are pre-simulated
+against Fermi and Kipseli state overrides from the same block context:
+
+```sh
+python taker.py --eth-rpc-url <your-rpc-url> \
+    --dual-bundle --stream --send --skip-setup --once \
+    --pair weth/usdc --notional-usd 1 \
+    --min-priority-gwei 5
+```
+
 Same against Bebop:
 
 ```sh
@@ -77,8 +97,8 @@ python taker.py --eth-rpc-url <your-rpc-url> \
     --min-priority-gwei 5 --interval-secs 3
 ```
 
-Flags: `--contract {fermi|bebop|kipseli}` `--pair {weth/usdc|weth/usdt}`
-`--notional-usd N` `--slippage-bps N`
+Flags: `--contract {fermi|bebop|kipseli}` `--dual-bundle`
+`--pair {weth/usdc|weth/usdt}` `--notional-usd N` `--slippage-bps N`
 `--min-priority-gwei N` `--interval-secs N` `--reserve-eth F` `--target-weth F`
 `--send` `--once` `--setup-only` `--skip-setup` `--stream`
 `--stream-region {eu|ap|us}` `--titan-url URL`. The ETH/USDC mid used to size
@@ -89,7 +109,8 @@ tx the script prints a `[dropped]` line and stops polling that hash.
 
 ## `src/main.rs` (Rust taker)
 
-Same flags, same behavior:
+The Rust taker matches the single-contract Python behavior. The `--dual-bundle`
+mode is currently Python-only:
 
 ```sh
 cargo run --release -- --eth-rpc-url <your-rpc-url> \
@@ -103,7 +124,10 @@ cargo run --release -- --eth-rpc-url <your-rpc-url> \
 `--stream` subscribes to Titan's pAMM state-diff
 WebSocket at `wss://{eu,ap,us}.rpc.titanbuilder.xyz/ws/pamm_quote_stream`,
 pre-simulates each swap via `eth_call` with the stream's `stateOverride` +
-block overrides, and skips submission when the sim reverts.
+block overrides, and skips submission when the sim reverts. With
+`--dual-bundle`, the stream waits for Fermi and Kipseli state overrides from the
+same block context, pre-simulates both, checks aggregate token balance and gas
+budget, and posts them as one Titan bundle.
 
 ## Targets
 

src/main.rs

@@ -96,6 +96,10 @@ const KIPSELI_POOL: Address = address!("5cdbe59400cc2efdcc2b54acca4a99fe00dd588c
 const STABLE_DECIMALS: u32 = 6;
 const BEBOP_EXPIRY_SECS: u64 = 120;
 
+type StateFrame = (u64, u64, Value);
+type StateReceiver = watch::Receiver<Option<StateFrame>>;
+type StateSender = watch::Sender<Option<StateFrame>>;
+
 sol! {
     interface IERC20 {
         function balanceOf(address account) external view returns (uint256);
@@ -304,7 +308,10 @@ fn usd_to_units(usd: f64, decimals: u32) -> U256 {
 async fn fetch_binance_mid(http: &reqwest::Client) -> Result<f64> {
     let resp = http.get(BINANCE_TICKER).send().await?.error_for_status()?;
     let v: Value = resp.json().await?;
-    let price = v.get("price").and_then(|x| x.as_str()).context("missing price field")?;
+    let price = v
+        .get("price")
+        .and_then(|x| x.as_str())
+        .context("missing price field")?;
     Ok(price.parse::<f64>()?)
 }
 
@@ -409,7 +416,10 @@ async fn main() -> Result<()> {
         println!("warning: chain_id={chain_id}, expected 1 (mainnet)");
     }
 
-    println!("signer={me} chain_id={chain_id} contract={}", args.contract.label());
+    println!(
+        "signer={me} chain_id={chain_id} contract={}",
+        args.contract.label()
+    );
     print_balances(&provider, me).await?;
 
     if !args.skip_setup {
@@ -420,9 +430,12 @@ async fn main() -> Result<()> {
         }
     }
 
-    let http = reqwest::Client::builder().timeout(Duration::from_secs(5)).build()?;
-    let initial_mid =
-        fetch_binance_mid(&http).await.context("initial Binance ETH/USDC fetch")?;
+    let http = reqwest::Client::builder()
+        .timeout(Duration::from_secs(5))
+        .build()?;
+    let initial_mid = fetch_binance_mid(&http)
+        .await
+        .context("initial Binance ETH/USDC fetch")?;
     let eth_price = Arc::new(AtomicU64::new(initial_mid.to_bits()));
     tokio::spawn(refresh_binance_mid(http.clone(), eth_price.clone()));
 
@@ -435,7 +448,7 @@ async fn main() -> Result<()> {
     };
 
     let interval = Duration::from_secs(args.interval_secs);
-    let mut stream_rx: Option<watch::Receiver<Option<(u64, u64, Value)>>> = if args.stream {
+    let mut stream_rx: Option<StateReceiver> = if args.stream {
         let key = args
             .contract
             .stream_key()
@@ -452,8 +465,11 @@ async fn main() -> Result<()> {
             Pair::WethUsdc => Stable::Usdc,
             Pair::WethUsdt => Stable::Usdt,
         };
-        let dir =
-            if iter.is_multiple_of(2) { Direction::StableToWeth } else { Direction::WethToStable };
+        let dir = if iter.is_multiple_of(2) {
+            Direction::StableToWeth
+        } else {
+            Direction::WethToStable
+        };
         let eth_price_usd = f64::from_bits(eth_price.load(Ordering::Relaxed));
         if let Err(e) = trade_once(
             &provider,
@@ -495,7 +511,9 @@ async fn read_view<P: Provider, C: SolCall>(
     to: Address,
     call: C,
 ) -> Result<C::Return> {
-    let req = TransactionRequest::default().with_to(to).with_input(call.abi_encode());
+    let req = TransactionRequest::default()
+        .with_to(to)
+        .with_input(call.abi_encode());
     let bytes = provider.call(req).await?;
     Ok(C::abi_decode_returns(&bytes)?)
 }
@@ -505,20 +523,35 @@ async fn setup<P: Provider>(provider: &P, me: Address, args: &Cli) -> Result<()>
     let label = args.contract.label();
     let weth_target = U256::from(100u128) * U256::from(10u128.pow(18));
     let stable_target = U256::from(100_000u128) * U256::from(10u128.pow(STABLE_DECIMALS));
-    for (token, sym, target) in
-        [(WETH, "WETH", weth_target), (USDC, "USDC", stable_target), (USDT, "USDT", stable_target)]
-    {
-        let allowance =
-            read_view(provider, token, IERC20::allowanceCall { owner: me, spender }).await?;
+    for (token, sym, target) in [
+        (WETH, "WETH", weth_target),
+        (USDC, "USDC", stable_target),
+        (USDT, "USDT", stable_target),
+    ] {
+        let allowance = read_view(
+            provider,
+            token,
+            IERC20::allowanceCall { owner: me, spender },
+        )
+        .await?;
         if allowance >= target {
             println!("[setup] {sym} already approved");
             continue;
         }
         println!("[setup] approving {sym} -> {label}");
-        let calldata = IERC20::approveCall { spender, value: target }.abi_encode();
-        let req = TransactionRequest::default().with_to(token).with_input(calldata);
+        let calldata = IERC20::approveCall {
+            spender,
+            value: target,
+        }
+        .abi_encode();
+        let req = TransactionRequest::default()
+            .with_to(token)
+            .with_input(calldata);
         let receipt = provider.send_transaction(req).await?.get_receipt().await?;
-        println!("[setup]   {sym} approve mined block={:?}", receipt.block_number);
+        println!(
+            "[setup]   {sym} approve mined block={:?}",
+            receipt.block_number
+        );
     }
 
     let weth_bal = read_view(provider, WETH, IERC20::balanceOfCall { account: me }).await?;
@@ -558,25 +591,34 @@ async fn trade_once<P: Provider>(
     me: Address,
     stable: Stable,
     dir: Direction,
-    stream_rx: Option<&mut watch::Receiver<Option<(u64, u64, Value)>>>,
+    stream_rx: Option<&mut StateReceiver>,
     eth_price_usd: f64,
     mon_tx: Option<&mpsc::UnboundedSender<(B256, String, u64)>>,
 ) -> Result<()> {
     let nonce = provider.get_transaction_count(me).pending().await?;
     let fees = provider.estimate_eip1559_fees().await?;
-    let max_priority = fees.max_priority_fee_per_gas.max(args.min_priority_gwei * 1_000_000_000);
+    let max_priority = fees
+        .max_priority_fee_per_gas
+        .max(args.min_priority_gwei * 1_000_000_000);
     let max_fee = fees.max_fee_per_gas.max(max_priority * 3);
 
     let stable_units = usd_to_units(args.notional_usd, STABLE_DECIMALS);
-    let weth_units =
-        U256::from((args.notional_usd * 1e18 / eth_price_usd) as u128);
+    let weth_units = U256::from((args.notional_usd * 1e18 / eth_price_usd) as u128);
     let (token_in, token_out, amount_in, expected_out, label) = match dir {
-        Direction::StableToWeth => {
-            (stable.addr(), WETH, stable_units, weth_units, format!("{}->WETH", stable.sym()))
-        }
-        Direction::WethToStable => {
-            (WETH, stable.addr(), weth_units, stable_units, format!("WETH->{}", stable.sym()))
-        }
+        Direction::StableToWeth => (
+            stable.addr(),
+            WETH,
+            stable_units,
+            weth_units,
+            format!("{}->WETH", stable.sym()),
+        ),
+        Direction::WethToStable => (
+            WETH,
+            stable.addr(),
+            weth_units,
+            stable_units,
+            format!("WETH->{}", stable.sym()),
+        ),
     };
     let bps = U256::from(10_000);
     let slip = args.slippage_bps as u64;
@@ -604,8 +646,10 @@ async fn trade_once<P: Provider>(
             .abi_encode()
         }
         Contract::Bebop => {
-            let pending =
-                provider.get_block(BlockId::pending()).await?.context("no pending block")?;
+            let pending = provider
+                .get_block(BlockId::pending())
+                .await?
+                .context("no pending block")?;
             let expiry = U256::from(pending.header.timestamp + BEBOP_EXPIRY_SECS);
             IBebop::swapCall {
                 tokenIn: token_in,
@@ -642,7 +686,10 @@ async fn trade_once<P: Provider>(
             "time": format!("0x{timestamp_secs:x}"),
         });
         let params = json!([call, "latest", state_override, block_overrides]);
-        match provider.raw_request::<_, Bytes>("eth_call".into(), params).await {
+        match provider
+            .raw_request::<_, Bytes>("eth_call".into(), params)
+            .await
+        {
             Ok(_) => println!("[trade] state-override sim ok @ block {block_number}"),
             Err(e) => {
                 println!("[trade] state-override sim reverts, skipping: {e}");
@@ -676,7 +723,11 @@ async fn trade_once<P: Provider>(
         "params": [{ "txs": [raw_tx], "blockNumber": "0x0" }],
     });
     let resp = http.post(titan_url).json(&body).send().await?;
-    println!("[trade] titan status={} body={}", resp.status(), resp.text().await?);
+    println!(
+        "[trade] titan status={} body={}",
+        resp.status(),
+        resp.text().await?
+    );
     if let Some(mt) = mon_tx {
         let _ = mt.send((tx_hash, label.clone(), nonce));
     }
@@ -711,11 +762,7 @@ fn build_signed_tx(
     Ok((format!("0x{}", hex::encode(&raw)), tx_hash))
 }
 
-async fn run_state_stream(
-    region: StreamRegion,
-    contract: Address,
-    tx: watch::Sender<Option<(u64, u64, Value)>>,
-) {
+async fn run_state_stream(region: StreamRegion, contract: Address, tx: StateSender) {
     let url = region.ws_url();
     let contract_lc = format!("{contract:#x}").to_lowercase();
     loop {
@@ -725,7 +772,9 @@ async fn run_state_stream(
             let (_, mut reader) = ws.split();
             while let Some(msg) = reader.next().await {
                 let Message::Text(text) = msg? else { continue };
-                let Ok(v) = serde_json::from_str::<Value>(&text) else { continue };
+                let Ok(v) = serde_json::from_str::<Value>(&text) else {
+                    continue;
+                };
                 let Some(obj) = v.as_object() else { continue };
                 let Some(block_number) = obj.get("blockNumber").and_then(|b| b.as_u64()) else {
                     continue;
@@ -735,8 +784,8 @@ async fn run_state_stream(
                     .and_then(|t| t.as_u64())
                     .map(|ns| ns / 1_000_000_000)
                     .unwrap_or(0);
-                if let Some(entry) = obj.iter().find(|(k, _)| k.to_lowercase() == contract_lc) &&
-                    let Some(so) = entry.1.get("stateOverride")
+                if let Some(entry) = obj.iter().find(|(k, _)| k.to_lowercase() == contract_lc)
+                    && let Some(so) = entry.1.get("stateOverride")
                 {
                     let _ = tx.send(Some((block_number, timestamp_secs, so.clone())));
                 }