feat: throttle order update

src/opendex/complete.spec.ts

@@ -13,38 +13,82 @@ const testSchedulerSetup = () => {
   });
 };
 
+const assertGetOpenDEXcomplete = (
+  inputEvents: {
+    tradeInfo$: string;
+    getOpenDEXorders$: string;
+  },
+  expected: string,
+  inputValues: {
+    tradeInfo$: any;
+  }
+) => {
+  testScheduler.run(helpers => {
+    const { cold, hot, expectObservable } = helpers;
+    const createOpenDEXorders$ = () => {
+      return cold(inputEvents.getOpenDEXorders$, {
+        a: true,
+      });
+    };
+    const getTradeInfo$ = () => {
+      return hot(inputEvents.tradeInfo$, inputValues.tradeInfo$) as Observable<
+        TradeInfo
+      >;
+    };
+    const centralizedExchangePrice$ = (cold('') as unknown) as Observable<
+      BigNumber
+    >;
+    const CEX = (null as unknown) as Exchange;
+    const trade$ = getOpenDEXcomplete$({
+      CEX,
+      config: testConfig(),
+      loggers: getLoggers(),
+      tradeInfo$: getTradeInfo$,
+      createOpenDEXorders$,
+      centralizedExchangePrice$,
+    });
+    expectObservable(trade$).toBe(expected, { a: true });
+  });
+};
+
 describe('getOpenDEXcomplete$', () => {
   beforeEach(testSchedulerSetup);
 
   test('when trade info present it creates new OpenDEX orders', () => {
-    testScheduler.run(helpers => {
-      const { cold, hot, expectObservable } = helpers;
-      const inputEvents = {
-        tradeInfo$: 'a ^ 1000ms a 1500ms a 500ms b',
-        getOpenDEXorders$: '1s a|',
-      };
-      const expected = '2s a 1500ms a';
-      const createOpenDEXorders$ = () => {
-        return cold(inputEvents.getOpenDEXorders$, {
-          a: true,
-        });
-      };
-      const getTradeInfo$ = () => {
-        return hot(inputEvents.tradeInfo$) as Observable<TradeInfo>;
-      };
-      const centralizedExchangePrice$ = (cold('') as unknown) as Observable<
-        BigNumber
-      >;
-      const CEX = (null as unknown) as Exchange;
-      const trade$ = getOpenDEXcomplete$({
-        CEX,
-        config: testConfig(),
-        loggers: getLoggers(),
-        tradeInfo$: getTradeInfo$,
-        createOpenDEXorders$,
-        centralizedExchangePrice$,
-      });
-      expectObservable(trade$).toBe(expected, { a: true });
-    });
+    const inputEvents = {
+      tradeInfo$: 'a ^ 1000ms a 1500ms b 500ms b',
+      getOpenDEXorders$: '1s a|',
+    };
+    const expected = '2s a 1500ms a';
+    const inputValues = {
+      tradeInfo$: {
+        a: {
+          price: new BigNumber('10000'),
+        },
+        b: {
+          price: new BigNumber('10010.1'),
+        },
+      },
+    };
+    assertGetOpenDEXcomplete(inputEvents, expected, inputValues);
+  });
+
+  test('filters by shouldCreateOpenDEXorders', () => {
+    const inputEvents = {
+      tradeInfo$: 'a ^ 1000ms a 1500ms a 500ms b',
+      getOpenDEXorders$: '1s a|',
+    };
+    const inputValues = {
+      tradeInfo$: {
+        a: {
+          price: new BigNumber('10000'),
+        },
+        b: {
+          price: new BigNumber('10010.1'),
+        },
+      },
+    };
+    const expected = '2s a 2001ms a';
+    assertGetOpenDEXcomplete(inputEvents, expected, inputValues);
   });
 });

src/opendex/complete.ts

@@ -1,7 +1,7 @@
 import { BigNumber } from 'bignumber.js';
 import { Exchange } from 'ccxt';
