How to Build an AI Agent That Reads News and Trades Automatically

Why News Trading With AI Works

The crypto market reacts immediately to news. But "immediately" means different things for different participants:

• Institutional news desks: under 100ms
• Professional traders: 1–5 seconds
• Regular traders: 30 seconds – 5 minutes
• Most people: 5–30 minutes (after reading a notification)

An AI agent can process news and execute a trade in 2–5 seconds. That puts you ahead of most retail traders without needing institutional infrastructure.

Architecture Overview

News Sources (RSS, APIs, Twitter)
        ↓
News Fetcher (runs every 30s)
        ↓
Deduplication (skip already-processed items)
        ↓
AI Analysis (GPT-4o-mini for speed)
        ↓
Signal Filter (only high-confidence signals)
        ↓
Risk Check (position limits, drawdown limits)
        ↓
Trade Executor (ccxt → exchange)
        ↓
Telegram Notification

Step 1: News Sources

import asyncio
import aiohttp
import feedparser
from dataclasses import dataclass
from datetime import datetime

@dataclass
class NewsItem:
    title: str
    summary: str
    url: str
    published: datetime
    source: str

NEWS_SOURCES = [
    # Crypto news RSS feeds
    {"url": "https://cointelegraph.com/rss", "source": "CoinTelegraph"},
    {"url": "https://decrypt.co/feed", "source": "Decrypt"},
    {"url": "https://coindesk.com/arc/outboundfeeds/rss/", "source": "CoinDesk"},
    {"url": "https://www.theblock.co/rss.xml", "source": "TheBlock"},
    {"url": "https://bitcoinmagazine.com/feed", "source": "BitcoinMagazine"},
    # Macro news that affects crypto
    {"url": "https://feeds.reuters.com/reuters/businessNews", "source": "Reuters"},
]

processed_urls = set()  # Track already-processed items

async def fetch_news() -> list[NewsItem]:
    """Fetch latest news from all sources."""
    all_items = []
    
    for source in NEWS_SOURCES:
        try:
            feed = feedparser.parse(source["url"])
            for entry in feed.entries[:5]:  # Last 5 items per source
                if entry.link not in processed_urls:
                    published = datetime(*entry.published_parsed[:6]) if hasattr(entry, 'published_parsed') else datetime.now()
                    
                    all_items.append(NewsItem(
                        title=entry.title,
                        summary=getattr(entry, 'summary', '')[:500],
                        url=entry.link,
                        published=published,
                        source=source["source"]
                    ))
                    processed_urls.add(entry.link)
        except Exception as e:
            print(f"Error fetching {source['source']}: {e}")
    
    return sorted(all_items, key=lambda x: x.published, reverse=True)

Step 2: AI Analysis Pipeline

import json
from openai import OpenAI

client = OpenAI()

TRACKED_ASSETS = {
    "Bitcoin": "BTC",
    "BTC": "BTC",
    "Ethereum": "ETH",
    "ETH": "ETH",
    "Solana": "SOL",
    "SOL": "SOL",
}

def analyze_news_for_trading(item: NewsItem) -> dict | None:
    """
    Analyze a news item for trading signals.
    Returns None if not actionable, signal dict if actionable.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Fast and cheap for screening
        messages=[
            {
                "role": "system",
                "content": """You are a crypto trading analyst specializing in news-driven trades.

Analyze news for SHORT-TERM price impact (1-4 hours after the news).

Return JSON:
{
  "tradeable": true/false,
  "asset": "BTC/ETH/SOL/null",
  "direction": "buy/sell/null",
  "confidence": 0.0-1.0,
  "impact_size": "small/medium/large",
  "time_window": "minutes/hours",
  "reasoning": "brief"
}

Only mark tradeable=true for clear, significant news. High bar: confidence must be 0.8+.

Examples of TRADEABLE news:
- "BlackRock increases Bitcoin holdings by $1B" → buy BTC, confidence: 0.85
- "SEC sues major crypto exchange" → sell affected assets, confidence: 0.90
- "Ethereum network upgrade successful" → buy ETH, confidence: 0.80

