modules/local_executor.py

"""
modules/local_executor.py — Garden Angel Local On-Chain Execution v1.6
──────────────────────────────────────────────────────────────────────────────
v1.6 — modules/oracle.py (the Cloudflare Worker relay client) is deleted;
  LocalExecutor is now the only execute_trade() implementation, no more
  fallback. ExecutionResult moves to modules/execution_types.py (import
  updated below) so this file doesn't import a now-deleted module.

v1.5 (2026-07-11) — FIX/DIAGNOSTIC: both legs' getAmountsOut() quotes are now
  pinned to the SAME explicit block_identifier (read once via
  w3.eth.block_number at the top of _execute_sync) instead of each
  independently resolving "latest". The block number is now embedded in
  both skip-reason messages. Prompted by production evidence that doesn't
  add up under the old code: two consecutive WBTC/USDT rejections 82
  seconds apart logged byte-identical quotes ("expected back $1,398.00 on
  $1,464.74 in, fee $1.32") even though scanner.py's own
  reserve_block_ts tracking showed the sell-leg pool had genuinely traded
  in between — i.e. this method's "live" quote was not actually reflecting
  current chain state at the time it ran. Pinning the block closes one
  possible cause (the two legs racing against different blocks a few
  seconds apart) outright. It does not by itself rule out the other:
  BSC_RPC_URL is a public free-tier Ankr endpoint (see this deployment's
  config), and some free-tier RPC gateways cache eth_call responses
  server-side independent of the requested block tag. If a future skip
  message still shows the SAME block number across two attempts more than
  a few seconds apart on BSC (~3s block time), that's on-the-record proof
  the RPC endpoint itself is serving stale reads and BSC_RPC_URL needs to
  move off the free tier — not something this file can fix on its own.
  See scanner.py's matching v18.21 changelog for a related, independently
  confirmed loan-sizing bug found while chasing this (the loan-liquidity
  cap was applied against each pool's TOTAL two-sided liquidity instead of
  the single reserve side actually being traded against, sizing every
  loan at ~2x the intended percentage of real pool depth).

v1.4 (2026-07-10) — NEW: optional `qwen=` constructor param
  (modules/qwen_client.py). When configured (QWEN_API_KEY set as a Space
  secret), a CONFIRMED on-chain execution's Telegram alert gets a short
  AI-generated context note appended — e.g. flagging anything unusual
  about the pair/spread. Explicitly advisory-only, confirmed with the
  operator: Qwen is only ever asked AFTER the trade has already executed
  on-chain, so it can never influence the BUY/HOLD decision, sizing, or
  whether execute_trade() runs — all unchanged, still 100% the existing
  spread/fee/liquidity math and the on-chain minProfit gate. Best-effort:
  any Qwen failure/timeout just means no commentary that cycle — the
  plain trade alert still sends exactly as it did before this existed.

v1.3 (2026-07-10) — FIX: the pre-flight go/no-go check was priced as if the
  loan principal were at risk, when startArbitrage is an ATOMIC flash-loan
  round trip — a trade that lands below the on-chain minProfit gate reverts
  entirely, costing only gas (cents on BSC), never the loan. The old check
  demanded the DOUBLE worst-case slippage haircut still beat amount_in
  (an effective ~1%+ spread after DEX fees) before even attempting — a bar
  confirmed in production to reject essentially everything, including
  spreads a real arb bot would happily race for. Now: attempt whenever the
  EXPECTED live quotes clear the caller's profit floor after the 0.09%
  flash-loan fee; keep the worst-case haircuts on minIntermediateOut/
  minFinalOut (execution protection); set on-chain minProfit to the
  caller's floor so the contract atomically guarantees "net at least the
  floor, or revert for pennies." Also: skip-reasons now log in USD instead
  of raw wei, so log lines are readable at a glance.

v1.2 (2026-07-10) — NEW: optional `telegram=` constructor param. When a
  TelegramClient is passed in, a CONFIRMED on-chain execution fires an
  explicit "✅ LIVE TRADE EXECUTED" alert with the tx hash/BscScan link —
  deliberately separate from ReportChannel's every-scan reporting (most
  users leave that off, since it's noisy) so a real trade is never buried.
  Best-effort: a notification failure never affects the already-returned
  trade result. See bot.py's matching change (passes self._tg through).

v1.1 (2026-07-10) — FIX: BSC's "WBTC" scan pair is actually BTCB (Binance-Peg
  Bitcoin), a different token than Ethereum's real WBTC — constants.py's
  BSC_TOKENS keys it "BTCB", not "WBTC". _resolve_token() aliased BNB→WBNB
  but had no equivalent WBTC→BTCB alias, so a live BSC WBTC BUY signal
  (confirmed in production: pancakeswap, net +$493.54) failed to resolve
  and was silently dropped — silently because execute_trade()'s
  `except ValueError` branch also didn't log anything. Both are fixed:
  WBTC now aliases to BTCB, and every resolution failure is logged at
  ERROR so this can never fail invisibly again.

Purpose: this is "fix hunt". Until now, every BUY signal Scanner computed was
handed to OracleClient, which POSTs to a third-party Cloudflare Worker
(garden-angel-production.elghaly.workers.dev/execute) to actually execute the
trade on-chain. That Worker's /execute route has never been shown to work in
this deployment — gac-defi-bot's src/index.js /execute-signal handler (the
Worker-side counterpart) fails closed with 403/501 (see that repo's
SECURITY_AUDIT.md), so effectively every execute_trade() call against it has
returned confirmed=False. Net effect: this bot has been "hunting" — computing
real BUY signals — for as long as it's been deployed, while confirming
approximately zero of them on-chain, because the only thing standing between
"found an opportunity" and "executed it" was a Worker endpoint that was never
wired up on the other end.

LocalExecutor removes that dependency entirely. It implements the exact same
interface OracleClient exposes — async execute_trade(...) -> ExecutionResult,
importing the identical dataclass from modules/oracle.py so callers can't
tell the two apart — but instead of an HTTP round trip to a third-party
Worker, it calls ContractManager.start_arbitrage() directly against the
FlashArbitrageV2 contract this deployment already owns and controls on BSC —
the same contract the Wallet tab already reads from.

Wired in from bot.py, opt-in only:
    LOCAL_EXECUTION_ENABLED=true   — required. Unset/false leaves bot.py's
                                      existing OracleClient (Worker-based)
                                      path completely unchanged; nothing in
                                      this file runs until this is explicitly
                                      turned on.
    DRY_RUN=true                    — respected exactly like the rest of
                                      this deployment (PayoutManager, etc.):
                                      LIVE router quotes are still fetched
                                      so the numbers are real, but no
                                      transaction is ever built, signed, or
                                      broadcast. Returns confirmed=False with
                                      a clear "(dry run)" message so Scanner
                                      never credits a dry-run as a real
                                      execution.

Safety properties:
  - EVM-only, hard-enforced. A `chain` kwarg other than "ARB" or "BSC" (or
    unset — every call site targets Arbitrum today, "BSC" kept only as a
    legacy-compatible value) is refused up front with confirmed=False.
    This never silently attempts an execution against the wrong chain's
    contract/router registry. (v1.6 FIX, 2026-07-15 — this note previously
    said "BSC-only", stale since the ARB migration; the code below has
    accepted ARB since that migration, only the docstring hadn't caught up.)
  - Every write goes through the SAME ContractManager/Web3Client the Wallet
    tab already uses for reads — one signing key (PAYOUT_PRIVATE_KEY), one
    verified RPC resolution path (see web3_client.py v1.2), no second,
    divergent execution path to keep in sync.
  - Slippage-protected quoting: minIntermediateOut/minFinalOut are computed
    from LIVE getAmountsOut() router quotes at execution time — not from
    Scanner's pre-trade spread estimate — with a SLIPPAGE_BPS haircut
    (constants.py CFG["SLIPPAGE_BPS"], default 50 = 0.50%) applied before
    being sent on-chain.
  - minProfit is the caller's own floor (min_profit_usd/net_profit_usd —
    never invented, never loosened to whatever quotes happen to show),
    enforced ATOMICALLY by the contract: the flash-loan round trip either
    nets at least the floor or the entire transaction reverts, costing only
    gas (a few cents on Arbitrum), never principal. The pre-flight check
    (v1.3) only skips trades whose EXPECTED live quotes can't clear the
    floor after the 0.05% flash-loan fee (AAVE_FLASHLOAN_FEE_BPS=5, see
    constants.py — this note previously said 0.09%, out of sync with that
    constant and with this file's own error-message text below) — a
    doomed-by-construction tx is never sent, but a genuine opportunity is
    attempted even though it might lose the race and revert for pennies.
  - Never fabricates a tx_hash or a realized_profit_usd. A transaction that
    reverts (receipt.status == 0) is reported confirmed=False. This module
    does not attempt to decode a realized-profit event — the deployed
    FlashArbitrageV2's exact event ABI has not been independently confirmed
    — so realized_profit_usd is left None on success; Scanner's
    _notify_payout() already falls back to its own pre-trade estimate
    whenever realized_profit_usd is None, so this is an existing, safe
    contract, not a new gap.
  - Every blocking web3.py call (getAmountsOut, build_transaction, sign,
    send_raw_transaction, wait_for_transaction_receipt) runs inside
    asyncio.to_thread() — execute_trade() is an async method called from
    Scanner's async _notify_payout(), and web3.py's HTTPProvider is fully
    synchronous; calling it directly would block the whole bot's event loop
    (Telegram polling, keep-alive pings, other in-flight scans) for the
    entire duration of an on-chain round trip.

Required env vars (only exercised when LOCAL_EXECUTION_ENABLED=true):
    FLASH_ARBITRAGE_CONTRACT_ADDRESS, PAYOUT_PRIVATE_KEY, ARB_RPC_URL
    — all already required by ContractManager/Web3Client. LocalExecutor
      introduces no new required env vars beyond LOCAL_EXECUTION_ENABLED
      and the existing DRY_RUN flag documented above.
"""

