feat: enable USDT/DAI trading pair (#100)

* feat: enable USDT/DAI trading pair

* remove timeout from price feed

* add more verbose logging to unrecoverable error

* attempt to recover from ccxt.NetworkError

Author: Karl Ranna

src/__snapshots__/config.spec.ts.snap

@@ -1,6 +1,6 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`checkConfigOptions LIVE_CEX disabled does not allow LTC/LTC trading pair 1`] = `"Invalid trading pair LTC/LTC. Supported trading pairs are: ETH/BTC, BTC/USDT, BTC/DAI, LTC/BTC, LTC/USDT"`;
+exports[`checkConfigOptions LIVE_CEX disabled does not allow LTC/LTC trading pair 1`] = `"Invalid trading pair LTC/LTC. Supported trading pairs are: ETH/BTC, BTC/USDT, BTC/DAI, LTC/BTC, LTC/USDT, USDT/DAI"`;
 
 exports[`checkConfigOptions LIVE_CEX disabled requires TEST_CENTRALIZED_EXCHANGE_BASEASSET_BALANCE 1`] = `"Incomplete configuration. Please add the following options to .env or as environment variables: TEST_CENTRALIZED_EXCHANGE_BASEASSET_BALANCE"`;
 

src/arby.ts

@@ -111,8 +111,10 @@ export const startArby = ({
               CEX,
             })
           ).pipe(
-            catchError(() => {
-              loggers.global.info('Unrecoverable error. Cleaning up.');
+            catchError(e => {
+              loggers.global.info(
+                `Unrecoverable error: ${JSON.stringify(e)} - cleaning up.`
+              );
               return cleanup$({
                 config,
                 loggers,

src/centralized/exchange-price.ts

@@ -1,11 +1,6 @@
 import BigNumber from 'bignumber.js';
 import { Observable, throwError } from 'rxjs';
-import {
-  catchError,
-  timeout,
-  share,
-  distinctUntilChanged,
-} from 'rxjs/operators';
+import { catchError, share, distinctUntilChanged } from 'rxjs/operators';
 import WebSocket from 'ws';
 import { Config } from '../config';
 import { Logger } from '../logger';
@@ -55,8 +50,6 @@ const getCentralizedExchangePrice$ = ({
     };
   });
   return priceObservable.pipe(
-    // if we have not received a price value in 20 seconds we'll error
-    timeout(10000),
     catchError(() => {
       return throwError(errors.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR);
     }),

src/config.spec.ts

@@ -102,6 +102,18 @@ describe('checkConfigOptions', () => {
       });
     });
 
+    it('allows USDT/DAI trading pair', () => {
+      const config = checkConfigOptions({
+        ...validLiveCEXenabledConf,
+        ...{ BASEASSET: 'USDT', QUOTEASSET: 'DAI' },
+      });
+      expect(config).toEqual({
+        ...config,
+        CEX: 'KRAKEN',
+        LIVE_CEX: true,
+      });
+    });
+
     it('allows ETH/BTC trading pair', () => {
       const config = checkConfigOptions({
         ...validLiveCEXenabledConf,

src/config.ts

@@ -90,6 +90,7 @@ const ALLOWED_TRADING_PAIRS: string[] = [
   'BTC/DAI',
   'LTC/BTC',
   'LTC/USDT',
+  'USDT/DAI',
 ];
 
 const checkConfigOptions = (dotEnvConfig: DotenvParseOutput): Config => {

src/opendex/catch-error.spec.ts

@@ -1,5 +1,5 @@
 import { status } from '@grpc/grpc-js';
-import { AuthenticationError, Exchange } from 'ccxt';
+import { AuthenticationError, Exchange, NetworkError } from 'ccxt';
 import { TestScheduler } from 'rxjs/testing';
 import { errors } from '../opendex/errors';
 import { ArbyStore, getArbyStore } from '../store';
@@ -64,6 +64,34 @@ const assertCatchOpenDEXerror = ({
 
 const ASSERTIONS_PER_TEST = 3;
 
+const assertCEXerror = (error: any) => {
+  return () => {
+    // 1 extra assertion after assertCatchOpenDEXerror
+    expect.assertions(ASSERTIONS_PER_TEST + 1);
+    const inputEvents = '1s #';
+    const inputError = error;
+    const expected = '';
+    const expectedSubscriptions = {
+      input$: ['^ 999ms !', '7001ms ^ 999ms !'],
+      cleanup$: ['1s ^ 1s !', '8001ms ^ 1s !'],
+    };
+    const store = {
+      ...getArbyStore(),
+      ...{ resetLastOrderUpdatePrice: jest.fn() },
+    };
+    const unsubscribe = '10s !';
+    assertCatchOpenDEXerror({
+      inputEvents,
+      inputError,
+      expected,
+      unsubscribe,
+      expectedSubscriptions,
+      store,
+    });
+    expect(store.resetLastOrderUpdatePrice).toHaveBeenCalledTimes(2);
+  };
+};
+
 describe('catchOpenDEXerror', () => {
   beforeEach(testSchedulerSetup);
 
@@ -189,31 +217,15 @@ describe('catchOpenDEXerror', () => {
     });
   });
 
-  it('cancels orders, updates store lastPriceUpdate, retries CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR', () => {
-    // 1 extra assertion after assertCatchOpenDEXerror
-    expect.assertions(ASSERTIONS_PER_TEST + 1);
-    const inputEvents = '1s #';
-    const inputError = errors.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR;
-    const expected = '';
-    const expectedSubscriptions = {
-      input$: ['^ 999ms !', '7001ms ^ 999ms !'],
-      cleanup$: ['1s ^ 1s !', '8001ms ^ 1s !'],
-    };
-    const store = {
-      ...getArbyStore(),
-      ...{ resetLastOrderUpdatePrice: jest.fn() },
-    };
-    const unsubscribe = '10s !';
-    assertCatchOpenDEXerror({
-      inputEvents,
-      inputError,
-      expected,
-      unsubscribe,
-      expectedSubscriptions,
-      store,
-    });
-    expect(store.resetLastOrderUpdatePrice).toHaveBeenCalledTimes(2);
-  });
+  it(
+    'cancels orders, updates store lastPriceUpdate, retries CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR',
+    assertCEXerror(errors.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR)
+  );
+
+  it(
+    'cancels orders, updates store lastPriceUpdate, retries NetworkError',
+    assertCEXerror(new NetworkError('CEX connection lost'))
+  );
 
   it('retries recoverable gRPC errors', () => {
     const recoverableGRPCerrors = [

src/opendex/catch-error.ts

@@ -1,5 +1,5 @@
 import { status } from '@grpc/grpc-js';
-import { AuthenticationError, Exchange } from 'ccxt';
+import { AuthenticationError, Exchange, NetworkError } from 'ccxt';
 import { concat, Observable, throwError, timer } from 'rxjs';
 import { ignoreElements, mergeMap, retryWhen } from 'rxjs/operators';
 import { ArbyStore } from 'src/store';
@@ -62,7 +62,8 @@ const catchOpenDEXerror = (
               logMessage(loggers.opendex);
               return timer(RETRY_INTERVAL);
             } else if (
-              e.code === errorCodes.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR
+              e.code === errorCodes.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR ||
+              e instanceof NetworkError
             ) {
               logMessage(loggers.centralized);
               store.resetLastOrderUpdatePrice();