Refactor jobs for deriving value & prices (#234)

* Create job for converting fees

* Introduce new job for determining fill values

* Shortcut fee conversion for fills without fees

* Refactor job for deriving fill prices

* Fix tests

Author: cbovis

config/default.js

@@ -19,13 +19,20 @@ module.exports = {
     apiKey: process.env.ETHPLORER_API_KEY,
   },
   jobs: {
+    convertFees: {
+      apiDelayMs: 100,
+      batchSize: 100,
+    },
     createFills: {
       batchSize: 500,
       processOldestFirst: true,
     },
-    updateFillPrices: {
+    deriveFillPrices: {
       batchSize: 250,
-      processOldestFirst: true,
+    },
+    determineFillValues: {
+      apiDelayMs: 100,
+      batchSize: 100,
     },
     updateFillRates: {
       batchSize: 100,

src/fills/get-base-token.js

@@ -1,33 +1,20 @@
 const _ = require('lodash');
 
 const { BASE_TOKENS } = require('../constants');
+const { getToken } = require('../tokens/token-cache');
 
-const getBaseToken = (fill, tokens) => {
-  const { makerToken, takerToken } = fill;
+const getBaseToken = fill => {
+  const address = _(BASE_TOKENS)
+    .keys()
+    .find(baseToken => {
+      return fill.makerToken === baseToken || fill.takerToken === baseToken;
+    });
 
-  return (
-    _(BASE_TOKENS)
-      .keys()
-      .map(baseToken => {
-        if (
-          _.has(tokens, makerToken) &&
-          tokens[makerToken].address === baseToken
-        ) {
-          return tokens[makerToken];
-        }
+  if (address === undefined) {
+    return null;
+  }
 
-        if (
-          _.has(tokens, takerToken) &&
-          tokens[takerToken].address === baseToken
-        ) {
-          return tokens[takerToken];
-        }
-
-        return null;
-      })
-      .compact()
-      .head() || null
-  );
+  return getToken(address);
 };
 
 module.exports = getBaseToken;

src/fills/get-base-token.test.js

@@ -1,4 +1,5 @@
 const getBaseToken = require('./get-base-token');
+const tokenCache = require('../tokens/token-cache');
 
 const tokens = {
   '0x123': { decimals: 18, symbol: 'ZRX' },
@@ -15,6 +16,10 @@ const tokens = {
   '0x7866': { decimals: 18, symbol: 'OMG' },
 };
 
+beforeAll(() => {
+  tokenCache.initialise(tokens);
+});
+
 it('should return WETH in WETH/OMG pair', () => {
   const fill = {
     makerToken: '0x7866',

src/jobs/convert-fees.js

@@ -0,0 +1,86 @@
+const bluebird = require('bluebird');
+const signale = require('signale');
+
+const { ZRX_TOKEN_ADDRESS } = require('../constants');
+const { getToken } = require('../tokens/token-cache');
+const Fill = require('../model/fill');
+const formatTokenAmount = require('../tokens/format-token-amount');
+const getConversionRate = require('../rates/get-conversion-rate');
+
+const logger = signale.scope('convert fees');
+
+const convertAmount = (amount, conversionRate) => {
+  if (amount === 0) {
+    return 0;
+  }
+
+  const zrxToken = getToken(ZRX_TOKEN_ADDRESS);
+  const formattedAmount = formatTokenAmount(amount, zrxToken);
+
+  return formattedAmount * conversionRate;
+};
+
+const convertFees = async ({ apiDelayMs, batchSize }) => {
+  // Fetch a batch of fills for processing where the maker or taker
+  // fee have not yet been converted.
+  const fills = await Fill.find({
+    $or: [
+      { 'conversions.USD.makerFee': null },
+      { 'conversions.USD.takerFee': null },
+    ],
+  })
+    .limit(batchSize)
+    .lean();
+
+  logger.info(`found ${fills.length} fills without converted fees`);
+
+  if (fills.length === 0) {
+    return;
+  }
+
+  await bluebird.mapSeries(fills, async fill => {
+    if (fill.makerFee + fill.takerFee === 0) {
+      await Fill.updateOne(
+        { _id: fill._id },
+        {
+          $set: {
+            'conversions.USD.makerFee': 0,
+            'conversions.USD.takerFee': 0,
+          },
+        },
+      );
+
+      logger.success(`Skipped fee conversion for fill ${fill._id}`);
+    } else {
+      logger.time(`Fetch ZRX conversion rate for ${fill.date}`);
+      const conversionRate = await getConversionRate('ZRX', 'USD', fill.date);
+      logger.timeEnd(`Fetch ZRX conversion rate for ${fill.date}`);
+
+      if (conversionRate === undefined) {
+        logger.warn(`Unable to fetch ZRX conversion rate for ${fill.date}`);
+        return;
+      }
+
+      const makerFee = convertAmount(fill.makerFee, conversionRate);
+      const takerFee = convertAmount(fill.takerFee, conversionRate);
+
+      await Fill.updateOne(
+        { _id: fill._id },
+        {
+          $set: {
+            'conversions.USD.makerFee': makerFee,
+            'conversions.USD.takerFee': takerFee,
+            'rates.ZRX.USD': conversionRate,
+          },
+        },
+      );
+
+      logger.success(`Converted fees for fill ${fill._id}`);
+
+      // limit the number of API requests made per second to avoid errors
+      await bluebird.delay(apiDelayMs);
+    }
+  });
+};
+
+module.exports = convertFees;

src/jobs/create-fills/create-fill.js

@@ -46,9 +46,17 @@ const createFill = async event => {
     takerAddress: taker,
   });
 
+  const conversions =
+    paidMakerFee + paidTakerFee === 0
+      ? {
+          USD: { makerFee: 0, takerFee: 0 },
+        }
+      : undefined;
+
   const fill = {
     blockHash,
     blockNumber,
+    conversions,
     date,
     feeRecipient,
     logIndex,

src/jobs/update-fill-prices/get-prices-for-fill.js src/jobs/derive-fill-prices/get-prices-for-fill.js

@@ -8,21 +8,23 @@ const getPrices = require('./get-prices-for-fill');
 const getRelayerPrices = require('./get-relayer-prices');
 const Relayer = require('../../model/relayer');
 const tokenCache = require('../../tokens/token-cache');
+const withTransaction = require('../../util/with-transaction');
 
-const logger = signale.scope('update fill prices');
+const logger = signale.scope('derive fill prices');
 
-const updateFillPrices = async ({ batchSize, processOldestFirst }) => {
+const deriveFillPrices = async ({ batchSize }) => {
   const fills = await Fill.find(
     {
-      'rates.saved': true,
+      // Determine whether prices can be derived
+      'conversions.USD.amount': { $ne: null },
       'tokenSaved.maker': true,
       'tokenSaved.taker': true,
+
+      // Determine whether prices have already been derived
       'prices.saved': { $in: [null, false] },
     },
     '_id conversions date makerAmount makerToken relayerId takerAmount takerToken',
-  )
-    .sort({ date: processOldestFirst ? 1 : -1 })
-    .limit(batchSize);
+  ).limit(batchSize);
 
   if (fills.length === 0) {
     logger.info('no fills were found without prices');
@@ -40,7 +42,7 @@ const updateFillPrices = async ({ batchSize, processOldestFirst }) => {
   )(fills);
 
   if (fillPrices.length === 0) {
-    logger.warn(`unable to generate prices for ${fillPrices.length} fills`);
+    logger.warn(`unable to derive prices for ${fillPrices.length} fills`);
     return;
   }
 
@@ -79,14 +81,15 @@ const updateFillPrices = async ({ batchSize, processOldestFirst }) => {
     },
   }));
 
-  // TODO: Investigate using the new MongoDB transactions feature to ensure consistency
-  await Fill.bulkWrite(fillOperations);
+  await withTransaction(async session => {
+    await Fill.bulkWrite(fillOperations, { session });
 
-  if (relayerOperations.length > 0) {
-    await Relayer.bulkWrite(relayerOperations);
-  }
+    if (relayerOperations.length > 0) {
+      await Relayer.bulkWrite(relayerOperations, { session });
+    }
+  });
 
-  logger.success(`updated prices for ${fillPrices.length} fills`);
+  logger.success(`derived prices of ${fillPrices.length} fills`);
 };
 
-module.exports = updateFillPrices;
+module.exports = deriveFillPrices;

src/jobs/update-fill-prices/get-prices-for-fill.test.js src/jobs/derive-fill-prices/get-prices-for-fill.test.js

@@ -0,0 +1,91 @@
+const _ = require('lodash');
+const bluebird = require('bluebird');
+const signale = require('signale');
+
+const { BASE_TOKENS } = require('../constants');
+const Fill = require('../model/fill');
+const formatTokenAmount = require('../tokens/format-token-amount');
+const getBaseToken = require('../fills/get-base-token');
+const getConversionRate = require('../rates/get-conversion-rate');
+const normalizeSymbol = require('../tokens/normalize-symbol');
+
+const logger = signale.scope('determine fill values');
+
+const getTokenValue = (fill, baseToken) => {
+  if (fill.makerToken === baseToken.address) {
+    return formatTokenAmount(fill.makerAmount, baseToken);
+  }
+
+  return formatTokenAmount(fill.takerAmount, baseToken);
+};
+
+const determineFillValues = async ({ apiDelayMs, batchSize }) => {
+  const baseTokens = _.keys(BASE_TOKENS);
+  const fills = await Fill.find({
+    'conversions.USD.amount': null,
+    $or: [
+      { makerToken: { $in: baseTokens } },
+      { takerToken: { $in: baseTokens } },
+    ],
+  })
+    .limit(batchSize)
+    .lean();
+
+  logger.info(`found ${fills.length} measurable fills without a value`);
+
+  if (fills.length === 0) {
+    return;
+  }
+
+  await bluebird.mapSeries(fills, async fill => {
+    const baseToken = getBaseToken(fill);
+
+    if (baseToken === null || baseToken === undefined) {
+      logger.warn(`unable to determine base token of fill ${fill._id}`);
+
+      return;
+    }
+
+    const normalisedSymbol = normalizeSymbol(baseToken.symbol);
+
+    logger.time(
+      `fetch conversion rate for ${normalisedSymbol} on ${fill.date}`,
+    );
+    const conversionRate = await getConversionRate(
+      normalisedSymbol,
+      'USD',
+      fill.date,
+    );
+    logger.timeEnd(
+      `fetch conversion rate for ${normalisedSymbol} on ${fill.date}`,
+    );
+
+    if (conversionRate === undefined) {
+      logger.warn(
+        `unable to determine conversion rate of ${normalisedSymbol} on ${fill.date}`,
+      );
+
+      return;
+    }
+
+    const tokenValue = getTokenValue(fill, baseToken);
+    const usdValue = tokenValue * conversionRate;
+
+    await Fill.updateOne(
+      { _id: fill._id },
+      {
+        $set: {
+          'conversions.USD.amount': usdValue,
+          [`rates.data.${normalisedSymbol}.USD`]: conversionRate,
+        },
+      },
+    );
+
+    logger.success(`determined fill value for ${fill._id}`);
+
+    // limit the number of API requests made per second to avoid errors
+    await bluebird.delay(apiDelayMs);
+  });
+};
+
+module.exports = determineFillValues;

src/jobs/update-fill-prices/get-relayer-prices.js src/jobs/derive-fill-prices/get-relayer-prices.js

@@ -4,12 +4,13 @@ const backfillRelayerRelationships = require('./backfill-relayer-relationships')
 const cacheRelayerMetrics = require('./cache-relayer-metrics');
 const cacheTokenMetrics = require('./cache-token-metrics');
 const cacheTokenStats = require('./cache-token-stats');
+const convertFees = require('./convert-fees');
 const createFills = require('./create-fills');
+const deriveFillPrices = require('./derive-fill-prices');
+const determineFillValues = require('./determine-fill-values');
 const getMissingTokenImages = require('./get-missing-token-images');
 const getNewArticles = require('./get-new-articles');
 const resolveTokens = require('./resolve-tokens');
-const updateFillPrices = require('./update-fill-prices');
-const updateFillRates = require('./update-fill-rates');
 const updateFillStatuses = require('./update-fill-statuses');
 const updateRelayerStats = require('./update-relayer-stats');
 const updateTokenPrices = require('./update-token-prices');
@@ -19,12 +20,13 @@ const jobFns = {
   cacheRelayerMetrics,
   cacheTokenMetrics,
   cacheTokenStats,
+  convertFees,
   createFills,
+  deriveFillPrices,
+  determineFillValues,
   getMissingTokenImages,
   getNewArticles,
   resolveTokens,
-  updateFillPrices,
-  updateFillRates,
   updateFillStatuses,
   updateRelayerStats,
   updateTokenPrices,