AI Agents for Crypto Research: Build Your Own Alpha Generator

What Is Alpha in Crypto?

Alpha is information or analysis advantage that lets you trade profitably against the market. In crypto, genuine alpha sources:

1. Protocol upgrades before announcement: GitHub commits reveal roadmap
2. Governance outcomes: Voting results affect token price
3. Whale accumulation: Large wallets buying = smart money signal
4. Developer activity: Commit frequency predicts protocol health
5. Narrative shifts: Twitter/X topics spreading before mainstream

An AI research agent can monitor all of these simultaneously, 24/7.

Building the Alpha Generation Agent

import anthropic
import httpx
import asyncio
from datetime import datetime, timedelta
import feedparser

claude = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

class CryptoAlphaAgent:
    """
    Autonomous agent that discovers crypto alpha by analyzing:
    - GitHub activity for DeFi protocols
    - Governance proposals and voting
    - On-chain large wallet movements
    - Social momentum on X/Twitter
    - Developer activity metrics
    """
    
    def __init__(self):
        self.discoveries = []
        self.last_run = None
    
    async def check_github_activity(self, repos: list[dict]) -> list[dict]:
        """Monitor GitHub repos for significant commits"""
        
        signals = []
        async with httpx.AsyncClient() as client:
            for repo in repos:
                response = await client.get(
                    f"https://api.github.com/repos/{repo['owner']}/{repo['name']}/commits",
                    params={"since": (datetime.now() - timedelta(hours=24)).isoformat()},
                    headers={"Authorization": f"token {GITHUB_TOKEN}"}
                )
                
                commits = response.json() if response.status_code == 200 else []
                
                if len(commits) >= 5:  # High activity day
                    # Get commit messages
                    messages = [c['commit']['message'][:200] for c in commits[:10]]
                    
                    # Use Claude to assess significance
                    analysis = claude.messages.create(
                        model="claude-3-5-sonnet-20241022",
                        max_tokens=200,
                        messages=[{
                            "role": "user",
                            "content": f"""GitHub commits for {repo['name']} in last 24h:
{chr(10).join(messages)}

Is there anything significant here that could affect the token price?
Respond with: SIGNIFICANT or NOT_SIGNIFICANT, followed by one sentence of reasoning."""
                        }]
                    )
                    
                    result = analysis.content[0].text
                    
                    if "SIGNIFICANT" in result and "NOT_" not in result:
                        signals.append({
                            'type': 'github_activity',
                            'protocol': repo['name'],
                            'commits': len(commits),
                            'analysis': result,
                            'confidence': 'MEDIUM',
                        })
        
        return signals
    
    async def check_governance_proposals(self, protocols: list[str]) -> list[dict]:
        """Monitor governance proposals on Snapshot.org"""
        
        signals = []
        async with httpx.AsyncClient() as client:
            for protocol in protocols:
                response = await client.post(
                    'https://hub.snapshot.org/graphql',
                    json={
                        'query': '''
                        query ($space: String!) {
                            proposals(
                                first: 5
                                skip: 0
                                where: { space: $space, state: "active" }
                                orderBy: "created"
                                orderDirection: desc
                            ) {
                                id
                                title
                                body
                                choices
                                start
                                end
                                scores
                                votes
                                state
                            }
                        }''',
                        'variables': {'space': protocol}
                    }
                )
                
                proposals = response.json().get('data', {}).get('proposals', [])
                
                for proposal in proposals:
                    # High vote count = significant governance event
                    if proposal.get('votes', 0) > 1000:
                        analysis = claude.messages.create(
                            model="claude-3-5-sonnet-20241022",
                            max_tokens=150,
                            messages=[{
                                "role": "user",
                                "content": f"""Governance proposal for {protocol}:
Title: {proposal['title']}
Votes: {proposal['votes']}
Choices: {proposal['choices']}

Will this proposal, if passed, be bullish or bearish for the token price?
One sentence answer."""
                            }]
                        )
                        
                        signals.append({
                            'type': 'governance',
                            'protocol': protocol,
                            'proposal': proposal['title'],
                            'votes': proposal['votes'],
                            'analysis': analysis.content[0].text,
                        })
        
        return signals
    
    async def check_defillama_tvl_changes(self, min_change_pct: float = 10) -> list[dict]:
        """Detect significant TVL movements that precede price moves"""
        
        async with httpx.AsyncClient() as client:
            response = await client.get('https://api.llama.fi/protocols')
            protocols = response.json()
        
        signals = []
        for p in protocols[:100]:  # Top 100 protocols by TVL
            change_24h = p.get('change_1d', 0) or 0
            change_7d = p.get('change_7d', 0) or 0
            
            if abs(change_24h) >= min_change_pct:
                direction = "📈 INFLOW" if change_24h > 0 else "📉 OUTFLOW"
                
                signals.append({
                    'type': 'tvl_movement',
                    'protocol': p['name'],
                    'tvl': p.get('tvl', 0),
                    'change_24h': change_24h,
                    'change_7d': change_7d,
                    'direction': direction,
                    'token': p.get('symbol', 'N/A'),
                })
        
        return sorted(signals, key=lambda x: abs(x['change_24h']), reverse=True)[:5]
    
    async def synthesize_daily_alpha(self) -> str:
        """Synthesize all signals into actionable alpha report"""
        
        # Gather all signals
        github_signals = await self.check_github_activity([
            {'owner': 'aave', 'name': 'aave-v3-core'},
            {'owner': 'Uniswap', 'name': 'v3-core'},
            {'owner': 'curvefi', 'name': 'curve-contract'},
        ])
        
        governance_signals = await self.check_governance_proposals([
            'aave.eth', 'uniswap', 'compound-governance'
        ])
        
        tvl_signals = await self.check_defillama_tvl_changes(min_change_pct=15)
        
        all_signals = github_signals + governance_signals + tvl_signals
        
        if not all_signals:
            return "No significant alpha signals detected in the last 24 hours."
        
        # Use Claude to synthesize into a trading report
        signals_text = json.dumps(all_signals, indent=2)
        
        response = claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=800,
            messages=[{
                "role": "user",
                "content": f"""You are a crypto hedge fund researcher. Analyze these signals and write a concise alpha report:

{signals_text}

Format as:
## 🔍 Daily Alpha Report — {datetime.now().strftime('%B %d, %Y')}

### High Conviction Ideas
[1-3 specific trade ideas with brief reasoning]

### Watch List
[2-3 things to monitor]

### Key Risk
[1 major risk today]

Be specific. Use coin names. Give actionable ideas."""
            }]
        )
        
        return response.content[0].text
    
    async def run_daily(self):
        """Generate and send daily alpha report"""
        
        print("🔍 Generating alpha report...")
        report = await self.synthesize_daily_alpha()
        
        # Send via Telegram
        import httpx
        async with httpx.AsyncClient() as client:
            await client.post(
                f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage',
                json={
                    'chat_id': CHAT_ID,
                    'text': report,
                    'parse_mode': 'Markdown'
                }
            )
        
        print(f"✅ Alpha report sent at {datetime.now().strftime('%H:%M')}")
        return report

