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

ExchangeUnion · Dec 23, 2020 · ab1d7d6 · ab1d7d6
1 parent 8bb5b1d
commit ab1d7d6
Show file tree

Hide file tree

Showing 10 changed files with 11,549 additions and 138 deletions.
diff --git a/.env-example b/.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
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/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"

diff --git a/src/arby.spec.ts b/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,

diff --git a/src/arby.ts b/src/arby.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) {

diff --git a/src/db/cex-fee.ts b/src/db/cex-fee.ts
@@ -0,0 +1,41 @@
+import {
+  DataTypes,
+  Model,
+  ModelAttributes,
+  ModelOptions,
+  Sequelize,
+} from 'sequelize';
+import { ModelCtor } from 'sequelize/types/lib/model';
+
+type CexFeeCreationAttributes = {
+  orderId?: string;
+  tradeId?: string;
+  type: 'taker' | 'maker';
+  currency: string;
+  rate: number;
+  cost: number;
+};
+
+type CexFeeAttributes = CexFeeCreationAttributes;
+
+export interface CexFeeInstance
+  extends Model<CexFeeAttributes, CexFeeCreationAttributes>,
+    CexFeeAttributes {}
+
+export function CexFee(sequelize: Sequelize): ModelCtor<CexFeeInstance> {
+  const attributes: ModelAttributes<CexFeeInstance> = {
+    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: 'cexfees',
+    timestamps: false,
+  };
+
+  return sequelize.define<CexFeeInstance>('CexFee', attributes, options);
+}
diff --git a/src/db/cex-order.ts b/src/db/cex-order.ts
@@ -0,0 +1,59 @@
+import {
+  DataTypes,
+  Model,
+  ModelAttributes,
+  ModelOptions,
+  Sequelize,
+} from 'sequelize';
+import { ModelCtor } from 'sequelize/types/lib/model';
+
+type CexOrderCreationAttributes = {
+  id: string;
+  datetime: string;
+  timestamp: number;
+  lastTradeTimestamp: number;
+  status: 'open' | 'closed' | 'canceled';
+  symbol: string;
+  type: string;
+  side: 'buy' | 'sell';
+  price: number;
+  average?: number;
+  amount: number;
+  filled: number;
+  remaining: number;
+  cost: number;
+  info: any;
+};
+
+type CexOrderAttributes = CexOrderCreationAttributes;
+
+export interface CexOrderInstance
+  extends Model<CexOrderAttributes, CexOrderCreationAttributes>,
+    CexOrderAttributes {}
+
+export function CexOrder(sequelize: Sequelize): ModelCtor<CexOrderInstance> {
+  const attributes: ModelAttributes<CexOrderInstance> = {
+    id: { type: DataTypes.STRING, primaryKey: true },
+    datetime: { type: DataTypes.STRING, allowNull: false },
+    timestamp: { type: DataTypes.BIGINT, allowNull: false },
+    lastTradeTimestamp: { type: DataTypes.BIGINT, allowNull: false },
+    status: { type: DataTypes.STRING, allowNull: false },
+    symbol: { type: DataTypes.STRING, allowNull: false },
+    type: { type: DataTypes.STRING, allowNull: false },
+    side: { type: DataTypes.STRING, allowNull: false },
+    price: { type: DataTypes.DOUBLE, allowNull: false },
+    average: { type: DataTypes.DOUBLE, allowNull: true },
+    amount: { type: DataTypes.DOUBLE, allowNull: false },
+    filled: { type: DataTypes.DOUBLE, allowNull: false },
+    remaining: { type: DataTypes.DOUBLE, allowNull: false },
+    cost: { type: DataTypes.DOUBLE, allowNull: false },
+    info: { type: DataTypes.STRING, allowNull: false },
+  };
+
+  const options: ModelOptions = {
+    tableName: 'cexorders',
+    timestamps: false,
+  };
+
+  return sequelize.define<CexOrderInstance>('CexOrder', attributes, options);
+}
diff --git a/src/db/cex-trade.ts b/src/db/cex-trade.ts
@@ -0,0 +1,53 @@
+import {
+    DataTypes,
+    Model,
+    ModelAttributes,
+    ModelOptions,
+    Sequelize,
+} from 'sequelize';
+import { ModelCtor } from 'sequelize/types/lib/model';
+
+type CexTradeCreationAttributes = {
+  id: string;
+  orderId: string;
+  amount: number;
+  datetime: string;
+  info: any;
+  price: number;
+  timestamp: number;
+  type?: 'market' | 'limit';
+  side: 'buy' | 'sell';
+  symbol: string;
+  takerOrMaker: 'taker' | 'maker';
+  cost: number;
+};
+
+type CexTradeAttributes = CexTradeCreationAttributes;
+
+export interface CexTradeInstance
+    extends Model<CexTradeAttributes, CexTradeCreationAttributes>,
+        CexTradeAttributes {}
+
+export function CexTrade(sequelize: Sequelize): ModelCtor<CexTradeInstance> {
+    const attributes: ModelAttributes<CexTradeInstance> = {
+        id: { type: DataTypes.STRING, primaryKey: true },
+        orderId: { type: DataTypes.STRING, allowNull: false },
+        amount: { type: DataTypes.DOUBLE, allowNull: false },
+        datetime: { type: DataTypes.STRING, allowNull: false },
+        info: { type: DataTypes.STRING, allowNull: false },
+        price: { type: DataTypes.DOUBLE, allowNull: false },
+        timestamp: { type: DataTypes.BIGINT, allowNull: false },
+        type: { type: DataTypes.STRING, allowNull: false },
+        side: { type: DataTypes.STRING, allowNull: false },
+        symbol: { type: DataTypes.STRING, allowNull: false },
+        takerOrMaker: { type: DataTypes.STRING, allowNull: false },
+        cost: { type: DataTypes.DOUBLE, allowNull: false },
+    };
+
+    const options: ModelOptions = {
+        tableName: 'cextrades',
+        timestamps: false,
+    };
+
+    return sequelize.define<CexTradeInstance>('CexTrade', attributes, options);
+}