Google Cloud DeFi Analytics: Scale Institutional Yield Farming with BigQuery

# Core architecture components
SERVICES = {
    'data_ingestion': 'Cloud Functions + Pub/Sub',
    'data_warehouse': 'BigQuery',
    'real_time_processing': 'Dataflow',
    'machine_learning': 'Vertex AI',
    'api_layer': 'Cloud Run',
    'monitoring': 'Cloud Monitoring + Logging',
    'security': 'Identity and Access Management'
}

import functions_framework
from google.cloud import bigquery
from google.cloud import pubsub_v1
import requests
import json
from datetime import datetime

@functions_framework.http
def collect_defi_data(request):
    """Collect yield farming data from multiple DeFi protocols"""
    
    # Initialize BigQuery and Pub/Sub clients
    bigquery_client = bigquery.Client()
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path('your-project', 'defi-updates')
    
    protocols = {
        'uniswap_v3': 'https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3',
        'compound': 'https://api.compound.finance/api/v2/ctoken',
        'aave': 'https://aave-api-v2.aave.com/data/liquidity/v2'
    }
    
    collected_data = []
    
    for protocol_name, api_url in protocols.items():
        try:
            # Fetch protocol-specific data
            if protocol_name == 'uniswap_v3':
                pool_data = fetch_uniswap_pools(api_url)
            elif protocol_name == 'compound':
                pool_data = fetch_compound_markets(api_url)
            elif protocol_name == 'aave':
                pool_data = fetch_aave_reserves(api_url)
            
            # Standardize data format
            standardized_data = standardize_protocol_data(protocol_name, pool_data)
            collected_data.extend(standardized_data)
            
            # Publish real-time updates
            for data_point in standardized_data:
                message_data = json.dumps(data_point).encode('utf-8')
                publisher.publish(topic_path, message_data)
            
        except Exception as e:
            print(f"Error collecting data from {protocol_name}: {e}")
    
    # Batch insert to BigQuery
    if collected_data:
        insert_to_bigquery(bigquery_client, collected_data)
    
    return f"Collected {len(collected_data)} data points from {len(protocols)} protocols"

def fetch_uniswap_pools(api_url):
    """Fetch Uniswap V3 pool data with APY calculations"""
    query = """
    {
      pools(first: 100, orderBy: totalValueLockedUSD, orderDirection: desc) {
        id
        token0 { symbol, decimals }
        token1 { symbol, decimals }
        feeTier
        totalValueLockedUSD
        volumeUSD
        feeGrowthGlobal0X128
        feeGrowthGlobal1X128
      }
    }
    """
    
    response = requests.post(api_url, json={'query': query})
    pools = response.json()['data']['pools']
    
    # Calculate APY for each pool
    for pool in pools:
        daily_volume = float(pool['volumeUSD'])
        tvl = float(pool['totalValueLockedUSD'])
        fee_tier = int(pool['feeTier']) / 10000  # Convert to percentage
        
        # Estimate APY based on fee collection
        if tvl > 0:
            daily_fees = daily_volume * fee_tier
            annual_fees = daily_fees * 365
            pool['estimated_apy'] = (annual_fees / tvl) * 100
        else:
            pool['estimated_apy'] = 0
    
    return pools

def standardize_protocol_data(protocol, raw_data):
    """Convert protocol-specific data to standardized format"""
    standardized = []
    timestamp = datetime.utcnow().isoformat()
    
    for item in raw_data:
        if protocol == 'uniswap_v3':
            standardized.append({
                'timestamp': timestamp,
                'protocol': protocol,
                'pool_id': item['id'],
                'asset_pair': f"{item['token0']['symbol']}/{item['token1']['symbol']}",
                'tvl_usd': float(item['totalValueLockedUSD']),
                'apy_percent': item['estimated_apy'],
                'volume_24h': float(item['volumeUSD']),
                'fee_tier': int(item['feeTier']) / 10000
            })
    
    return standardized

-- Create dataset for DeFi analytics
CREATE SCHEMA IF NOT EXISTS `defi_analytics`
OPTIONS (
  description = 'Institutional DeFi yield farming analytics',
  location = 'US'
);

-- Pool performance tracking table
CREATE TABLE IF NOT EXISTS `defi_analytics.pool_snapshots` (
  timestamp TIMESTAMP NOT NULL,
  protocol STRING NOT NULL,
  pool_id STRING NOT NULL,
  asset_pair STRING NOT NULL,
  tvl_usd NUMERIC(20,2),
  apy_percent NUMERIC(8,4),
  volume_24h_usd NUMERIC(20,2),
  fee_tier NUMERIC(6,4),
  impermanent_loss_risk NUMERIC(8,4),
  liquidity_depth NUMERIC(20,2)
)
PARTITION BY DATE(timestamp)
CLUSTER BY protocol, asset_pair
OPTIONS (
  description = 'Time-series data for DeFi pool performance'
);

-- Portfolio positions table
CREATE TABLE IF NOT EXISTS `defi_analytics.portfolio_positions` (
  position_id STRING NOT NULL,
  wallet_address STRING NOT NULL,
  protocol STRING NOT NULL,
  pool_id STRING NOT NULL,
  asset_pair STRING NOT NULL,
  entry_timestamp TIMESTAMP NOT NULL,
  entry_price_usd NUMERIC(20,8),
  position_size_usd NUMERIC(20,2),
  current_value_usd NUMERIC(20,2),
  accrued_fees_usd NUMERIC(20,2),
  impermanent_loss_usd NUMERIC(20,2),
  net_pnl_usd NUMERIC(20,2),
  is_active BOOLEAN NOT NULL DEFAULT TRUE
)
PARTITION BY DATE(entry_timestamp)
CLUSTER BY wallet_address, protocol
OPTIONS (
  description = 'Active and historical portfolio positions'
);

