feat: adding sequelize along with new db type 'Order' (#85)

.env-example

@@ -13,4 +13,4 @@ CEX_BASEASSET=BTC
 CEX_QUOTEASSET=USDT
 TEST_CENTRALIZED_EXCHANGE_BASEASSET_BALANCE=10
 TEST_CENTRALIZED_EXCHANGE_QUOTEASSET_BALANCE=100000
-LIVE_CEX=false
+TEST_MODE=true

package.json

@@ -59,6 +59,8 @@
     "moment": "2.26.0",
     "ramda": "0.27.0",
     "rxjs": "6.5.5",
+    "sequelize": "^6.3.5",
+    "sqlite3": "^5.0.0",
     "uuid": "8.1.0",
     "winston": "3.2.1",
     "ws": "7.3.0"

src/arby.spec.ts

@@ -4,6 +4,7 @@ import { startArby } from '../src/arby';
 import { Config } from '../src/config';
 import { InitCEXResponse } from './centralized/ccxt/init';
 import { getLoggers } from './test-utils';
+import { InitDBResponse } from './db/db';
 
 let testScheduler: TestScheduler;
 
@@ -15,6 +16,7 @@ type AssertStartArbyParams = {
     shutdown$: string;
     cleanup$: string;
     initCEX$: string;
+    initDB$: string;
   };
   verifyMarkets?: () => boolean;
 };
