10 Real AI Agent Use Cases in Crypto and Finance (With Code)

From Theory to Practice

There is no shortage of articles saying "AI agents will transform finance." Most do not show you the code.

Here are 10 real use cases with working implementation examples.

1. Whale Alert Monitor

Monitor large on-chain transactions and get Telegram alerts:

from web3 import Web3
from telegram import Bot
import asyncio

w3 = Web3(Web3.HTTPProvider(os.getenv('ETH_RPC')))
bot = Bot(token=os.getenv('TELEGRAM_TOKEN'))

WHALE_THRESHOLD_ETH = 100  # Alert on moves > 100 ETH

async def monitor_whale_transactions():
    last_block = w3.eth.block_number
    
    while True:
        current_block = w3.eth.block_number
        
        for block_num in range(last_block + 1, current_block + 1):
            block = w3.eth.get_block(block_num, full_transactions=True)
            
            for tx in block['transactions']:
                value_eth = w3.from_wei(tx['value'], 'ether')
                
                if value_eth >= WHALE_THRESHOLD_ETH:
                    await bot.send_message(
                        chat_id=os.getenv('CHAT_ID'),
                        text=f"🐋 WHALE ALERT\n"
                             f"Amount: {value_eth:.0f} ETH (${value_eth * get_eth_price():,.0f})\n"
                             f"From: {tx['from'][:10]}...\n"
                             f"To: {tx['to'][:10]}...\n"
                             f"TX: https://etherscan.io/tx/{tx['hash'].hex()}"
                    )
        
        last_block = current_block
        await asyncio.sleep(12)  # New block every ~12 seconds

asyncio.run(monitor_whale_transactions())

2. Automated DeFi Liquidation Bot

Monitor under-collateralized loans and liquidate for profit:

from web3 import Web3
from eth_abi import decode

AAVE_POOL_ADDRESS = "0x87870Bca3F3fD6335C3F4ce8392D69350B4fA4E2"
MIN_HEALTH_FACTOR = 1.05  # Liquidate when health factor < 1.05

async def find_liquidatable_positions():
    """
    Monitor Aave for under-collateralized positions.
    When health factor < 1.0, position can be liquidated for profit.
    """
    pool = w3.eth.contract(address=AAVE_POOL_ADDRESS, abi=AAVE_ABI)
    
    # Listen for Borrow events to track positions
    borrow_filter = pool.events.Borrow.create_filter(fromBlock='latest')
    
    while True:
        for event in borrow_filter.get_new_entries():
            user = event['args']['user']
            account_data = pool.functions.getUserAccountData(user).call()
            health_factor = account_data[5] / 10**18
            
            if health_factor < 1.0:
                print(f"Liquidatable: {user} | HF: {health_factor:.3f}")
                # Execute liquidation for profit
                execute_liquidation(user, account_data)
        
        await asyncio.sleep(1)

3. Automated Research Report Agent

Generate daily market research reports automatically:

async def generate_daily_research():
    """Generate comprehensive daily market analysis."""
    
    # 1. Gather data
    btc_data = await fetch_market_data("BTC")
    eth_data = await fetch_market_data("ETH")
    news = await fetch_top_crypto_news(limit=20)
    onchain = await fetch_onchain_metrics(["BTC", "ETH"])
    
    # 2. Generate analysis with GPT-4o
    report = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Generate a professional daily crypto market report.

Market Data: {json.dumps({
    'BTC': btc_data, 
    'ETH': eth_data,
    'on_chain': onchain
}, indent=2)}

Top News: {chr(10).join(news[:10])}