import functions_framework
from google.cloud import bigquery
from google.cloud import aiplatform
import pandas as pd
import numpy as np

@functions_framework.cloud_event
def optimize_yield_allocation(cloud_event):
    """Analyze current positions and recommend optimal yield allocations"""
    
    client = bigquery.Client()
    
    # Get current portfolio positions
    current_positions = get_current_positions(client)
    
    # Get latest yield data across protocols
    yield_opportunities = get_yield_opportunities(client)
    
    # Calculate optimization recommendations
    recommendations = calculate_optimal_allocation(
        current_positions, 
        yield_opportunities
    )
    
    # Risk assessment for each recommendation
    risk_scored_recommendations = assess_rebalancing_risk(
        client, 
        recommendations
    )
    
    # Store recommendations in BigQuery
    store_recommendations(client, risk_scored_recommendations)
    
    return f"Generated {len(risk_scored_recommendations)} optimization recommendations"

def get_yield_opportunities(client):
    """Query latest yield opportunities across all protocols"""
    query = """
    WITH latest_snapshots AS (
      SELECT 
        protocol,
        pool_id,
        asset_pair,
        tvl_usd,
        apy_percent,
        volume_24h_usd,
        impermanent_loss_risk,
        ROW_NUMBER() OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp DESC
        ) as rn
      FROM `defi_analytics.pool_snapshots`
      WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
        AND tvl_usd > 1000000  -- Minimum liquidity threshold
        AND apy_percent > 0
    )
    SELECT *
    FROM latest_snapshots
    WHERE rn = 1
      AND tvl_usd > 1000000  -- Institutional liquidity requirements
    ORDER BY apy_percent DESC
    """
    
    return client.query(query).to_dataframe()

def calculate_optimal_allocation(current_positions, opportunities):
    """Calculate optimal yield allocation using risk-adjusted returns"""
    recommendations = []
    
    # Group opportunities by asset pair for comparison
    opportunities_by_pair = opportunities.groupby('asset_pair')
    
    for asset_pair, pair_opportunities in opportunities_by_pair:
        # Find current position in this asset pair
        current_position = current_positions[
            current_positions['asset_pair'] == asset_pair
        ]
        
        if not current_position.empty:
            current_apy = current_position.iloc[0]['current_apy']
            current_protocol = current_position.iloc[0]['protocol']
            position_size = current_position.iloc[0]['position_size_usd']
            
            # Find best alternative opportunity
            best_opportunity = pair_opportunities.loc[
                pair_opportunities['apy_percent'].idxmax()
            ]
            
            # Calculate potential improvement
            apy_improvement = best_opportunity['apy_percent'] - current_apy
            
            # Only recommend if improvement exceeds gas cost threshold
            gas_cost_threshold = calculate_gas_cost_threshold(position_size)
            annual_improvement = position_size * (apy_improvement / 100)
            
            if annual_improvement > gas_cost_threshold:
                recommendations.append({
                    'asset_pair': asset_pair,
                    'current_protocol': current_protocol,
                    'recommended_protocol': best_opportunity['protocol'],
                    'recommended_pool_id': best_opportunity['pool_id'],
                    'current_apy': current_apy,
                    'target_apy': best_opportunity['apy_percent'],
                    'apy_improvement': apy_improvement,
                    'position_size_usd': position_size,
                    'estimated_annual_gain': annual_improvement,
                    'tvl_target': best_opportunity['tvl_usd'],
                    'impermanent_loss_risk': best_opportunity['impermanent_loss_risk']
                })
    
    return recommendations

def assess_rebalancing_risk(client, recommendations):
    """Assess risk factors for each rebalancing recommendation"""
    risk_query = """
    SELECT 
      protocol,
      pool_id,
      asset_pair,
      STDDEV(apy_percent) as apy_volatility,
      AVG(volume_24h_usd) as avg_volume,
      MIN(tvl_usd) as min_tvl_30d
    FROM `defi_analytics.pool_snapshots`
    WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
    GROUP BY protocol, pool_id, asset_pair
    """
    
    risk_data = client.query(risk_query).to_dataframe()
    
    for rec in recommendations:
        # Find risk metrics for target protocol
        target_risk = risk_data[
            (risk_data['protocol'] == rec['recommended_protocol']) &
            (risk_data['pool_id'] == rec['recommended_pool_id'])
        ]
        
        if not target_risk.empty:
            rec['apy_volatility'] = float(target_risk.iloc[0]['apy_volatility'])
            rec['liquidity_stability'] = float(target_risk.iloc[0]['min_tvl_30d'])
            rec['volume_consistency'] = float(target_risk.iloc[0]['avg_volume'])
            
            # Calculate composite risk score (0-100, lower is better)
            risk_score = calculate_composite_risk_score(rec)
            rec['risk_score'] = risk_score
            
            # Risk-adjusted recommendation priority
            rec['priority_score'] = rec['estimated_annual_gain'] / (1 + risk_score/100)
    
    return sorted(recommendations, key=lambda x: x['priority_score'], reverse=True)

from google.cloud import functions_v1
import math

