Pyth Network

# Pyth Hermes API URL
PYTH_PRICE_SERVICE_URL=https://hermes.pyth.network

# SOL/USD Price Feed ID
PYTH_PRICE_FEED_ID_SOL_USD=0xef0d8b6fda2ceba41da15d4095d1da392a0d2f8ed0c6c7bc0f4cfac8c280b56d

┌─────────────────────────────────────────────────┐
│              Price Service                       │
│  ┌─────────────────────────────────────────┐    │
│  │         SSE Streaming (Primary)          │    │
│  │   Real-time price updates via EventSource│    │
│  └─────────────────────┬───────────────────┘    │
│                        │                        │
│                        │ Fallback               │
│                        ▼                        │
│  ┌─────────────────────────────────────────┐    │
│  │         Polling (Fallback)               │    │
│  │   10-second interval REST API polling    │    │
│  └─────────────────────────────────────────┘    │
└─────────────────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────┐
│           In-Memory Price Store                  │
│   solPrice: number                              │
│   lastUpdated: Date                             │
└─────────────────────────────────────────────────┘

// infrastructure/external/pyth/price-service.ts
 
class PriceService {
  private solPrice: number = 100; // Default fallback
  private lastUpdated: Date | null = null;
  
  async initialize(): Promise<void> {
    console.log('📊 Initializing price service...');
    
    // Fetch price immediately on startup
    await this.fetchPrice();
    
    // Start SSE streaming, fallback to polling if fails
    try {
      await this.startStreaming();
    } catch (error) {
      console.warn('Failed to start SSE, falling back to polling');
      this.fallbackToPolling();
    }
    
    console.log('✅ Price service initialized');
  }
  
  // Get current price (fast, in-memory)
  getSolPrice(): number {
    return this.solPrice;
  }
}

private async startStreaming(): Promise<void> {
  const streamUrl = `${this.baseUrl}/v2/updates/price/stream?ids[]=${this.solUsdFeedId}`;
  
  // Use eventsource package for Node.js SSE support
  const EventSourceModule = await import('eventsource');
  this.eventSource = new EventSourceModule.EventSource(streamUrl);
  
  this.eventSource.onopen = () => {
    this.connectionState = 'connected';
    this.reconnectAttempts = 0;
    console.log('📡 SSE connection established');
    this.startHealthMonitoring();
    this.startProactiveReconnection();
  };
  
  this.eventSource.onmessage = (event: MessageEvent) => {
    this.parsePriceUpdate(event.data);
    this.lastMessageTime = new Date();
  };
  
  this.eventSource.onerror = (error: Event) => {
    this.handleStreamError(error);
  };
}

private parsePriceUpdate(data: string): void {
  const response = JSON.parse(data);
  
  // Pyth SSE format: { parsed: [{ id, price: { price, expo, ... } }] }
  const message = response.parsed?.[0];
  
  if (!message?.price) {
    throw new Error('Invalid price data');
  }
  
  // Normalize: price * 10^expo
  const price = parseFloat(message.price.price);
  const expo = message.price.expo || -8;
  const actualPrice = price * Math.pow(10, expo);
  
  // Only log if price changed significantly (> $0.05)
  const priceChanged = Math.abs(actualPrice - this.solPrice) > 0.05;
  
  this.solPrice = actualPrice;
  this.lastUpdated = new Date();
  
  if (priceChanged) {
    console.log(`💰 Price updated: $${actualPrice.toFixed(2)} SOL/USD`);
  }
}

private async reconnect(): Promise<void> {
  if (this.reconnectAttempts >= this.maxReconnectAttempts) {
    this.fallbackToPolling();
    return;
  }
  
  // Exponential backoff: 1s, 2s, 4s, 8s, 16s
  const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts);
  this.reconnectAttempts++;
  
  console.log(`🔄 Reconnecting in ${delay}ms...`);
  await new Promise(resolve => setTimeout(resolve, delay));
  
  await this.startStreaming();
}

private startHealthMonitoring(): void {
  this.healthCheckInterval = setInterval(() => {
    if (!this.lastMessageTime || this.connectionState !== 'connected') {
      return;
    }
    
    const timeSinceLastMessage = Date.now() - this.lastMessageTime.getTime();
    
    // Reconnect if no messages for 30 seconds
    if (timeSinceLastMessage > 30000) {
      console.warn('⚠️ SSE connection stale, reconnecting...');
      this.stopStreaming();
      this.reconnect();
    }
  }, 30000);
}

private startProactiveReconnection(): void {
  const reconnectDelay = 23 * 60 * 60 * 1000; // 23 hours
  
  this.proactiveReconnectInterval = setTimeout(() => {
    console.log('🔄 Proactive reconnection before 24h timeout...');
    this.stopStreaming();
    this.reconnectAttempts = 0;
    this.reconnect();
  }, reconnectDelay);
}

private fallbackToPolling(): void {
  this.connectionState = 'fallback';
  this.stopStreaming();
  
  // Poll every 10 seconds
  this.updateInterval = setInterval(() => {
    this.fetchPrice();
  }, 10000);
  
  console.log('⚠️ Falling back to polling mode');
}
 
private async fetchPrice(): Promise<void> {
  const url = `${this.baseUrl}/v2/updates/price/latest?ids[]=${this.solUsdFeedId}`;
  
  const response = await fetch(url);
  const data = await response.json();
  
  const priceData = data.parsed?.[0];
  if (!priceData?.price) {
    throw new Error('Invalid price data');
  }
  
  const price = parseFloat(priceData.price.price);
  const expo = priceData.price.expo || -8;
  this.solPrice = price * Math.pow(10, expo);
  this.lastUpdated = new Date();
}

// In JupiterPriceProvider
private getSolUsdPrice(): number {
  const priceService = PriceService.getInstance();
  return priceService.getSolPrice();
}
 
// Convert USD to SOL
const priceUsd = tokenData.usdPrice;
const solUsdPrice = this.getSolUsdPrice();
const priceSol = priceUsd / solUsdPrice;

interface PythPriceResponse {
  parsed: Array<{
    id: string;          // Feed ID
    price: {
      price: string;     // Price value (needs expo conversion)
      conf: string;      // Confidence interval
      expo: number;      // Exponent (usually -8)
      publish_time: number;
    };
  }>;
}