feat: replace mock APY sources with real protocol fetchers (closes #157)

src/agent/scanner.ts

@@ -5,74 +5,150 @@
 import { logger } from '../utils/logger';
 import { YieldProtocol, ProtocolRate } from './types';
 import db from '../db';
+import { fetchWithRetry } from '../utils/fetchWithRetry';
 
-const PROTOCOLS = ['Blend', 'Stellar DEX', 'Luma'];
 const ASSET_SYMBOL = 'USDC';
-const MINIMUM_TVL = 10000; // Minimum TVL to consider a protocol
+const MINIMUM_TVL = 10000;
+
+// Metrics tracking
+const metrics: Record<string, { duration: number; failures: number; lastFetched: number }> = {};
+
+function recordMetric(name: string, duration: number, failed: boolean) {
+  const prev = metrics[name] || { duration: 0, failures: 0, lastFetched: 0 };
+  metrics[name] = {
+    duration,
+    failures: failed ? prev.failures + 1 : 0,
+    lastFetched: Date.now(),
+  };
+}
+
+function isStale(name: string, maxAgeMs = 300000): boolean {
+  const m = metrics[name];
+  if (!m) return true;
+  return Date.now() - m.lastFetched > maxAgeMs;
+}
 
 /**
- * Fetch APY from Blend testnet
+ * Fetch APY from Blend protocol (real)
  */
 async function fetchBlendApy(): Promise<YieldProtocol | null> {
+  const start = Date.now();
   try {
-    // Mock implementation - in production, call actual Blend API
-    // https://testnet-api.blend.capital/api/v1/pool/GBUQWP3BOUZX34PISXEAMBNIZJLNCLVNX77MHAHVXHVVB4CMYAOK6BAC
+    const network = process.env.STELLAR_NETWORK?.toLowerCase() === 'mainnet'
+      ? 'https://api.blend.capital'
+      : 'https://testnet-api.blend.capital';
+
+    const poolId = process.env.BLEND_POOL_ID || 'GBUQWP3BOUZX34PISXEAMBNIZJLNCLVNX77MHAHVXHVVB4CMYAOK6BAC';
+
+    const data = await fetchWithRetry(
+      `${network}/api/v1/pool/${poolId}`,
+      { timeout: 5000, retries: 3 }
+    );
+
+    // Extract USDC reserve APY and TVL from response
+    const reserve = data?.reserves?.find((r: any) =>
+      r.asset?.code === 'USDC' || r.asset?.symbol === 'USDC'
+    );
+
+    const apyRate = reserve?.supplyApy
+      ? parseFloat(reserve.supplyApy) * 100
+      : null;
+
+    const tvl = reserve?.totalSupply
+      ? parseFloat(reserve.totalSupply)
+      : null;
 
-    const apyRate = 4.25;
-    const tvl = 50000000;
+    if (apyRate === null) throw new Error('Could not parse Blend APY from response');
+
+    recordMetric('Blend', Date.now() - start, false);
 
     return {
       name: 'Blend',
       apy: apyRate,
-      tvl,
+      tvl: tvl ?? undefined,
       assetSymbol: ASSET_SYMBOL,
       lastUpdated: new Date(),
       isAvailable: true,
     };
   } catch (error) {
-    const errorMessage =
-      error instanceof Error ? error.message : 'Unknown error fetching Blend APY';
+    recordMetric('Blend', Date.now() - start, true);
+    const errorMessage = error instanceof Error ? error.message : 'Unknown error';
     logger.error('Blend APY fetch failed', { error: errorMessage });
     return null;
   }
 }
 
 /**
- * Fetch APY from Stellar DEX pools
+ * Fetch APY from Stellar DEX pools (real via Horizon)
  */
 async function fetchStellarDexApy(): Promise<YieldProtocol | null> {
+  const start = Date.now();
   try {
-    // Mock implementation - in production, aggregate DEX pool rates
-    // Could use SoroswapRouter or other DEX aggregators
+    const horizonUrl = process.env.HORIZON_URL || 'https://horizon.stellar.org';
+    const usdcIssuer = process.env.USDC_ISSUER || 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN';
+
+    const data = await fetchWithRetry(
+      `${horizonUrl}/liquidity_pools?reserves=${ASSET_SYMBOL}:${usdcIssuer}&limit=10&order=desc`,
+      { timeout: 5000, retries: 3 }
+    );
 
-    const apyRate = 3.85;
-    const tvl = 25000000;
+    const pools = data?._embedded?.records || [];
+    if (pools.length === 0) throw new Error('No Stellar DEX pools found');
+
+    // Aggregate: weighted average fee APY by TVL
+    let totalTvl = 0;
+    let weightedApy = 0;
+
+    for (const pool of pools) {
+      const tvlValue = parseFloat(pool.total_shares || '0');
+      const feeApy = parseFloat(pool.fee_bp || '30') / 10000 * 365;
+      totalTvl += tvlValue;
+      weightedApy += feeApy * tvlValue;
+    }
+
+    const apyRate = totalTvl > 0 ? weightedApy / totalTvl : 0;
+
+    recordMetric('Stellar DEX', Date.now() - start, false);
 
     return {
       name: 'Stellar DEX',
       apy: apyRate,
-      tvl,
+      tvl: totalTvl,
       assetSymbol: ASSET_SYMBOL,
       lastUpdated: new Date(),
       isAvailable: true,
     };
   } catch (error) {
-    const errorMessage =
-      error instanceof Error ? error.message : 'Unknown error fetching Stellar DEX APY';
+    recordMetric('Stellar DEX', Date.now() - start, true);
+    const errorMessage = error instanceof Error ? error.message : 'Unknown error';
     logger.error('Stellar DEX APY fetch failed', { error: errorMessage });
     return null;
   }
 }
 
 /**
- * Fetch APY from Luma
+ * Fetch APY from Luma (real)
  */
 async function fetchLumaApy(): Promise<YieldProtocol | null> {
+  const start = Date.now();
   try {
-    // Mock implementation - in production, call Luma API
+    const lumaUrl = process.env.LUMA_API_URL || 'https://api.luma.finance';
 
-    const apyRate = 4.10;
-    const tvl = 35000000;
+    const data = await fetchWithRetry(
+      `${lumaUrl}/v1/rates?asset=${ASSET_SYMBOL}`,
+      { timeout: 5000, retries: 3 }
+    );
+
+    const rate = data?.rates?.find((r: any) =>
+      r.asset === ASSET_SYMBOL || r.symbol === ASSET_SYMBOL
+    );
+
+    if (!rate) throw new Error('USDC rate not found in Luma response');
+
+    const apyRate = parseFloat(rate.apy) * 100;
+    const tvl = rate.tvl ? parseFloat(rate.tvl) : undefined;
+
+    recordMetric('Luma', Date.now() - start, false);
 
     return {
       name: 'Luma',
@@ -83,16 +159,15 @@
       isAvailable: true,
     };
   } catch (error) {
-    const errorMessage =
-      error instanceof Error ? error.message : 'Unknown error fetching Luma APY';
+    recordMetric('Luma', Date.now() - start, true);
+    const errorMessage = error instanceof Error ? error.message : 'Unknown error';
     logger.error('Luma APY fetch failed', { error: errorMessage });
     return null;
   }
 }
 
 /**
  * Scan all protocol APY rates
- * Uses Promise.allSettled to continue even if one protocol fails
  */
 export async function scanAllProtocols(): Promise<YieldProtocol[]> {
   const fetchPromises = [
@@ -102,7 +177,6 @@
   ];
 
   const results = await Promise.allSettled(fetchPromises);
-
   const protocols: YieldProtocol[] = [];
 
   for (const result of results) {
@@ -115,59 +189,52 @@
     }
   }
 
-  // Sort by APY descending (highest first)
   protocols.sort((a, b) => b.apy - a.apy);
-
-  // Filter by minimum TVL
   const filtered = protocols.filter(p => !p.tvl || p.tvl >= MINIMUM_TVL);
 
+  // Log metrics
   logger.info('Protocol scan complete', {
     protocols: filtered.length,
     topApy: filtered[0]?.apy,
     topProtocol: filtered[0]?.name,
+    metrics: Object.entries(metrics).map(([name, m]) => ({
+      name,
+      fetchDurationMs: m.duration,
+      failures: m.failures,
+      stale: isStale(name),
+    })),
   });
 
-  // Save snapshot to database
   await saveProtocolRates(filtered);
-
   return filtered;
 }
 
