Integrates datastream client

Author: mohamed-mehany

packages/composites/glv-token/package.json

@@ -36,7 +36,7 @@
   "dependencies": {
     "@chainlink/external-adapter-framework": "2.7.0",
     "decimal.js": "^10.3.1",
-    "ethers": "^5.4.6",
+    "ethers": "^6.15.0",
     "tslib": "2.4.1"
   }
 }

packages/composites/glv-token/src/config/index.ts

@@ -26,18 +26,18 @@ export const config = new AdapterConfig(
       required: true,
       default: '0x6a9505D0B44cFA863d9281EA5B0b34cB36243b45',
     },
-    TIINGO_ADAPTER_URL: {
-      description: 'URL of Tiingo EA',
+    DATA_ENGINE_BASE_URL: {
+      description: 'URL of DataEngine',
       type: 'string',
       required: true,
     },
-    NCFX_ADAPTER_URL: {
-      description: 'URL of NCFX EA',
+    DATA_ENGINE_USER_ID: {
+      description: 'User ID of DataEngine',
       type: 'string',
       required: true,
     },
-    COINMETRICS_ADAPTER_URL: {
-      description: 'URL of Coinmetrics EA',
+    DATA_ENGINE_USER_SECRET: {
+      description: 'Secret key for DataEngine',
       type: 'string',
       required: true,
     },

packages/composites/glv-token/src/datastreams/client.ts

@@ -62,17 +62,21 @@ export class DataStreamsHttpClient {
     quote_asset?: string
     feed_type?: string
   }): Promise<FeedsResponse> {
-    const qs = new URLSearchParams()
-    if (params?.base_asset) qs.set('base_asset', params.base_asset)
-    if (params?.quote_asset) qs.set('quote_asset', params.quote_asset)
-    if (params?.feed_type) qs.set('feed_type', params.feed_type)
-    const path = `/api/v1/feeds${qs.toString() ? `?${qs.toString()}` : ''}`
+    const query = new URLSearchParams()
+    if (params?.base_asset) query.set('base_asset', params.base_asset)
+    if (params?.quote_asset) query.set('quote_asset', params.quote_asset)
+    if (params?.feed_type) query.set('feed_type', params.feed_type)
+    const path = `/api/v1/feeds${query.toString() ? `?${query.toString()}` : ''}`
     return this.request<FeedsResponse>('GET', path)
   }
 
   async resolveFeedId(base: string, quote: string, feedType?: string): Promise<string> {
-    const r = await this.listFeeds({ base_asset: base, quote_asset: quote, feed_type: feedType })
-    const ids = (r.feeds ?? [])
+    const response = await this.listFeeds({
+      base_asset: base,
+      quote_asset: quote,
+      feed_type: feedType,
+    })
+    const ids = (response.feeds ?? [])
       .map((f) => (f.feedID ?? f.feedid ?? '').toLowerCase())
       .filter(Boolean)
     if (ids.length === 0) throw new Error(`No feed found for ${base}/${quote}`)
@@ -84,14 +88,14 @@ export class DataStreamsHttpClient {
     feedId: string,
   ): Promise<{ fullReportHex: string; observationsTimestamp: number; validFromTimestamp: number }> {
     const path = `/api/v1/reports/latest?feedID=${encodeURIComponent(feedId)}`
-    const r = await this.request<LatestReportResponse>('GET', path)
-    const rr = r.report
-    const full = rr.fullReport ?? rr.fullreport
-    if (!full) throw new Error(`Missing fullReport for ${feedId}`)
+    const response = await this.request<LatestReportResponse>('GET', path)
+    const rawReport = response.report
+    const fullReport = rawReport.fullReport ?? rawReport.fullreport
+    if (!fullReport) throw new Error(`Missing fullReport for ${feedId}`)
     return {
-      fullReportHex: full,
-      observationsTimestamp: rr.observationsTimestamp,
-      validFromTimestamp: rr.validFromTimestamp,
+      fullReportHex: fullReport,
+      observationsTimestamp: rawReport.observationsTimestamp,
+      validFromTimestamp: rawReport.validFromTimestamp,
     }
   }
 }

packages/composites/glv-token/src/transport/base.ts

@@ -1,19 +1,18 @@
-import { ethers, utils } from 'ethers'
-import { SubscriptionTransport } from '@chainlink/external-adapter-framework/transports/abstract/subscription'
-import { TransportDependencies } from '@chainlink/external-adapter-framework/transports'
+import { EndpointContext } from '@chainlink/external-adapter-framework/adapter'
 import { ResponseCache } from '@chainlink/external-adapter-framework/cache/response'
-import { Requester } from '@chainlink/external-adapter-framework/util/requester'
-import {
-  EndpointContext,
-  LwbaResponseDataFields,
-} from '@chainlink/external-adapter-framework/adapter'
+import { TransportDependencies } from '@chainlink/external-adapter-framework/transports'
+import { SubscriptionTransport } from '@chainlink/external-adapter-framework/transports/abstract/subscription'
 import { AdapterResponse, makeLogger } from '@chainlink/external-adapter-framework/util'
+import { Requester } from '@chainlink/external-adapter-framework/util/requester'
 import { AdapterDataProviderError } from '@chainlink/external-adapter-framework/validation/error'
+import { TypeFromDefinition } from '@chainlink/external-adapter-framework/validation/input-params'
+import { ethers } from 'ethers'
 import glvAbi from '../config/glvReaderAbi.json'
-import { BaseEndpointTypes, inputParameters } from '../endpoint/price'
+import { DataStreamsHttpClient } from '../datastreams/client'
+import { decodeReport } from '../datastreams/decode'
 import { BaseEndpointTypesLwba } from '../endpoint/lwba'
+import { BaseEndpointTypes, inputParameters } from '../endpoint/price'
 import {
-  mapParameter,
   mapSymbol,
   Market,
   median,
@@ -23,7 +22,6 @@ import {
   toFixed,
   Token,
 } from './utils'
-import { TypeFromDefinition } from '@chainlink/external-adapter-framework/validation/input-params'
 
 const logger = makeLogger('GlvBaseTransport')
 
@@ -54,10 +52,10 @@ export abstract class BaseGlvTransport<
   name!: string
   responseCache!: ResponseCache<T>
   requester!: Requester
-  provider!: ethers.providers.JsonRpcProvider
+  provider!: ethers.JsonRpcProvider
   glvReaderContract!: ethers.Contract
   settings!: T['Settings']
-
+  dataStreamsClient!: DataStreamsHttpClient
   tokensMap: Record<string, Token> = {}
   marketsMap: Record<string, Market> = {}
   decimals: Record<string, number> = {}
@@ -70,7 +68,7 @@ export abstract class BaseGlvTransport<
   ): Promise<void> {
     await super.initialize(dependencies, adapterSettings, endpointName, transportName)
     this.settings = adapterSettings
-    this.provider = new ethers.providers.JsonRpcProvider(
+    this.provider = new ethers.JsonRpcProvider(
       adapterSettings.ARBITRUM_RPC_URL,
       adapterSettings.ARBITRUM_CHAIN_ID,
     )
@@ -85,6 +83,13 @@ export abstract class BaseGlvTransport<
     await this.tokenInfo()
     await this.marketInfo()
 
+    this.dataStreamsClient = new DataStreamsHttpClient({
+      baseUrl: adapterSettings.DATA_ENGINE_BASE_URL,
+      userId: adapterSettings.DATA_ENGINE_USER_ID,
+      userSecret: adapterSettings.DATA_ENGINE_USER_SECRET,
+      timeoutMs: adapterSettings.GLV_INFO_API_TIMEOUT_MS,
+      requester: dependencies.requester,
+    })
     if (this.settings.METADATA_REFRESH_INTERVAL_MS > 0) {
       setInterval(() => {
         this.tokenInfo()
@@ -195,8 +200,8 @@ export abstract class BaseGlvTransport<
       this.glvReaderContract.getGlvTokenPrice(...glvTokenPriceContractParams, false),
     ])
 
-    const maximizedPrice = Number(utils.formatUnits(maximizedPriceRaw, SIGNED_PRICE_DECIMALS))
-    const minimizedPrice = Number(utils.formatUnits(minimizedPriceRaw, SIGNED_PRICE_DECIMALS))
+    const maximizedPrice = Number(ethers.formatUnits(maximizedPriceRaw, SIGNED_PRICE_DECIMALS))
+    const minimizedPrice = Number(ethers.formatUnits(minimizedPriceRaw, SIGNED_PRICE_DECIMALS))
     const result = median([minimizedPrice, maximizedPrice])
 
     const timestamps = {
@@ -217,74 +222,64 @@ export abstract class BaseGlvTransport<
   private async fetchPrices(assets: string[], dataRequestedTimestamp: number) {
     const priceData = {} as PriceData
 
-    const sources = [
-      { url: this.settings.TIINGO_ADAPTER_URL, name: 'tiingo' },
-      { url: this.settings.COINMETRICS_ADAPTER_URL, name: 'coinmetrics' },
-      { url: this.settings.NCFX_ADAPTER_URL, name: 'ncfx' },
-    ]
-
     const priceProviders: Record<string, string[]> = {}
-    const promises = []
-
-    for (let i = 0; i < sources.length; i++) {
-      const source = sources[i]
-      const assetPromises = assets.map(async (asset) => {
-        const mappedAsset = mapParameter(source.name, asset)
-        const base = this.unwrapAsset(mappedAsset)
-        const requestConfig = {
-          url: source.url,
-          method: 'POST',
-          data: {
-            data: {
-              endpoint: 'crypto-lwba',
-              base,
-              quote: 'USD',
+    const sources: Source[] = [{ name: 'data-streams', url: this.settings.DATA_ENGINE_BASE_URL }]
+
+    await Promise.all(
+      assets.map(async (asset) => {
+        const base = this.unwrapAsset(asset)
+        const feedId = await this.dataStreamsClient.resolveFeedId(base, 'USD', 'Crypto')
+        const { fullReportHex } = await this.dataStreamsClient.getLatestReport(feedId)
+
+        // Decode (V3 expected: price, bid, ask; decoder handles V2–V10)
+        const decoded = decodeReport(fullReportHex, feedId)
+        const scale = 10 ** SIGNED_PRICE_DECIMALS
+        const toNum = (x: any | undefined) => (x === undefined ? undefined : Number(x) / scale)
+
+        const v3Bid = (decoded as any).bid
+        const v3Ask = (decoded as any).ask
+        const v3Price = (decoded as any).price
+
+        const bidNum = toNum(v3Bid ?? v3Price)
+        const askNum = toNum(v3Ask ?? v3Price)
+
+        if (bidNum === undefined || askNum === undefined) {
+          throw new AdapterDataProviderError(
+            { statusCode: 502, message: `Could not decode bid/ask for ${asset}` },
+            {
+              providerDataRequestedUnixMs: dataRequestedTimestamp,
+              providerDataReceivedUnixMs: Date.now(),
+              providerIndicatedTimeUnixMs: undefined,
             },
-          },
-        }
-
-        try {
-          const response = await this.requester.request<{ data: LwbaResponseDataFields['Data'] }>(
-            JSON.stringify(requestConfig),
-            requestConfig,
-          )
-          const { bid, ask } = response.response.data.data
-
-          priceData[asset] = {
-            bids: [...(priceData[asset]?.bids || []), bid],
-            asks: [...(priceData[asset]?.asks || []), ask],
-          }
-
-          priceProviders[asset] = priceProviders[asset]
-            ? [...new Set([...priceProviders[asset], source.name])]
-            : [source.name]
-        } catch (error) {
-          const e = error as Error
-          logger.error(
-            `Error fetching data for ${asset} from ${source.name}, url - ${source.url}: ${e.message}`,
           )
         }
-      })
 
-      promises.push(...assetPromises)
-    }
+        // Store raw numbers for median calc
+        priceData[asset] = {
+          bids: [...(priceData[asset]?.bids || []), bidNum],
+          asks: [...(priceData[asset]?.asks || []), askNum],
+        }
 
-    await Promise.all(promises)
+        // Track that Data Streams responded for this *base* key
+        priceProviders[base] = priceProviders[base] ? priceProviders[base] : []
+        if (!priceProviders[base].includes(sources[0].name)) {
+          priceProviders[base].push(sources[0].name)
+        }
+      }),
+    )
 
     this.validateRequiredResponses(priceProviders, sources, assets, dataRequestedTimestamp)
 
     const medianValues = this.calculateMedian(assets, priceData)
-
     const prices: Record<string, Record<string, string | number>> = {}
 
-    medianValues.map(
-      (v) =>
-        (prices[v.asset] = {
-          ...v,
-          ask: toFixed(v.ask, this.decimals[v.asset as keyof typeof this.decimals]),
-          bid: toFixed(v.bid, this.decimals[v.asset as keyof typeof this.decimals]),
-        }),
-    )
+    medianValues.forEach((v) => {
+      prices[v.asset] = {
+        ...v,
+        ask: toFixed(v.ask, this.decimals[v.asset as keyof typeof this.decimals]),
+        bid: toFixed(v.bid, this.decimals[v.asset as keyof typeof this.decimals]),
+      }
+    })
 
     return {
       prices,

packages/composites/glv-token/src/transport/utils.ts

@@ -52,24 +52,3 @@ export interface Market {
 export function mapSymbol(address: string, symbolMap: Record<string, any>) {
   return symbolMap[address]
 }
-
-const adapterParamOverride: Record<string, Record<string, string>> = {
-  coinmetrics: {
-    TAO: 'tao_bittensor',
-    SPX6900: 'spx',
-  },
-  tiingo: {
-    FLOKI: 'floki2',
-    SPX6900: 'spx',
-  },
-  ncfx: {
-    SPX6900: 'spx',
-  },
-}
-
-export function mapParameter(source: string, param: string) {
-  if (source in adapterParamOverride && param in adapterParamOverride[source]) {
-    return adapterParamOverride[source][param]
-  }
-  return param
-}

packages/composites/glv-token/test/integration/adapter.test.ts

@@ -1,25 +1,25 @@
+import {
+  LwbaResponseDataFields,
+  validateLwbaResponse,
+} from '@chainlink/external-adapter-framework/adapter'
 import {
   TestAdapter,
   setEnvVariables,
 } from '@chainlink/external-adapter-framework/util/testing-utils'
+import { ethers } from 'ethers'
 import * as nock from 'nock'
 import {
   mockCoinmetricsEAResponseSuccess,
   mockNCFXEAResponseSuccess,
   mockTiingoEAResponseSuccess,
 } from './fixtures'
-import { ethers } from 'ethers'
-import {
-  LwbaResponseDataFields,
-  validateLwbaResponse,
-} from '@chainlink/external-adapter-framework/adapter'
 
 jest.mock('ethers', () => ({
   ...jest.requireActual('ethers'),
   ethers: {
     providers: {
-      JsonRpcProvider: function (): ethers.providers.JsonRpcProvider {
-        return {} as ethers.providers.JsonRpcProvider
+      JsonRpcProvider: function (): ethers.JsonRpcProvider {
+        return {} as ethers.JsonRpcProvider
       },
     },
     Contract: function () {