Merge pull request #81 from bitrinjani/feature/analytics

Analytics plugin system

.gitignore

@@ -16,4 +16,5 @@ config.json
 .idea
 .python-version
 datastore/
-reports/
+reports/
+sandbox/

config_default.json

@@ -21,6 +21,11 @@
       "ttl": 3000
     }
   },
+  "analytics": {
+    "enabled": false,
+    "plugin": "SimpleSpreadStatHandler.js",
+    "initialHistory": { "minutes": 30 }
+  },
   "brokers": [
     {
       "broker": "Coincheck",

package.json

@@ -19,6 +19,7 @@
   "dependencies": {
     "@bitr/castable": "^1.0.1",
     "@bitr/chronodb": "^1.0.2",
+    "@bitr/logger": "^1.0.0",
     "chalk": "^2.3.0",
     "date-fns": "^1.28.5",
     "decimal.js": "^7.3.0",
@@ -32,8 +33,10 @@
     "node-fetch": "^1.7.3",
     "pino": "^4.10.2",
     "reflect-metadata": "^0.1.10",
+    "simple-statistics": "^5.2.1",
     "split2": "^2.2.0",
-    "uuid": "^3.1.0"
+    "uuid": "^3.1.0",
+    "zeromq": "^4.6.0"
   },
   "devDependencies": {
     "@types/chalk": "^2.2.0",
@@ -49,6 +52,7 @@
     "@types/pino": "^4.7.0",
     "@types/split2": "^2.1.6",
     "@types/uuid": "^3.4.2",
+    "@types/zeromq": "^4.5.3",
     "coveralls": "^3.0.0",
     "cpy-cli": "^1.0.1",
     "jest": "^21.0.0",
@@ -77,6 +81,8 @@
       "**/*.{ts,tsx}",
       "!src/index.ts",
       "!src/transport/index.ts",
+      "!src/analytics/index.ts",
+      "!src/container.config.ts",
       "!**/__tests__/**",
       "!**/node_modules/**",
       "!**/vendor/**",

plugins/SimpleSpreadStatHandler.js

@@ -0,0 +1,55 @@
+const _ = require('lodash');
+const ss = require('simple-statistics');
+const { getLogger } = require('@bitr/logger');
+
+const precision = 3;
+
+class SimpleSpreadStatHandler {
+  // Constructor is called when initial snapshot of spread stat history has arrived.
+  constructor(history) {
+    this.log = getLogger(this.constructor.name);
+    const profitPercentHistory = history.map(x => x.bestCase.profitPercentAgainstNotional);
+    this.sampleSize = profitPercentHistory.length;
+    this.profitPercentMean = this.sampleSize != 0 ? ss.mean(profitPercentHistory) : 0;
+    this.profitPercentVariance = this.sampleSize != 0 ? ss.sampleVariance(profitPercentHistory) : 0;
+  }
+
+  // The method is called each time new spread stat has arrived, by default every 3 seconds.
+  // Return value: part of ConfigRoot or undefined.
+  // If part of ConfigRoot is returned, the configuration will be merged. If undefined is returned, no update will be made.
+  async handle(spreadStat) {
+    const newData = spreadStat.bestCase.profitPercentAgainstNotional;
+    // add new data to mean
+    this.profitPercentMean = ss.addToMean(this.profitPercentMean, this.sampleSize, newData);
+    // add new data to variance
+    this.profitPercentVariance = ss.combineVariances(
+      this.profitPercentVariance,
+      this.profitPercentMean,
+      this.sampleSize,
+      0,
+      newData,
+      1
+    );
+
+    this.sampleSize++;
+
+    // set μ + σ to minTargetProfitPercent
+    const n = this.sampleSize;
+    const mean = this.profitPercentMean;
+    const standardDeviation = Math.sqrt(this.profitPercentVariance * n/(n-1));
+    const minTargetProfitPercent = _.round(mean + standardDeviation, precision);
+    if (_.isNaN(minTargetProfitPercent)) {
+      return undefined;
+    }
+    this.log.info(
+      `μ: ${_.round(mean, precision)}, σ: ${_.round(
+        standardDeviation,
+        precision
+      )}, n: ${n} => minTargetProfitPercent: ${minTargetProfitPercent}`
+    );
+    const config = { minTargetProfitPercent };
+    return config;
+  }
+}
+
+module.exports = SimpleSpreadStatHandler;

src/AppRoot.ts

@@ -1,7 +1,6 @@
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import t from './intl';
 import 'reflect-metadata';
-import container from './container';
 import symbols from './symbols';
 import { BrokerAdapter, ConfigStore } from './types';
 import { Container } from 'inversify';
@@ -15,7 +14,7 @@ export default class AppRoot {
   private readonly log = getLogger(this.constructor.name);
   private services: { start: () => Promise<void>; stop: () => Promise<void> }[];
 
-  constructor(private readonly ioc: Container = container) {}
+  constructor(private readonly ioc: Container) {}
 
   async start(): Promise<void> {
     try {

src/Arbitrager.ts

@@ -1,4 +1,4 @@
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import { injectable, inject } from 'inversify';
 import * as _ from 'lodash';
 import { ConfigStore, Quote } from './types';

src/Bitflyer/BrokerAdapterImpl.ts

@@ -11,7 +11,7 @@
   Quote,
   BrokerConfigType
 } from '../types';
-import { getLogger } from '../logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import BrokerApi from './BrokerApi';
 import { ChildOrdersParam, SendChildOrderRequest, ChildOrder, BoardResponse } from './types';

src/BrokerAdapterRouter.ts

@@ -1,5 +1,5 @@
 import { Broker, BrokerAdapter, BrokerMap, Order, Quote } from './types';
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import { injectable, multiInject } from 'inversify';
 import symbols from './symbols';

src/Coincheck/BrokerAdapterImpl.ts

@@ -1,4 +1,4 @@
-import { getLogger } from '../logger';
+import { getLogger } from '@bitr/logger';
 import { addMinutes } from 'date-fns';
 import * as _ from 'lodash';
 import BrokerApi from './BrokerApi';

src/JsonConfigStore.ts

@@ -1,24 +1,85 @@
 import { injectable } from 'inversify';
-import { ConfigStore, ConfigRoot } from './types';
-import { getConfigRoot } from './configUtil';
+import { ConfigStore, ConfigRoot, ConfigRequest } from './types';
+import { getConfigRoot, getConfigPath } from './configUtil';
 import ConfigValidator from './ConfigValidator';
 import { setTimeout } from 'timers';
+import { socket, Socket } from 'zeromq';
+import { configStoreSocketUrl } from './constants';
+import { parseBuffer } from './util';
+import * as _ from 'lodash';
+import * as fs from 'fs';
+import { promisify } from 'util';
+import { getLogger } from '@bitr/logger';
+
+const writeFile = promisify(fs.writeFile);
 
 @injectable()
 export default class JsonConfigStore implements ConfigStore {
-  private TTL = 5 * 1000;
+  private readonly log = getLogger(this.constructor.name);
+  private timer: NodeJS.Timer;
+  private readonly server: Socket;
+  private readonly TTL = 5 * 1000;
   private cache?: ConfigRoot;
 
-  constructor(private readonly configValidator: ConfigValidator) {}
+  constructor(private readonly configValidator: ConfigValidator) {
+    this.server = socket('rep');
+    this.server.on('message', message => this.messageHandler(message));
+    this.server.bindSync(configStoreSocketUrl);
+  }
 
   get config(): ConfigRoot {
     if (this.cache) {
       return this.cache;
     }
     const config = getConfigRoot();
     this.configValidator.validate(config);
-    this.cache = config;
-    setTimeout(() => (this.cache = undefined), this.TTL);
+    this.updateCache(config);
     return config;
   }
+
+  async set(config: ConfigRoot) {
+    this.configValidator.validate(config);
+    await writeFile(getConfigPath(), JSON.stringify(config, undefined, 2));
+    this.updateCache(config);
+  }
+
+  close() {
+    this.server.close();
+  }
+
+  private async messageHandler(message: Buffer) {
+    const parsed = parseBuffer<ConfigRequest>(message);
+    if (parsed === undefined) {
+      this.log.debug(`Invalid message received. Message: ${message.toString()}`);
+      this.server.send(JSON.stringify({ success: false, reason: 'invalid message' }));
+      return;
+    }
+    switch (parsed.type) {
+      case 'set':
+        try {
+          const newConfig = parsed.data;
+          await this.set(_.merge({}, getConfigRoot(), newConfig));
+          this.server.send(JSON.stringify({ success: true }));
+          this.log.debug(`Config updated with ${JSON.stringify(newConfig)}`);
+        } catch (ex) {
+          this.server.send(JSON.stringify({ success: false, reason: 'invalid config' }));
+          this.log.warn(`Failed to update config. Error: ${ex.message}`);
+          this.log.debug(ex.stack);
+        }
+        break;
+      case 'get':
+        this.server.send(JSON.stringify({ success: true, data: getConfigRoot() }));
+        break;
+      default:
+        this.log.warn(`ConfigStore received an invalid message. Message: ${parsed}`);
+        this.server.send(JSON.stringify({ success: false, reason: 'invalid message type' }));
+        break;
+    }
+  }
+
+  private updateCache(config: ConfigRoot) {
+    this.cache = config;
+    clearTimeout(this.timer);
+    this.timer = setTimeout(() => (this.cache = undefined), this.TTL);
+  }
 } /* istanbul ignore next */

src/MainLimitChecker.ts

@@ -5,7 +5,7 @@ import {
   LimitChecker,
   OrderPair
 } from './types';
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import t from './intl';
 import PositionService from './PositionService';

src/OpportunitySearcher.ts

@@ -1,4 +1,4 @@
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import { injectable, inject } from 'inversify';
 import * as _ from 'lodash';
 import { ConfigStore, SpreadAnalysisResult, ActivePairStore, Quote, OrderPair } from './types';

src/PairTrader.ts

@@ -1,4 +1,4 @@
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import { injectable, inject } from 'inversify';
 import * as _ from 'lodash';
 import OrderImpl from './OrderImpl';

src/PositionService.ts

@@ -1,6 +1,6 @@
 import { injectable, inject } from 'inversify';
 import { ConfigStore, BrokerConfig, BrokerMap, BrokerPosition } from './types';
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import Decimal from 'decimal.js';
 import BrokerPositionImpl from './BrokerPositionImpl';

src/Quoine/BrokerAdapterImpl.ts

@@ -11,7 +11,7 @@
   BrokerConfigType
 } from '../types';
 import BrokerApi from './BrokerApi';
-import { getLogger } from '../logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import { PriceLevelsResponse, SendOrderRequest, OrdersResponse, CashMarginTypeStrategy } from './types';
 import { timestampToDate, toExecution, toQuote } from '../util';

src/QuoteAggregator.ts

@@ -1,6 +1,6 @@
 import { injectable, inject } from 'inversify';
 import { ConfigStore, BrokerConfig, QuoteSide, Broker, Quote } from './types';
-import { getLogger } from './logger';
+import { getLogger } from '@bitr/logger';
 import * as _ from 'lodash';
 import symbols from './symbols';
 import QuoteImpl from './QuoteImpl';