Feat/DF-20368 combine tp and icap (#3449)

Co-authored-by: Michael Xiao <[email protected]>

.changeset/lazy-ducks-pay.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/icap-adapter': minor
+---
+
+Seperate ICAP from TP

.changeset/popular-queens-yawn.md

@@ -0,0 +1,5 @@
+---
+'@chainlink/tp-adapter': minor
+---
+
+Combined TP and ICAP EAs into a single EA and removed ICAP.URL must have query param appended as selector in bridge URL, eg: https://<tp-ea>:8080?streamName=icapThis change will save subscription costs as all data for both DPs is sent on 1 WS connection and each additional connection requires additional subscriptions (and cost).Should be backwards compatible for TP ONLY

packages/sources/icap/src/endpoint/price.ts

@@ -1,12 +1,42 @@
-import { generateInputParams, GeneratePriceOptions, generateTransport } from '@chainlink/tp-adapter'
+import { GeneratePriceOptions } from '@chainlink/tp-adapter'
 import { ForexPriceEndpoint } from '@chainlink/external-adapter-framework/adapter'
+import { InputParameters } from '@chainlink/external-adapter-framework/validation'
+import {
+  priceEndpointInputParametersDefinition,
+  PriceEndpointInputParametersDefinition,
+} from '@chainlink/external-adapter-framework/adapter'
+import { generateTransport } from '../transport/price'
 
 const options: GeneratePriceOptions = {
   sourceName: 'icapSource',
   streamName: 'IC',
   sourceOptions: ['BGK', 'GBL', 'HKG', 'JHB'],
 }
 
+export const generateInputParams = (
+  generatePriceOptions: GeneratePriceOptions,
+): InputParameters<PriceEndpointInputParametersDefinition> =>
+  new InputParameters(
+    {
+      ...priceEndpointInputParametersDefinition,
+      [generatePriceOptions.sourceName]: {
+        description: `Source of price data for this price pair on the ${generatePriceOptions.streamName} stream`,
+        default: 'GBL',
+        required: false,
+        type: 'string',
+        ...(generatePriceOptions.sourceOptions
+          ? { options: generatePriceOptions.sourceOptions }
+          : {}),
+      },
+    },
+    [
+      {
+        base: 'EUR',
+        quote: 'USD',
+      },
+    ],
+  )
+
 const inputParameters = generateInputParams(options)
 const transport = generateTransport(options)
 

packages/sources/icap/src/transport/price.ts

@@ -0,0 +1,210 @@
+import Decimal from 'decimal.js'
+import { WebSocketTransport } from '@chainlink/external-adapter-framework/transports/websocket'
+import { makeLogger, ProviderResult } from '@chainlink/external-adapter-framework/util'
+import { BaseEndpointTypes, GeneratePriceOptions } from '@chainlink/tp-adapter'
+
+const logger = makeLogger('TpIcapPrice')
+
+type WsMessage = {
+  msg: 'auth' | 'sub'
+  pro?: string
+  rec: string // example: FXSPTEURUSDSPT:GBL.BIL.QTE.RTM!IC
+  sta: number
+  img?: number
+  fvs?: {
+    CCY1?: string // example: "EUR"
+    CCY2?: string // example: "USD"
+    ACTIV_DATE?: string // example: "2023-03-06"
+    TIMACT?: string // example: "15:00:00"
+    BID?: number
+    ASK?: number
+    MID_PRICE?: number
+  }
+}
+
+export type WsTransportTypes = BaseEndpointTypes & {
+  Provider: {
+    WsMessage: WsMessage
+  }
+}
+
+const isNum = (i: number | undefined) => typeof i === 'number'
+
+let providerDataStreamEstablishedUnixMs: number
+
+/*
+TP and ICAP EAs currently do not receive asset prices during off-market hours. When a heartbeat message is received during these hours,
+we update the TTL of cache entries that EA is requested to provide a price during off-market hours.
+ */
+const updateTTL = async (transport: WebSocketTransport<WsTransportTypes>, ttl: number) => {
+  const params = await transport.subscriptionSet.getAll()
+  transport.responseCache.writeTTL(transport.name, params, ttl)
+}
+
+export const generateTransport = (generatePriceOptions: GeneratePriceOptions) => {
+  const tpTransport = new WebSocketTransport<WsTransportTypes>({
+    url: ({ adapterSettings: { WS_API_ENDPOINT } }) => WS_API_ENDPOINT,
+    handlers: {
+      open: (connection, { adapterSettings: { WS_API_USERNAME, WS_API_PASSWORD } }) => {
+        logger.debug('Opening WS connection')
+
+        return new Promise((resolve) => {
+          connection.addEventListener('message', (event: MessageEvent) => {
+            const { msg, sta } = JSON.parse(event.data.toString())
+            if (msg === 'auth' && sta === 1) {
+              logger.info('Got logged in response, connection is ready')
+              providerDataStreamEstablishedUnixMs = Date.now()
+              resolve()
+            }
+          })
+          const options = {
+            msg: 'auth',
+            user: WS_API_USERNAME,
+            pass: WS_API_PASSWORD,
+            mode: 'broadcast',
+          }
+          connection.send(JSON.stringify(options))
+        })
+      },
+      message: (message, context) => {
+        logger.debug({ msg: 'Received message from WS', message })
+
+        const providerDataReceivedUnixMs = Date.now()
+
+        if (!('msg' in message) || message.msg === 'auth') return []
+
+        const { fvs, rec, sta } = message
+
+        if (!fvs || !rec || sta !== 1) {
+          logger.debug({ msg: 'Missing expected field `fvs` or `rec` from `sub` message', message })
+          return []
+        }
+
+        // Check for a heartbeat message, refresh the TTLs of all requested entries in the cache
+        if (rec.includes('HBHHH')) {
+          logger.debug({
+            msg: 'Received heartbeat message from WS, updating TTLs of active entries',
+            message,
+          })
+          updateTTL(tpTransport, context.adapterSettings.CACHE_MAX_AGE)
+          return []
+        }
+
+        const ticker = parseRec(rec)
+        if (!ticker) {
+          logger.debug({ msg: `Invalid symbol: ${rec}`, message })
+          return []
+        }
+
+        if (ticker.stream !== generatePriceOptions.streamName) {
+          logger.debug({
+            msg: `Only ${generatePriceOptions.streamName} forex prices accepted on this adapter. Filtering out this message.`,
+            message,
+          })
+          return []
+        }
+
+        const { ASK, BID, MID_PRICE } = fvs
+
+        if (!isNum(MID_PRICE) && !(isNum(BID) && isNum(ASK))) {
+          const errorMessage = '`sub` message did not include required price fields'
+          logger.debug({ errorMessage, message })
+          return []
+        }
+
+        const result =
+          MID_PRICE ||
+          new Decimal(ASK as number)
+            .add(BID as number)
+            .div(2)
+            .toNumber()
+
+        const response = {
+          result,
+          data: {
+            result,
+          },
+          timestamps: {
+            providerDataReceivedUnixMs,
+            providerDataStreamEstablishedUnixMs,
+            providerIndicatedTimeUnixMs: undefined,
+          },
+        }
+
+        // Cache both the base and the full ticker string. The full ticker is to
+        // accomodate cases where there are multiple instruments for a single base
+        // (e.g. forwards like CEFWDXAUUSDSPT06M:LDN.BIL.QTE.RTM!TP, CEFWDXAUUSDSPT02Y:LDN.BIL.QTE.RTM!TP, CEFWDXAUUSDSPT03M:LDN.BIL.QTE.RTM!TP, etc).
+        // It is expected that for such cases, the exact ticker will be provided as
+        // an override.
+        // e.g. request body = {"data":{"endpoint":"forex","from":"CHF","to":"USD","overrides":{"tp":{"CHF":"FXSPTUSDAEDSPT:GBL.BIL.QTE.RTM!TP"}}}}
+        return [
+          {
+            params: {
+              base: ticker.base,
+              quote: ticker.quote,
+              [generatePriceOptions.sourceName]: ticker.source,
+            },
+            response,
+          },
+          {
+            params: {
+              base: rec,
+              quote: ticker.quote,
+              [generatePriceOptions.sourceName]: ticker.source,
+            },
+            response,
+          },
+        ] as unknown as ProviderResult<WsTransportTypes>[]
+      },
+    },
+  })
+  return tpTransport
+}
+
+// mapping OTRWTS to WTIUSD specifically for caching with quote = USD
+const marketBaseQuoteOverrides: Record<string, string> = {
+  CEOILOTRWTS: 'CEOILWTIUSD',
+}
+
+type Ticker = {
+  market: string
+  base: string
+  quote: string
+  source: string
+  stream: string
+}
+
+/*
+For example, if rec = 'FXSPTCHFSEKSPT:GBL.BIL.QTE.RTM!IC', then the parsed output is
+{
+  market: 'FXSPT',
+  base: 'CHF',
+  quote: 'SEK',
+  source: 'GBL',
+  stream: 'IC'
+}
+*/
+export const parseRec = (rec: string): Ticker | null => {
+  const [symbol, rec1] = rec.split(':')
+  if (!rec1) {
+    return null
+  }
+
+  const [sources, stream] = rec1.split('!')
+  if (!stream) {
+    return null
+  }
+
+  let marketBaseQuote = symbol.slice(0, 11)
+  if (marketBaseQuote in marketBaseQuoteOverrides) {
+    marketBaseQuote = marketBaseQuoteOverrides[marketBaseQuote]
+  }
+
+  return {
+    market: marketBaseQuote.slice(0, 5),
+    base: marketBaseQuote.slice(5, 8),
+    quote: marketBaseQuote.slice(8, 11),
+    source: sources.split('.')[0],
+    stream,
+  }
+}

packages/sources/tp/README.md

@@ -8,9 +8,9 @@ This document was generated automatically. Please see [README Generator](../../s
 
 ### Concurrent connections
 
-Context: TP and ICAP EAs use the same credentials, and often there are issues with the set of credentials not having concurrent (ie: 2+) connections enabled.
+Context: Often there are issues with the set of credentials not having concurrent (ie: 2+) connections enabled.
 
-- With both TP and ICAP EAs off, try the following commands to check if this is the case:
+- With all EA instances off, try the following commands to check if this is the case:
 
 ```bash
 wscat -c 'ws://json.mktdata.portal.apac.parametasolutions.com:12000'
@@ -56,11 +56,12 @@ Supported names for this endpoint are: `commodities`, `forex`, `price`.
 
 ### Input Params
 
-| Required? |   Name   |    Aliases     |                        Description                        |  Type  | Options | Default | Depends On | Not Valid With |
-| :-------: | :------: | :------------: | :-------------------------------------------------------: | :----: | :-----: | :-----: | :--------: | :------------: |
-|    ✅     |   base   | `coin`, `from` |      The symbol of symbols of the currency to query       | string |         |         |            |                |
-|    ✅     |  quote   | `market`, `to` |         The symbol of the currency to convert to          | string |         |         |            |                |
-|           | tpSource |                | Source of price data for this price pair on the TP stream | string |         |  `GBL`  |            |                |
+| Required? |    Name    |    Aliases     |                      Description                       |  Type  |  Options   | Default | Depends On | Not Valid With |
+| :-------: | :--------: | :------------: | :----------------------------------------------------: | :----: | :--------: | :-----: | :--------: | :------------: |
+|    ✅     |    base    | `coin`, `from` |     The symbol of symbols of the currency to query     | string |            |         |            |                |
+|    ✅     |   quote    | `market`, `to` |        The symbol of the currency to convert to        | string |            |         |            |                |
+|           | streamName |    `source`    |                       TP or ICAP                       | string | `IC`, `TP` |  `TP`   |            |                |
+|           | sourceName |   `tpSource`   | Source of price data for this price pair on the stream | string |            |  `GBL`  |            |                |
 
 ### Example
 
@@ -71,7 +72,9 @@ Request:
   "data": {
     "endpoint": "price",
     "base": "EUR",
-    "quote": "USD"
+    "quote": "USD",
+    "streamName": "TP",
+    "sourceName": "GBL"
   }
 }
 ```

packages/sources/tp/docs/known-issues.md

@@ -2,9 +2,9 @@
 
 ### Concurrent connections
 
-Context: TP and ICAP EAs use the same credentials, and often there are issues with the set of credentials not having concurrent (ie: 2+) connections enabled.
+Context: Often there are issues with the set of credentials not having concurrent (ie: 2+) connections enabled.
 
-- With both TP and ICAP EAs off, try the following commands to check if this is the case:
+- With all EA instances off, try the following commands to check if this is the case:
 
 ```bash
 wscat -c 'ws://json.mktdata.portal.apac.parametasolutions.com:12000'

packages/sources/tp/package.json

@@ -38,6 +38,9 @@
     "@types/jest": "27.5.2",
     "@types/node": "16.18.96",
     "@types/sinonjs__fake-timers": "8.1.5",
+    "@types/supertest": "2.0.16",
+    "mock-socket": "9.3.1",
+    "supertest": "6.2.4",
     "typescript": "5.0.4"
   }
 }

packages/sources/tp/src/config/includes.json

@@ -1,4 +1,15 @@
 [
+  {
+    "from": "CAD",
+    "to": "USD",
+    "includes": [
+      {
+        "from": "USD",
+        "to": "CAD",
+        "inverse": true
+      }
+    ]
+  },
   {
     "from": "AED",
     "to": "USD",

packages/sources/tp/src/endpoint/index.ts

@@ -1 +1,6 @@
-export { generateInputParams, priceEndpoint, GeneratePriceOptions } from './price'
+export {
+  generateInputParams,
+  priceEndpoint,
+  GeneratePriceOptions,
+  BaseEndpointTypes,
+} from './price'