Agentic Trading CG

agent_trade1.py

import os
import json
import time
import asyncio
import requests
import pandas as pd
from datetime import datetime, timedelta
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# =============================================================================
# Configuration
# =============================================================================
CG_API_KEY = os.getenv("CG_PRO_API_KEY") or os.getenv("CG_DEMO_API_KEY")
CG_BASE_URL = "https://pro-api.coingecko.com/api/v3/onchain"  # Pro API for paid features
CG_DEMO_URL = "https://api.coingecko.com/api/v3/onchain"      # Demo API for free endpoints

# Use Pro API if Pro key available, otherwise Demo
if os.getenv("CG_PRO_API_KEY"):
    BASE_URL = CG_BASE_URL
    API_KEY = os.getenv("CG_PRO_API_KEY")
    print("✓ Using CoinGecko Pro API")
else:
    BASE_URL = CG_DEMO_URL
    API_KEY = os.getenv("CG_DEMO_API_KEY")
    print("✓ Using CoinGecko Demo API")

# Blockscout API for transaction verification
BLOCKSCOUT_BASE_URL = "https://eth.blockscout.com/api/v2"

# Headers for CoinGecko API
HEADERS = {
    "accept": "application/json",
    "x-cg-pro-api-key": API_KEY
}

# =============================================================================
# Helper Functions
# =============================================================================
def cg_request(endpoint: str, params: dict = None) -> dict:
    """Make a request to CoinGecko API with error handling and rate limiting."""
    url = f"{BASE_URL}{endpoint}"
    
    for attempt in range(3):
        try:
            response = requests.get(url, headers=HEADERS, params=params, timeout=30)
            
            if response.status_code == 429:
                print(f"⚠ Rate limited. Waiting {2 ** attempt} seconds...")
                time.sleep(2 ** attempt)
                continue
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            if attempt == 2:
                print(f"✗ API request failed: {e}")
                raise
            time.sleep(1)
    
    return {}

agent_trade2.py

def get_trending_pools(network: str = "eth", limit: int = 10) -> pd.DataFrame:
    Fetch trending pools on a specific network.
    Args:

        network: Network ID (e.g. 'base', 'eth')
        limit: Maximum number of pools to return

    Returns:
        DataFrame with trending pool data

    print(f"Fetching trending pools on {network}...")
    endpoint = f"/networks/{network}/trending_pools"

    params = {"include": "base_token,quote_token"}
    data = cg_request(endpoint, params)
    pools = data.get("data", [])[:limit]   
    pool_data = []

    for pool in pools:
        attr = pool.get("attributes", {})
        relationships = pool.get("relationships", {})     

        pool_data.append({
            "pool_address": attr.get("address", ""),
            "pool_name": attr.get("name", "Unknown"),
            "dex": relationships.get("dex", {}).get("data", {}).get("id", "unknown"),
            "price_usd": float(attr.get("base_token_price_usd", 0) or 0),
            "volume_24h": float(attr.get("volume_usd", {}).get("h24", 0) or 0),
            "liquidity_usd": float(attr.get("reserve_in_usd", 0) or 0),
            "price_change_24h": float(attr.get("price_change_percentage", {}).get("h24", 0) or 0),
        })

    return pd.DataFrame(pool_data)