from __future__ import annotations

import asyncio
import logging
import time
from typing import Any, Optional

from web3 import Web3

from modules.contract_manager import (
    ContractManager, get_contract_manager, resolve_router,
    get_contract_manager_v3, resolve_v3_router,
    UNISWAP_V3_QUOTER_V2,
)
from modules.execution_types import ExecutionResult

logger = logging.getLogger("garden_angel.local_executor")

ROUTER_QUOTE_ABI = [
    {
        "inputs": [
            {"name": "amountIn", "type": "uint256"},
            {"name": "path", "type": "address[]"},
        ],
        "name": "getAmountsOut",
        "outputs": [{"name": "amounts", "type": "uint256[]"}],
        "stateMutability": "view",
        "type": "function",
    }
]

# Uniswap V3 QuoterV2 — same simulate-via-eth_call pattern as
# ROUTER_QUOTE_ABI above, used only for the pre-flight go/no-go check on a
# V3-involving leg. See modules/scanner.py's identical ABI/usage for the
# discovery-side counterpart.
V3_QUOTER_ABI = [
    {
        "inputs": [
            {
                "components": [
                    {"name": "tokenIn", "type": "address"},
                    {"name": "tokenOut", "type": "address"},
                    {"name": "amountIn", "type": "uint256"},
                    {"name": "fee", "type": "uint24"},
                    {"name": "sqrtPriceLimitX96", "type": "uint160"},
                ],
                "name": "params",
                "type": "tuple",
            }
        ],
        "name": "quoteExactInputSingle",
        "outputs": [
            {"name": "amountOut", "type": "uint256"},
            {"name": "sqrtPriceX96After", "type": "uint160"},
            {"name": "initializedTicksCrossed", "type": "uint32"},
            {"name": "gasEstimate", "type": "uint256"},
        ],
        "stateMutability": "nonpayable",
        "type": "function",
    }
]

