wip: dynamic minimum quantity

Author: Karl Ranna

src/arby.ts

@@ -102,6 +102,7 @@ export const startArby = ({
           loggers.global.info('Starting. Hello, Arby.');
           logConfig(config, loggers.global);
           verifyMarkets(config, CEXmarkets);
+          store.setMarkets(CEXmarkets);
           const tradeComplete$ = trade$({
             config,
             loggers,

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

@@ -10,6 +10,7 @@ const MINIMUM_ORDER_SIZE: MinimumCEXquantities = {
   ETH: new BigNumber('0.05'),
   DAI: new BigNumber('15'),
   USDT: new BigNumber('15'),
+  USD: new BigNumber('15'),
 };
 
 const getMinimumOrderSize = (asset: string): BigNumber => {

src/centralized/order-builder.ts

@@ -1,6 +1,7 @@
 import BigNumber from 'bignumber.js';
-import { merge, Observable, of } from 'rxjs';
-import { filter, map, mergeMap, repeat, take } from 'rxjs/operators';
+import { curry } from 'ramda';
+import { empty, merge, Observable, of } from 'rxjs';
+import { map, mergeMap, repeat, take } from 'rxjs/operators';
 import { Config } from '../config';
 import { OrderSide } from '../constants';
 import { Logger } from '../logger';
@@ -27,9 +28,7 @@ type GetOrderBuilderParams = {
   accumulateOrderFillsForQuoteAssetReceived: (
     config: Config
   ) => (source: Observable<SwapSuccess>) => Observable<BigNumber>;
-  quantityAboveMinimum: (
-    asset: string
-  ) => (filledQuantity: BigNumber) => boolean;
+  minimumQuantity$: Observable<BigNumber>;
   store: ArbyStore;
 };
 
@@ -38,13 +37,40 @@ type CEXorder = {
   side: OrderSide;
 };
 
+const quantityAboveMinimum = curry(
+  (
+    store: ArbyStore,
+    logger: Logger,
+    assetToTradeOnCEX: string,
+    minimumQuantity$: Observable<BigNumber>,
+    quantity: BigNumber
+  ) => {
+    logger.info(
+      `Swap success. Accumulated ${assetToTradeOnCEX} quantity: ${quantity.toFixed()}`
+    );
+    store.resetLastOrderUpdatePrice();
+    return minimumQuantity$.pipe(
+      take(1),
+      mergeMap(minimumQuantity => {
+        if (quantity.isGreaterThanOrEqualTo(minimumQuantity)) {
+          return of(quantity);
+        }
+        logger.info(
+          `Will not execute CEX order because ${quantity.toFixed()} is below the minimum allowed CEX quantity ${minimumQuantity.toFixed()}`
+        );
+        return empty();
+      })
+    );
+  }
+);
+
 const getOrderBuilder$ = ({
   config,
   logger,
   getOpenDEXswapSuccess$,
   accumulateOrderFillsForBaseAssetReceived,
   accumulateOrderFillsForQuoteAssetReceived,
-  quantityAboveMinimum,
+  minimumQuantity$,
   store,
 }: GetOrderBuilderParams): Observable<CEXorder> => {
   const {
@@ -59,19 +85,17 @@ const getOrderBuilder$ = ({
     config.CEX_QUOTEASSET === 'BTC'
       ? config.CEX_BASEASSET
       : config.CEX_QUOTEASSET;
+  const filterMinimum = quantityAboveMinimum(
+    store,
+    logger,
+    assetToTradeOnCEX,
+    minimumQuantity$
+  );
   const receivedQuoteAssetOrder$ = receivedQuoteAssetSwapSuccess$.pipe(
     // accumulate OpenDEX order fills when receiving
     // quote asset
     accumulateOrderFillsForQuoteAssetReceived(config),
-    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)),
+    mergeMap(filterMinimum),
     map(quantity => {
       return { quantity, side: OrderSide.BUY };
     }),
@@ -84,15 +108,7 @@ const getOrderBuilder$ = ({
     // accumulate OpenDEX order fills when receiving
     // quote asset
     accumulateOrderFillsForBaseAssetReceived(config),
-    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)),
+    mergeMap(filterMinimum),
     map(quantity => {
       return { quantity, side: OrderSide.SELL };
     }),

src/centralized/order.ts

@@ -1,12 +1,14 @@
 import BigNumber from 'bignumber.js';
-import { Exchange } from 'ccxt';
-import { Observable, timer } from 'rxjs';
+import { Dictionary, Exchange, Market } from 'ccxt';
+import { combineLatest, Observable, timer } from 'rxjs';
 import {
   catchError,
   mergeMap,
   mergeMapTo,
   withLatestFrom,
+  map,
 } from 'rxjs/operators';
+import { ArbyStore } from 'src/store';
 import { Config } from '../config';
 import { Logger } from '../logger';
 import { getOpenDEXswapSuccess$ } from '../opendex/swap-success';
@@ -16,9 +18,7 @@ import {
 } from '../trade/accumulate-fills';
 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;
@@ -34,7 +34,7 @@ type GetCentralizedExchangeOrderParams = {
     getOpenDEXswapSuccess$,
     accumulateOrderFillsForBaseAssetReceived,
     accumulateOrderFillsForQuoteAssetReceived,
-    quantityAboveMinimum,
+    minimumQuantity$,
   }: GetOrderBuilderParams) => Observable<CEXorder>;
   centralizedExchangePrice$: Observable<BigNumber>;
   deriveCEXorderQuantity: (
@@ -55,13 +55,28 @@ const getCentralizedExchangeOrder$ = ({
   deriveCEXorderQuantity,
   store,
 }: GetCentralizedExchangeOrderParams): Observable<null> => {
+  const minimumBaseAssetQuantity$ = (store.selectState(
+    'markets'
+  ) as Observable<Dictionary<Market>>).pipe(
+    map(markets => {
+      const tradingPair = `${config.CEX_BASEASSET}/${config.CEX_QUOTEASSET}`;
+      const market = markets[tradingPair];
+      return new BigNumber(market.limits.amount.min);
+    })
+  );
+  const minimumQuantity$ = combineLatest([minimumBaseAssetQuantity$, centralizedExchangePrice$]).pipe(
+    map(([minimumQuantity, price]) => {
+      // TODO: only if BTC is baseAsset
+      return minimumQuantity.multipliedBy(price);
+    }),
+  );
   return getOrderBuilder$({
     config,
     logger,
     getOpenDEXswapSuccess$,
     accumulateOrderFillsForBaseAssetReceived,
     accumulateOrderFillsForQuoteAssetReceived,
-    quantityAboveMinimum,
+    minimumQuantity$,
     store,
   }).pipe(
     withLatestFrom(

src/store.spec.ts

@@ -1,4 +1,5 @@
 import BigNumber from 'bignumber.js';
+import { Dictionary, Market } from 'ccxt';
 import { getArbyStore } from './store';
 
 describe('ArbyStore', () => {
@@ -18,6 +19,24 @@ describe('ArbyStore', () => {
     });
   });
 
+  it('selectState returns empty initial markets', done => {
+    const { selectState } = getArbyStore();
+    selectState('markets').subscribe(markets => {
+      expect(markets).toEqual({});
+      done();
+    });
+  });
+
+  it('selectState returns updated markets', done => {
+    const { selectState, setMarkets } = getArbyStore();
+    const testMarkets = ({ 'BTC/USDT': true } as unknown) as Dictionary<Market>;
+    setMarkets(testMarkets);
+    selectState('markets').subscribe(markets => {
+      expect(markets).toEqual(testMarkets);
+      done();
+    });
+  });
+
   it('selectState returns updated last sell order price', done => {
     const { selectState, updateLastSellOrderUpdatePrice } = getArbyStore();
     const updatedPrice = new BigNumber('123');

src/store.ts

@@ -1,18 +1,21 @@
 import BigNumber from 'bignumber.js';
 import { BehaviorSubject, Subject, Observable } from 'rxjs';
 import { scan, pluck, distinctUntilKeyChanged } from 'rxjs/operators';
+import { Dictionary, Market } from 'ccxt';
 
 type ArbyStore = {
   updateLastSellOrderUpdatePrice: (price: BigNumber) => void;
   updateLastBuyOrderUpdatePrice: (price: BigNumber) => void;
   resetLastOrderUpdatePrice: () => void;
-  selectState: (stateKey: ArbyStoreDataKeys) => Observable<BigNumber>;
+  setMarkets: (markets: Dictionary<Market>) => void;
+  selectState: (stateKey: ArbyStoreDataKeys) => Observable<BigNumber | Dictionary<Market>>;
   stateChanges: () => Observable<ArbyStoreData>;
 };
 
 type ArbyStoreData = {
   lastSellOrderUpdatePrice: BigNumber;
   lastBuyOrderUpdatePrice: BigNumber;
+  markets: Dictionary<Market>;
 };
 
 type ArbyStoreDataKeys = keyof ArbyStoreData;
@@ -21,6 +24,7 @@ const getArbyStore = (): ArbyStore => {
   const initialState: ArbyStoreData = {
     lastSellOrderUpdatePrice: new BigNumber('0'),
     lastBuyOrderUpdatePrice: new BigNumber('0'),
+    markets: {},
   };
   const store = new BehaviorSubject(initialState);
   const stateUpdates = new Subject() as Subject<Partial<ArbyStoreData>>;
@@ -47,6 +51,11 @@ const getArbyStore = (): ArbyStore => {
       lastBuyOrderUpdatePrice: new BigNumber('0'),
     });
   };
+  const setMarkets = (markets: Dictionary<Market>) => {
+    stateUpdates.next({
+      markets,
+    });
+  };
   const selectState = (stateKey: ArbyStoreDataKeys) => {
     return store.pipe(distinctUntilKeyChanged(stateKey), pluck(stateKey));
   };
@@ -59,6 +68,7 @@ const getArbyStore = (): ArbyStore => {
     selectState,
     resetLastOrderUpdatePrice,
     stateChanges,
+    setMarkets,
   };
 };