From 25ba4a5aca5617889a001462a30fdb174f1ddc20 Mon Sep 17 00:00:00 2001 From: Louis Ponet Date: Wed, 13 May 2026 16:41:29 +0100 Subject: [PATCH] feat(taker): add dual Kipseli Fermi bundles Add a Python taker mode that builds one Titan bundle containing consecutive Kipseli and Fermi swaps. The dual mode reuses the existing swap sizing, calldata, signing, stream simulation, and bundle submission paths while requiring stream gating for live sends and checking aggregate balance/gas budget before posting.\n\nDocument the new setup and live-send examples with public-safe placeholders, clarify that dual mode is Python-only, and keep the Rust taker passing clippy with formatter and type-complexity cleanup. --- README.md | 36 +++- src/main.rs | 127 +++++++++----- taker.py | 491 ++++++++++++++++++++++++++++++++++++++++++++-------- 3 files changed, 535 insertions(+), 119 deletions(-) diff --git a/README.md b/README.md index aac589b..a89bf37 100644 --- a/README.md +++ b/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 `. 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 --contract fermi --setup-only ``` +For dual-bundle mode, setup grants approvals to both FermiSwapper and +KipseliGuard: + +```sh +python taker.py --eth-rpc-url --dual-bundle --setup-only +``` + Dry-run a single trade (no `--send`): ```sh @@ -68,6 +77,17 @@ python taker.py --eth-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 \ + --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 \ --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 \ @@ -103,7 +124,10 @@ cargo run --release -- --eth-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 diff --git a/src/main.rs b/src/main.rs index 81e9512..05c3a60 100644 --- a/src/main.rs +++ b/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>; +type StateSender = watch::Sender>; + 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 { 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::()?) } @@ -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>> = if args.stream { + let mut stream_rx: Option = 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( to: Address, call: C, ) -> Result { - 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(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( me: Address, stable: Stable, dir: Direction, - stream_rx: Option<&mut watch::Receiver>>, + 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( .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( "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( "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>, -) { +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::(&text) else { continue }; + let Ok(v) = serde_json::from_str::(&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()))); } diff --git a/taker.py b/taker.py index c0b015d..b14ff82 100644 --- a/taker.py +++ b/taker.py @@ -19,6 +19,11 @@ python taker.py --eth-rpc-url "$ETH_RPC_URL" \\ --contract kipseli --setup-only +Dual-bundle setup grants approvals to both FermiSwapper and KipseliGuard:: + + python taker.py --eth-rpc-url "$ETH_RPC_URL" \\ + --dual-bundle --setup-only + Single-shot dry-run (no ``--send``) — prints the calldata and tx hash that would be submitted but does not POST to Titan:: @@ -40,6 +45,15 @@ --pair weth/usdc --notional-usd 2 \\ --min-priority-gwei 5 --interval-secs 3 +One bundle containing Kipseli + Fermi swaps. Live ``--dual-bundle --send`` +requires ``--stream`` so both swaps are pre-simulated against matching block +state overrides:: + + python taker.py --eth-rpc-url "$ETH_RPC_URL" \ + --dual-bundle --stream --send --skip-setup --once \ + --pair weth/usdc --notional-usd 1 \ + --min-priority-gwei 5 + Same against Bebop:: python taker.py --eth-rpc-url "$ETH_RPC_URL" \\ @@ -386,24 +400,31 @@ async def send_with_account(w3: AsyncWeb3, account, tx: dict) -> dict: async def setup(w3: AsyncWeb3, account, args) -> None: me = account.address - spender = args.contract.address - label = args.contract.label + spenders = ( + ( + (FERMI_SWAPPER, Contract.FERMI.label), + (KIPSELI_GUARD, Contract.KIPSELI.label), + ) + if args.dual_bundle + else ((args.contract.address, args.contract.label),) + ) weth_target = 100 * 10**18 stable_target = 100_000 * 10**STABLE_DECIMALS - for token, sym, target in ( - (WETH, "WETH", weth_target), - (USDC, "USDC", stable_target), - (USDT, "USDT", stable_target), - ): - allowance = await read_uint256(w3, token, cd_allowance(me, spender)) - if allowance >= target: - print(f"[setup] {sym} already approved") - continue - print(f"[setup] approving {sym} -> {label}") - receipt = await send_with_account( - w3, account, {"to": token, "data": cd_approve(spender, target)} - ) - print(f"[setup] {sym} approve mined block={receipt.get('blockNumber')}") + for spender, label in spenders: + for token, sym, target in ( + (WETH, "WETH", weth_target), + (USDC, "USDC", stable_target), + (USDT, "USDT", stable_target), + ): + allowance = await read_uint256(w3, token, cd_allowance(me, spender)) + if allowance >= target: + print(f"[setup] {sym} already approved -> {label}") + continue + print(f"[setup] approving {sym} -> {label}") + receipt = await send_with_account( + w3, account, {"to": token, "data": cd_approve(spender, target)} + ) + print(f"[setup] {sym} approve mined block={receipt.get('blockNumber')}") weth_bal = await read_uint256(w3, WETH, cd_balance_of(me)) target_wei = int(args.target_weth * 1e18) @@ -453,22 +474,7 @@ def build_signed_tx( return "0x" + raw.hex(), "0x" + keccak(raw).hex() -async def trade_once( - w3: AsyncWeb3, - http: aiohttp.ClientSession, - account, - args, - chain_id: int, - iter_: int, - state: Optional[StateStream], - watcher: Optional[TxMonitor], -) -> None: - me = account.address - nonce = await w3.eth.get_transaction_count(me, "pending") - max_fee, max_priority = await estimate_eip1559(w3) - max_priority = max(max_priority, args.min_priority_gwei * 1_000_000_000) - max_fee = max(max_fee, max_priority * 3) - +def trade_params(args, iter_: int): stable = Stable.USDC if args.pair is Pair.WETH_USDC else Stable.USDT direction = ( Direction.STABLE_TO_WETH if iter_ % 2 == 0 else Direction.WETH_TO_STABLE @@ -490,8 +496,36 @@ async def trade_once( slip_lo = max(10_000 - slip, 0) slip_hi = 10_000 + slip min_out = expected_out * slip_lo // bps + return ( + direction, + token_in, + token_out, + amount_in, + stable_units, + weth_units, + slip_lo, + slip_hi, + min_out, + label, + ) + - if args.contract is Contract.FERMI: +async def build_swap_calldata( + w3: AsyncWeb3, + contract: Contract, + direction: Direction, + token_in: str, + token_out: str, + amount_in: int, + min_out: int, + stable_units: int, + weth_units: int, + slip_lo: int, + slip_hi: int, + recipient: str, +) -> bytes: + bps = 10_000 + if contract is Contract.FERMI: # Fermi's amountSpecified is signed: positive = exact tokenIn, negative # = exact tokenOut. Trade is always denominated in stable units, so flip # the sign on WETH-input legs to mean "exact stable output". @@ -501,36 +535,151 @@ async def trade_once( else: amount_specified = -stable_units amount_check = weth_units * slip_hi // bps - calldata = cd_fermi_swap(token_in, token_out, amount_specified, amount_check, me) - elif args.contract is Contract.BEBOP: + return cd_fermi_swap( + token_in, token_out, amount_specified, amount_check, recipient + ) + if contract is Contract.BEBOP: pending = await w3.eth.get_block("pending") expiry = pending["timestamp"] + BEBOP_EXPIRY_SECS - calldata = cd_bebop_swap(token_in, token_out, amount_in, min_out, expiry) - else: - calldata = cd_kipseli_swap(token_in, amount_in, token_out, min_out) + return cd_bebop_swap(token_in, token_out, amount_in, min_out, expiry) + return cd_kipseli_swap(token_in, amount_in, token_out, min_out) - if state is not None: + +async def next_state_frame(state: StateStream): + await state.changed() + frame = state.borrow_and_update() + while frame is None: await state.changed() frame = state.borrow_and_update() - while frame is None: - await state.changed() - frame = state.borrow_and_update() - block_number, timestamp_secs, state_override = frame - call = { - "from": me, - "to": args.contract.address, - "data": "0x" + calldata.hex(), - } - block_overrides = { - "number": hex(block_number), - "time": hex(timestamp_secs), - } - params = [call, "latest", state_override, block_overrides] - response = await w3.provider.make_request("eth_call", params) - if response.get("error"): - print(f"[trade] state-override sim reverts, skipping: {response['error']}") + return frame + + +async def state_sim_ok( + w3: AsyncWeb3, + me: str, + to: str, + calldata: bytes, + frame, + prefix: str, +) -> bool: + block_number, timestamp_secs, state_override = frame + call = { + "from": me, + "to": to, + "data": "0x" + calldata.hex(), + } + block_overrides = { + "number": hex(block_number), + "time": hex(timestamp_secs), + } + params = [call, "latest", state_override, block_overrides] + response = await w3.provider.make_request("eth_call", params) + if response.get("error"): + print(f"{prefix} state-override sim reverts, skipping: {response['error']}") + return False + print(f"{prefix} state-override sim ok @ block {block_number}") + return True + + +async def dual_preflight_ok( + w3: AsyncWeb3, + me: str, + direction: Direction, + token_in: str, + stable_units: int, + weth_units: int, + slip_hi: int, + max_fee: int, +) -> bool: + gas_budget = (Contract.KIPSELI.gas_limit + Contract.FERMI.gas_limit) * max_fee + eth_balance = await w3.eth.get_balance(me) + if eth_balance < gas_budget: + print(f"[dual] ETH {eth_balance} < gas budget {gas_budget}, skipping") + return False + if direction is Direction.STABLE_TO_WETH: + stable_balance = await read_uint256(w3, token_in, cd_balance_of(me)) + needed = stable_units * 2 + if stable_balance < needed: + print( + f"[dual] stable balance {stable_balance} < dual input {needed}, skipping" + ) + return False + else: + weth_balance = await read_uint256(w3, WETH, cd_balance_of(me)) + fermi_max_input = weth_units * slip_hi // 10_000 + needed = weth_units + fermi_max_input + if weth_balance < needed: + print( + f"[dual] WETH balance {weth_balance} < dual max input {needed}, skipping" + ) + return False + return True + + +async def send_titan_bundle( + http: aiohttp.ClientSession, titan_url: str, raw_txs: list, prefix: str +) -> None: + # Titan accepts `blockNumber: 0x0` as "include in any block within validity". + body = { + "jsonrpc": "2.0", + "id": 1, + "method": "eth_sendBundle", + "params": [{"txs": raw_txs, "blockNumber": "0x0"}], + } + async with http.post(titan_url, json=body) as resp: + text = await resp.text() + print(f"{prefix} titan status={resp.status} body={text}") + + +async def trade_once( + w3: AsyncWeb3, + http: aiohttp.ClientSession, + account, + args, + chain_id: int, + iter_: int, + state: Optional[StateStream], + watcher: Optional[TxMonitor], +) -> None: + me = account.address + nonce = await w3.eth.get_transaction_count(me, "pending") + max_fee, max_priority = await estimate_eip1559(w3) + max_priority = max(max_priority, args.min_priority_gwei * 1_000_000_000) + max_fee = max(max_fee, max_priority * 3) + + ( + direction, + token_in, + token_out, + amount_in, + stable_units, + weth_units, + slip_lo, + slip_hi, + min_out, + label, + ) = trade_params(args, iter_) + calldata = await build_swap_calldata( + w3, + args.contract, + direction, + token_in, + token_out, + amount_in, + min_out, + stable_units, + weth_units, + slip_lo, + slip_hi, + me, + ) + + if state is not None: + frame = await next_state_frame(state) + if not await state_sim_ok( + w3, me, args.contract.address, calldata, frame, "[trade]" + ): return - print(f"[trade] state-override sim ok @ block {block_number}") raw_tx, tx_hash = build_signed_tx( account, @@ -548,20 +697,134 @@ async def trade_once( ) if not args.send: return - # Titan accepts `blockNumber: 0x0` as "include in any block within validity". - body = { - "jsonrpc": "2.0", - "id": 1, - "method": "eth_sendBundle", - "params": [{"txs": [raw_tx], "blockNumber": "0x0"}], - } - async with http.post(args.titan_url, json=body) as resp: - text = await resp.text() - print(f"[trade] titan status={resp.status} body={text}") + await send_titan_bundle(http, args.titan_url, [raw_tx], "[trade]") if watcher is not None: watcher.track(tx_hash, label, nonce) +async def trade_dual_once( + w3: AsyncWeb3, + http: aiohttp.ClientSession, + account, + args, + chain_id: int, + iter_: int, + state: Optional[StateStream], + watcher: Optional[TxMonitor], +) -> None: + me = account.address + nonce = await w3.eth.get_transaction_count(me, "pending") + max_fee, max_priority = await estimate_eip1559(w3) + max_priority = max(max_priority, args.min_priority_gwei * 1_000_000_000) + max_fee = max(max_fee, max_priority * 3) + + ( + direction, + token_in, + token_out, + amount_in, + stable_units, + weth_units, + slip_lo, + slip_hi, + min_out, + label, + ) = trade_params(args, iter_) + kipseli_calldata = await build_swap_calldata( + w3, + Contract.KIPSELI, + direction, + token_in, + token_out, + amount_in, + min_out, + stable_units, + weth_units, + slip_lo, + slip_hi, + me, + ) + fermi_calldata = await build_swap_calldata( + w3, + Contract.FERMI, + direction, + token_in, + token_out, + amount_in, + min_out, + stable_units, + weth_units, + slip_lo, + slip_hi, + me, + ) + + if state is not None: + ( + block_number, + timestamp_secs, + fermi_override, + kipseli_override, + ) = await next_state_frame(state) + kipseli_ok = await state_sim_ok( + w3, + me, + Contract.KIPSELI.address, + kipseli_calldata, + (block_number, timestamp_secs, kipseli_override), + "[dual] kipseli", + ) + fermi_ok = await state_sim_ok( + w3, + me, + Contract.FERMI.address, + fermi_calldata, + (block_number, timestamp_secs, fermi_override), + "[dual] fermi", + ) + if not (kipseli_ok and fermi_ok): + return + + if args.send and not await dual_preflight_ok( + w3, me, direction, token_in, stable_units, weth_units, slip_hi, max_fee + ): + return + + raw_kipseli, kipseli_tx_hash = build_signed_tx( + account, + chain_id, + nonce, + Contract.KIPSELI.gas_limit, + max_fee, + max_priority, + Contract.KIPSELI.address, + kipseli_calldata, + ) + raw_fermi, fermi_tx_hash = build_signed_tx( + account, + chain_id, + nonce + 1, + Contract.FERMI.gas_limit, + max_fee, + max_priority, + Contract.FERMI.address, + fermi_calldata, + ) + print( + f"[dual] {label} nonce={nonce}/{nonce + 1} amount_in={amount_in} " + f"min_out={min_out} max_fee={max_fee} max_prio={max_priority} " + f"kipseli_tx_hash={kipseli_tx_hash} fermi_tx_hash={fermi_tx_hash}" + ) + if not args.send: + return + await send_titan_bundle( + http, args.titan_url, [raw_kipseli, raw_fermi], "[dual]" + ) + if watcher is not None: + watcher.track(kipseli_tx_hash, f"{label} Kipseli", nonce) + watcher.track(fermi_tx_hash, f"{label} Fermi", nonce + 1) + + async def run_state_stream( region: StreamRegion, contract: str, state: StateStream ) -> None: @@ -591,13 +854,71 @@ async def run_state_stream( and isinstance(val, dict) and "stateOverride" in val ): - state.set((block_number, timestamp_secs, val["stateOverride"])) + state.set( + (block_number, timestamp_secs, val["stateOverride"]) + ) break except Exception as e: print(f"[stream] disconnected: {e}") await asyncio.sleep(5) +async def run_dual_state_stream(region: StreamRegion, state: StateStream) -> None: + url = region.ws_url + fermi_lc = FERMI_SWAPPER.lower() + kipseli_lc = KIPSELI_POOL.lower() + by_block = {} + while True: + try: + async with websockets.connect(url) as ws: + print(f"[stream] connected: {url}") + async for msg in ws: + if isinstance(msg, bytes): + continue + try: + v = json.loads(msg) + except ValueError: + continue + if not isinstance(v, dict): + continue + block_number = v.get("blockNumber") + if not isinstance(block_number, int): + continue + ts = v.get("timestamp") + timestamp_secs = ( + (ts // 1_000_000_000) if isinstance(ts, int) else 0 + ) + block_state = by_block.setdefault( + block_number, {"timestamp_secs": timestamp_secs} + ) + for k, val in v.items(): + if not isinstance(k, str) or not isinstance(val, dict): + continue + state_override = val.get("stateOverride") + if not isinstance(state_override, dict): + continue + k_lc = k.lower() + if k_lc == fermi_lc: + block_state["fermi"] = state_override + elif k_lc == kipseli_lc: + block_state["kipseli"] = state_override + if "fermi" in block_state and "kipseli" in block_state: + state.set( + ( + block_number, + block_state["timestamp_secs"], + block_state["fermi"], + block_state["kipseli"], + ) + ) + for old_block in list(by_block): + if old_block < block_number - 2: + del by_block[old_block] + except Exception as e: + print(f"[stream] disconnected: {e}") + await asyncio.sleep(5) + + def _enum_arg(cls, name: str): def parse(s: str): try: @@ -621,6 +942,11 @@ def parse_args() -> argparse.Namespace: default=Contract.FERMI, help="Target pAMM contract (fermi|bebop|kipseli).", ) + p.add_argument( + "--dual-bundle", + action="store_true", + help="Submit one Titan bundle containing both Kipseli and Fermi swaps.", + ) p.add_argument( "--notional-usd", type=float, default=1.0, help="USD notional per trade (stable side).", @@ -690,6 +1016,12 @@ def parse_args() -> argparse.Namespace: args = p.parse_args() if args.setup_only and args.skip_setup: p.error("--setup-only conflicts with --skip-setup") + if args.dual_bundle and args.contract is Contract.BEBOP: + p.error( + "--dual-bundle submits Kipseli + Fermi; Bebop is not part of this mode" + ) + if args.dual_bundle and args.send and not args.stream: + p.error("--dual-bundle --send requires --stream") return args @@ -715,7 +1047,8 @@ async def amain() -> None: sys.exit(f"failed to fetch ETH/USDC mid from Binance: {e}") asyncio.create_task(refresh_binance_mid(args)) - print(f"signer={me} chain_id={chain_id} contract={args.contract.label}") + label = "Kipseli+Fermi" if args.dual_bundle else args.contract.label + print(f"signer={me} chain_id={chain_id} contract={label}") await print_balances(w3, me) if not args.skip_setup: @@ -726,11 +1059,14 @@ async def amain() -> None: state: Optional[StateStream] = None if args.stream: - key = args.contract.stream_key - if key is None: - raise RuntimeError(f"--stream not supported for {args.contract.label}") state = StateStream() - asyncio.create_task(run_state_stream(args.stream_region, key, state)) + if args.dual_bundle: + asyncio.create_task(run_dual_state_stream(args.stream_region, state)) + else: + key = args.contract.stream_key + if key is None: + raise RuntimeError(f"--stream not supported for {args.contract.label}") + asyncio.create_task(run_state_stream(args.stream_region, key, state)) watcher: Optional[TxMonitor] = None if args.send: @@ -742,7 +1078,14 @@ async def amain() -> None: iter_ = 0 while True: try: - await trade_once(w3, http, account, args, chain_id, iter_, state, watcher) + if args.dual_bundle: + await trade_dual_once( + w3, http, account, args, chain_id, iter_, state, watcher + ) + else: + await trade_once( + w3, http, account, args, chain_id, iter_, state, watcher + ) except Exception as e: print(f"[iter {iter_}] error: {e}") iter_ += 1