@functions_framework.http
def calculate_impermanent_loss_scenarios(request):
    """Calculate impermanent loss across different price scenarios"""
    
    client = bigquery.Client()
    
    # Get current positions
    positions_query = """
    SELECT *
    FROM `defi_analytics.portfolio_positions`
    WHERE is_active = TRUE
    """
    positions = client.query(positions_query).to_dataframe()
    
    # Calculate IL scenarios for each position
    il_scenarios = []
    
    for _, position in positions.iterrows():
        scenarios = generate_il_scenarios(position)
        il_scenarios.extend(scenarios)
    
    # Store scenarios in BigQuery
    store_il_scenarios(client, il_scenarios)
    
    return f"Calculated IL scenarios for {len(positions)} positions"

def generate_il_scenarios(position):
    """Generate impermanent loss scenarios for different price movements"""
    scenarios = []
    
    # Price change scenarios: -50% to +200% in 10% increments
    price_changes = [i/100 for i in range(-50, 201, 10)]
    
    for price_change in price_changes:
        # Calculate impermanent loss
        il_percentage = calculate_impermanent_loss(price_change)
        
        # Calculate fees collected over time periods
        daily_fees = estimate_daily_fees(position)
        
        # Time scenarios: 1 day to 365 days
        for days in [1, 7, 30, 90, 180, 365]:
            total_fees = daily_fees * days
            il_amount = position['position_size_usd'] * (il_percentage / 100)
            net_result = total_fees - abs(il_amount)
            
            scenarios.append({
                'position_id': position['position_id'],
                'asset_pair': position['asset_pair'],
                'protocol': position['protocol'],
                'price_change_percent': price_change * 100,
                'time_days': days,
                'impermanent_loss_usd': il_amount,
                'collected_fees_usd': total_fees,
                'net_result_usd': net_result,
                'breakeven_days': calculate_breakeven_days(il_amount, daily_fees),
                'scenario_timestamp': datetime.utcnow()
            })
    
    return scenarios

def calculate_impermanent_loss(price_ratio):
    """Calculate impermanent loss percentage for given price ratio"""
    if price_ratio <= -1:  # Protect against invalid ratios
        return -100
    
    k = 1 + price_ratio  # Price multiplier
    
    # Impermanent loss formula for 50/50 pools
    il_percentage = ((2 * math.sqrt(k)) / (1 + k) - 1) * 100
    
    return il_percentage

def estimate_daily_fees(position):
    """Estimate daily fee collection based on historical data"""
    # This would query historical fee data for the specific pool
    # Simplified calculation for example
    base_daily_yield = position['position_size_usd'] * 0.0001  # 0.01% daily
    return base_daily_yield

from flask import Flask, jsonify, request
from google.cloud import bigquery
import pandas as pd
from datetime import datetime, timedelta

app = Flask(__name__)
client = bigquery.Client()

@app.route('/api/portfolio/summary/<wallet_address>')
def get_portfolio_summary(wallet_address):
    """Get comprehensive portfolio summary for a wallet"""
    
    query = """
    WITH position_summary AS (
      SELECT
        COUNT(*) as active_positions,
        SUM(position_size_usd) as total_invested,
        SUM(current_value_usd) as current_value,
        SUM(accrued_fees_usd) as total_fees,
        SUM(impermanent_loss_usd) as total_il,
        SUM(net_pnl_usd) as total_pnl
      FROM `defi_analytics.portfolio_positions`
      WHERE wallet_address = @wallet_address
        AND is_active = TRUE
    ),
    protocol_breakdown AS (
      SELECT
        protocol,
        COUNT(*) as positions_count,
        SUM(position_size_usd) as invested_amount,
        SUM(current_value_usd) as current_value,
        AVG(CASE 
          WHEN position_size_usd > 0 
          THEN (net_pnl_usd / position_size_usd) * 100 
          ELSE 0 
        END) as avg_return_pct
      FROM `defi_analytics.portfolio_positions`
      WHERE wallet_address = @wallet_address
        AND is_active = TRUE
      GROUP BY protocol
    )
    SELECT 
      ps.*,
      ARRAY_AGG(
        STRUCT(
          pb.protocol,
          pb.positions_count,
          pb.invested_amount,
          pb.current_value,
          pb.avg_return_pct
        )
      ) as protocol_breakdown
    FROM position_summary ps
    CROSS JOIN protocol_breakdown pb
    GROUP BY ps.active_positions, ps.total_invested, ps.current_value, 
             ps.total_fees, ps.total_il, ps.total_pnl
    """
    
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("wallet_address", "STRING", wallet_address)
        ]
    )
    
    result = client.query(query, job_config=job_config).to_dataframe()
    
    if result.empty:
        return jsonify({"error": "No positions found for wallet"}), 404
    
    summary = result.iloc[0]
    
    return jsonify({
        "wallet_address": wallet_address,
        "summary": {
            "active_positions": int(summary['active_positions']),
            "total_invested_usd": float(summary['total_invested']),
            "current_value_usd": float(summary['current_value']),
            "total_fees_usd": float(summary['total_fees']),
            "total_impermanent_loss_usd": float(summary['total_il']),
            "net_pnl_usd": float(summary['total_pnl']),
            "total_return_pct": ((float(summary['current_value']) / float(summary['total_invested'])) - 1) * 100 if summary['total_invested'] > 0 else 0
        },
        "protocol_breakdown": summary['protocol_breakdown']
    })

