Building AI Agents

import requests
import time
from datetime import datetime
from typing import Dict, List, Optional
import os

class PortfolioMonitorAgent:
    """Autonomous agent that monitors crypto portfolios using Octav API"""

    def __init__(self, api_key: str, check_interval: int = 300):
        self.api_key = api_key
        self.base_url = 'https://api.octav.fi'
        self.check_interval = check_interval  # seconds
        self.headers = {'Authorization': f'Bearer {api_key}'}
        self.last_values: Dict[str, float] = {}

    def get_portfolio(self, address: str) -> Optional[Dict]:
        """Fetch current portfolio data"""
        try:
            response = requests.get(
                f'{self.base_url}/v1/portfolio?addresses={address}',
                headers=self.headers,
                timeout=30
            )
            response.raise_for_status()
            return response.json()[0]
        except requests.exceptions.RequestException as e:
            print(f"Error fetching portfolio: {e}")
            return None

    def check_balance_threshold(self, address: str, min_balance: float = 1000):
        """Alert if portfolio falls below threshold"""
        portfolio = self.get_portfolio(address)
        if not portfolio:
            return

        current_value = float(portfolio['networth'])

        if current_value < min_balance:
            self.send_alert(
                f"⚠️ Low Balance Alert\n"
                f"Address: {address[:10]}...\n"
                f"Current: ${current_value:,.2f}\n"
                f"Threshold: ${min_balance:,.2f}"
            )

    def check_significant_change(self, address: str, threshold_pct: float = 5.0):
        """Alert on significant portfolio value changes"""
        portfolio = self.get_portfolio(address)
        if not portfolio:
            return

        current_value = float(portfolio['networth'])
        last_value = self.last_values.get(address)

        if last_value:
            change = current_value - last_value
            change_pct = (change / last_value) * 100

            if abs(change_pct) >= threshold_pct:
                direction = "📈" if change > 0 else "📉"
                self.send_alert(
                    f"{direction} Significant Change Detected\n"
                    f"Address: {address[:10]}...\n"
                    f"Change: {change_pct:+.2f}%\n"
                    f"From: ${last_value:,.2f}\n"
                    f"To: ${current_value:,.2f}"
                )

        self.last_values[address] = current_value

    def generate_daily_summary(self, address: str):
        """Generate and send daily portfolio summary"""
        portfolio = self.get_portfolio(address)
        if not portfolio:
            return

        # Calculate chain distribution
        chains_summary = []
        for chain_key, chain in portfolio['chains'].items():
            chains_summary.append(
                f"  {chain['name']}: ${float(chain['value']):,.2f}"
            )

        # Calculate protocol distribution
        protocols_summary = []
        for protocol_key, protocol in portfolio['assetByProtocols'].items():
            protocols_summary.append(
                f"  {protocol['name']}: ${float(protocol['value']):,.2f}"
            )

        summary = f"""
📊 Daily Portfolio Summary
Address: {address[:10]}...
Total Value: ${float(portfolio['networth']):,.2f}

Chains:
{chr(10).join(chains_summary)}

Protocols:
{chr(10).join(protocols_summary)}

Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        """

        self.send_alert(summary.strip())

    def monitor(self, addresses: List[str], config: Dict):
        """Main monitoring loop"""
        print(f"Starting portfolio monitor for {len(addresses)} addresses")
        print(f"Check interval: {self.check_interval}s")

        while True:
            try:
                for address in addresses:
                    # Run configured checks
                    if config.get('check_threshold'):
                        self.check_balance_threshold(
                            address,
                            config.get('min_balance', 1000)
                        )

                    if config.get('check_changes'):
                        self.check_significant_change(
                            address,
                            config.get('change_threshold', 5.0)
                        )

                    # Daily summary at midnight
                    if config.get('daily_summary'):
                        now = datetime.now()
                        if now.hour == 0 and now.minute < 5:
                            self.generate_daily_summary(address)

                    # Respect rate limits
                    time.sleep(2)

                time.sleep(self.check_interval)

            except KeyboardInterrupt:
                print("\nMonitoring stopped by user")
                break
            except Exception as e:
                print(f"Error in monitoring loop: {e}")
                time.sleep(60)  # Wait before retrying

    def send_alert(self, message: str):
        """Send alert notification (implement your preferred method)"""
        print(f"\n{'='*50}")
        print(message)
        print(f"{'='*50}\n")

        # TODO: Implement your notification method:
        # - Email via SendGrid/Mailgun
        # - SMS via Twilio
        # - Slack/Discord webhook
        # - Telegram bot
        # - Push notification service

# Usage Example
if __name__ == "__main__":
    agent = PortfolioMonitorAgent(
        api_key=os.getenv('OCTAV_API_KEY'),
        check_interval=300  # Check every 5 minutes
    )

    # Configure monitoring
    config = {
        'check_threshold': True,
        'min_balance': 5000,
        'check_changes': True,
        'change_threshold': 3.0,  # Alert on 3%+ changes
        'daily_summary': True
    }

    # Start monitoring
    addresses = [
        '0x123...',  # Your addresses
        '0x456...',
    ]

    agent.monitor(addresses, config)