1inch API Python Tutorial: Master DEX Trading Fast
Learn to use the 1inch API with Python to fetch real-time swap quotes, execute DEX trades, and automate DeFi workflows — complete with working code examples and error handling.
Learn to use the 1inch API with Python to fetch real-time swap quotes, execute DEX trades, and automate DeFi workflows — complete with working code examples and error handling.
The 1inch aggregator routes your trades across dozens of DEXes simultaneously — Uniswap, Curve, Balancer, and more — finding the best price with the least slippage. Its API opens that same routing engine to your Python scripts, letting you build automated DeFi tools that rival what institutional traders run on centralized venues like Binance or Bybit. Whether you're building a simple swap bot or a full portfolio rebalancer, the 1inch API is one of the most powerful on-chain execution tools available right now.
Since 2023, the 1inch Swap API v6.0 requires authentication. Go to portal.1inch.dev, create an account, and generate an API key under the Swap product. The free tier handles testing and moderate production traffic comfortably. Once you have your key, install the required packages:
pip install requests web3 python-dotenv
import requests
import os
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv("ONEINCH_API_KEY") # Store in .env, never hardcode
BASE_URL = "https://api.1inch.dev/swap/v6.0"
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"accept": "application/json"
}
# Chain IDs for major supported networks
CHAIN_IDS = {
"ethereum": 1,
"bsc": 56,
"polygon": 137,
"arbitrum": 42161,
"optimism": 10,
"base": 8453
}
Never commit your API key to Git. Store it in a .env file and add .env to .gitignore immediately. If you accidentally expose a key, rotate it from the 1inch portal right away — treat it like a private key.
The quote endpoint is your most-used call. It returns the expected output amount, the full routing path, and estimated gas — all without committing any transaction. This makes it ideal for displaying prices, calculating expected value before execution, or building a monitoring loop that watches for favorable rates.
def get_quote(src_token: str, dst_token: str, amount: int, chain_id: int = 1) -> dict:
"""
Fetch a swap quote from 1inch.
amount: raw token units (already multiplied by token decimals)
"""
url = f"{BASE_URL}/{chain_id}/quote"
params = {
"src": src_token,
"dst": dst_token,
"amount": str(amount)
}
response = requests.get(url, headers=HEADERS, params=params)
response.raise_for_status()
return response.json()
# Example: Quote 100 USDC -> WETH on Ethereum mainnet
USDC = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
WETH = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
amount_usdc = 100 * 10**6 # 100 USDC (6 decimals)
quote = get_quote(USDC, WETH, amount_usdc)
eth_received = int(quote["dstAmount"]) / 10**18
print(f"100 USDC -> {eth_received:.6f} ETH")
print(f"Estimated gas: {quote.get('estimatedGas', 'N/A')}")
print(f"Route: {[p[0]['name'] for p in quote.get('protocols', [])]}")
The protocols field in the response is genuinely useful. On large trades you'll often see splits — for example, 60% routed through Uniswap V3 and 40% through Curve — because the aggregator found that splitting reduces price impact more than executing on a single pool. This is execution quality you simply can't replicate by calling any single DEX directly.
A price monitor is the first practical tool most developers build with the quote API. Traders on platforms like OKX and Coinbase are used to clean real-time price feeds; this gives you the same for on-chain DeFi rates. You can also use this as a signal layer alongside tools like VoiceOfChain, which tracks wallet movements and on-chain momentum — when a large wallet starts accumulating, the 1inch quote for that token often starts shifting before it shows up on Binance or KuCoin order books.
import time
from datetime import datetime
def monitor_price(
src_token: str,
dst_token: str,
amount: int,
chain_id: int = 1,
interval: int = 30
):
"""Poll 1inch for price quotes at a fixed interval."""
print(f"Monitoring every {interval}s — Ctrl+C to stop.\n")
prev_amount = None
while True:
try:
quote = get_quote(src_token, dst_token, amount, chain_id)
dst_amount = int(quote["dstAmount"])
gas = quote.get("estimatedGas", "N/A")
ts = datetime.now().strftime("%H:%M:%S")
change = ""
if prev_amount:
diff_pct = (dst_amount - prev_amount) / prev_amount * 100
change = f" ({diff_pct:+.3f}%)"
print(f"[{ts}] dst={dst_amount}{change} | gas={gas}")
prev_amount = dst_amount
except requests.HTTPError as e:
print(f"API error {e.response.status_code}: {e.response.text}")
except Exception as e:
print(f"Unexpected: {e}")
time.sleep(interval)
# Monitor native ETH -> USDC every 30 seconds on Ethereum
ETH_NATIVE = "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE"
monitor_price(ETH_NATIVE, USDC, 1 * 10**18, interval=30)
The swap endpoint goes one step further than quote — it returns the actual transaction calldata ready to sign and broadcast. You provide your wallet address, slippage tolerance, and amount; the API returns a tx object with everything needed. This pattern works across all EVM chains supported by 1inch — same Python code, just a different chain_id. Before you can swap, you need to approve the 1inch router to spend your source token. The current v6 router address is 0x111111125421cA6dc452d289314280a0f8842A65.
from web3 import Web3
RPC_URL = os.getenv("ETH_RPC_URL") # Alchemy, Infura, or your own node
PRIVATE_KEY = os.getenv("WALLET_PRIVATE_KEY")
w3 = Web3(Web3.HTTPProvider(RPC_URL))
WALLET = w3.eth.account.from_key(PRIVATE_KEY).address
ROUTER_V6 = "0x111111125421cA6dc452d289314280a0f8842A65"
def get_swap_calldata(
src_token: str,
dst_token: str,
amount: int,
slippage: float = 1.0,
chain_id: int = 1
) -> dict:
"""Get swap transaction data from 1inch. slippage is a percentage (1.0 = 1%)."""
url = f"{BASE_URL}/{chain_id}/swap"
params = {
"src": src_token,
"dst": dst_token,
"amount": str(amount),
"from": WALLET,
"slippage": slippage,
"disableEstimate": "false"
}
response = requests.get(url, headers=HEADERS, params=params)
response.raise_for_status()
return response.json()
def execute_swap(src_token: str, dst_token: str, amount: int, slippage: float = 1.0) -> str:
swap_data = get_swap_calldata(src_token, dst_token, amount, slippage)
tx = swap_data["tx"]
transaction = {
"from": WALLET,
"to": Web3.to_checksum_address(tx["to"]),
"data": tx["data"],
"value": int(tx["value"]),
"gas": int(int(tx["gas"]) * 1.25), # 25% buffer for safety
"gasPrice": int(tx["gasPrice"]),
"nonce": w3.eth.get_transaction_count(WALLET)
}
signed = w3.eth.account.sign_transaction(transaction, PRIVATE_KEY)
tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
print(f"Submitted: https://etherscan.io/tx/{tx_hash.hex()}")
return tx_hash.hex()
# Execute: 100 USDC -> WETH with 1% slippage
tx_hash = execute_swap(USDC, WETH, 100 * 10**6, slippage=1.0)
Always approve the 1inch v6 router (0x111111125421cA6dc452d289314280a0f8842A65) to spend your source token before calling execute_swap. Call approve() on the ERC-20 contract with this address and the amount you plan to swap. Without approval the transaction will revert.
The 1inch API returns predictable HTTP status codes. A 400 means bad parameters — wrong token address or invalid amount. A 429 means you're hitting rate limits, which requires backing off rather than retrying immediately. A 500 usually means the aggregator can't find a viable route for that pair, which happens with low-liquidity tokens. Here's a retry decorator that handles all three cases gracefully:
from functools import wraps
def with_retry(max_retries: int = 3, backoff: float = 2.0):
"""Decorator: automatic retry with exponential backoff."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.HTTPError as e:
code = e.response.status_code
if code == 400:
body = e.response.json()
raise ValueError(f"Bad request: {body.get('description', 'check params')}")
elif code == 429:
wait = backoff ** (attempt + 1)
print(f"Rate limited. Backing off {wait:.1f}s...")
time.sleep(wait)
elif code >= 500:
print(f"Server error {code}, retry {attempt + 1}/{max_retries}")
time.sleep(backoff)
else:
raise
except requests.ConnectionError:
print(f"Connection lost, retry {attempt + 1}/{max_retries}")
time.sleep(backoff)
raise RuntimeError(f"All {max_retries} retries exhausted")
return wrapper
return decorator
@with_retry(max_retries=3, backoff=2.0)
def safe_quote(src, dst, amount, chain_id=1):
return get_quote(src, dst, amount, chain_id)
# Usage
try:
quote = safe_quote(USDC, WETH, 100 * 10**6)
print(f"Got quote: {quote['dstAmount']}")
except ValueError as e:
print(f"Invalid parameters: {e}")
except RuntimeError as e:
print(f"API unreachable: {e}")
| Code | Meaning | Action |
|---|---|---|
| 200 | Success | Parse response normally |
| 400 | Bad params (invalid token, amount, or address) | Log error body, do not retry |
| 429 | Rate limit exceeded | Exponential backoff; reduce polling frequency |
| 500 | No route found or internal aggregator error | Retry with backoff; verify token has liquidity |
| 503 | API temporarily unavailable | Wait 10–30s then retry |
The 1inch API is one of the most practical additions to any DeFi Python stack. The quote endpoint gives you a real-time window into on-chain liquidity with zero gas cost, the swap endpoint abstracts away the complexity of routing across dozens of protocols, and the error handling patterns here will keep your bot running reliably under real-world conditions. Pair this execution layer with a signal platform like VoiceOfChain — which surfaces on-chain wallet movements before they hit centralized exchanges like Binance or Bybit — and you have the foundation of a serious automated DeFi trading system built entirely in Python.