-/**
- * Normalize STELLAR_NETWORK to valid network label
- * Supports: mainnet, testnet, futurenet
- */
 function normalizeNetwork(): string {
-  const network = process.env.STELLAR_NETWORK?.toLowerCase()
-
-  const validNetworks = ['mainnet', 'testnet', 'futurenet']
+  const network = process.env.STELLAR_NETWORK?.toLowerCase();
+  const validNetworks = ['mainnet', 'testnet', 'futurenet'];
   if (!network || !validNetworks.includes(network)) {
     throw new Error(
       `Invalid STELLAR_NETWORK: "${process.env.STELLAR_NETWORK}". Must be one of: ${validNetworks.join(', ')}`
-    )
+    );
   }
-
-  // Map to uppercase for database storage
-  return network.toUpperCase()
+  return network.toUpperCase();
 }
 
 /**
  * Save protocol rates to database for historical tracking
  */
 async function saveProtocolRates(protocols: YieldProtocol[]): Promise<void> {
   try {
-    const networkLabel = normalizeNetwork()
-
+    const networkLabel = normalizeNetwork();
     for (const protocol of protocols) {
       await db.protocolRate.create({
         data: {
           protocolName: protocol.name,
           assetSymbol: protocol.assetSymbol,
-          // Prisma Decimal type wrapper isn't exposed consistently across tooling;
-          // passing number values is enough for Prisma to coerce.
           supplyApy: protocol.apy as any,
           tvl: protocol.tvl === undefined ? undefined : (protocol.tvl as any),
           network: networkLabel as any,
+          rawResponse: JSON.stringify({ fetchedAt: new Date(), source: protocol.name }),
         },
       });
     }
@@ -178,26 +245,16 @@
   }
 }
 
