Upload builderbrain/arc_bridge.py

builderbrain/arc_bridge.py

@@ -0,0 +1,347 @@
+"""
+Arc/Circle Integration Bridge
+=============================
+
+Handles:
+- Gateway + CCTP for cross-chain USDC movement
+- Nanopayments for per-trade micropayments
+- Wallets (embedded + strategy wallets)
+- Paymaster for gas abstraction
+- USYC yield on idle capital
+- Contract interactions for on-chain tracking
+
+This is the settlement layer for BuilderBrain.
+"""
+
+import json
+import hashlib
+from dataclasses import dataclass, asdict
+from typing import Dict, Optional, List, Any
+from datetime import datetime
+
+
+@dataclass
+class NanopaymentConfig:
+    """Configuration for Arc Nanopayments."""
+    enabled: bool = True
+    fee_per_trade_bps: float = 5.0  # 5 bps per trade
+    fee_per_insight_usd: float = 0.01  # 1c per insight
+    min_payment_usd: float = 0.001  # 0.1c minimum
+    max_payment_usd: float = 10.0
+    settlement_interval_sec: int = 3600  # hourly batching
+
+
+@dataclass
+class WalletConfig:
+    """Embedded wallet configuration."""
+    wallet_type: str = "embedded"  # 'embedded', 'eoa', 'smart_contract'
+    chain_id: int = 42161  # Arbitrum (Arc default)
+    gas_abstraction: bool = True  # Paymaster enabled
+    daily_limit_usd: float = 1000.0
+    max_single_trade_usd: float = 500.0
+
+
+@dataclass
+class USYCConfig:
+    """USYC yield configuration for idle capital."""
+    enabled: bool = True
+    apy: float = 0.043  # 4.3% current
+    auto_rotate_threshold: float = 0.70  # rotate 70% to USYC when risk-off
+    risk_off_signals: List[str] = None
+    
+    def __post_init__(self):
+        if self.risk_off_signals is None:
+            self.risk_off_signals = [
+                "vix_spike",
+                "correlation_breakdown",
+                "drawdown_15pct",
+            ]
+
+
+@dataclass
+class GatewayRoute:
+    """A cross-chain USDC route via Gateway + CCTP."""
+    from_chain: str
+    to_chain: str
+    amount_usd: float
+    estimated_time_sec: int
+    fee_usd: float
+    cctp_burn_tx: Optional[str] = None
+    cctp_mint_tx: Optional[str] = None
+
+
+class ArcBridge:
+    """
+    Bridge to Arc/Circle infrastructure.
+    
+    In production, this would:
+    - Call Arc Gateway API for CCTP transfers
+    - Submit Nanopayment batches
+    - Interact with USYC contract for yield
+    - Use Paymaster for gas abstraction
+    
+    For hackathon, we simulate the integration patterns
+    with structured logging and mock transactions.
+    """
+    
+    SUPPORTED_CHAINS = [
+        "arbitrum", "base", "optimism", "ethereum", "polygon"
+    ]
+    
+    def __init__(
+        self,
+        nanopayment: Optional[NanopaymentConfig] = None,
+        wallet: Optional[WalletConfig] = None,
+        usyc: Optional[USYCConfig] = None,
+    ):
+        self.nanopayment = nanopayment or NanopaymentConfig()
+        self.wallet = wallet or WalletConfig()
+        self.usyc = usyc or USYCConfig()
+        
+        # Simulated state
+        self.balances: Dict[str, float] = {"usdc": 0.0, "usyc": 0.0}
+        self.pending_payments: List[Dict] = []
+        self.gateway_routes: List[GatewayRoute] = []
+        self.paymaster_ops: List[Dict] = []
+        self.contract_calls: List[Dict] = []
+    
+    # ────────────────────────────── Gateway + CCTP ──────────────────────────────
+    
+    def route_usdc(
+        self,
+        from_chain: str,
+        to_chain: str,
+        amount_usd: float,
+        purpose: str = "trading",
+    ) -> GatewayRoute:
+        """
+        Route USDC between chains via Gateway + CCTP.
+        
+        In production: call Arc Gateway API, initiate CCTP burn,
+        wait for attestation, mint on destination.
+        """
+        if from_chain not in self.SUPPORTED_CHAINS:
+            raise ValueError(f"Unsupported chain: {from_chain}")
+        if to_chain not in self.SUPPORTED_CHAINS:
+            raise ValueError(f"Unsupported chain: {to_chain}")
+        
+        # Simulate CCTP timing and fees
+        route = GatewayRoute(
+            from_chain=from_chain,
+            to_chain=to_chain,
+            amount_usd=amount_usd,
+            estimated_time_sec=1800 if from_chain == "ethereum" else 600,
+            fee_usd=amount_usd * 0.0005,  # 5 bps CCTP fee
+        )
+        
+        # Simulate burn
+        route.cctp_burn_tx = f"0x{hashlib.sha256(f'burn_{amount_usd}'.encode()).hexdigest()[:40]}"
+        
+        self.gateway_routes.append(route)
+        
+        # Update balances
+        self.balances["usdc"] -= amount_usd + route.fee_usd
+        
+        return route
+    
+    def get_balance(self, chain: str = "arbitrum") -> Dict:
+        """Get USDC balance on a chain."""
+        return {
+            "chain": chain,
+            "usdc": self.balances.get("usdc", 0),
+            "usyc": self.balances.get("usyc", 0),
+            "total_usd": self.balances.get("usdc", 0) + self.balances.get("usyc", 0),
+        }
+    
+    # ──────────────────��─────────── Nanopayments ──────────────────────────────
+    
+    def charge_insight_fee(
+        self,
+        user_id: str,
+        trace_id: str,
+        amount_usd: Optional[float] = None,
+    ) -> Dict:
+        """Charge per-insight fee via Nanopayments."""
+        if not self.nanopayment.enabled:
+            return {"status": "skipped", "reason": "nanopayments_disabled"}
+        
+        amount = amount_usd or self.nanopayment.fee_per_insight_usd
+        amount = max(amount, self.nanopayment.min_payment_usd)
+        
+        payment = {
+            "type": "insight",
+            "user_id": user_id,
+            "trace_id": trace_id,
+            "amount_usd": amount,
+            "timestamp": datetime.utcnow().isoformat(),
+            "status": "pending_batch",
+        }
+        
+        self.pending_payments.append(payment)
+        return payment
+    
+    def charge_trade_fee(
+        self,
+        user_id: str,
+        trade_id: str,
+        notional_usd: float,
+    ) -> Dict:
+        """Charge per-trade fee via Nanopayments."""
+        if not self.nanopayment.enabled:
+            return {"status": "skipped", "reason": "nanopayments_disabled"}
+        
+        amount = notional_usd * self.nanopayment.fee_per_trade_bps / 10000
+        amount = max(amount, self.nanopayment.min_payment_usd)
+        
+        payment = {
+            "type": "trade",
+            "user_id": user_id,
+            "trade_id": trade_id,
+            "amount_usd": amount,
+            "notional_usd": notional_usd,
+            "timestamp": datetime.utcnow().isoformat(),
+            "status": "pending_batch",
+        }
+        
+        self.pending_payments.append(payment)
+        return payment
+    
+    def batch_settle(self) -> Dict:
+        """Batch settle pending nanopayments."""
+        if not self.pending_payments:
+            return {"settled": 0, "total_usd": 0}
+        
+        total = sum(p["amount_usd"] for p in self.pending_payments)
+        count = len(self.pending_payments)
+        
+        # Simulate batch settlement
+        batch_tx = f"0x{hashlib.sha256(f'batch_{count}'.encode()).hexdigest()[:40]}"
+        
+        for p in self.pending_payments:
+            p["status"] = "settled"
+            p["batch_tx"] = batch_tx
+        
+        self.pending_payments = []
+        
+        return {
+            "settled": count,
+            "total_usd": total,
+            "batch_tx": batch_tx,
+            "avg_fee_usd": total / count if count > 0 else 0,
+        }
+    
+    # ────────────────────────────── Paymaster ──────────────────────────────
+    
+    def sponsor_transaction(
+        self,
+        user_id: str,
+        tx_data: Dict,
+    ) -> Dict:
+        """
+        Sponsor gas for a user transaction via Paymaster.
+        
+        All UX-facing actions are gas-abstracted and USDC-denominated.
+        """
+        if not self.wallet.gas_abstraction:
+            return {"status": "rejected", "reason": "gas_abstraction_disabled"}
+        
+        # Estimate gas cost in USDC
+        gas_estimate_usd = tx_data.get("gas_estimate", 0.50)
+        
+        op = {
+            "user_id": user_id,
+            "tx_data": tx_data,
+            "gas_sponsored_usd": gas_estimate_usd,
+            "timestamp": datetime.utcnow().isoformat(),
+            "status": "sponsored",
+            "paymaster_tx": f"0x{hashlib.sha256(json.dumps(tx_data).encode()).hexdigest()[:40]}",
+        }
+        
+        self.paymaster_ops.append(op)
+        return op
+    
+    # ────────────────────────────── USYC ──────────────────────────────
+    
+    def rotate_to_usyc(self, amount_usd: float) -> Dict:
+        """Rotate idle USDC to USYC for yield."""
+        if not self.usyc.enabled:
+            return {"status": "skipped", "reason": "usyc_disabled"}
+        
+        if amount_usd > self.balances.get("usdc", 0):
+            amount_usd = self.balances.get("usdc", 0)
+        
+        self.balances["usdc"] -= amount_usd
+        self.balances["usyc"] += amount_usd
+        
+        return {
+            "status": "rotated",
+            "amount_usd": amount_usd,
+            "estimated_apy": self.usyc.apy,
+            "projected_annual_yield": amount_usd * self.usyc.apy,
+        }
+    
+    def rotate_from_usyc(self, amount_usd: float) -> Dict:
+        """Rotate USYC back to USDC for trading."""
+        if amount_usd > self.balances.get("usyc", 0):
+            amount_usd = self.balances.get("usyc", 0)
+        
+        self.balances["usyc"] -= amount_usd
+        self.balances["usdc"] += amount_usd
+        
+        return {
+            "status": "rotated",
+            "amount_usd": amount_usd,
+        }
+    
+    def auto_risk_off_rotation(self, risk_signal: str) -> Dict:
+        """Auto-rotate to USYC when risk-off signals trigger."""
+        if risk_signal not in self.usyc.risk_off_signals:
+            return {"status": "no_action", "reason": "signal_not_configured"}
+        
+        usdc_balance = self.balances.get("usdc", 0)
+        rotate_amount = usdc_balance * self.usyc.auto_rotate_threshold
+        
+        return self.rotate_to_usyc(rotate_amount)
+    
+    # ────────────────────────────── Contract Tracking ──────────────────────────────
+    
+    def record_strategy_metrics(
+        self,
+        strategy_id: str,
+        sharpe: float,
+        drawdown: float,
+        win_rate: float,
+        pnl_usd: float,
+    ) -> Dict:
+        """
+        Record on-chain strategy performance metrics.
+        
+        In production: call Arc contract to update strategy leaderboard.
+        """
+        metrics = {
+            "strategy_id": strategy_id,
+            "sharpe": sharpe,
+            "drawdown": drawdown,
+            "win_rate": win_rate,
+            "pnl_usd": pnl_usd,
+            "timestamp": datetime.utcnow().isoformat(),
+            "contract_tx": f"0x{hashlib.sha256(f'{strategy_id}_{pnl_usd}'.encode()).hexdigest()[:40]}",
+        }
+        
+        self.contract_calls.append(metrics)
+        return metrics
+    
+    # ────────────────────────────── Stats ──────────────────────────────
+    
+    def stats(self) -> Dict:
+        """Bridge usage statistics."""
+        return {
+            "balances": self.balances,
+            "total_gateway_routes": len(self.gateway_routes),
+            "total_pending_payments": len(self.pending_payments),
+            "total_paymaster_ops": len(self.paymaster_ops),
+            "total_contract_calls": len(self.contract_calls),
+            "nanopayment_config": asdict(self.nanopayment),
+            "wallet_config": asdict(self.wallet),
+            "usyc_config": asdict(self.usyc),
+        }