Verify CEX assets (#111)

* rename initBinance$ -> initCEX$

* return markets from initCEX$

* fixup! rename initBinance$ -> initCEX$

* verify CEX assets

* fixup! verify CEX assets

Author: Karl Ranna

src/arby.spec.ts

@@ -2,8 +2,8 @@ import { Observable } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { startArby } from '../src/arby';
 import { Config } from '../src/config';
+import { InitCEXResponse } from './centralized/ccxt/init';
 import { getLoggers } from './test-utils';
-import { Exchange } from 'ccxt';
 
 let testScheduler: TestScheduler;
 
@@ -14,11 +14,16 @@ type AssertStartArbyParams = {
     getTrade$: string;
     shutdown$: string;
     cleanup$: string;
-    initBinance$: string;
+    initCEX$: string;
   };
+  verifyMarkets?: () => boolean;
 };
 
-const assertStartArby = ({ expected, inputEvents }: AssertStartArbyParams) => {
+const assertStartArby = ({
+  expected,
+  inputEvents,
+  verifyMarkets,
+}: AssertStartArbyParams) => {
   testScheduler.run(helpers => {
     const { cold, expectObservable } = helpers;
     const config$ = cold(inputEvents.config$) as Observable<Config>;
@@ -29,9 +34,9 @@ const assertStartArby = ({ expected, inputEvents }: AssertStartArbyParams) => {
     const cleanup$ = () => {
       return cold(inputEvents.cleanup$);
     };
-    const initBinance$ = () => {
-      return (cold(inputEvents.initBinance$) as unknown) as Observable<
-        Exchange
+    const initCEX$ = () => {
+      return (cold(inputEvents.initCEX$) as unknown) as Observable<
+        InitCEXResponse
       >;
     };
     const arby$ = startArby({
@@ -40,9 +45,10 @@ const assertStartArby = ({ expected, inputEvents }: AssertStartArbyParams) => {
       shutdown$,
       trade$: getTrade$,
       cleanup$,
-      initBinance$,
+      initCEX$,
+      verifyMarkets: verifyMarkets ? verifyMarkets : () => true,
     });
-    expectObservable(arby$).toBe(expected);
+    expectObservable(arby$).toBe(expected, undefined, { message: 'error' });
   });
 };
 
@@ -56,7 +62,7 @@ describe('startArby', () => {
   it('waits for valid configuration before starting', () => {
     const inputEvents = {
       config$: '1000ms a',
-      initBinance$: '1s a',
+      initCEX$: '1s a',
       getTrade$: 'b',
       shutdown$: '',
       cleanup$: '',
@@ -68,10 +74,28 @@ describe('startArby', () => {
     });
   });
 
+  it('errors when verifyMarkets fails', () => {
+    const inputEvents = {
+      config$: '1000ms a',
+      initCEX$: '1s a',
+      getTrade$: 'b',
+      shutdown$: '',
+      cleanup$: '',
+    };
+    const expected = '2s #';
+    assertStartArby({
+      inputEvents,
+      expected,
+      verifyMarkets: () => {
+        throw { message: 'error' };
+      },
+    });
+  });
+
   it('performs cleanup when shutting down gracefully', () => {
     const inputEvents = {
       config$: 'a',
-      initBinance$: '1s a',
+      initCEX$: '1s a',
       getTrade$: '500ms b',
       shutdown$: '10s c',
       cleanup$: '2s a',
@@ -86,7 +110,7 @@ describe('startArby', () => {
   it('performs cleanup when getTrade$ errors', () => {
     const inputEvents = {
       config$: 'a',
-      initBinance$: '1s a',
+      initCEX$: '1s a',
       getTrade$: '500ms #',
       shutdown$: '10s c',
       cleanup$: '2s a',

src/arby.ts

@@ -1,8 +1,11 @@
-import { Exchange } from 'ccxt';
 import { concat, Observable } from 'rxjs';
 import { catchError, mergeMap, takeUntil } from 'rxjs/operators';
 import { getExchange } from './centralized/ccxt/exchange';
-import { initBinance$, InitBinanceParams } from './centralized/ccxt/init';
+import {
+  initCEX$,
+  InitCEXparams,
+  InitCEXResponse,
+} from './centralized/ccxt/init';
 import { loadMarkets$ } from './centralized/ccxt/load-markets';
 import { getCentralizedExchangePrice$ } from './centralized/exchange-price';
 import { getCentralizedExchangeOrder$ } from './centralized/order';
@@ -12,10 +15,12 @@ import { Logger, Loggers } from './logger';
 import { catchOpenDEXerror } from './opendex/catch-error';
 import { getOpenDEXcomplete$ } from './opendex/complete';
 import { removeOpenDEXorders$ } from './opendex/remove-orders';
+import { getArbyStore } from './store';
 import { getCleanup$, GetCleanupParams } from './trade/cleanup';
 import { getNewTrade$, GetTradeParams } from './trade/trade';
 import { getStartShutdown$ } from './utils';
-import { getArbyStore } from './store';
+import { Dictionary, Market } from 'ccxt';
+import { verifyMarkets } from './centralized/verify-markets';
 
 type StartArbyParams = {
   config$: Observable<Config>;
@@ -32,11 +37,12 @@ type StartArbyParams = {
     config,
     removeOpenDEXorders$,
   }: GetCleanupParams) => Observable<unknown>;
-  initBinance$: ({
+  initCEX$: ({
     getExchange,
     config,
     loadMarkets$,
-  }: InitBinanceParams) => Observable<Exchange>;
+  }: InitCEXparams) => Observable<InitCEXResponse>;
+  verifyMarkets: (config: Config, CEXmarkets: Dictionary<Market>) => boolean;
 };
 
 const logConfig = (config: Config, logger: Logger) => {
@@ -79,21 +85,23 @@ export const startArby = ({
   shutdown$,
   trade$,
   cleanup$,
-  initBinance$,
+  initCEX$,
+  verifyMarkets,
 }: StartArbyParams): Observable<any> => {
   const store = getArbyStore();
   return config$.pipe(
     mergeMap(config => {
-      const CEX$ = initBinance$({
+      const CEX$ = initCEX$({
         config,
         loadMarkets$,
         getExchange,
       });
       return CEX$.pipe(
-        mergeMap(CEX => {
+        mergeMap(({ markets: CEXmarkets, exchange: CEX }) => {
           const loggers = getLoggers(config);
           loggers.global.info('Starting. Hello, Arby.');
           logConfig(config, loggers.global);
+          verifyMarkets(config, CEXmarkets);
           const tradeComplete$ = trade$({
             config,
             loggers,
@@ -145,7 +153,8 @@ if (!module.parent) {
     getLoggers,
     shutdown$: getStartShutdown$(),
     cleanup$: getCleanup$,
-    initBinance$,
+    initCEX$,
+    verifyMarkets,
   }).subscribe({
     error: error => {
       if (error.message) {

src/centralized/__snapshots__/minimum-order-quantity-filter.spec.ts.snap

@@ -0,0 +1,3 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`quantityAboveMinimum throws for unknown asset KILRAU 1`] = `"Could not retrieve minimum order quantity for KILRAU"`;

src/centralized/__snapshots__/verify-markets.spec.ts.snap

@@ -0,0 +1,5 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`verifyMarkets throws when BTC/USD trading pair is inactive 1`] = `"The configured trading pair BTC/USD does not exist on Binance"`;
+
+exports[`verifyMarkets throws when BTC/USD trading pair not exist 1`] = `"The configured trading pair BTC/USD does not exist on Binance"`;

src/centralized/ccxt/init.spec.ts

@@ -2,7 +2,7 @@ import { Dictionary, Exchange, Market } from 'ccxt';
 import { Observable, of } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { testConfig } from '../../test-utils';
-import { initBinance$ } from './init';
+import { initCEX$ } from './init';
 
 let testScheduler: TestScheduler;
 
@@ -21,12 +21,14 @@ const assertInitBinance = (
     };
     const config = testConfig();
     const getExchange = () => ('a' as unknown) as Exchange;
-    const centralizedExchangeOrder$ = initBinance$({
+    const centralizedExchangeOrder$ = initCEX$({
       config,
       loadMarkets$,
       getExchange,
     });
-    expectObservable(centralizedExchangeOrder$).toBe(expected);
+    expectObservable(centralizedExchangeOrder$).toBe(expected, {
+      a: { exchange: 'a', markets: 'a' },
+    });
   });
 };
 
@@ -54,7 +56,7 @@ it('initializes once and loads markets', done => {
   const getExchange = jest.fn(() => {
     return (null as unknown) as Exchange;
   });
-  const CEX = initBinance$({
+  const CEX = initCEX$({
     loadMarkets$,
     config: testConfig(),
     getExchange,

src/centralized/ccxt/init.ts

@@ -1,21 +1,33 @@
 import { Dictionary, Exchange, Market } from 'ccxt';
 import { Observable } from 'rxjs';
-import { mapTo } from 'rxjs/operators';
+import { map } from 'rxjs/operators';
 import { Config } from '../../config';
 
-type InitBinanceParams = {
+type InitCEXparams = {
   config: Config;
   loadMarkets$: (exchange: Exchange) => Observable<Dictionary<Market>>;
   getExchange: (config: Config) => Exchange;
 };
 
-const initBinance$ = ({
+type InitCEXResponse = {
+  markets: Dictionary<Market>;
+  exchange: Exchange;
+};
+
+const initCEX$ = ({
   getExchange,
   config,
   loadMarkets$,
-}: InitBinanceParams): Observable<Exchange> => {
+}: InitCEXparams): Observable<InitCEXResponse> => {
   const exchange = getExchange(config);
-  return loadMarkets$(exchange).pipe(mapTo(exchange));
+  return loadMarkets$(exchange).pipe(
+    map(markets => {
+      return {
+        markets,
+        exchange,
+      };
+    })
+  );
 };
 
-export { initBinance$, InitBinanceParams };
+export { initCEX$, InitCEXparams, InitCEXResponse };

src/centralized/ccxt/load-markets.ts

@@ -1,5 +1,5 @@
-import { Exchange, Market, Dictionary } from 'ccxt';
-import { from, Observable, defer } from 'rxjs';
+import { Dictionary, Exchange, Market } from 'ccxt';
+import { defer, from, Observable } from 'rxjs';
 
 const loadMarkets$ = (exchange: Exchange): Observable<Dictionary<Market>> => {
   return defer(() => from(exchange.loadMarkets()));

src/centralized/derive-order-quantity.spec.ts

@@ -1,5 +1,5 @@
 import BigNumber from 'bignumber.js';
-import { OrderSide, Asset } from '../constants';
+import { OrderSide } from '../constants';
 import { deriveCEXorderQuantity } from './derive-order-quantity';
 import { CEXorder } from './order-builder';
 import { testConfig } from '../test-utils';
@@ -15,8 +15,8 @@ describe('deriveCEXorderQuantity', () => {
       quantity: new BigNumber('0.00104227'),
       side: OrderSide.BUY,
     };
-    const CEX_BASEASSET: Asset = 'BTC';
-    const CEX_QUOTEASSET: Asset = 'USDT';
+    const CEX_BASEASSET = 'BTC';
+    const CEX_QUOTEASSET = 'USDT';
     const config = {
       ...testConfig(),
       CEX_BASEASSET,
@@ -32,8 +32,8 @@ describe('deriveCEXorderQuantity', () => {
       quantity: new BigNumber('10.94381840'),
       side: OrderSide.BUY,
     };
-    const CEX_BASEASSET: Asset = 'ETH';
-    const CEX_QUOTEASSET: Asset = 'BTC';
+    const CEX_BASEASSET = 'ETH';
+    const CEX_QUOTEASSET = 'BTC';
     const config = {
       ...testConfig(),
       CEX_BASEASSET,

src/centralized/minimum-order-quantity-filter.spec.ts

@@ -37,4 +37,12 @@ describe('quantityAboveMinimum', () => {
     );
     expect(quantityAboveMinimum('USDT')(quantity)).toEqual(false);
   });
+
+  it('throws for unknown asset KILRAU', () => {
+    expect.assertions(1);
+    const quantity = new BigNumber('1');
+    expect(() => {
+      quantityAboveMinimum('KILRAU')(quantity);
+    }).toThrowErrorMatchingSnapshot();
+  });
 });

src/centralized/minimum-order-quantity-filter.ts

@@ -1,11 +1,8 @@
 import BigNumber from 'bignumber.js';
-import { Asset } from '../constants';
+import { errors } from '../opendex/errors';
 
 type MinimumCEXquantities = {
-  BTC: BigNumber;
-  ETH: BigNumber;
-  DAI: BigNumber;
-  USDT: BigNumber;
+  [key: string]: BigNumber;
 };
 
 const MINIMUM_ORDER_SIZE: MinimumCEXquantities = {
@@ -15,9 +12,17 @@ const MINIMUM_ORDER_SIZE: MinimumCEXquantities = {
   USDT: new BigNumber('15'),
 };
 
-const quantityAboveMinimum = (asset: Asset) => {
+const getMinimumOrderSize = (asset: string): BigNumber => {
+  const minimumOrderSize = MINIMUM_ORDER_SIZE[asset];
+  if (!minimumOrderSize) {
+    throw errors.CEX_INVALID_MINIMUM_ORDER_QUANTITY(asset);
+  }
+  return minimumOrderSize;
+};
+
+const quantityAboveMinimum = (asset: string) => {
   return (quantity: BigNumber): boolean => {
-    return quantity.isGreaterThanOrEqualTo(MINIMUM_ORDER_SIZE[asset]);
+    return quantity.isGreaterThanOrEqualTo(getMinimumOrderSize(asset));
   };
 };