# Run daily
agent = CryptoAlphaAgent()
import schedule
schedule.every().day.at("07:00").do(lambda: asyncio.run(agent.run_daily()))

while True:
    schedule.run_pending()
    import time; time.sleep(60)

The Compounding Knowledge Edge

The alpha generation agent gets better over time because you can save its outputs, track which signals led to profitable trades, and fine-tune the prompts based on what actually worked.

class AlphaTracker:
    """Track which alpha signals were profitable"""
    
    def log_signal_outcome(self, signal: dict, outcome_pct: float):
        """Record whether a signal led to profit"""
        record = {
            **signal,
            'outcome_pct': outcome_pct,
            'profitable': outcome_pct > 0,
        }
        
        with open('alpha_outcomes.jsonl', 'a') as f:
            f.write(json.dumps(record) + '\n')
    
    def get_best_signal_types(self) -> dict:
        """Which signal types have the best track record?"""
        import pandas as pd
        
        records = [json.loads(l) for l in open('alpha_outcomes.jsonl')]
        df = pd.DataFrame(records)
        
        return df.groupby('type').agg({
            'profitable': 'mean',
            'outcome_pct': 'mean',
        }).to_dict()

The best alpha comes from combining signals no one else is combining. A GitHub commit is meaningless alone. A GitHub commit + governance vote + TVL inflow + whale accumulation in the same protocol in 24 hours — that's the kind of multi-signal confluence that professional crypto funds look for.