@app.route('/api/recommendations/<wallet_address>')
def get_optimization_recommendations(wallet_address):
    """Get current yield optimization recommendations"""
    
    query = """
    SELECT 
      asset_pair,
      current_protocol,
      recommended_protocol,
      current_apy,
      target_apy,
      apy_improvement,
      position_size_usd,
      estimated_annual_gain,
      risk_score,
      priority_score
    FROM `defi_analytics.optimization_recommendations`
    WHERE wallet_address = @wallet_address
      AND created_timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
    ORDER BY priority_score DESC
    LIMIT 10
    """
    
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("wallet_address", "STRING", wallet_address)
        ]
    )
    
    recommendations = client.query(query, job_config=job_config).to_dataframe()
    
    return jsonify({
        "recommendations": recommendations.to_dict('records'),
        "total_potential_gain": recommendations['estimated_annual_gain'].sum(),
        "average_apy_improvement": recommendations['apy_improvement'].mean()
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

from google.cloud import monitoring_v3
from google.cloud import logging

def monitor_portfolio_risk(event, context):
    """Monitor portfolio risk metrics and trigger alerts"""
    
    client = bigquery.Client()
    monitoring_client = monitoring_v3.MetricServiceClient()
    logging_client = logging.Client()
    
    # Check concentration risk
    concentration_alerts = check_concentration_risk(client)
    
    # Check impermanent loss risk
    il_alerts = check_impermanent_loss_risk(client)
    
    # Check protocol risk
    protocol_alerts = check_protocol_risk(client)
    
    # Check liquidity risk
    liquidity_alerts = check_liquidity_risk(client)
    
    # Process all alerts
    all_alerts = concentration_alerts + il_alerts + protocol_alerts + liquidity_alerts
    
    for alert in all_alerts:
        # Log to Cloud Logging
        logging_client.logger('defi-risk-alerts').log_struct(alert)
        
        # Create monitoring metric
        create_risk_metric(monitoring_client, alert)
        
        # Send notification if critical
        if alert['severity'] == 'CRITICAL':
            send_risk_notification(alert)

def check_concentration_risk(client):
    """Check for over-concentration in single assets or protocols"""
    query = """
    WITH portfolio_total AS (
      SELECT SUM(position_size_usd) as total_portfolio_value
      FROM `defi_analytics.portfolio_positions`
      WHERE is_active = TRUE
    ),
    concentrations AS (
      SELECT
        'asset_pair' as concentration_type,
        asset_pair as identifier,
        SUM(position_size_usd) as concentrated_amount,
        (SUM(position_size_usd) / pt.total_portfolio_value) * 100 as concentration_pct
      FROM `defi_analytics.portfolio_positions` pp
      CROSS JOIN portfolio_total pt
      WHERE pp.is_active = TRUE
      GROUP BY asset_pair, pt.total_portfolio_value
      
      UNION ALL
      
      SELECT
        'protocol' as concentration_type,
        protocol as identifier,
        SUM(position_size_usd) as concentrated_amount,
        (SUM(position_size_usd) / pt.total_portfolio_value) * 100 as concentration_pct
      FROM `defi_analytics.portfolio_positions` pp
      CROSS JOIN portfolio_total pt
      WHERE pp.is_active = TRUE
      GROUP BY protocol, pt.total_portfolio_value
    )
    SELECT *
    FROM concentrations
    WHERE concentration_pct > 25  -- Alert if >25% concentration
    ORDER BY concentration_pct DESC
    """
    
    concentrations = client.query(query).to_dataframe()
    
    alerts = []
    for _, conc in concentrations.iterrows():
        severity = 'CRITICAL' if conc['concentration_pct'] > 50 else 'WARNING'
        
        alerts.append({
            'alert_type': 'CONCENTRATION_RISK',
            'severity': severity,
            'message': f"High concentration: {conc['concentration_pct']:.1f}% in {conc['identifier']}",
            'concentration_type': conc['concentration_type'],
            'identifier': conc['identifier'],
            'concentration_percent': conc['concentration_pct'],
            'timestamp': datetime.utcnow().isoformat()
        })
    
    return alerts

# terraform/main.tf
provider "google" {
  project = var.project_id
  region  = var.region
}

# BigQuery dataset and tables
resource "google_bigquery_dataset" "defi_analytics" {
  dataset_id  = "defi_analytics"
  description = "Institutional DeFi yield farming analytics"
  location    = "US"
}

# Cloud Functions for data collection
resource "google_cloudfunctions2_function" "defi_collector" {
  name     = "defi-data-collector"
  location = var.region

  build_config {
    runtime     = "python311"
    entry_point = "collect_defi_data"
    source {
      storage_source {
        bucket = google_storage_bucket.functions_bucket.name
        object = google_storage_bucket_object.collector_source.name
      }
    }
  }

  service_config {
    max_instance_count = 100
    available_memory   = "512M"
    timeout_seconds    = 300
    environment_variables = {
      PROJECT_ID = var.project_id
    }
  }

  event_trigger {
    trigger_region = var.region
    event_type     = "google.cloud.pubsub.topic.v1.messagePublished"
    pubsub_topic   = google_pubsub_topic.defi_trigger.id
  }
}

# Cloud Run service for API
resource "google_cloud_run_v2_service" "defi_api" {
  name     = "defi-analytics-api"
  location = var.region

  template {
    scaling {
      min_instance_count = 1
      max_instance_count = 10
    }

    containers {
      image = "gcr.io/${var.project_id}/defi-analytics-api"
      
      resources {
        limits = {
          cpu    = "2000m"
          memory = "4Gi"
        }
      }

      env {
        name  = "PROJECT_ID"
        value = var.project_id
      }
    }
  }
}

# Pub/Sub topics for real-time updates
resource "google_pubsub_topic" "defi_updates" {
  name = "defi-updates"
}

resource "google_pubsub_topic" "defi_trigger" {
  name = "defi-data-trigger"
}

# Cloud Scheduler for periodic data collection
resource "google_cloud_scheduler_job" "defi_scheduler" {
  name      = "defi-data-collection"
  schedule  = "*/5 * * * *"  # Every 5 minutes
  time_zone = "UTC"

  pubsub_target {
    topic_name = google_pubsub_topic.defi_trigger.id
    data       = base64encode("trigger")
  }
}

-- Optimized query for real-time yield comparison
CREATE OR REPLACE VIEW `defi_analytics.optimized_yield_view` AS
WITH latest_data AS (
  SELECT 
    protocol,
    pool_id,
    asset_pair,
    apy_percent,
    tvl_usd,
    impermanent_loss_risk,
    timestamp,
    ROW_NUMBER() OVER (
      PARTITION BY protocol, pool_id 
      ORDER BY timestamp DESC
    ) as rn
  FROM `defi_analytics.pool_snapshots`
  WHERE DATE(timestamp) >= DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
),
ranked_yields AS (
  SELECT 
    *,
    RANK() OVER (
      PARTITION BY asset_pair 
      ORDER BY apy_percent DESC
    ) as yield_rank
  FROM latest_data
  WHERE rn = 1
    AND tvl_usd > 1000000  -- Institutional minimum
)
SELECT 
  asset_pair,
  protocol,
  pool_id,
  apy_percent,
  tvl_usd,
  impermanent_loss_risk,
  yield_rank,
  CASE 
    WHEN yield_rank = 1 THEN 'BEST'
    WHEN yield_rank <= 3 THEN 'COMPETITIVE'
    ELSE 'LOWER'
  END as yield_tier
FROM ranked_yields
WHERE yield_rank <= 5;  -- Top 5 yields per asset pair

from google.cloud import aiplatform
from google.cloud import bigquery
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
import joblib

def train_yield_prediction_model():
    """Train ML model to predict DeFi yield changes"""
    
    # Initialize Vertex AI
    aiplatform.init(project='your-project-id', location='us-central1')
    
    client = bigquery.Client()
    
    # Fetch training data with features
    training_query = """
    WITH hourly_features AS (
      SELECT 
        timestamp,
        protocol,
        pool_id,
        asset_pair,
        apy_percent,
        tvl_usd,
        volume_24h_usd,
        impermanent_loss_risk,
        LAG(apy_percent, 1) OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp
        ) as prev_apy,
        LAG(tvl_usd, 1) OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp
        ) as prev_tvl,
        LAG(volume_24h_usd, 1) OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp
        ) as prev_volume,
        LEAD(apy_percent, 4) OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp
        ) as future_apy  -- Predict 4 hours ahead
      FROM `defi_analytics.pool_snapshots`
      WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
    )
    SELECT 
      *,
      -- Feature engineering
      SAFE_DIVIDE(apy_percent - prev_apy, prev_apy) * 100 as apy_change_pct,
      SAFE_DIVIDE(tvl_usd - prev_tvl, prev_tvl) * 100 as tvl_change_pct,
      SAFE_DIVIDE(volume_24h_usd - prev_volume, prev_volume) * 100 as volume_change_pct,
      EXTRACT(HOUR FROM timestamp) as hour_of_day,
      EXTRACT(DAYOFWEEK FROM timestamp) as day_of_week
    FROM hourly_features
    WHERE prev_apy IS NOT NULL 
      AND future_apy IS NOT NULL
      AND apy_percent > 0
    """
    
    df = client.query(training_query).to_dataframe()
    
    # Prepare features and target
    feature_columns = [
        'apy_percent', 'tvl_usd', 'volume_24h_usd', 'impermanent_loss_risk',
        'apy_change_pct', 'tvl_change_pct', 'volume_change_pct',
        'hour_of_day', 'day_of_week'
    ]
    
    # Handle missing values and infinite values
    df = df.replace([np.inf, -np.inf], np.nan).dropna()
    
    X = df[feature_columns].values
    y = df['future_apy'].values
    
    # Scale features
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)
    
    # Train Random Forest model
    model = RandomForestRegressor(
        n_estimators=100,
        max_depth=10,
        random_state=42,
        n_jobs=-1
    )
    
    model.fit(X_scaled, y)
    
    # Save model and scaler to Cloud Storage
    save_model_to_gcs(model, scaler, feature_columns)
    
    return model, scaler

def save_model_to_gcs(model, scaler, feature_columns):
    """Save trained model to Google Cloud Storage"""
    
    from google.cloud import storage
    import pickle
    
    bucket_name = 'your-defi-models-bucket'
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    
    # Save model
    model_blob = bucket.blob('models/yield_predictor_v1.pkl')
    model_blob.upload_from_string(pickle.dumps(model))
    
    # Save scaler
    scaler_blob = bucket.blob('models/yield_scaler_v1.pkl')
    scaler_blob.upload_from_string(pickle.dumps(scaler))
    
    # Save feature columns
    features_blob = bucket.blob('models/feature_columns_v1.pkl')
    features_blob.upload_from_string(pickle.dumps(feature_columns))

@functions_framework.http
def predict_yield_changes(request):
    """Cloud Function to generate yield predictions"""
    
    # Load model from GCS
    model, scaler, feature_columns = load_model_from_gcs()
    
    client = bigquery.Client()
    
    # Get latest pool data for prediction
    current_data_query = """
    WITH latest_snapshots AS (
      SELECT 
        *,
        ROW_NUMBER() OVER (
          PARTITION BY protocol, pool_id 
          ORDER BY timestamp DESC
        ) as rn
      FROM `defi_analytics.pool_snapshots`
      WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 HOUR)
    )
    SELECT *
    FROM latest_snapshots
    WHERE rn = 1
      AND tvl_usd > 1000000
    """
    
    current_data = client.query(current_data_query).to_dataframe()
    
    predictions = []
    
    for _, row in current_data.iterrows():
        # Prepare features for prediction
        features = prepare_prediction_features(row)
        
        if features is not None:
            # Scale features
            features_scaled = scaler.transform([features])
            
            # Make prediction
            predicted_apy = model.predict(features_scaled)[0]
            
            # Calculate confidence based on historical volatility
            confidence = calculate_prediction_confidence(row, predicted_apy)
            
            predictions.append({
                'protocol': row['protocol'],
                'pool_id': row['pool_id'],
                'asset_pair': row['asset_pair'],
                'current_apy': row['apy_percent'],
                'predicted_apy_4h': predicted_apy,
                'apy_change_forecast': predicted_apy - row['apy_percent'],
                'confidence_score': confidence,
                'prediction_timestamp': datetime.utcnow().isoformat()
            })
    
    # Store predictions in BigQuery
    store_predictions(client, predictions)
    
    return jsonify({
        'predictions_generated': len(predictions),
        'top_opportunities': sorted(predictions, 
            key=lambda x: x['apy_change_forecast'], reverse=True)[:5]
    })

import requests
from google.cloud import functions_v1

@functions_framework.http
def assess_protocol_security(request):
    """Assess DeFi protocol security risks"""
    
    client = bigquery.Client()
    
    protocols = get_active_protocols(client)
    security_assessments = []
    
    for protocol in protocols:
        assessment = {
            'protocol': protocol['name'],
            'contract_address': protocol['contract_address'],
            'audit_score': get_audit_score(protocol),
            'tvl_history_score': calculate_tvl_stability_score(protocol),
            'code_quality_score': assess_code_quality(protocol),
            'governance_score': evaluate_governance(protocol),
            'liquidity_risk_score': calculate_liquidity_risk(protocol)
        }
        
        # Calculate composite security score
        weights = {
            'audit_score': 0.3,
            'tvl_history_score': 0.2,
            'code_quality_score': 0.2,
            'governance_score': 0.15,
            'liquidity_risk_score': 0.15
        }
        
        composite_score = sum(
            assessment[key] * weights[key] 
            for key in weights.keys()
        )
        
        assessment['composite_security_score'] = composite_score
        assessment['risk_tier'] = categorize_risk_tier(composite_score)
        
        security_assessments.append(assessment)
    
    # Store security assessments
    store_security_assessments(client, security_assessments)
    
    return jsonify({
        'assessments_completed': len(security_assessments),
        'high_risk_protocols': [
            a for a in security_assessments 
            if a['risk_tier'] == 'HIGH_RISK'
        ]
    })

def get_audit_score(protocol):
    """Calculate audit score based on security audits"""
    
    # Check major audit firms (simplified example)
    audit_sources = {
        'consensys': f"https://consensys.net/diligence/audits/{protocol['name']}",
        'certik': f"https://certik.org/projects/{protocol['name']}",
        'openzeppelin': f"https://openzeppelin.com/security-audits/{protocol['name']}"
    }
    
    audit_count = 0
    recent_audits = 0
    
    for source, url in audit_sources.items():
        try:
            # In production, you'd implement proper API calls
            # This is a simplified check
            audit_data = check_audit_source(source, protocol['name'])
            if audit_data:
                audit_count += 1
                if audit_data.get('recent', False):
                    recent_audits += 1
        except:
            continue
    
    # Score based on audit count and recency
    base_score = min(audit_count * 25, 75)  # Max 75 for multiple audits
    recency_bonus = recent_audits * 12.5    # Up to 25 bonus for recent audits
    
    return min(base_score + recency_bonus, 100)

def calculate_tvl_stability_score(protocol):
    """Score protocol based on TVL stability and growth"""
    
    client = bigquery.Client()
    
    tvl_query = """
    SELECT 
      DATE(timestamp) as date,
      AVG(tvl_usd) as daily_avg_tvl
    FROM `defi_analytics.pool_snapshots`
    WHERE protocol = @protocol_name
      AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 90 DAY)
    GROUP BY DATE(timestamp)
    ORDER BY date
    """
    
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("protocol_name", "STRING", protocol['name'])
        ]
    )
    
    tvl_data = client.query(tvl_query, job_config=job_config).to_dataframe()
    
    if len(tvl_data) < 30:  # Need at least 30 days of data
        return 30  # Low score for insufficient data
    
    # Calculate TVL volatility (coefficient of variation)
    tvl_volatility = tvl_data['daily_avg_tvl'].std() / tvl_data['daily_avg_tvl'].mean()
    
    # Calculate growth trend
    growth_rate = ((tvl_data['daily_avg_tvl'].iloc[-1] / tvl_data['daily_avg_tvl'].iloc[0]) - 1) * 100
    
    # Score calculation (lower volatility and positive growth = higher score)
    volatility_score = max(0, 100 - (tvl_volatility * 1000))  # Penalize high volatility
    growth_score = min(max(growth_rate, -50), 50) + 50  # Normalize growth to 0-100
    
    return (volatility_score * 0.7) + (growth_score * 0.3)

