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