@@ -39,13 +41,19 @@ const assertStartArby = ({
         InitCEXResponse
       >;
     };
+    const initDB$ = () => {
+      return (cold(inputEvents.initDB$) as unknown) as Observable<
+        InitDBResponse
+      >;
+    };
     const arby$ = startArby({
       config$,
       getLoggers,
       shutdown$,
       trade$: getTrade$,
       cleanup$,
       initCEX$,
+      initDB$,
       verifyMarkets: verifyMarkets ? verifyMarkets : () => true,
     });
     expectObservable(arby$).toBe(expected, undefined, { message: 'error' });
@@ -63,11 +71,12 @@ describe('startArby', () => {
     const inputEvents = {
       config$: '1000ms a',
       initCEX$: '1s a',
+      initDB$: '1s a',
       getTrade$: 'b',
       shutdown$: '',
       cleanup$: '',
     };
-    const expected = '2s b';
+    const expected = '3s b';
     assertStartArby({
       inputEvents,
       expected,
@@ -78,11 +87,12 @@ describe('startArby', () => {
     const inputEvents = {
       config$: '1000ms a',
       initCEX$: '1s a',
+      initDB$: '1s a',
       getTrade$: 'b',
       shutdown$: '',
       cleanup$: '',
     };
-    const expected = '2s #';
+    const expected = '3s #';
     assertStartArby({
       inputEvents,
       expected,
@@ -96,11 +106,12 @@ describe('startArby', () => {
     const inputEvents = {
       config$: 'a',
       initCEX$: '1s a',
+      initDB$: '1s a',
       getTrade$: '500ms b',
       shutdown$: '10s c',
       cleanup$: '2s a',
     };
-    const expected = '1500ms b 11499ms a';
+    const expected = '2500ms b 11499ms a';
     assertStartArby({
       inputEvents,
       expected,
@@ -111,11 +122,12 @@ describe('startArby', () => {
     const inputEvents = {
       config$: 'a',
       initCEX$: '1s a',
+      initDB$: '1s a',
       getTrade$: '500ms #',
       shutdown$: '10s c',
       cleanup$: '2s a',
     };
-    const expected = '3500ms a';
+    const expected = '4500ms a';
     assertStartArby({
       inputEvents,
       expected,

src/arby.ts

@@ -21,6 +21,8 @@ import { getNewTrade$, GetTradeParams } from './trade/trade';
 import { getStartShutdown$ } from './utils';
 import { Dictionary, Market } from 'ccxt';
 import { verifyMarkets } from './centralized/verify-markets';
+import { initDB$, InitDBparams, InitDBResponse, closeDB$ } from './db/db';
+import { OrderRepository } from './db/order-repository';
 
 type StartArbyParams = {
   config$: Observable<Config>;
@@ -43,6 +45,10 @@ type StartArbyParams = {
     loadMarkets$,
   }: InitCEXparams) => Observable<InitCEXResponse>;
   verifyMarkets: (config: Config, CEXmarkets: Dictionary<Market>) => boolean;
+  initDB$: ({
+    dataDir: string,
+    logger: Logger,
+  }: InitDBparams) => Observable<InitDBResponse>;
 };
 
 const logConfig = (config: Config, logger: Logger) => {
@@ -86,54 +92,67 @@ export const startArby = ({
   trade$,
   cleanup$,
   initCEX$,
+  initDB$,
   verifyMarkets,
 }: StartArbyParams): Observable<any> => {
   const store = getArbyStore();
   return config$.pipe(
     mergeMap(config => {
-      const CEX$ = initCEX$({
-        config,
-        loadMarkets$,
-        getExchange,
+      const loggers = getLoggers(config);
+      const db$ = initDB$({
+        logger: loggers.db,
+        dataDir: config.DATA_DIR,
       });
-      return CEX$.pipe(
-        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$({
+      return db$.pipe(
+        mergeMap((models: InitDBResponse) => {
+          const CEX$ = initCEX$({
             config,
-            loggers,
-            getOpenDEXcomplete$,
-            shutdown$,
-            getCentralizedExchangeOrder$,
-            catchOpenDEXerror,
-            getCentralizedExchangePrice$,
-            CEX,
-            store,
-          }).pipe(takeUntil(shutdown$));
-          return concat(
-            tradeComplete$,
-            cleanup$({
-              config,
-              loggers,
-              removeOpenDEXorders$,
-              removeCEXorders$,
-              CEX,
-            })
-          ).pipe(
-            catchError(e => {
-              loggers.global.info(
-                `Unrecoverable error: ${JSON.stringify(e)} - cleaning up.`
-              );
-              return cleanup$({
+            loadMarkets$,
+            getExchange,
+          });
+          return CEX$.pipe(
+            mergeMap(({ markets: CEXmarkets, exchange: CEX }) => {
+              loggers.global.info('Starting. Hello, Arby.');
+              logConfig(config, loggers.global);
+              verifyMarkets(config, CEXmarkets);
+              const orderRepository = new OrderRepository(models, loggers.db);
+              const tradeComplete$ = trade$({
                 config,
                 loggers,
-                removeOpenDEXorders$,
-                removeCEXorders$,
+                getOpenDEXcomplete$,
+                shutdown$,
+                getCentralizedExchangeOrder$,
+                catchOpenDEXerror,
+                getCentralizedExchangePrice$,
                 CEX,
-              });
+                store,
+                saveOrder$: orderRepository.saveOrder$,
+              }).pipe(takeUntil(shutdown$));
+              return concat(
+                tradeComplete$,
+                cleanup$({
+                  config,
+                  loggers,
+                  removeOpenDEXorders$,
+                  removeCEXorders$,
+                  closeDB$,
+                  CEX,
+                })
+              ).pipe(
+                catchError(e => {
+                  loggers.global.info(
+                    `Unrecoverable error: ${JSON.stringify(e)} - cleaning up.`
+                  );
+                  return cleanup$({
+                    config,
+                    loggers,
+                    removeOpenDEXorders$,
+                    removeCEXorders$,
+                    CEX,
+                    closeDB$,
+                  });
+                })
+              );
             })
           );
         })
@@ -155,6 +174,7 @@ if (!module.parent) {
     cleanup$: getCleanup$,
     initCEX$,
     verifyMarkets,
+    initDB$,
   }).subscribe({
     error: error => {
       if (error.message) {

src/centralized/execute-order.spec.ts

@@ -7,6 +7,7 @@ import { OrderSide } from '../constants';
 import { Config } from '../config';
 import { Observable } from 'rxjs';
 import { Order, Exchange } from 'ccxt';
+import { OrderInstance } from '../db/order';
 
 let testScheduler: TestScheduler;
 
@@ -25,6 +26,9 @@ const assertExecuteCEXorder = (
     const createOrder$ = () => {
       return (cold(inputEvents.createOrder$) as unknown) as Observable<Order>;
     };
+    const saveOrder$ = () => {
+      return (cold('') as unknown) as Observable<OrderInstance>;
+    };
     const CEX = (null as unknown) as Exchange;
     const CEXorder$ = executeCEXorder$({
       CEX,
@@ -33,6 +37,7 @@ const assertExecuteCEXorder = (
       price: inputEvents.price,
       order: inputEvents.order,
       createOrder$,
+      saveOrder$,
     });
     expectObservable(CEXorder$, inputEvents.unsubscribe).toBe(expected, {
       a: null,

src/centralized/execute-order.ts

@@ -13,13 +13,16 @@ import { Config } from '../config';
 import { Logger } from '../logger';
 import { CreateOrderParams } from './ccxt/create-order';
 import { CEXorder } from './order-builder';
+import { SaveOrderParams } from '../db/order-repository';
+import { OrderInstance } from '../db/order';
 
 type ExecuteCEXorderParams = {
   CEX: Exchange;
   config: Config;
   logger: Logger;
   price: BigNumber;
   order: CEXorder;
+  saveOrder$: ({ order }: SaveOrderParams) => Observable<OrderInstance>;
   createOrder$: ({
     config,
     exchange,
@@ -35,6 +38,7 @@ const executeCEXorder$ = ({
   price,
   order,
   createOrder$,
+  saveOrder$,
 }: ExecuteCEXorderParams): Observable<null> => {
   if (!config.TEST_MODE) {
     logger.info(
@@ -46,11 +50,12 @@ const executeCEXorder$ = ({
       side: order.side,
       quantity: order.quantity,
     }).pipe(
-      tap(order =>
+      tap(order => {
         logger.info(
           `Centralized exchange order finished: ${JSON.stringify(order)}`
-        )
-      ),
+        );
+        saveOrder$({ order });
+      }),
       catchError((e, caught) => {
         logger.warn(`Failed to execute CEX order: ${e}. Retrying in 1000ms`);
         return timer(1000).pipe(mergeMapTo(caught));

src/centralized/order.spec.ts

@@ -6,6 +6,7 @@ import { CEXorder } from './order-builder';
 import BigNumber from 'bignumber.js';
 import { Exchange } from 'ccxt';
 import { getArbyStore } from '../store';
+import { OrderInstance } from '../db/order';
 
 let testScheduler: TestScheduler;
 
@@ -34,6 +35,9 @@ const assertCentralizedExchangeOrder = (
     const centralizedExchangePrice$ = (cold(
       inputEvents.centralizedExchangePrice$
     ) as unknown) as Observable<BigNumber>;
+    const saveOrder$ = () => {
+      return (cold('') as unknown) as Observable<OrderInstance>;
+    };
     const CEX = (null as unknown) as Exchange;
     const deriveCEXorderQuantity = (order: any) => order;
     const store = getArbyStore();
@@ -46,6 +50,7 @@ const assertCentralizedExchangeOrder = (
       centralizedExchangePrice$,
       deriveCEXorderQuantity,
       store,
+      saveOrder$,
     });
     expectObservable(centralizedExchangeOrder$, inputEvents.unsubscribe).toBe(
       expected

src/centralized/order.ts

@@ -19,6 +19,8 @@ import { ExecuteCEXorderParams } from './execute-order';
 import { quantityAboveMinimum } from './minimum-order-quantity-filter';
 import { CEXorder, GetOrderBuilderParams } from './order-builder';
 import { ArbyStore } from 'src/store';
+import { SaveOrderParams } from '../db/order-repository';
+import { OrderInstance } from '../db/order';
 
 type GetCentralizedExchangeOrderParams = {
   CEX: Exchange;
@@ -28,6 +30,7 @@ type GetCentralizedExchangeOrderParams = {
     logger,
     price,
     order,
+    saveOrder$,
   }: ExecuteCEXorderParams) => Observable<null>;
   getOrderBuilder$: ({
     config,
@@ -43,12 +46,14 @@ type GetCentralizedExchangeOrderParams = {
     config: Config
   ) => CEXorder;
   store: ArbyStore;
+  saveOrder$: ({ order }: SaveOrderParams) => Observable<OrderInstance>;
 };
 
 const getCentralizedExchangeOrder$ = ({
   CEX,
   logger,
   config,
+  saveOrder$,
   executeCEXorder$,
   getOrderBuilder$,
   centralizedExchangePrice$,
@@ -79,6 +84,7 @@ const getCentralizedExchangeOrder$ = ({
         logger,
         price,
         order: deriveCEXorderQuantity(order, price, config),
+        saveOrder$,
       });
     })
   );