from web3 import Web3
from eth_account import Account
import json

class MultiSigWalletManager:
    """Manage multi-signature wallet interactions for institutional DeFi"""
    
    def __init__(self, web3_provider, multisig_address):
        self.w3 = Web3(Web3.HTTPProvider(web3_provider))
        self.multisig_address = multisig_address
        
        # Load multi-sig contract ABI
        with open('multisig_abi.json', 'r') as f:
            self.multisig_abi = json.load(f)
        
        self.multisig_contract = self.w3.eth.contract(
            address=multisig_address,
            abi=self.multisig_abi
        )
    
    def propose_yield_rebalance(self, recommendation):
        """Propose a yield farming rebalance transaction"""
        
        # Build transaction data for the rebalance
        transaction_data = self.build_rebalance_transaction(recommendation)
        
        # Calculate gas estimate
        gas_estimate = self.estimate_gas_cost(transaction_data)
        
        # Create proposal
        proposal = {
            'transaction_data': transaction_data,
            'gas_estimate': gas_estimate,
            'recommendation_id': recommendation['recommendation_id'],
            'expected_apy_improvement': recommendation['apy_improvement'],
            'risk_score': recommendation['risk_score'],
            'proposal_timestamp': datetime.utcnow().isoformat()
        }
        
        # Submit to multi-sig for approval
        return self.submit_multisig_proposal(proposal)
    
    def build_rebalance_transaction(self, recommendation):
        """Build the actual rebalancing transaction"""
        
        # Example: Moving from Compound to Aave
        if recommendation['current_protocol'] == 'compound' and recommendation['recommended_protocol'] == 'aave':
            return self.build_compound_to_aave_migration(recommendation)
        elif recommendation['current_protocol'] == 'aave' and recommendation['recommended_protocol'] == 'uniswap_v3':
            return self.build_aave_to_uniswap_migration(recommendation)
        else:
            raise ValueError(f"Unsupported migration: {recommendation['current_protocol']} to {recommendation['recommended_protocol']}")
    
    def build_compound_to_aave_migration(self, recommendation):
        """Build transaction for Compound -> Aave migration"""
        
        # Load protocol contract ABIs
        compound_abi = self.load_protocol_abi('compound')
        aave_abi = self.load_protocol_abi('aave')
        
        transactions = []
        
        # Step 1: Withdraw from Compound
        compound_contract = self.w3.eth.contract(
            address=recommendation['current_pool_address'],
            abi=compound_abi
        )
        
        withdraw_tx = compound_contract.functions.redeemUnderlying(
            recommendation['position_size_wei']
        ).buildTransaction({
            'from': self.multisig_address,
            'gas': 200000,
            'gasPrice': self.w3.eth.gas_price
        })
        
        transactions.append(('withdraw_compound', withdraw_tx))
        
        # Step 2: Deposit to Aave
        aave_contract = self.w3.eth.contract(
            address=recommendation['target_pool_address'],
            abi=aave_abi
        )
        
        deposit_tx = aave_contract.functions.deposit(
            recommendation['asset_address'],
            recommendation['position_size_wei'],
            self.multisig_address,
            0  # referralCode
        ).buildTransaction({
            'from': self.multisig_address,
            'gas': 300000,
            'gasPrice': self.w3.eth.gas_price
        })
        
        transactions.append(('deposit_aave', deposit_tx))
        
        return transactions

