Real-Time Crypto Price Feeds for AI Agents: TWAP, OHLCV & Oracle Design

Price data quality is the most underappreciated input variable in AI trading systems. A language model that reasons brilliantly about market conditions and trade sizing can still lose money consistently if it is working from stale, manipulated, or incorrectly aggregated price data. The failure mode is not obvious: a 0.3% price error does not look catastrophic until it is compounded across hundreds of trades, or until a flash loan attack moves a DEX price 12% for one block and your agent fires an entry order at the wrong level. Understanding the different types of price data — spot, TWAP, and OHLCV — and knowing when to use each is a foundational skill for any agent that touches financial markets.

Why Price Data Quality Matters for AI Agents

Three concrete failure modes illustrate why this matters at a system level:

• Wrong trade, wrong price: Your agent uses a DEX spot price to decide to buy ETH because it appears cheap versus Chainlink. The DEX price is depressed because of a large sell order that landed 30 seconds ago. By the time the buy executes, the DEX has partially recovered and the agent buys at a worse-than-expected price, negating the edge.
• Flash loan manipulation: A sophisticated attacker uses a flash loan to temporarily move a Uniswap V2 pool price 8% in a single block. Your agent's stop-loss logic reads this as a genuine price drop and closes a position. The attacker exits, price recovers, and your agent has been forced out of a good position at the bottom.
• Stale data liquidation: Your agent monitors a DeFi position using a cached price that is 45 seconds old. The actual price has already crossed the liquidation threshold. The auto-repay transaction fires 45 seconds too late, and the position is liquidated at the penalty rate.

Spot Price vs TWAP: When to Use Each

These are not interchangeable. They measure fundamentally different things and have different manipulation profiles:

OHLCV Candles: How Agents Use Time-Series Price Data

OHLCV (Open, High, Low, Close, Volume) candles are the standard format for time-series price data. Each candle represents one time period and captures four price points plus traded volume. AI agents use different candle intervals for different purposes:

• 1-minute candles: High-frequency signal generation, scalping entries, volatility spikes. Noisy but immediate.
• 5-minute candles: Short-term trend identification, momentum signals, breakout detection. The most widely used timeframe for intraday trading agents.
• 1-hour candles: Regime detection ("are we trending or ranging?"), swing trade setups, macro context for lower-timeframe signals. Less noise, slower signal.
• Daily candles: Portfolio rebalancing decisions, long-horizon position sizing, fundamental context. Useful for agents managing multi-week positions.

The standard pattern is to use a higher-timeframe candle for context and a lower-timeframe candle for entry timing. An agent might use 1-hour candles to confirm an uptrend, then use 5-minute candles to find a pullback entry within that trend.

Price Manipulation Attacks on DEX Oracles and How to Defend

Uniswap V2 spot prices are among the most widely used price references in DeFi, and among the most frequently attacked. The attack is straightforward: borrow a large amount of one asset, dump it into the pool to move the price, trigger your target smart contract at the manipulated price, then restore the pool in the same transaction. Uniswap V3 and later versions offer built-in TWAP oracles via the observe() function, which makes manipulation require sustained capital commitment across multiple blocks.

For AI agents, three defensive techniques apply:

• Minimum TWAP window of 5 minutes. An attack requires holding an abnormal price for the full window. Flash loans are limited to one block (<15 seconds), so a 5-minute TWAP is inherently immune to flash loan manipulation.
• Outlier rejection. If a new price reading is more than 3% away from the preceding 10-reading average, flag it as a potential anomaly rather than accepting it immediately.
• Multi-source validation. Cross-check a Uniswap TWAP against a Chainlink price feed. If the deviation exceeds 2%, halt trading until the discrepancy resolves.

Batch Price Fetching: 100 Tokens in One Call

Sequential price fetching is a common performance bottleneck in multi-asset agents. If your agent monitors 100 tokens and makes one HTTP call per token, that is 100 sequential round trips. At 50ms average latency per call, that is 5 seconds to refresh your universe — long enough for significant price movement in volatile markets.

from purpleflea import PurpleFleaClient
import os

pf = PurpleFleaClient(api_key=os.environ["PURPLEFLEA_API_KEY"])

# --- Spot price: single token ---
async def get_spot(symbol: str) -> dict:
    return await pf.prices.spot(symbol)
    # {"symbol": "ETH", "price_usd": 2841.30, "timestamp": 1740000000}

# --- TWAP: 30-minute window ---
async def get_twap(symbol: str, window_minutes: int = 30) -> dict:
    return await pf.prices.twap(symbol, window_minutes=window_minutes)
    # {"symbol": "ETH", "twap_usd": 2836.80, "window_minutes": 30,
    #  "samples": 360, "from_timestamp": 1739998200}

# --- OHLCV candles: last 24 1-hour candles ---
async def get_ohlcv(symbol: str, interval: str = "1h", limit: int = 24) -> list:
    return await pf.prices.ohlcv(symbol, interval=interval, limit=limit)
    # [{"t": 1739998200, "o": 2810.0, "h": 2856.5, "l": 2805.0, "c": 2841.3, "v": 142500.0}, ...]

# --- Batch fetch: 100 tokens in a single HTTP call ---
WATCHLIST = [
    "ETH", "BTC", "SOL", "AVAX", "ARB", "OP", "MATIC", "LINK",
    "UNI", "AAVE", "CRV", "MKR", "COMP", "SNX", "BAL", "YFI",
    # ... up to 100 tokens
]