Examples of NOT tradeable:
- General market analysis articles
- Price predictions from analysts
- Old news being recycled"""
            },
            {
                "role": "user",
                "content": f"Source: {item.source}\nTitle: {item.title}\nSummary: {item.summary}"
            }
        ],
        response_format={"type": "json_object"},
        temperature=0.1
    )
    
    result = json.loads(response.choices[0].message.content)
    
    if not result.get("tradeable") or result.get("confidence", 0) < 0.8:
        return None
    
    return result

Step 3: Risk Management Layer

import ccxt
from dataclasses import dataclass

@dataclass
class RiskParams:
    max_position_pct: float = 0.05    # Max 5% of portfolio per trade
    max_daily_trades: int = 10        # Max 10 trades per day
    max_daily_loss_pct: float = 0.03  # Stop trading if down 3% today
    min_confidence: float = 0.85      # Only trade high-confidence signals

class RiskManager:
    def __init__(self, exchange: ccxt.Exchange, params: RiskParams):
        self.exchange = exchange
        self.params = params
        self.daily_trades = 0
        self.daily_pnl = 0.0
        self._reset_day()
    
    def _reset_day(self):
        self.daily_trades = 0
        self.daily_pnl = 0.0
        self.last_reset = datetime.now().date()
    
    def _check_daily_reset(self):
        if datetime.now().date() > self.last_reset:
            self._reset_day()
    
    def approve_trade(self, signal: dict) -> tuple[bool, str, float]:
        """
        Returns (approved, reason, position_size_usd)
        """
        self._check_daily_reset()
        
        if self.daily_trades >= self.params.max_daily_trades:
            return False, f"Daily trade limit ({self.params.max_daily_trades}) reached", 0
        
        if self.daily_pnl < -self.params.max_daily_loss_pct:
            return False, f"Daily loss limit ({self.params.max_daily_loss_pct * 100}%) reached", 0
        
        if signal.get("confidence", 0) < self.params.min_confidence:
            return False, f"Confidence too low: {signal['confidence']:.2f}", 0
        
        # Calculate position size
        balance = self.exchange.fetch_balance()
        total_usd = sum(
            balance['total'].get('USDT', 0) + balance['total'].get('BUSD', 0)
        )
        position_size = total_usd * self.params.max_position_pct
        
        # Scale by confidence and impact size
        confidence_multiplier = signal.get("confidence", 0.8)
        impact_multiplier = {"small": 0.5, "medium": 1.0, "large": 1.5}.get(
            signal.get("impact_size", "medium"), 1.0
        )
        final_size = position_size * confidence_multiplier * min(impact_multiplier, 1.5)
        
        return True, "Approved", final_size

Step 4: The Main Agent Loop

import ccxt
import asyncio
from telegram import Bot

async def news_trading_agent():
    exchange = ccxt.binance({
        'apiKey': os.getenv('BINANCE_API_KEY'),
        'secret': os.getenv('BINANCE_SECRET'),
    })
    risk = RiskManager(exchange, RiskParams())
    telegram = Bot(token=os.getenv('TELEGRAM_TOKEN'))
    chat_id = int(os.getenv('TELEGRAM_CHAT_ID'))
    
    print("📰 News trading agent started")
    
    while True:
        # Fetch latest unprocessed news
        news_items = await fetch_news()
        
        for item in news_items:
            # AI analysis
            signal = analyze_news_for_trading(item)
            
            if not signal:
                continue
            
            print(f"📊 Signal: {signal['direction'].upper()} {signal['asset']} "
                  f"(confidence: {signal['confidence']:.0%})")
            
            # Risk check
            approved, reason, size_usd = risk.approve_trade(signal)
            
            if not approved:
                print(f"⛔ Trade blocked: {reason}")
                continue
            
            # Execute trade
            symbol = f"{signal['asset']}/USDT"
            ticker = exchange.fetch_ticker(symbol)
            amount = size_usd / ticker['last']
            
            try:
                if signal['direction'] == 'buy':
                    order = exchange.create_market_buy_order(symbol, amount)
                else:
                    order = exchange.create_market_sell_order(symbol, amount)
                
                # Notify via Telegram
                await telegram.send_message(
                    chat_id=chat_id,
                    text=(
                        f"{'🟢' if signal['direction'] == 'buy' else '🔴'} "
                        f"NEWS TRADE EXECUTED\n\n"
                        f"Asset: {signal['asset']}\n"
                        f"Action: {signal['direction'].upper()}\n"
                        f"Size: ${size_usd:.0f}\n"
                        f"Confidence: {signal['confidence']:.0%}\n\n"
                        f"News: {item.title}\n"
                        f"Reason: {signal['reasoning']}"
                    )
                )
                risk.daily_trades += 1
                
            except Exception as e:
                print(f"Trade execution failed: {e}")
        
        await asyncio.sleep(30)  # Poll every 30 seconds

# Run
asyncio.run(news_trading_agent())

Adding Premium News Sources

Basic RSS feeds are free but slow. For an extra edge:

• Chainanalysis Market Intel — On-chain data + news
• Santiment — Crypto-specific news API with sentiment scores
• Benzinga Pro — Real-time news feed with crypto section
• TheBlock API — Deep crypto news

Performance Expectations

Based on this strategy framework:

• Trade frequency: 1–5 trades per week (only genuine signals)
• Win rate: 55–65% (directional accuracy on strong signals)
• Average trade duration: 1–4 hours
• Return: 15–40% annually with proper risk management

The key is patience. Do not lower the confidence threshold to trade more — that destroys the edge.

Get the Full Code

The complete news trading agent with additional sources, better deduplication, and performance logging is available on our Tools page.