@functions_framework.http
def execute_approved_rebalances(request):
    """Execute multi-sig approved rebalancing transactions"""
    
    client = bigquery.Client()
    
    # Get approved rebalance proposals
    approved_proposals = get_approved_proposals(client)
    
    execution_results = []
    
    for proposal in approved_proposals:
        try:
            # Initialize wallet manager
            wallet_manager = MultiSigWalletManager(
                web3_provider='https://mainnet.infura.io/v3/YOUR_API_KEY',
                multisig_address=proposal['wallet_address']
            )
            
            # Execute the rebalance
            result = wallet_manager.execute_rebalance(proposal)
            
            execution_results.append({
                'proposal_id': proposal['proposal_id'],
                'status': 'SUCCESS',
                'transaction_hash': result['tx_hash'],
                'gas_used': result['gas_used'],
                'execution_timestamp': datetime.utcnow().isoformat()
            })
            
            # Update position in BigQuery
            update_position_after_rebalance(client, proposal, result)
            
        except Exception as e:
            execution_results.append({
                'proposal_id': proposal['proposal_id'],
                'status': 'FAILED',
                'error': str(e),
                'execution_timestamp': datetime.utcnow().isoformat()
            })
    
    # Store execution results
    store_execution_results(client, execution_results)
    
    return jsonify({
        'executed_rebalances': len(execution_results),
        'successful_executions': len([r for r in execution_results if r['status'] == 'SUCCESS']),
        'failed_executions': len([r for r in execution_results if r['status'] == 'FAILED'])
    })

