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

Author: rsercano

.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-lock.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"

package.json

@@ -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.spec.ts

@@ -21,6 +21,7 @@ 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 } from './db/db';
 
 type StartArbyParams = {
   config$: Observable<Config>;
@@ -43,6 +44,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 +91,63 @@ 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(() => {
+          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 tradeComplete$ = trade$({
                 config,
                 loggers,
-                removeOpenDEXorders$,
-                removeCEXorders$,
+                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$({
+                    config,
+                    loggers,
+                    removeOpenDEXorders$,
+                    removeCEXorders$,
+                    CEX,
+                  });
+                })
+              );
             })
           );
         })
@@ -155,6 +169,7 @@ if (!module.parent) {
     cleanup$: getCleanup$,
     initCEX$,
     verifyMarkets,
+    initDB$,
   }).subscribe({
     error: error => {
       if (error.message) {

src/arby.ts

@@ -0,0 +1,79 @@
+import { from, Observable, of } from 'rxjs';
+import { Sequelize, ModelCtor } from 'sequelize';
+import { Order, OrderInstance } from './order';
+import { Fee, FeeInstance } from './fee';
+import { Trade, TradeInstance } from './trade';
+import { Logger } from '../logger';
+import { mergeMap } from 'rxjs/operators';
+
+type InitDBparams = {
+  dataDir?: string;
+  logger: Logger;
+};
+
+type InitDBResponse = {
+  Order: ModelCtor<OrderInstance>;
+  Fee: ModelCtor<FeeInstance>;
+  Trade: ModelCtor<TradeInstance>;
+};
+
+const createModels = (sequelize: Sequelize): InitDBResponse => {
+  const models = {
+    Order: Order(sequelize),
+    Fee: Fee(sequelize),
+    Trade: Trade(sequelize),
+  };
+
+  models.Order.hasMany(models.Trade, {
+    as: 'orderTrades',
+    foreignKey: 'orderId',
+    constraints: true,
+  });
+  models.Trade.belongsTo(models.Order, {
+    as: 'orderInstance',
+    constraints: true,
+    foreignKey: 'orderId',
+  });
+
+  models.Fee.belongsTo(models.Order, {
+    foreignKey: 'orderId',
+    constraints: true,
+  });
+  models.Fee.belongsTo(models.Trade, {
+    foreignKey: 'tradeId',
+    constraints: true,
+  });
+
+  return models;
+};
+
+const initDB$ = ({
+  logger,
+  dataDir,
+}: InitDBparams): Observable<InitDBResponse> => {
+  const sequelize = new Sequelize({
+    storage: dataDir ? `${dataDir}/arby.db` : undefined,
+    logging: logger.trace,
+    dialect: 'sqlite',
+  });
+
+  return of(createModels(sequelize)).pipe(
+    mergeMap(models => {
+      return from(sequelize.authenticate()).pipe(
+        mergeMap(() => {
+          return from(models.Order.sync()).pipe(
+            mergeMap(() => {
+              return from(models.Trade.sync()).pipe(
+                mergeMap(() => {
+                  return from(models.Fee.sync()).pipe(mergeMap(() => of(models)));
+                })
+              );
+            })
+          );
+        })
+      );
+    })
+  );
+};
+
+export { initDB$, InitDBparams, InitDBResponse };

src/db/db.ts

@@ -0,0 +1,37 @@
+import {
+  DataTypes,
+  Model,
+  ModelAttributes,
+  ModelOptions,
+  Sequelize,
+} from 'sequelize';
+import { ModelCtor } from 'sequelize/types/lib/model';
+
+type FeeAttributes = {
+  orderId?: string;
+  tradeId?: string;
+  type: 'taker' | 'maker';
+  currency: string;
+  rate: number;
+  cost: number;
+};
+
+export interface FeeInstance extends Model<FeeAttributes>, FeeAttributes {}
+
+export function Fee(sequelize: Sequelize): ModelCtor<FeeInstance> {
+  const attributes: ModelAttributes<FeeInstance> = {
+    orderId: { type: DataTypes.STRING, allowNull: true },
+    tradeId: { type: DataTypes.STRING, allowNull: true },
+    type: { type: DataTypes.STRING, allowNull: false },
+    currency: { type: DataTypes.STRING, allowNull: false },
+    rate: { type: DataTypes.DOUBLE, allowNull: false },
+    cost: { type: DataTypes.DOUBLE, allowNull: false },
+  };
+
+  const options: ModelOptions = {
+    tableName: 'fees',
+    timestamps: false,
+  };
+
+  return sequelize.define<FeeInstance>('Fee', attributes, options);
+}