async def get_all_prices() -> dict[str, float]:
    # One HTTP call returns all prices — 10x faster than sequential
    results = await pf.prices.batch_spot(symbols=WATCHLIST)
    return {r["symbol"]: r["price_usd"] for r in results}
    # {"ETH": 2841.30, "BTC": 94200.50, "SOL": 188.40, ...}

Multi-Source Validation: Chainlink vs DEX TWAP

The most robust price pipeline for a production agent cross-references at least two independent sources and flags divergence. Chainlink is an off-chain oracle network: data comes from professional data providers who aggregate CEX prices. Uniswap V3 TWAP is on-chain: derived purely from DEX activity. They should agree within ~1% under normal conditions. When they diverge by more, something unusual is happening — either genuine CEX/DEX price discovery lag, or an oracle manipulation attempt.

import asyncio

MAX_DEVIATION_PCT = 2.0  # Flag if sources diverge by more than 2%

async def get_validated_price(pf, symbol: str) -> dict:
    """Fetch price from Chainlink and Uniswap V3 TWAP. Flag if divergence > 2%."""
    chainlink, dex_twap = await asyncio.gather(
        pf.prices.chainlink(symbol),
        pf.prices.twap(symbol, window_minutes=30, source="uniswap_v3"),
    )

    cl_price = chainlink["price_usd"]
    dex_price = dex_twap["twap_usd"]
    deviation_pct = abs(cl_price - dex_price) / cl_price * 100

    if deviation_pct > MAX_DEVIATION_PCT:
        return {
            "symbol": symbol,
            "status": "ANOMALY",
            "chainlink_usd": cl_price,
            "dex_twap_usd": dex_price,
            "deviation_pct": deviation_pct,
            "action": "halt_trading",  # Do not trade on anomalous prices
        }

    # Use the average of both sources as the canonical price
    canonical = (cl_price + dex_price) / 2
    return {
        "symbol": symbol,
        "status": "OK",
        "price_usd": canonical,
        "deviation_pct": deviation_pct,
    }


async def anomaly_detection_loop(pf, symbols: list[str], poll_interval: int = 30):
    """Continuously validate prices and alert on anomalies."""
    while True:
        tasks = [get_validated_price(pf, s) for s in symbols]
        results = await asyncio.gather(*tasks)

        anomalies = [r for r in results if r["status"] == "ANOMALY"]
        if anomalies:
            for a in anomalies:
                print(f"PRICE ANOMALY: {a['symbol']} — CL: ${a['chainlink_usd']:.2f}, "
                      f"DEX TWAP: ${a['dex_twap_usd']:.2f}, deviation: {a['deviation_pct']:.2f}%")

        await asyncio.sleep(poll_interval)

Use Case: Take-Profit Agent Using TWAP Monitoring

This is a complete, practical example: an agent that monitors ETH's TWAP against an entry price and triggers a take-profit sell when it detects a sustained 20% gain. Using TWAP instead of spot prevents the agent from taking profit on a brief spike that immediately reverses — a common failure mode of spot-based take-profit logic.

import asyncio
from purpleflea import PurpleFleaClient
import os

pf = PurpleFleaClient(api_key=os.environ["PURPLEFLEA_API_KEY"], mnemonic=os.environ["AGENT_MNEMONIC"])

async def run_take_profit_agent(
    symbol: str,
    entry_price_usd: float,
    take_profit_pct: float,
    eth_amount: float,
    twap_window_minutes: int = 30,
):
    """Monitor TWAP and sell when it confirms a take-profit level."""
    target_price = entry_price_usd * (1 + take_profit_pct / 100)
    print(f"Monitoring {symbol}. Entry: ${entry_price_usd:,.2f}. "
          f"TP target: ${target_price:,.2f} (+{take_profit_pct}%)")

    while True:
        # Get validated price (cross-checks Chainlink + Uniswap TWAP)
        price_data = await get_validated_price(pf, symbol)

        if price_data["status"] == "ANOMALY":
            print(f"Price anomaly detected for {symbol}. Skipping this cycle.")
            await asyncio.sleep(60)
            continue

        twap = price_data["price_usd"]
        gain_pct = (twap - entry_price_usd) / entry_price_usd * 100
        print(f"{symbol} TWAP: ${twap:,.2f} ({gain_pct:+.2f}% from entry)")

        if twap >= target_price:
            print(f"Take-profit target reached. Selling {eth_amount} {symbol}...")
            sell_result = await pf.trading.close_position(symbol=symbol)
            print(f"Sold. Net PnL: ${sell_result['realized_pnl_usd']:+,.2f}")
            return sell_result

        await asyncio.sleep(60)  # Poll every 60 seconds


# Example: enter long ETH at $2,400, take profit when 30-min TWAP hits +20%
if __name__ == "__main__":
    asyncio.run(run_take_profit_agent(
        symbol="ETH",
        entry_price_usd=2400.00,
        take_profit_pct=20.0,  # exit when TWAP confirms +20%
        eth_amount=2.5,
        twap_window_minutes=30,
    ))

A reliable price oracle pipeline is the foundation of every competent AI trading agent. The distinction between spot and TWAP should be internalized at the architecture level — not decided per-trade. Batch fetching eliminates the latency penalty of monitoring large token universes. Multi-source validation with automatic anomaly detection keeps your agent safe from the manipulation attempts that will inevitably target any on-chain price reference it uses. Purple Flea's price oracle API provides all three primitives — spot, TWAP, and OHLCV — with batch and multi-source support built in, so you can focus on trading logic rather than data infrastructure.