How to Track DePIN Adoption with Ollama: Real-World Usage Metrics

# Install Ollama on Linux/macOS
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for Data Analysis
ollama pull llama2:13b

# Verify installation
ollama list

import requests
import json
import subprocess
from datetime import datetime, timedelta

class DePINTracker:
    def __init__(self, network_endpoints):
        self.endpoints = network_endpoints
        self.metrics = {}
        
    def collect_node_data(self, endpoint):
        """Fetch node statistics from DePIN network API"""
        try:
            response = requests.get(f"{endpoint}/api/v1/stats")
            return response.json()
        except requests.RequestException as e:
            print(f"Error fetching data from {endpoint}: {e}")
            return None
    
    def analyze_with_ollama(self, data_payload):
        """Process network data using Ollama"""
        prompt = f"""
        Analyze this DePIN network data and extract key adoption metrics:
        {json.dumps(data_payload, indent=2)}
        
        Focus on:
        1. Active node count trends
        2. Geographic distribution patterns
        3. Hardware utilization rates
        4. Token reward distribution
        5. Network capacity growth
        
        Provide specific numbers and percentage changes.
        """
        
        # Send to Ollama for analysis
        result = subprocess.run([
            'ollama', 'run', 'llama2:13b', prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Example usage
networks = [
    "https://api.helium.com",
    "https://api.filecoin.io", 
    "https://api.render.com"
]

tracker = DePINTracker(networks)

def calculate_active_nodes(self):
    """Identify truly active infrastructure nodes"""
    active_criteria = {
        'uptime_threshold': 0.95,  # 95% uptime requirement
        'data_processed': 1000,    # Minimum daily data processing
        'rewards_earned': 0.1      # Minimum token rewards
    }
    
    active_nodes = []
    for node in self.node_data:
        if (node['uptime'] >= active_criteria['uptime_threshold'] and
            node['daily_data'] >= active_criteria['data_processed'] and
            node['rewards'] >= active_criteria['rewards_earned']):
            active_nodes.append(node)
    
    return len(active_nodes)

def analyze_geographic_spread(self):
    """Calculate geographic distribution metrics"""
    regions = {}
    for node in self.node_data:
        region = node.get('region', 'unknown')
        if region not in regions:
            regions[region] = 0
        regions[region] += 1
    
    # Calculate distribution index (0 = centralized, 1 = perfectly distributed)
    total_nodes = sum(regions.values())
    expected_per_region = total_nodes / len(regions)
    
    distribution_score = 1 - sum(
        abs(count - expected_per_region) / total_nodes 
        for count in regions.values()
    ) / 2
    
    return {
        'regions': regions,
        'distribution_score': distribution_score,
        'geographic_diversity': len(regions)
    }

def track_utilization_metrics(self):
    """Monitor network capacity and revenue patterns"""
    metrics = {
        'total_capacity': 0,
        'used_capacity': 0,
        'revenue_generated': 0,
        'avg_session_duration': 0
    }
    
    for node in self.node_data:
        metrics['total_capacity'] += node.get('max_capacity', 0)
        metrics['used_capacity'] += node.get('current_usage', 0)
        metrics['revenue_generated'] += node.get('earnings', 0)
    
    # Calculate utilization rate
    if metrics['total_capacity'] > 0:
        metrics['utilization_rate'] = (
            metrics['used_capacity'] / metrics['total_capacity']
        ) * 100
    
    return metrics

def generate_adoption_report(self, historical_data):
    """Create comprehensive adoption analysis using Ollama"""
    
    # Prepare data summary for analysis
    data_summary = {
        'current_metrics': self.get_current_metrics(),
        'historical_trends': historical_data,
        'growth_rates': self.calculate_growth_rates(historical_data)
    }
    
    analysis_prompt = f"""
    Based on this DePIN network data spanning 90 days:
    {json.dumps(data_summary, indent=2)}
    
    Generate a detailed adoption report covering:
    
    1. **Growth Trajectory**: Calculate month-over-month growth rates
    2. **Quality Indicators**: Assess if growth represents real usage
    3. **Geographic Expansion**: Identify new regions with significant adoption
    4. **Revenue Trends**: Analyze earning patterns and sustainability
    5. **Predictions**: Forecast next quarter's adoption metrics
    
    Include specific percentages and actionable insights.
    Format the response as structured sections.
    """
    
    # Process with Ollama
    result = subprocess.run([
        'ollama', 'run', 'llama2:13b', analysis_prompt
    ], capture_output=True, text=True)
    
    return result.stdout

def detect_adoption_anomalies(self):
    """Identify unusual patterns in adoption metrics"""
    
    anomalies = []
    current_metrics = self.get_current_metrics()
    
    # Check for sudden spikes or drops
    if abs(current_metrics['node_growth'] - self.avg_growth) > (2 * self.growth_std):
        anomalies.append({
            'type': 'growth_anomaly',
            'severity': 'high',
            'description': f"Node growth rate of {current_metrics['node_growth']}% deviates significantly from average"
        })
    
    # Geographic concentration alerts
    geo_metrics = self.analyze_geographic_spread()
    if geo_metrics['distribution_score'] < 0.6:
        anomalies.append({
            'type': 'centralization_risk',
            'severity': 'medium',
            'description': f"Geographic distribution score of {geo_metrics['distribution_score']:.2f} indicates centralization"
        })
    
    return anomalies

import streamlit as st
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def create_adoption_dashboard():
    """Build real-time DePIN adoption dashboard"""
    
    st.title("DePIN Network Adoption Tracker")
    
    # Load current data
    tracker = DePINTracker(networks)
    current_data = tracker.collect_all_metrics()
    
    # Key metrics row
    col1, col2, col3, col4 = st.columns(4)
    
    with col1:
        st.metric(
            "Active Nodes", 
            current_data['active_nodes'],
            delta=current_data['node_change']
        )
    
    with col2:
        st.metric(
            "Network Utilization",
            f"{current_data['utilization_rate']:.1f}%",
            delta=f"{current_data['utilization_change']:.1f}%"
        )
    
    with col3:
        st.metric(
            "Geographic Regions",
            current_data['region_count'],
            delta=current_data['new_regions']
        )
    
    with col4:
        st.metric(
            "Daily Revenue",
            f"${current_data['daily_revenue']:,.2f}",
            delta=f"${current_data['revenue_change']:,.2f}"
        )
    
    # Trend charts
    create_trend_charts(current_data['historical'])
    
    # Ollama-generated insights
    st.header("AI-Generated Insights")
    insights = tracker.generate_adoption_report(current_data['historical'])
    st.write(insights)

def create_trend_charts(historical_data):
    """Generate adoption trend visualizations"""
    
    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=('Node Growth', 'Geographic Distribution', 
                       'Utilization Rate', 'Revenue Trends')
    )
    
    # Node growth over time
    fig.add_trace(
        go.Scatter(
            x=historical_data['dates'],
            y=historical_data['node_counts'],
            mode='lines+markers',
            name='Active Nodes'
        ),
        row=1, col=1
    )
    
    # Geographic diversity
    fig.add_trace(
        go.Scatter(
            x=historical_data['dates'],
            y=historical_data['region_diversity'],
            mode='lines+markers',
            name='Geographic Diversity Score'
        ),
        row=1, col=2
    )
    
    # Utilization trends
    fig.add_trace(
        go.Scatter(
            x=historical_data['dates'],
            y=historical_data['utilization_rates'],
            mode='lines+markers',
            name='Network Utilization %'
        ),
        row=2, col=1
    )
    
    # Revenue progression
    fig.add_trace(
        go.Scatter(
            x=historical_data['dates'],
            y=historical_data['daily_revenues'],
            mode='lines+markers',
            name='Daily Revenue'
        ),
        row=2, col=2
    )
    
    fig.update_layout(height=600, showlegend=False)
    st.plotly_chart(fig, use_container_width=True)

def validate_genuine_activity(self, node_data):
    """Filter out test nodes and artificial activity"""
    
    genuine_nodes = []
    
    for node in node_data:
        # Check for consistent operation patterns
        if self.has_consistent_operation(node):
            # Verify economic participation
            if self.has_meaningful_earnings(node):
                # Confirm geographic authenticity
                if self.verify_location_authenticity(node):
                    genuine_nodes.append(node)
    
    return genuine_nodes

def has_consistent_operation(self, node):
    """Check if node shows consistent operation patterns"""
    uptime_history = node.get('uptime_history', [])
    
    # Require at least 30 days of data
    if len(uptime_history) < 30:
        return False
    
    # Check for realistic uptime patterns (not perfect 100%)
    avg_uptime = sum(uptime_history) / len(uptime_history)
    return 0.85 <= avg_uptime <= 0.99

def has_meaningful_earnings(self, node):
    """Verify node generates meaningful revenue"""
    earnings = node.get('total_earnings', 0)
    operating_days = node.get('operating_days', 0)
    
    if operating_days == 0:
        return False
    
    daily_average = earnings / operating_days
    
    # Minimum threshold to cover basic operating costs
    return daily_average >= 0.05  # Adjust based on network economics

def calculate_network_health_score(self):
    """Generate comprehensive network health assessment"""
    
    health_factors = {
        'decentralization': self.calculate_decentralization_score(),
        'sustainability': self.calculate_sustainability_score(),
        'growth_quality': self.calculate_growth_quality_score(),
        'economic_viability': self.calculate_economic_score()
    }
    
    # Weighted average (adjust weights based on network priorities)
    weights = {
        'decentralization': 0.3,
        'sustainability': 0.25,
        'growth_quality': 0.25,
        'economic_viability': 0.2
    }
    
    health_score = sum(
        health_factors[factor] * weights[factor]
        for factor in health_factors
    )
    
    return {
        'overall_score': health_score,
        'factor_scores': health_factors,
        'recommendations': self.generate_health_recommendations(health_factors)
    }

class HeliumTracker(DePINTracker):
    """Specialized tracker for Helium IoT network"""
    
    def __init__(self):
        super().__init__(["https://api.helium.io"])
        self.network_type = "iot_wireless"
    
    def get_hotspot_metrics(self):
        """Fetch Helium hotspot performance data"""
        response = requests.get("https://api.helium.io/v1/hotspots")
        hotspots = response.json()['data']
        
        metrics = {
            'total_hotspots': len(hotspots),
            'active_hotspots': len([h for h in hotspots if h['status']['online'] == 'online']),
            'avg_rewards': sum(h.get('rewards_24h', 0) for h in hotspots) / len(hotspots),
            'coverage_hexes': len(set(h['location'] for h in hotspots if h.get('location')))
        }
        
        return metrics

class FilecoinTracker(DePINTracker):
    """Monitor Filecoin storage network adoption"""
    
    def __init__(self):
        super().__init__(["https://api.filscan.io"])
        self.network_type = "storage"
    
    def get_storage_metrics(self):
        """Track Filecoin storage provider statistics"""
        
        # Get network storage capacity
        response = requests.get("https://api.filscan.io/api/v1/network/stats")
        network_stats = response.json()
        
        # Get active storage providers
        providers_response = requests.get("https://api.filscan.io/api/v1/miners")
        providers = providers_response.json()['data']
        
        metrics = {
            'total_storage_power': network_stats['total_power'],
            'active_providers': len([p for p in providers if p['is_active']]),
            'utilization_rate': network_stats['used_storage'] / network_stats['total_storage'] * 100,
            'avg_deal_size': network_stats['avg_deal_size']
        }
        
        return metrics

import schedule
import time
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib

def generate_weekly_report():
    """Create and send weekly adoption summary"""
    
    # Collect data from all tracked networks
    tracker = DePINTracker(networks)
    week_data = tracker.get_weekly_summary()
    
    # Generate analysis with Ollama
    report_prompt = f"""
    Create a concise weekly DePIN adoption report based on this data:
    {json.dumps(week_data, indent=2)}
    
    Include:
    1. Key growth highlights
    2. Notable geographic expansions
    3. Performance concerns
    4. Week-over-week percentage changes
    5. Top 3 recommendations for next week
    
    Keep the report under 500 words and include specific metrics.
    """
    
    report = subprocess.run([
        'ollama', 'run', 'llama2:13b', report_prompt
    ], capture_output=True, text=True).stdout
    
    # Send email report
    send_report_email(report, week_data)

def send_report_email(report_content, metrics_data):
    """Email the weekly report to stakeholders"""
    
    msg = MIMEMultipart()
    msg['From'] = "your-tracker@company.com"
    msg['To'] = "team@company.com"
    msg['Subject'] = f"DePIN Adoption Report - Week {datetime.now().strftime('%Y-%W')}"
    
    # Create HTML email with metrics table
    html_content = f"""
    <h2>Weekly DePIN Adoption Summary</h2>
    
    <h3>Key Metrics</h3>
    <table border="1">
        <tr><th>Metric</th><th>Current</th><th>Change</th></tr>
        <tr><td>Active Nodes</td><td>{metrics_data['active_nodes']}</td><td>{metrics_data['node_change']:+.1%}</td></tr>
        <tr><td>Network Utilization</td><td>{metrics_data['utilization']:.1%}</td><td>{metrics_data['util_change']:+.1%}</td></tr>
        <tr><td>Geographic Regions</td><td>{metrics_data['regions']}</td><td>{metrics_data['new_regions']:+d}</td></tr>
    </table>
    
    <h3>Analysis</h3>
    <pre>{report_content}</pre>
    """
    
    msg.attach(MIMEText(html_content, 'html'))
    
    # Send email
    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login("your-email@gmail.com", "your-password")
    server.send_message(msg)
    server.quit()

# Schedule weekly reports
schedule.every().monday.at("09:00").do(generate_weekly_report)

# Keep the scheduler running
while True:
    schedule.run_pending()
    time.sleep(3600)  # Check every hour

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class OptimizedDePINTracker:
    """High-performance tracker for large-scale DePIN networks"""
    
    def __init__(self, endpoints, max_concurrent=50):
        self.endpoints = endpoints
        self.max_concurrent = max_concurrent
        self.session = None
    
    async def collect_all_data(self):
        """Asynchronously collect data from all network endpoints"""
        
        async with aiohttp.ClientSession() as session:
            self.session = session
            
            # Create semaphore to limit concurrent requests
            semaphore = asyncio.Semaphore(self.max_concurrent)
            
            # Collect data from all endpoints concurrently
            tasks = [
                self.fetch_endpoint_data(semaphore, endpoint)
                for endpoint in self.endpoints
            ]
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Filter successful results
            valid_results = [r for r in results if not isinstance(r, Exception)]
            
            return valid_results
    
    async def fetch_endpoint_data(self, semaphore, endpoint):
        """Fetch data from single endpoint with rate limiting"""
        
        async with semaphore:
            try:
                async with self.session.get(f"{endpoint}/api/stats") as response:
                    return await response.json()
            except Exception as e:
                print(f"Error fetching {endpoint}: {e}")
                return None
    
    def process_large_dataset(self, data):
        """Efficiently process large volumes of network data"""
        
        # Use ThreadPoolExecutor for CPU-intensive processing
        with ThreadPoolExecutor(max_workers=4) as executor:
            
            # Split data into chunks for parallel processing
            chunk_size = len(data) // 4
            chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
            
            # Process chunks in parallel
            future_results = [
                executor.submit(self.analyze_data_chunk, chunk)
                for chunk in chunks
            ]
            
            # Combine results
            results = [future.result() for future in future_results]
            
            return self.merge_analysis_results(results)

import time
from functools import wraps

def rate_limit(calls_per_minute=60):
    """Decorator to enforce API rate limiting"""
    min_interval = 60.0 / calls_per_minute
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limit(calls_per_minute=30)
def fetch_network_data(endpoint):
    """Rate-limited network data collection"""
    return requests.get(endpoint).json()

def validate_metrics_quality(self, metrics_data):
    """Ensure collected metrics meet quality standards"""
    
    quality_checks = {
        'completeness': self.check_data_completeness(metrics_data),
        'consistency': self.check_data_consistency(metrics_data),
        'accuracy': self.check_data_accuracy(metrics_data),
        'timeliness': self.check_data_freshness(metrics_data)
    }
    
    # Calculate overall quality score
    quality_score = sum(quality_checks.values()) / len(quality_checks)
    
    if quality_score < 0.8:
        self.log_quality_issue(quality_checks, metrics_data)
        return False
    
    return True

def check_data_completeness(self, data):
    """Verify all required fields are present"""
    required_fields = ['node_count', 'utilization', 'revenue', 'timestamp']
    
    missing_fields = [field for field in required_fields if field not in data]
    
    return len(missing_fields) == 0