-/**
- * Get current on-chain APY for active user positions
- */
 export async function getCurrentOnChainApy(protocolName: string): Promise<number | null> {
   try {
     const latestRate = await db.protocolRate.findFirst({
-      where: {
-        protocolName,
-        assetSymbol: ASSET_SYMBOL,
-      },
-      orderBy: {
-        fetchedAt: 'desc',
-      },
+      where: { protocolName, assetSymbol: ASSET_SYMBOL },
+      orderBy: { fetchedAt: 'desc' },
     });
-
     if (!latestRate) {
       logger.warn(`No on-chain APY found for ${protocolName}`);
       return null;
     }
-
     return latestRate.supplyApy.toNumber();
   } catch (error) {
     logger.error('Failed to get current on-chain APY', {
@@ -208,9 +265,6 @@
   }
 }
 
-/**
- * Get best protocol from latest scan
- */
 export async function getBestProtocol(): Promise<YieldProtocol | null> {
   const protocols = await scanAllProtocols();
   return protocols.length > 0 ? protocols[0] : null;

src/utils/fetchWithRetry.ts

@@ -0,0 +1,69 @@
+/**
+ * Fetch with timeout, retry, and circuit breaker support
+ */
+
+interface FetchOptions {
+  timeout?: number;
+  retries?: number;
+  retryDelay?: number;
+}
+
+interface CircuitBreaker {
+  failures: number;
+  lastFailure: number;
+  isOpen: boolean;
+}
+
+const circuitBreakers: Record<string, CircuitBreaker> = {};
+const CIRCUIT_OPEN_DURATION = 60000; // 1 minute
+const FAILURE_THRESHOLD = 3;
+
+export async function fetchWithRetry(
+  url: string,
+  options: FetchOptions = {}
+): Promise<any> {
+  const { timeout = 5000, retries = 3, retryDelay = 1000 } = options;
+
+  // Check circuit breaker
+  const breaker = circuitBreakers[url] || { failures: 0, lastFailure: 0, isOpen: false };
+  if (breaker.isOpen) {
+    const timeSinceFailure = Date.now() - breaker.lastFailure;
+    if (timeSinceFailure < CIRCUIT_OPEN_DURATION) {
+      throw new Error(`Circuit breaker open for ${url}`);
+    }
+    breaker.isOpen = false;
+    breaker.failures = 0;
+  }
+
+  let lastError: Error = new Error('Unknown error');
+
+  for (let attempt = 0; attempt < retries; attempt++) {
+    try {
+      const controller = new AbortController();
+      const timer = setTimeout(() => controller.abort(), timeout);
+
+      const res = await fetch(url, { signal: controller.signal });
+      clearTimeout(timer);
+
+      if (!res.ok) throw new Error(`HTTP ${res.status}: ${res.statusText}`);
+
+      // Reset circuit breaker on success
+      circuitBreakers[url] = { failures: 0, lastFailure: 0, isOpen: false };
+
+      return await res.json();
+    } catch (err) {
+      lastError = err instanceof Error ? err : new Error(String(err));
+      if (attempt < retries - 1) {
+        await new Promise(r => setTimeout(r, retryDelay * (attempt + 1)));
+      }
+    }
+  }
+
+  // Trip circuit breaker
+  breaker.failures += 1;
+  breaker.lastFailure = Date.now();
+  if (breaker.failures >= FAILURE_THRESHOLD) breaker.isOpen = true;
+  circuitBreakers[url] = breaker;
+
+  throw lastError;
+}