Include:
1. Executive Summary (2 sentences)
2. BTC Analysis (technical + sentiment)
3. ETH Analysis
4. Key News Impact
5. Watchlist for Today
6. Risk Level: Low/Medium/High"""
        }]
    )
    
    report_text = report.choices[0].message.content
    
    # 3. Send to subscribers
    await send_telegram_message(report_text)
    # Or post to Substack, email list, etc.
    
    return report_text

4. Cross-Exchange Price Monitor

Real-time price comparison across 5 exchanges:

import ccxt.async_support as ccxt_async
import asyncio

exchanges = {
    'binance': ccxt_async.binance(),
    'bybit': ccxt_async.bybit(),
    'kraken': ccxt_async.kraken(),
    'okx': ccxt_async.okx(),
    'coinbase': ccxt_async.coinbase(),
}

async def monitor_prices(symbol: str = 'BTC/USDT'):
    while True:
        tasks = [
            exchange.fetch_ticker(symbol) 
            for exchange in exchanges.values()
        ]
        tickers = await asyncio.gather(*tasks, return_exceptions=True)
        
        prices = {}
        for (name, _), ticker in zip(exchanges.items(), tickers):
            if not isinstance(ticker, Exception):
                prices[name] = ticker['last']
        
        if prices:
            max_price = max(prices.values())
            min_price = min(prices.values())
            spread_pct = (max_price - min_price) / min_price * 100
            
            print(f"\n{symbol} Prices:")
            for exchange, price in sorted(prices.items(), key=lambda x: x[1]):
                indicator = "📉" if price == min_price else ("📈" if price == max_price else "  ")
                print(f"  {indicator} {exchange}: ${price:,.2f}")
            print(f"  Spread: {spread_pct:.3f}%")
            
            if spread_pct > 0.3:
                print(f"  ⚡ ARBITRAGE OPPORTUNITY: {spread_pct:.2f}% spread")
        
        await asyncio.sleep(1)

asyncio.run(monitor_prices())

5. Smart Contract Event Monitor + Alert

Watch any smart contract for specific events:

async def monitor_contract_events(
    contract_address: str,
    event_name: str,
    abi: list,
    alert_filter: callable = None
):
    """Generic contract event monitor."""
    contract = w3.eth.contract(address=contract_address, abi=abi)
    event = getattr(contract.events, event_name)
    event_filter = event.create_filter(fromBlock='latest')
    
    print(f"Watching {event_name} on {contract_address[:10]}...")
    
    while True:
        for evt in event_filter.get_new_entries():
            if alert_filter is None or alert_filter(evt):
                await bot.send_message(
                    chat_id=CHAT_ID,
                    text=f"📣 {event_name} EVENT\n{format_event(evt)}"
                )
        await asyncio.sleep(5)

6. Sentiment-Based Position Sizer

Adjust position sizes based on market sentiment:

def get_sentiment_adjusted_size(base_size: float, asset: str) -> float:
    """
    Increase position size when sentiment confirms the trade.
    Decrease when sentiment contradicts.
    """
    sentiment = aggregate_sentiment(asset, sources=['twitter', 'reddit', 'news'])
    
    # Sentiment score: -1.0 (max bearish) to +1.0 (max bullish)
    score = sentiment['composite_score']
    
    # Adjustment multiplier: 0.5x to 1.5x
    multiplier = 1.0 + (score * 0.5)  # ±50% adjustment
    
    adjusted_size = base_size * multiplier
    print(f"Sentiment: {score:+.2f} → Size multiplier: {multiplier:.2f}x")
    
    return adjusted_size

7. Automated Portfolio Rebalancer

Keep allocations on target without manual work:

TARGET_ALLOCATION = {
    'BTC': 0.40,    # 40%
    'ETH': 0.30,    # 30%
    'SOL': 0.10,    # 10%
    'USDC': 0.20,   # 20% stable
}

async def rebalance_if_needed(threshold: float = 0.05):
    """Rebalance if any asset drifts by more than threshold from target."""
    current = await get_portfolio_allocations()
    trades_needed = []
    
    for asset, target in TARGET_ALLOCATION.items():
        current_pct = current.get(asset, 0)
        drift = current_pct - target
        
        if abs(drift) > threshold:
            trades_needed.append({
                'asset': asset,
                'action': 'sell' if drift > 0 else 'buy',
                'drift': drift,
            })
    
    if trades_needed:
        print(f"Rebalancing {len(trades_needed)} assets...")
        for trade in trades_needed:
            await execute_rebalance_trade(trade)

8. Automated Tax Lot Tracker

Track cost basis for every trade automatically:

from dataclasses import dataclass
from datetime import datetime

@dataclass
class TaxLot:
    asset: str
    amount: float
    cost_basis: float  # price paid per unit
    date_acquired: datetime
    exchange: str

tax_lots: list[TaxLot] = []

def record_trade(asset: str, side: str, amount: float, price: float, exchange: str):
    if side == 'buy':
        tax_lots.append(TaxLot(
            asset=asset,
            amount=amount,
            cost_basis=price,
            date_acquired=datetime.now(),
            exchange=exchange
        ))
    elif side == 'sell':
        # FIFO method
        calculate_gain_loss(asset, amount, price)

def calculate_gain_loss(asset: str, sell_amount: float, sell_price: float):
    lots = [l for l in tax_lots if l.asset == asset and l.amount > 0]
    lots.sort(key=lambda x: x.date_acquired)  # FIFO
    
    total_gain = 0
    remaining = sell_amount
    
    for lot in lots:
        if remaining <= 0:
            break
        used = min(remaining, lot.amount)
        gain = (sell_price - lot.cost_basis) * used
        total_gain += gain
        lot.amount -= used
        remaining -= used
        print(f"  Lot {lot.date_acquired.date()}: ${gain:+,.2f} gain on {used:.4f} {asset}")
    
    print(f"Total gain/loss: ${total_gain:+,.2f}")

9. Fear & Greed Index Trader

Simple but effective: trade against extreme sentiment:

import requests

def get_fear_greed_index() -> dict:
    """Fetch current Crypto Fear & Greed Index."""
    response = requests.get("https://api.alternative.me/fng/?limit=1")
    data = response.json()['data'][0]
    return {'value': int(data['value']), 'classification': data['value_classification']}

def fear_greed_strategy(current_price: float) -> str:
    """Contrarian strategy: buy extreme fear, sell extreme greed."""
    fgi = get_fear_greed_index()
    
    if fgi['value'] <= 15:    # Extreme Fear
        return 'BUY'
    elif fgi['value'] >= 85:  # Extreme Greed
        return 'SELL'
    return 'HOLD'

10. Yield Optimizer Agent

Monitor 20+ DeFi protocols and move funds to the best rate:

PROTOCOLS = [
    {"name": "Aave USDC", "api": "aave_api", "min_tvl": 10_000_000},
    {"name": "Compound USDC", "api": "compound_api", "min_tvl": 5_000_000},
    {"name": "Morpho USDC", "api": "morpho_api", "min_tvl": 1_000_000},
]

async def optimize_yield(current_protocol: str, current_balance: float):
    """Move funds to highest-yielding protocol above minimum TVL."""
    rates = {}
    
    for protocol in PROTOCOLS:
        rate = await fetch_protocol_apy(protocol['api'])
        if await get_protocol_tvl(protocol['api']) >= protocol['min_tvl']:
            rates[protocol['name']] = rate
    
    best = max(rates, key=rates.get)
    improvement = rates[best] - rates.get(current_protocol, 0)
    
    # Only move if improvement > gas cost equivalent
    if improvement > 0.02 and best != current_protocol:
        print(f"Moving funds: {current_protocol} ({rates[current_protocol]:.1%}) → {best} ({rates[best]:.1%})")
        await migrate_funds(current_protocol, best, current_balance)

Combine Them

The real power comes from combining these. A complete system might use:

• Whale Monitor (#1) to feed into Sentiment (#6)
• News Agent (#3) to adjust the Rebalancer (#7)
• Yield Optimizer (#10) for the cash portion of the portfolio

All 10 use cases are implemented in our open-source tools. Download from the Tools page.