from google.cloud import monitoring_v3
import json

def setup_defi_monitoring_dashboards():
    """Set up comprehensive DeFi monitoring dashboards"""
    
    client = monitoring_v3.DashboardsServiceClient()
    project_name = f"projects/{PROJECT_ID}"
    
    # Portfolio Performance Dashboard
    portfolio_dashboard = create_portfolio_dashboard()
    client.create_dashboard(
        parent=project_name,
        dashboard=portfolio_dashboard
    )
    
    # Risk Monitoring Dashboard
    risk_dashboard = create_risk_dashboard()
    client.create_dashboard(
        parent=project_name,
        dashboard=risk_dashboard
    )
    
    # System Health Dashboard
    system_dashboard = create_system_health_dashboard()
    client.create_dashboard(
        parent=project_name,
        dashboard=system_dashboard
    )

def create_portfolio_dashboard():
    """Create portfolio performance monitoring dashboard"""
    
    dashboard_config = {
        "displayName": "DeFi Portfolio Performance",
        "mosaicLayout": {
            "tiles": [
                {
                    "width": 6,
                    "height": 4,
                    "widget": {
                        "title": "Total Portfolio Value",
                        "scorecard": {
                            "timeSeriesQuery": {
                                "timeSeriesFilter": {
                                    "filter": 'resource.type="cloud_function" AND metric.type="custom.googleapis.com/defi/portfolio_value"',
                                    "aggregation": {
                                        "alignmentPeriod": "300s",
                                        "perSeriesAligner": "ALIGN_MEAN",
                                        "crossSeriesReducer": "REDUCE_SUM"
                                    }
                                }
                            },
                            "sparkChartView": {
                                "sparkChartType": "SPARK_LINE"
                            }
                        }
                    }
                },
                {
                    "xPos": 6,
                    "width": 6,
                    "height": 4,
                    "widget": {
                        "title": "Average Portfolio APY",
                        "scorecard": {
                            "timeSeriesQuery": {
                                "timeSeriesFilter": {
                                    "filter": 'metric.type="custom.googleapis.com/defi/average_apy"',
                                    "aggregation": {
                                        "alignmentPeriod": "3600s",
                                        "perSeriesAligner": "ALIGN_MEAN"
                                    }
                                }
                            },
                            "sparkChartView": {
                                "sparkChartType": "SPARK_LINE"
                            }
                        }
                    }
                },
                {
                    "yPos": 4,
                    "width": 12,
                    "height": 4,
                    "widget": {
                        "title": "Protocol Distribution",
                        "pieChart": {
                            "timeSeriesQuery": {
                                "timeSeriesFilter": {
                                    "filter": 'metric.type="custom.googleapis.com/defi/protocol_allocation"',
                                    "aggregation": {
                                        "alignmentPeriod": "3600s",
                                        "perSeriesAligner": "ALIGN_MEAN",
                                        "crossSeriesReducer": "REDUCE_SUM",
                                        "groupByFields": ["metric.label.protocol"]
                                    }
                                }
                            }
                        }
                    }
                }
            ]
        }
    }
    
    return monitoring_v3.Dashboard(dashboard_config)