_DEFAULT_SLIPPAGE_BPS = 50  # 0.50% — matches constants.py CFG["SLIPPAGE_BPS"]
_RECEIPT_TIMEOUT_SECS = 180  # matches the rest of this deployment's fund-moving-call budget


def _bps_haircut(amount: int, bps: int) -> int:
    """amount * (1 - bps/10000), integer-safe (no float rounding on-chain-bound values)."""
    return (amount * (10_000 - bps)) // 10_000


class _DryRunSkip(Exception):
    """Internal signal only — the dry-run path never reaches a real send/receipt."""


class LocalExecutor:
    """
    Drop-in replacement for OracleClient. Same execute_trade() signature
    (plus additive optional kwargs) and the same ExecutionResult return
    type — no HTTP, no third party, calls ContractManager.start_arbitrage()
    directly.
    """

    def __init__(
        self,
        contract_manager: Optional[ContractManager] = None,
        dry_run: bool = True,
        slippage_bps: int = _DEFAULT_SLIPPAGE_BPS,
        token_registry: Optional[dict] = None,
        telegram: Optional[Any] = None,
        qwen: Optional[Any] = None,
    ) -> None:
        self._cm = contract_manager or get_contract_manager()
        self._dry_run = bool(dry_run)
        self._slippage_bps = int(slippage_bps)
        # Optional TelegramClient (duck-typed via Any to avoid importing
        # modules.telegram_client here — this module has no other reason
        # to depend on it). When set, a CONFIRMED on-chain execution fires
        # an explicit alert — deliberately separate from ReportChannel's
        # every-scan reporting (which most users leave off, since it's
        # noisy) so "a real trade just executed" is never buried in scan
        # spam. Best-effort: a notification failure never affects the
        # trade result already returned to the caller.
        self._telegram = telegram
        # v1.4 — optional QwenClient (modules/qwen_client.py). Advisory
        # ONLY: called after a trade is already confirmed on-chain, purely
        # to append a short context note to the Telegram alert above. Can
        # never influence the trade that already happened. Inert/None-safe
        # if not passed or not configured (QWEN_API_KEY unset).
        self._qwen = qwen

        if token_registry is None:
            from constants import BSC_TOKENS  # single source of truth for BSC token addresses
            token_registry = BSC_TOKENS
        self._tokens = {
            sym.upper(): Web3.to_checksum_address(info["address"])
            for sym, info in token_registry.items()
        }

        logger.warning(
            "[LocalExecutor] ACTIVE — BUY signals will be executed DIRECTLY "
            "on-chain against %s via ContractManager, bypassing the "
            "Cloudflare Worker /execute path entirely. dry_run=%s "
            "slippage_bps=%d",
            self._cm.contract.address, self._dry_run, self._slippage_bps,
        )

    def _resolve_token(self, symbol: str) -> str:
        sym = symbol.upper()
        # Migrated 2026-07 to Arbitrum One. Native gas token is ETH, which has
        # no ERC-20 address; the contract only ever moves wrapped WETH. WBTC on
        # Arbitrum is real WBTC (8 decimals), so no BTCB-style alias is needed.
        if sym in ("ETH", "WETH"):
            sym = "WETH"
        if sym not in self._tokens:
            raise ValueError(
                f"LocalExecutor: unknown token symbol '{symbol}' — not in "
                f"ARB_TOKENS. Known: {', '.join(sorted(self._tokens))}."
            )
        return self._tokens[sym]

    async def execute_trade(
        self,
        base_asset: str,
        stable_asset: str,
        target_dex: str,
        amount: float,
        net_profit_usd: float,
        sell_dex: Optional[str] = None,
        chain: Optional[str] = None,
        min_profit_usd: Optional[float] = None,
        buy_fee_tier: int = 0,
        sell_fee_tier: int = 0,
    ) -> ExecutionResult:
        t0 = time.monotonic()

        if chain and chain.upper() not in ("ARB", "BSC"):
            return ExecutionResult(
                confirmed=False,
                error=f"LocalExecutor (EVM) runs on Arbitrum One, got chain={chain!r}. "
                      f"Solana signals are handled by modules/solana_arb.py.",
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        if not sell_dex:
            return ExecutionResult(
                confirmed=False,
                error="LocalExecutor requires sell_dex (the exit-leg DEX) "
                      "— caller did not provide one.",
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        # Uniswap V3 leg(s) — a SEPARATE contract/path (FlashArbitrageV3.sol,
        # see that file's own header for why FlashArbitrageV2 can't run
        # this). Dispatched here rather than woven into the V2-only path
        # below so a live-trading deployment's existing, already-proven V2
        # execution is completely untouched by this addition. The common
        # case (buy_fee_tier == sell_fee_tier == 0, i.e. every trade before
        # this existed) falls straight through to the unchanged code below.
        if buy_fee_tier or sell_fee_tier:
            return await self._execute_trade_mixed(
                base_asset, stable_asset, target_dex, sell_dex, amount,
                net_profit_usd, min_profit_usd, buy_fee_tier, sell_fee_tier, t0,
            )

        try:
            stable_addr = self._resolve_token(stable_asset)
            base_addr = self._resolve_token(base_asset)
            router_buy = resolve_router(target_dex)
            router_sell = resolve_router(sell_dex)
        except ValueError as exc:
            # FIX — this branch previously returned without logging anything,
            # so a token/router resolution failure (e.g. the WBTC/BTCB gap
            # this exact bug fixed) was completely silent: no error line in
            # the HF Space console, nothing to grep for. A real BUY signal
            # could vanish here with zero trace. Now logged at ERROR so a
            # resolution failure is always visible, same as every other
            # failure path in this method.
            logger.error(
                "[LocalExecutor] Token/router resolution failed for "
                "%s/%s on %s→%s: %s",
                base_asset, stable_asset, target_dex, sell_dex, exc,
            )
            return ExecutionResult(
                confirmed=False, error=str(exc),
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        floor_usd = min_profit_usd if min_profit_usd is not None else max(net_profit_usd, 0.0)

        try:
            tx_hash, realized = await asyncio.to_thread(
                self._execute_sync,
                stable_addr, base_addr, router_buy, router_sell,
                amount, floor_usd,
            )
        except _DryRunSkip as exc:
            logger.info("[LocalExecutor] %s", exc)
            return ExecutionResult(
                confirmed=False, error=f"(dry run) {exc}",
                duration_ms=(time.monotonic() - t0) * 1000,
            )
        except Exception as exc:
            duration_ms = (time.monotonic() - t0) * 1000
            logger.error(
                "[LocalExecutor] Execution failed for %s/%s on %s→%s: %s",
                base_asset, stable_asset, target_dex, sell_dex, exc,
            )
            return ExecutionResult(
                confirmed=False, error=str(exc)[:300],
                duration_ms=duration_ms,
            )

        duration_ms = (time.monotonic() - t0) * 1000
        logger.info(
            "[LocalExecutor] Execution confirmed on-chain in %.0fms — tx=%s",
            duration_ms, tx_hash,
        )
        await self._notify_confirmed(
            tx_hash, base_asset, stable_asset, target_dex, sell_dex,
            amount, net_profit_usd, realized,
        )
        return ExecutionResult(
            confirmed=True,
            tx_hash=tx_hash,
            realized_profit_usd=realized,
            status_raw="confirmed",
            duration_ms=duration_ms,
        )

    # ------------------------------------------------------------------
    # Uniswap V3 mixed-leg execution — separate contract, opt-in only.
    # ------------------------------------------------------------------

    async def _execute_trade_mixed(
        self,
        base_asset: str, stable_asset: str,
        target_dex: str, sell_dex: str,
        amount: float, net_profit_usd: float, min_profit_usd: Optional[float],
        buy_fee_tier: int, sell_fee_tier: int,
        t0: float,
    ) -> ExecutionResult:
        """Mirrors execute_trade()'s V2-only path above, but against
        FlashArbitrageV3.sol (a SEPARATE, independently-deployed contract —
        see that file's own header). Fails safe and explicitly: if the
        operator hasn't deployed+configured it, this returns confirmed=False
        with a clear reason, exactly like every other resolution failure in
        this module — it never falls back to silently skipping the trade
        without a trace, and it never attempts to force the V2-only path
        with mismatched router interfaces.
        """
        import constants  # local import — same lazy-import convention _resolve_token uses

        if not constants.v3_execution_enabled():
            return ExecutionResult(
                confirmed=False,
                error="Uniswap V3 leg found but V3_EXECUTION_ENABLED is not "
                      "'true' — reported only, not executed. Deploy "
                      "Contrect/FlashArbitrageV3.sol and set "
                      "FLASH_ARBITRAGE_V3_CONTRACT_ADDRESS + "
                      "V3_EXECUTION_ENABLED=true to go live on this leg.",
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        cm_v3 = get_contract_manager_v3()
        if cm_v3 is None:
            return ExecutionResult(
                confirmed=False,
                error="Uniswap V3 leg found but FLASH_ARBITRAGE_V3_CONTRACT_ADDRESS "
                      "is unset/unreachable — deploy Contrect/FlashArbitrageV3.sol "
                      "yourself first (see that file's header).",
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        try:
            stable_addr = self._resolve_token(stable_asset)
            base_addr = self._resolve_token(base_asset)
            router_buy = resolve_v3_router(target_dex) if buy_fee_tier else resolve_router(target_dex)
            router_sell = resolve_v3_router(sell_dex) if sell_fee_tier else resolve_router(sell_dex)
        except ValueError as exc:
            logger.error(
                "[LocalExecutor] V3 token/router resolution failed for "
                "%s/%s on %s→%s: %s",
                base_asset, stable_asset, target_dex, sell_dex, exc,
            )
            return ExecutionResult(
                confirmed=False, error=str(exc),
                duration_ms=(time.monotonic() - t0) * 1000,
            )

        floor_usd = min_profit_usd if min_profit_usd is not None else max(net_profit_usd, 0.0)

        try:
            tx_hash, realized = await asyncio.to_thread(
                self._execute_sync_mixed,
                cm_v3, stable_addr, base_addr,
                router_buy, bool(buy_fee_tier), int(buy_fee_tier),
                router_sell, bool(sell_fee_tier), int(sell_fee_tier),
                amount, floor_usd,
            )
        except _DryRunSkip as exc:
            logger.info("[LocalExecutor] %s", exc)
            return ExecutionResult(
                confirmed=False, error=f"(dry run) {exc}",
                duration_ms=(time.monotonic() - t0) * 1000,
            )
        except Exception as exc:
            duration_ms = (time.monotonic() - t0) * 1000
            logger.error(
                "[LocalExecutor] V3 execution failed for %s/%s on %s→%s: %s",
                base_asset, stable_asset, target_dex, sell_dex, exc,
            )
            return ExecutionResult(
                confirmed=False, error=str(exc)[:300],
                duration_ms=duration_ms,
            )

        duration_ms = (time.monotonic() - t0) * 1000
        logger.info(
            "[LocalExecutor] V3 execution confirmed on-chain in %.0fms — tx=%s",
            duration_ms, tx_hash,
        )
        await self._notify_confirmed(
            tx_hash, base_asset, stable_asset, target_dex, sell_dex,
            amount, net_profit_usd, realized,
        )
        return ExecutionResult(
            confirmed=True,
            tx_hash=tx_hash,
            realized_profit_usd=realized,
            status_raw="confirmed",
            duration_ms=duration_ms,
        )

    def _execute_sync_mixed(
        self,
        cm_v3,
        stable_addr: str, base_addr: str,
        router_buy: str, buy_is_v3: bool, buy_fee_tier: int,
        router_sell: str, sell_is_v3: bool, sell_fee_tier: int,
        amount: float, floor_usd: float,
    ) -> tuple[str, Optional[float]]:
        """Runs inside asyncio.to_thread(). Mirrors _execute_sync's pre-
        flight/slippage/on-chain-minProfit structure exactly, but quotes a
        V3 leg via QuoterV2.quoteExactInputSingle instead of getAmountsOut()
        (V3 has no two-address `path` — it's single-hop per fee tier)."""
        w3 = cm_v3.w3
        amount_in = cm_v3.assets.scale_amount(stable_addr, amount)
        floor_wei = cm_v3.assets.scale_amount(stable_addr, max(floor_usd, 0.0))
        block_number = w3.eth.block_number

        def _quote_leg(is_v3: bool, router: str, fee: int, token_in: str, token_out: str, amount_in_raw: int) -> int:
            if is_v3:
                quoter = w3.eth.contract(
                    address=Web3.to_checksum_address(UNISWAP_V3_QUOTER_V2), abi=V3_QUOTER_ABI,
                )
                params = (
                    Web3.to_checksum_address(token_in), Web3.to_checksum_address(token_out),
                    amount_in_raw, fee, 0,
                )
                # Pinned to the same explicit block as the V2 leg's
                # getAmountsOut() call below (see _execute_sync's v1.5
                # note on why both legs must share one consistent block).
                out = quoter.functions.quoteExactInputSingle(params).call(
                    block_identifier=block_number
                )
                return int(out[0])
            router_c = w3.eth.contract(address=router, abi=ROUTER_QUOTE_ABI)
            quote = router_c.functions.getAmountsOut(
                amount_in_raw, [token_in, token_out]
            ).call(block_identifier=block_number)
            return int(quote[-1])

        expected_intermediate = _quote_leg(buy_is_v3, router_buy, buy_fee_tier, stable_addr, base_addr, amount_in)
        min_intermediate_out = _bps_haircut(expected_intermediate, self._slippage_bps)

        expected_final = _quote_leg(sell_is_v3, router_sell, sell_fee_tier, base_addr, stable_addr, expected_intermediate)
        min_final_out = _bps_haircut(expected_final, 2 * self._slippage_bps)

        from constants import AAVE_FLASHLOAN_FEE_BPS
        flash_fee_wei = (amount_in * AAVE_FLASHLOAN_FEE_BPS) // 10_000
        expected_profit_wei = expected_final - amount_in - flash_fee_wei
        dec = cm_v3.assets.decimals(stable_addr)

        def _usd(wei: int) -> str:
            return f"${wei / 10 ** dec:,.2f}"

        if expected_profit_wei <= 0:
            raise RuntimeError(
                f"V3 live quotes unprofitable after the 0.05% flash-loan fee "
                f"(expected back {_usd(expected_final)} on {_usd(amount_in)} in, "
                f"fee {_usd(flash_fee_wei)}) — spread too thin, skipping."
            )
        if floor_wei > 0 and expected_profit_wei < floor_wei:
            raise RuntimeError(
                f"V3 expected profit {_usd(expected_profit_wei)} is below the "
                f"configured floor {_usd(floor_wei)} — skipping rather than "
                "sending a trade that would revert on the contract's own "
                "minProfit gate."
            )

        min_profit = floor_wei if floor_wei > 0 else 1

        if self._dry_run:
            raise _DryRunSkip(
                f"would call startArbitrageMixed(asset={stable_addr}, "
                f"amount={amount_in}, routerBuy={router_buy} v3={buy_is_v3} "
                f"fee={buy_fee_tier}, routerSell={router_sell} v3={sell_is_v3} "
                f"fee={sell_fee_tier}, intermediateToken={base_addr}, "
                f"minIntermediateOut={min_intermediate_out}, "
                f"minFinalOut={min_final_out}, minProfit={min_profit}) — "
                f"DRY_RUN=true, nothing broadcast."
            )

        tx_hash = cm_v3.start_arbitrage_mixed(
            asset=stable_addr,
            amount=amount_in,
            router_buy=router_buy, buy_is_v3=buy_is_v3, buy_fee_tier=buy_fee_tier,
            router_sell=router_sell, sell_is_v3=sell_is_v3, sell_fee_tier=sell_fee_tier,
            intermediate_token=base_addr,
            min_intermediate_out=min_intermediate_out,
            min_final_out=min_final_out,
            min_profit=min_profit,
        )

        receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=_RECEIPT_TIMEOUT_SECS)
        if receipt.status != 1:
            raise RuntimeError(f"Transaction {tx_hash} reverted on-chain (status=0).")

        return tx_hash, None

    async def _notify_confirmed(
        self,
        tx_hash: str, base_asset: str, stable_asset: str,
        target_dex: str, sell_dex: str,
        amount: float, net_profit_usd: float, realized: Optional[float],
    ) -> None:
        """Best-effort Telegram alert for a CONFIRMED execution — see
        __init__'s self._telegram docstring note. Never raises: a
        notification failure must never look like the trade itself
        failed, since by this point it has already succeeded on-chain."""
        if self._telegram is None:
            return
        profit_usd = realized if realized is not None else net_profit_usd
        profit_line = (
            f"${realized:,.2f} (realized)" if realized is not None
            else f"~${net_profit_usd:,.2f} (pre-trade estimate)"
        )
        # FIX (2026-07-15, operator went live — DRY_RUN=false) — this linked
        # to bscscan.com, a leftover from before the migration. ContractManager/
        # Web3Client (self._cm) only ever operate against Arbitrum in this
        # deployment (see execute_trade's own chain gate above), so every real
        # tx_hash here is an Arbitrum One transaction — bscscan.com can never
        # find it. This is exactly the link an operator opens the moment their
        # very first live trade confirms, to verify funds actually moved.
        text = (
            f"✅ LIVE TRADE EXECUTED\n"
            f"{base_asset}/{stable_asset} — {target_dex}→{sell_dex}\n"
            f"Loan: ${amount:,.2f} | Profit: {profit_line}\n"
            f"https://arbiscan.io/tx/{tx_hash}"
        )

        # v1.4 — advisory-only Qwen commentary, appended AFTER the trade
        # already confirmed on-chain (see __init__'s self._qwen docstring
        # note). approx spread_pct here is just profit/loan — a rough
        # proxy for the commentary prompt, not the scanner's precise
        # pre-trade spread_pct (which isn't available at this call site).
        # Never allowed to delay or block the core alert: if this fails,
        # times out, or isn't configured, the plain alert above still
        # sends exactly as it always has.
        if self._qwen is not None and getattr(self._qwen, "enabled", False):
            try:
                approx_spread_pct = (profit_usd / amount * 100) if amount else 0.0
                commentary = await self._qwen.analyze_opportunity(
                    base_asset=base_asset, stable_asset=stable_asset,
                    buy_dex=target_dex, sell_dex=sell_dex,
                    spread_pct=approx_spread_pct, loan_amount=amount,
                    net_profit_usd=profit_usd,
                )
                if commentary:
                    text += f"\n\n🤖 {commentary}"
            except Exception as exc:
                logger.warning("[LocalExecutor] Qwen commentary failed (non-fatal): %s", exc)

        try:
            await self._telegram.safe_send(text)
        except Exception as exc:
            logger.warning("[LocalExecutor] Confirmed-trade Telegram alert failed: %s", exc)

    def _execute_sync(
        self,
        stable_addr: str, base_addr: str,
        router_buy: str, router_sell: str,
        amount: float, floor_usd: float,
    ) -> tuple[str, Optional[float]]:
        """Runs inside asyncio.to_thread() — blocking web3.py calls only."""
        w3 = self._cm.w3

        amount_in = self._cm.assets.scale_amount(stable_addr, amount)
        floor_wei = self._cm.assets.scale_amount(stable_addr, max(floor_usd, 0.0))

        # v1.5 FIX/DIAGNOSTIC (2026-07-11) — pin both legs' getAmountsOut()
        # calls to the SAME explicit block, and surface that block number
        # in every skip/failure message. Prompted by production evidence:
        # two consecutive WBTC/USDT rejections 82s apart logged
        # byte-identical numbers ("expected back $1,398.00 on $1,464.74
        # in, fee $1.32") even though the sell-leg pool's own
        # reserve_block_ts (tracked by scanner.py) showed it had traded
        # in between — i.e. the quote this method used was NOT current
        # chain state at the time it ran. Two possible causes, both
        # addressed by this change: (1) the two legs racing against
        # different blocks if a new block lands between the calls — fixed
        # by pinning both to one explicit block_identifier instead of each
        # implicitly defaulting to "latest" independently; (2) the RPC
        # endpoint itself (BSC_RPC_URL — a public free-tier Ankr URL per
        # this deployment's config) caching eth_call responses server-side
        # regardless of block tag — NOT fixable from this file, but now
        # provable: if this block number stops advancing between
        # consecutive failures on the same pair, that's the smoking gun,
        # and BSC_RPC_URL needs to move off the free-tier endpoint.
        block_number = w3.eth.block_number

        buy_router = w3.eth.contract(address=router_buy, abi=ROUTER_QUOTE_ABI)
        sell_router = w3.eth.contract(address=router_sell, abi=ROUTER_QUOTE_ABI)

        buy_quote = buy_router.functions.getAmountsOut(
            amount_in, [stable_addr, base_addr]
        ).call(block_identifier=block_number)
        expected_intermediate = buy_quote[-1]
        min_intermediate_out = _bps_haircut(expected_intermediate, self._slippage_bps)

        # Quote the sell leg from the EXPECTED intermediate amount — the
        # profitability decision should be made on what the quotes say
        # will actually happen, not on the worst-case haircut chain.
        # Pinned to the SAME block as the buy-leg quote above (see v1.5
        # note) so both legs are evaluated against one consistent chain
        # state instead of each independently resolving "latest".
        sell_quote = sell_router.functions.getAmountsOut(
            expected_intermediate, [base_addr, stable_addr]
        ).call(block_identifier=block_number)
        expected_final = sell_quote[-1]
        # minFinalOut stays worst-case (both legs slipping to their
        # haircut limits) — this is EXECUTION protection, enforced by the
        # router/contract on-chain, not the go/no-go decision.
        min_final_out = _bps_haircut(expected_final, 2 * self._slippage_bps)

        # v1.3 — the go/no-go decision. Previous versions demanded the
        # DOUBLE-HAIRCUT worst case (min_final_out) still beat amount_in,
        # i.e. an effective ~1%+ spread after DEX fees, before even
        # attempting — a bar so high it near-never fires on real BSC
        # blue-chip pairs. That caution was misplaced: startArbitrage is
        # an ATOMIC flash-loan round trip. If reality lands below the
        # on-chain minProfit gate at execution time, the ENTIRE
        # transaction reverts — the loan never completes, no principal is
        # ever at risk, and the only cost is the gas of a reverted tx
        # (cents on BSC). So the right economics are: attempt whenever
        # the EXPECTED quotes clear the caller's profit floor after the
        # flash-loan fee, and let the atomic on-chain minProfit revert be
        # the (cheap) loss-taker when the price races away. Losing that
        # race occasionally costs cents; never entering it costs every
        # real opportunity.
        from constants import AAVE_FLASHLOAN_FEE_BPS  # single source of truth (5 bps on Arbitrum)
        flash_fee_wei = (amount_in * AAVE_FLASHLOAN_FEE_BPS) // 10_000
        expected_profit_wei = expected_final - amount_in - flash_fee_wei
        dec = self._cm.assets.decimals(stable_addr)

        def _usd(wei: int) -> str:
            return f"${wei / 10 ** dec:,.2f}"

        if expected_profit_wei <= 0:
            raise RuntimeError(
                f"Live quotes unprofitable after the 0.05% flash-loan fee "
                f"(expected back {_usd(expected_final)} on {_usd(amount_in)} in, "
                f"fee {_usd(flash_fee_wei)}, quoted @ block {block_number}) — "
                f"spread too thin, skipping."
            )
        if floor_wei > 0 and expected_profit_wei < floor_wei:
            raise RuntimeError(
                f"Expected profit {_usd(expected_profit_wei)} is below the "
                f"configured floor {_usd(floor_wei)} (quoted @ block "
                f"{block_number}) — skipping rather than sending a trade "
                f"that would revert on the contract's own minProfit gate."
            )

        # On-chain minProfit = the caller's floor, enforced atomically by
        # the contract: it either nets at least this or the whole flash
        # loan reverts for pennies of gas. Never loosened to whatever the
        # quotes happen to show — the floor is the operator's risk knob.
        min_profit = floor_wei if floor_wei > 0 else 1

        if self._dry_run:
            raise _DryRunSkip(
                f"would call startArbitrage(asset={stable_addr}, "
                f"amount={amount_in}, routerBuy={router_buy}, "
                f"routerSell={router_sell}, intermediateToken={base_addr}, "
                f"minIntermediateOut={min_intermediate_out}, "
                f"minFinalOut={min_final_out}, minProfit={min_profit}) — "
                f"DRY_RUN=true, nothing broadcast."
            )

        tx_hash = self._cm.start_arbitrage(
            asset=stable_addr,
            amount=amount_in,
            router_buy=router_buy,
            router_sell=router_sell,
            intermediate_token=base_addr,
            min_intermediate_out=min_intermediate_out,
            min_final_out=min_final_out,
            min_profit=min_profit,
        )

        receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=_RECEIPT_TIMEOUT_SECS)
        if receipt.status != 1:
            raise RuntimeError(f"Transaction {tx_hash} reverted on-chain (status=0).")

        # realized_profit_usd deliberately left None — see module docstring
        # (no independently-confirmed event ABI to decode it from). Scanner
        # falls back to its own pre-trade estimate for ledger crediting.
        return tx_hash, None


_local_executor: Optional[LocalExecutor] = None


def get_local_executor() -> LocalExecutor:
    global _local_executor
    if _local_executor is None:
        import os
        _local_executor = LocalExecutor(
            dry_run=os.environ.get("DRY_RUN", "false").strip().lower() == "true",
        )
    return _local_executor