-import { Observable } from 'rxjs';
-import { exhaustMap } from 'rxjs/operators';
+import { BehaviorSubject, 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';
@@ -14,6 +14,7 @@ import { getOpenDEXassets$ } from './assets';
 import { logAssetBalance, parseOpenDEXassets } from './assets-utils';
 import { CreateOpenDEXordersParams } from './create-orders';
 import { tradeInfoToOpenDEXorders } from './orders';
+import { shouldCreateOpenDEXorders } from './should-create-orders';
 import { getXudBalance$ } from './xud/balance';
 import { getXudClient$ } from './xud/client';
 import { createXudOrder$ } from './xud/create-order';
@@ -59,6 +60,7 @@ const getOpenDEXcomplete$ = ({
       xudTradingLimits$: getXudTradingLimits$,
     });
   };
+  const lastPriceUpdateStore = new BehaviorSubject(new BigNumber('0'));
   return tradeInfo$({
     config,
     loggers,
@@ -71,18 +73,31 @@ const getOpenDEXcomplete$ = ({
     // ignore new trade information when creating orders
     // is already in progress
     exhaustMap((tradeInfo: TradeInfo) => {
-      const getTradeInfo = () => {
-        return tradeInfo;
-      };
-      // create orders based on latest trade info
-      return createOpenDEXorders$({
-        config,
-        logger: loggers.opendex,
-        getTradeInfo,
-        getXudClient$,
-        createXudOrder$,
-        tradeInfoToOpenDEXorders,
-      });
+      const getTradeInfo = () => tradeInfo;
+      return lastPriceUpdateStore.pipe(
+        take(1),
+        mergeMap((lastPriceUpdate: BigNumber) => {
+          if (shouldCreateOpenDEXorders(tradeInfo.price, lastPriceUpdate)) {
+            // create orders based on latest trade info
+            return createOpenDEXorders$({
+              config,
+              logger: loggers.opendex,
+              getTradeInfo,
+              getXudClient$,
+              createXudOrder$,
+              tradeInfoToOpenDEXorders,
+            }).pipe(
+              mergeMap(() => {
+                // store the last price update
+                lastPriceUpdateStore.next(tradeInfo.price);
+                return of(true);
+              })
+            );
+          }
+          // do nothing in case orders do not need updating
+          return empty();
+        })
+      );
     })
   );
 };

src/opendex/should-create-orders.spec.ts

@@ -0,0 +1,28 @@
+import { shouldCreateOpenDEXorders } from './should-create-orders';
+import BigNumber from 'bignumber.js';
+
+describe('shouldCreateOpenDEXorders', () => {
+  it('returns true when price increases by 0.1%', () => {
+    const newPrice = new BigNumber('10000');
+    const lastPriceUpdate = new BigNumber('10010.1');
+    expect(shouldCreateOpenDEXorders(newPrice, lastPriceUpdate)).toEqual(true);
+  });
+
+  it('returns true when price decreases by 0.1%', () => {
+    const newPrice = new BigNumber('10000');
+    const lastPriceUpdate = new BigNumber('9989.9');
+    expect(shouldCreateOpenDEXorders(newPrice, lastPriceUpdate)).toEqual(true);
+  });
+
+  it('returns false when price increases less than 0.1%', () => {
+    const newPrice = new BigNumber('10000');
+    const lastPriceUpdate = new BigNumber('10009.9');
+    expect(shouldCreateOpenDEXorders(newPrice, lastPriceUpdate)).toEqual(false);
+  });
+
+  it('returns false when price decreases less than 0.1%', () => {
+    const newPrice = new BigNumber('10000');
+    const lastPriceUpdate = new BigNumber('9990.1');
+    expect(shouldCreateOpenDEXorders(newPrice, lastPriceUpdate)).toEqual(false);
+  });
+});

src/opendex/should-create-orders.ts

@@ -0,0 +1,17 @@
+import BigNumber from 'bignumber.js';
+
+// decide whether to update orders based on
+// last price update and new price
+const shouldCreateOpenDEXorders = (
+  newPrice: BigNumber,
+  lastPriceUpdate: BigNumber
+): boolean => {
+  const priceDiff = lastPriceUpdate.minus(newPrice).absoluteValue();
+  const maxPriceDiff = lastPriceUpdate.multipliedBy(new BigNumber('0.001'));
+  if (priceDiff.isGreaterThanOrEqualTo(maxPriceDiff)) {
+    return true;
+  }
+  return false;
+};
+
+export { shouldCreateOpenDEXorders };