def create_custom_metrics():
    """Create custom metrics for DeFi monitoring"""
    
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{PROJECT_ID}"
    
    metrics = [
        {
            "type": "custom.googleapis.com/defi/portfolio_value",
            "labels": [
                {"key": "wallet_address", "valueType": "STRING"},
                {"key": "currency", "valueType": "STRING"}
            ],
            "metricKind": "GAUGE",
            "valueType": "DOUBLE",
            "description": "Total portfolio value in USD"
        },
        {
            "type": "custom.googleapis.com/defi/average_apy",
            "labels": [
                {"key": "wallet_address", "valueType": "STRING"}
            ],
            "metricKind": "GAUGE",
            "valueType": "DOUBLE",
            "description": "Average APY across all positions"
        },
        {
            "type": "custom.googleapis.com/defi/impermanent_loss",
            "labels": [
                {"key": "wallet_address", "valueType": "STRING"},
                {"key": "protocol", "valueType": "STRING"}
            ],
            "metricKind": "GAUGE",
            "valueType": "DOUBLE",
            "description": "Total impermanent loss in USD"
        }
    ]
    
    for metric_config in metrics:
        descriptor = monitoring_v3.MetricDescriptor(metric_config)
        client.create_metric_descriptor(
            name=project_name,
            metric_descriptor=descriptor
        )

-- Partitioned and clustered table optimization
CREATE TABLE IF NOT EXISTS `defi_analytics.optimized_pool_snapshots` (
  snapshot_date DATE NOT NULL,
  timestamp TIMESTAMP NOT NULL,
  protocol STRING NOT NULL,
  pool_id STRING NOT NULL,
  asset_pair STRING NOT NULL,
  tvl_usd NUMERIC(20,2),
  apy_percent NUMERIC(8,4),
  volume_24h_usd NUMERIC(20,2),
  impermanent_loss_risk NUMERIC(8,4)
)
PARTITION BY snapshot_date
CLUSTER BY protocol, asset_pair, pool_id
OPTIONS (
  description = 'Cost-optimized DeFi pool snapshots',
  partition_expiration_days = 730  -- 2 years retention
);

-- Cost-effective materialized view for common queries
CREATE MATERIALIZED VIEW `defi_analytics.daily_protocol_summary`
PARTITION BY summary_date
CLUSTER BY protocol
AS
SELECT 
  DATE(timestamp) as summary_date,
  protocol,
  COUNT(DISTINCT pool_id) as active_pools,
  SUM(tvl_usd) as total_tvl_usd,
  AVG(apy_percent) as avg_apy_percent,
  STDDEV(apy_percent) as apy_volatility,
  SUM(volume_24h_usd) as total_volume_usd
FROM `defi_analytics.pool_snapshots`
WHERE DATE(timestamp) >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY summary_date, protocol;

-- Approximate aggregation for large-scale analytics
SELECT 
  protocol,
  asset_pair,
  APPROX_QUANTILES(apy_percent, 100)[OFFSET(50)] as median_apy,
  APPROX_QUANTILES(apy_percent, 100)[OFFSET(95)] as p95_apy,
  APPROX_COUNT_DISTINCT(pool_id) as unique_pools,
  -- Use approximate functions for better performance
  APPROX_AVG(tvl_usd) as avg_tvl,
  APPROX_STDDEV(apy_percent) as apy_stddev
FROM `defi_analytics.pool_snapshots`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY protocol, asset_pair
ORDER BY median_apy DESC;