Merge pull request #80 from ExchangeUnion/fix/price-throttle-update

fix: price throttle update

Author: Karl Ranna

src/arby.ts

@@ -15,6 +15,7 @@ import { removeOpenDEXorders$ } from './opendex/remove-orders';
 import { getCleanup$, GetCleanupParams } from './trade/cleanup';
 import { getNewTrade$, GetTradeParams } from './trade/trade';
 import { getStartShutdown$ } from './utils';
+import { getArbyStore } from './store';
 
 type StartArbyParams = {
   config$: Observable<Config>;
@@ -74,6 +75,7 @@ export const startArby = ({
   cleanup$,
   initBinance$,
 }: StartArbyParams): Observable<any> => {
+  const store = getArbyStore();
   return config$.pipe(
     mergeMap(config => {
       const CEX$ = initBinance$({
@@ -95,6 +97,7 @@ export const startArby = ({
             catchOpenDEXerror,
             getCentralizedExchangePrice$,
             CEX,
+            store,
           }).pipe(takeUntil(shutdown$));
           return concat(
             tradeComplete$,

src/centralized/order-builder.spec.ts

@@ -1,11 +1,12 @@
-import { getOrderBuilder$, CEXorder } from './order-builder';
-import { TestScheduler } from 'rxjs/testing';
-import { testConfig, getLoggers } from '../test-utils';
-import { Observable } from 'rxjs';
-import { SwapSuccess } from '../proto/xudrpc_pb';
-import { OrderSide, Asset } from '../constants';
 import BigNumber from 'bignumber.js';
+import { Observable } from 'rxjs';
+import { TestScheduler } from 'rxjs/testing';
 import { Config } from '../config';
+import { Asset, OrderSide } from '../constants';
+import { SwapSuccess } from '../proto/xudrpc_pb';
+import { ArbyStore, getArbyStore } from '../store';
+import { getLoggers, testConfig } from '../test-utils';
+import { CEXorder, getOrderBuilder$ } from './order-builder';
 
 let testScheduler: TestScheduler;
 
@@ -28,7 +29,8 @@ const assertOrderBuilder = (
     a: CEXorder;
   },
   config: Config,
-  expectedAssetToTradeOnCEX: Asset
+  expectedAssetToTradeOnCEX: Asset,
+  store?: ArbyStore
 ) => {
   testScheduler.run(helpers => {
     const { cold, expectObservable } = helpers;
@@ -59,6 +61,7 @@ const assertOrderBuilder = (
       accumulateOrderFillsForBaseAssetReceived: accumulateOrderFillsForAssetReceived,
       accumulateOrderFillsForQuoteAssetReceived: accumulateOrderFillsForAssetReceived,
       quantityAboveMinimum,
+      store: store ? store : getArbyStore(),
     });
     expectObservable(orderBuilder$, inputEvents.unsubscribe).toBe(
       expected,
@@ -83,6 +86,7 @@ describe('getCentralizedExchangeOrder$', () => {
   });
 
   it('accumulates buy and sell orders for ETHBTC', () => {
+    expect.assertions(6);
     const inputEvents = {
       receivedBaseAssetSwapSuccess$: '1s a',
       receivedQuoteAssetSwapSuccess$: '1400ms b',
@@ -117,17 +121,24 @@ describe('getCentralizedExchangeOrder$', () => {
       QUOTEASSET,
     };
     const expectedAssetToTradeOnCEX = BASEASSET;
+    const store = {
+      ...getArbyStore(),
+      ...{ resetLastOrderUpdatePrice: jest.fn() },
+    };
     assertOrderBuilder(
       inputEvents,
       inputValues,
       expected,
       expectedValues,
       config,
-      expectedAssetToTradeOnCEX
+      expectedAssetToTradeOnCEX,
+      store
     );
+    expect(store.resetLastOrderUpdatePrice).toHaveBeenCalledTimes(2);
   });
 
   it('accumulates buy and sell orders for BTCUSDT', () => {
+    expect.assertions(6);
     const inputEvents = {
       receivedBaseAssetSwapSuccess$: '1s a',
       receivedQuoteAssetSwapSuccess$: '1400ms b',
@@ -162,13 +173,19 @@ describe('getCentralizedExchangeOrder$', () => {
       QUOTEASSET,
     };
     const expectedAssetToTradeOnCEX = QUOTEASSET;
+    const store = {
+      ...getArbyStore(),
+      ...{ resetLastOrderUpdatePrice: jest.fn() },
+    };
     assertOrderBuilder(
       inputEvents,
       inputValues,
       expected,
       expectedValues,
       config,
-      expectedAssetToTradeOnCEX
+      expectedAssetToTradeOnCEX,
+      store
     );
+    expect(store.resetLastOrderUpdatePrice).toHaveBeenCalledTimes(2);
   });
 });

src/centralized/order-builder.ts

@@ -1,6 +1,6 @@
 import BigNumber from 'bignumber.js';
-import { merge, Observable } from 'rxjs';
-import { filter, map, repeat, take, tap } from 'rxjs/operators';
+import { merge, Observable, of } from 'rxjs';
+import { filter, map, repeat, take, tap, mergeMap } from 'rxjs/operators';
 import { Config } from '../config';
 import { OrderSide, Asset } from '../constants';
 import { Logger } from '../logger';
@@ -11,6 +11,7 @@ import {
 import { getXudClient$ } from '../opendex/xud/client';
 import { subscribeXudSwaps$ } from '../opendex/xud/subscribe-swaps';
 import { SwapSuccess } from '../proto/xudrpc_pb';
+import { ArbyStore } from '../store';
 
 type GetOrderBuilderParams = {
   config: Config;
@@ -29,6 +30,7 @@ type GetOrderBuilderParams = {
   quantityAboveMinimum: (
     asset: Asset
   ) => (filledQuantity: BigNumber) => boolean;
+  store: ArbyStore;
 };
 
 type CEXorder = {
@@ -43,6 +45,7 @@ const getOrderBuilder$ = ({
   accumulateOrderFillsForBaseAssetReceived,
   accumulateOrderFillsForQuoteAssetReceived,
   quantityAboveMinimum,
+  store,
 }: GetOrderBuilderParams): Observable<CEXorder> => {
   const {
     receivedBaseAssetSwapSuccess$,
@@ -58,10 +61,12 @@ const getOrderBuilder$ = ({
     // accumulate OpenDEX order fills when receiving
     // quote asset
     accumulateOrderFillsForQuoteAssetReceived(config),
-    tap((quantity: BigNumber) => {
+    mergeMap((quantity: BigNumber) => {
       logger.info(
         `Swap success. Accumulated ${assetToTradeOnCEX} quantity: ${quantity.toFixed()}`
       );
+      store.resetLastOrderUpdatePrice();
+      return of(quantity);
     }),
     // filter based on minimum CEX order quantity
     filter(quantityAboveMinimum(assetToTradeOnCEX)),

src/centralized/order.spec.ts

@@ -5,6 +5,7 @@ import { getCentralizedExchangeOrder$ } from './order';
 import { CEXorder } from './order-builder';
 import BigNumber from 'bignumber.js';
 import { Exchange } from 'ccxt';
+import { getArbyStore } from '../store';
 
 let testScheduler: TestScheduler;
 
@@ -35,6 +36,7 @@ const assertCentralizedExchangeOrder = (
     ) as unknown) as Observable<BigNumber>;
     const CEX = (null as unknown) as Exchange;
     const deriveCEXorderQuantity = (order: any) => order;
+    const store = getArbyStore();
     const centralizedExchangeOrder$ = getCentralizedExchangeOrder$({
       CEX,
       logger: getLoggers().centralized,
@@ -43,6 +45,7 @@ const assertCentralizedExchangeOrder = (
       executeCEXorder$,
       centralizedExchangePrice$,
       deriveCEXorderQuantity,
+      store,
     });
     expectObservable(centralizedExchangeOrder$, inputEvents.unsubscribe).toBe(
       expected

src/centralized/order.ts

@@ -18,6 +18,7 @@ import { createOrder$ } from './ccxt/create-order';
 import { ExecuteCEXorderParams } from './execute-order';
 import { quantityAboveMinimum } from './minimum-order-quantity-filter';
 import { CEXorder, GetOrderBuilderParams } from './order-builder';
+import { ArbyStore } from 'src/store';
 
 type GetCentralizedExchangeOrderParams = {
   CEX: Exchange;
@@ -41,6 +42,7 @@ type GetCentralizedExchangeOrderParams = {
     price: BigNumber,
     config: Config
   ) => CEXorder;
+  store: ArbyStore;
 };
 
 const getCentralizedExchangeOrder$ = ({
@@ -51,6 +53,7 @@ const getCentralizedExchangeOrder$ = ({
   getOrderBuilder$,
   centralizedExchangePrice$,
   deriveCEXorderQuantity,
+  store,
 }: GetCentralizedExchangeOrderParams): Observable<null> => {
   return getOrderBuilder$({
     config,
@@ -59,6 +62,7 @@ const getCentralizedExchangeOrder$ = ({
     accumulateOrderFillsForBaseAssetReceived,
     accumulateOrderFillsForQuoteAssetReceived,
     quantityAboveMinimum,
+    store,
   }).pipe(
     withLatestFrom(
       centralizedExchangePrice$.pipe(

src/opendex/catch-error.spec.ts

@@ -1,10 +1,10 @@
+import { status } from '@grpc/grpc-js';
+import { AuthenticationError, Exchange } from 'ccxt';
 import { TestScheduler } from 'rxjs/testing';
 import { errors } from '../opendex/errors';
-import { getLoggers, TestError, testConfig } from '../test-utils';
+import { ArbyStore, getArbyStore } from '../store';
+import { getLoggers, testConfig, TestError } from '../test-utils';
 import { catchOpenDEXerror } from './catch-error';
-import { status } from '@grpc/grpc-js';
-import { AuthenticationError, Exchange } from 'ccxt';
-
 let testScheduler: TestScheduler;
 const testSchedulerSetup = () => {
   testScheduler = new TestScheduler((actual, expected) => {
@@ -22,6 +22,7 @@ type AssertCatchOpenDEXerrorParams = {
   };
   expectedError?: TestError;
   unsubscribe: string;
+  store?: ArbyStore;
 };
 
 const assertCatchOpenDEXerror = ({
@@ -31,6 +32,7 @@ const assertCatchOpenDEXerror = ({
   inputError,
   expectedError,
   unsubscribe,
+  store,
 }: AssertCatchOpenDEXerrorParams) => {
   testScheduler.run(helpers => {
     const { cold, expectObservable, expectSubscriptions } = helpers;
@@ -43,7 +45,8 @@ const assertCatchOpenDEXerror = ({
       getLoggers(),
       config,
       getCleanup$,
-      CEX
+      CEX,
+      store ? store : getArbyStore()
     )(input$);
     expectObservable(output$, unsubscribe).toBe(
       expected,
@@ -186,23 +189,30 @@ describe('catchOpenDEXerror', () => {
     });
   });
 
-  it('cancels orders, retries CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR', () => {
-    expect.assertions(ASSERTIONS_PER_TEST);
+  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('retries recoverable gRPC errors', () => {

src/opendex/catch-error.ts

@@ -2,6 +2,7 @@ import { status } from '@grpc/grpc-js';
 import { AuthenticationError, Exchange } from 'ccxt';
 import { concat, Observable, throwError, timer } from 'rxjs';
 import { ignoreElements, mergeMap, retryWhen } from 'rxjs/operators';
+import { ArbyStore } from 'src/store';
 import { removeCEXorders$ } from '../centralized/remove-orders';
 import { Config } from '../config';
 import { MAX_RETRY_ATTEMPS, RETRY_INTERVAL } from '../constants';
@@ -19,7 +20,8 @@ const catchOpenDEXerror = (
     removeOpenDEXorders$,
     removeCEXorders$,
   }: GetCleanupParams) => Observable<unknown>,
-  CEX: Exchange
+  CEX: Exchange,
+  store: ArbyStore
 ) => {
   return (source: Observable<any>) => {
     return source.pipe(
@@ -63,6 +65,7 @@ const catchOpenDEXerror = (
               e.code === errorCodes.CENTRALIZED_EXCHANGE_PRICE_FEED_ERROR
             ) {
               logMessage(loggers.centralized);
+              store.resetLastOrderUpdatePrice();
               return concat(
                 getCleanup$({
                   config,

src/opendex/complete.spec.ts

@@ -2,6 +2,7 @@ import BigNumber from 'bignumber.js';
 import { Exchange } from 'ccxt';
 import { Observable } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
+import { getArbyStore } from '../store';
 import { getLoggers, testConfig } from '../test-utils';
 import { TradeInfo } from '../trade/info';
 import { getOpenDEXcomplete$ } from './complete';
@@ -39,13 +40,15 @@ const assertGetOpenDEXcomplete = (
       BigNumber
     >;
     const CEX = (null as unknown) as Exchange;
+    const store = getArbyStore();
     const trade$ = getOpenDEXcomplete$({
       CEX,
       config: testConfig(),
       loggers: getLoggers(),
       tradeInfo$: getTradeInfo$,
       createOpenDEXorders$,
       centralizedExchangePrice$,
+      store,
     });
     expectObservable(trade$).toBe(expected, { a: true });
   });

src/opendex/complete.ts

@@ -1,10 +1,11 @@
 import { BigNumber } from 'bignumber.js';
 import { Exchange } from 'ccxt';
-import { BehaviorSubject, empty, Observable, of } from 'rxjs';
+import { empty, Observable, of } from 'rxjs';
 import { exhaustMap, mergeMap, take } from 'rxjs/operators';
 import { getCentralizedExchangeAssets$ } from '../centralized/assets';
 import { Config } from '../config';
 import { Loggers } from '../logger';
+import { ArbyStore } from '../store';
 import {
   GetTradeInfoParams,
   TradeInfo,
@@ -39,6 +40,7 @@ type GetOpenDEXcompleteParams = {
     createXudOrder$,
   }: CreateOpenDEXordersParams) => Observable<boolean>;
   centralizedExchangePrice$: Observable<BigNumber>;
+  store: ArbyStore;
 };
 
 const getOpenDEXcomplete$ = ({
@@ -48,6 +50,7 @@ const getOpenDEXcomplete$ = ({
   tradeInfo$,
   createOpenDEXorders$,
   centralizedExchangePrice$,
+  store,
 }: GetOpenDEXcompleteParams): Observable<boolean> => {
   const openDEXassetsWithConfig = (config: Config) => {
     return getOpenDEXassets$({
@@ -60,7 +63,6 @@ const getOpenDEXcomplete$ = ({
       xudTradingLimits$: getXudTradingLimits$,
     });
   };
-  const lastPriceUpdateStore = new BehaviorSubject(new BigNumber('0'));
   return tradeInfo$({
     config,
     loggers,
@@ -74,7 +76,7 @@ const getOpenDEXcomplete$ = ({
     // is already in progress
     exhaustMap((tradeInfo: TradeInfo) => {
       const getTradeInfo = () => tradeInfo;
-      return lastPriceUpdateStore.pipe(
+      return store.selectState('lastOrderUpdatePrice').pipe(
         take(1),
         mergeMap((lastPriceUpdate: BigNumber) => {
           if (shouldCreateOpenDEXorders(tradeInfo.price, lastPriceUpdate)) {
@@ -89,7 +91,7 @@ const getOpenDEXcomplete$ = ({
             }).pipe(
               mergeMap(() => {
                 // store the last price update
-                lastPriceUpdateStore.next(tradeInfo.price);
+                store.updateLastOrderUpdatePrice(tradeInfo.price);
                 return of(true);
               })
             );