Add rebuildVolumeForAllMarkets() and migration to run it

knexfile.js

@@ -55,7 +55,7 @@ module.exports = {
   "prod-test": {
     client: "sqlite3",
     connection: {
-      filename: "./augur-1-3.db",
+      filename: "./augur-1-4.db",
     },
     migrations: {
       directory: "./src/migrations",

src/blockchain/log-processors/order-filled/update-volumetrics.ts

@@ -1,9 +1,9 @@
 import { Augur } from "augur.js";
 import BigNumber from "bignumber.js";
 import * as Knex from "knex";
-import { Address, Bytes32, TradesRow, ReportingState} from "../../../types";
-import { convertFixedPointToDecimal } from "../../../utils/convert-fixed-point-to-decimal";
 import { WEI_PER_ETHER } from "../../../constants";
+import { Address, Bytes32, ReportingState, TradesRow } from "../../../types";
+import { convertFixedPointToDecimal } from "../../../utils/convert-fixed-point-to-decimal";
 import { updateCategoryAggregationsOnMarketOpenInterestChanged } from "../category-aggregations";
 
 // volumeForTrade owns the business definition for the incremental financial
@@ -30,7 +30,7 @@ export function volumeForTrade(p: {
 }
 
 async function incrementMarketVolume(db: Knex, marketId: Address, amount: BigNumber, tradesRow: TradesRow<BigNumber>, isIncrease: boolean): Promise<void> {
-  const marketsRow: { minPrice: BigNumber, maxPrice: BigNumber, volume: BigNumber; shareVolume: BigNumber }|undefined = await db("markets").first("minPrice", "maxPrice", "volume", "shareVolume").where({ marketId });
+  const marketsRow: { minPrice: BigNumber, maxPrice: BigNumber, volume: BigNumber; shareVolume: BigNumber } | undefined = await db("markets").first("minPrice", "maxPrice", "volume", "shareVolume").where({ marketId });
   if (marketsRow === undefined) throw new Error(`No marketId for incrementMarketVolume: ${marketId}`);
   const newShareVolume = amount.plus(marketsRow.shareVolume);
   let vft = volumeForTrade({
@@ -44,9 +44,9 @@ async function incrementMarketVolume(db: Knex, marketId: Address, amount: BigNum
 }
 
 async function incrementOutcomeVolume(db: Knex, marketId: Address, outcome: number, amount: BigNumber, tradesRow: TradesRow<BigNumber>, isIncrease: boolean): Promise<void> {
-  const marketsRow: { minPrice: BigNumber, maxPrice: BigNumber }|undefined = await db("markets").first("minPrice", "maxPrice").where({ marketId });
+  const marketsRow: { minPrice: BigNumber, maxPrice: BigNumber } | undefined = await db("markets").first("minPrice", "maxPrice").where({ marketId });
   if (marketsRow === undefined) throw new Error(`No marketId for incrementOutcomeVolume: ${marketId}`);
-  const outcomesRow: { volume: BigNumber; shareVolume: BigNumber }|undefined = await db("outcomes").first("volume", "shareVolume").where({ marketId, outcome });
+  const outcomesRow: { volume: BigNumber; shareVolume: BigNumber } | undefined = await db("outcomes").first("volume", "shareVolume").where({ marketId, outcome });
   if (outcomesRow === undefined) throw new Error(`No outcome for incrementOutcomeVolume: marketId=${marketId} outcome=${outcome}`);
   const newShareVolume = amount.plus(outcomesRow.shareVolume);
   let vft = volumeForTrade({
@@ -69,7 +69,7 @@ export async function updateMarketOpenInterest(db: Knex, marketId: Address) {
     numTicks: BigNumber,
     openInterest: BigNumber,
     reportingState: ReportingState,
-  }|undefined = await db.first([
+  } | undefined = await db.first([
     "markets.category as category",
     "markets.numTicks as numTicks",
     "markets.openInterest as openInterest",
@@ -79,7 +79,7 @@ export async function updateMarketOpenInterest(db: Knex, marketId: Address) {
     .where({ "markets.marketId": marketId });
   if (marketRow == null) throw new Error(`No marketId for openInterest: ${marketId}`);
 
-  const shareTokenRow: { supply: BigNumber }|undefined = await db.first("supply").from("token_supply").join("tokens", "token_supply.token", "tokens.contractAddress").where({
+  const shareTokenRow: { supply: BigNumber } | undefined = await db.first("supply").from("token_supply").join("tokens", "token_supply.token", "tokens.contractAddress").where({
     marketId,
     symbol: "shares",
   });
@@ -103,7 +103,7 @@ export async function updateVolumetrics(db: Knex, augur: Augur, category: string
   if (shareTokenRow == null) throw new Error(`No shareToken found for market: ${marketId} outcome: ${outcome}`);
   const sharesOutstanding = augur.utils.convertOnChainAmountToDisplayAmount(new BigNumber(shareTokenRow.supply, 10), tickSize).toString();
   await db("markets").where({ marketId }).update({ sharesOutstanding });
-  const tradesRow: TradesRow<BigNumber>|undefined = await db.first("numCreatorShares", "numCreatorTokens", "numFillerTokens", "numFillerShares", "amount").from("trades")
+  const tradesRow: TradesRow<BigNumber> | undefined = await db.first("numCreatorShares", "numCreatorTokens", "numFillerTokens", "numFillerShares", "amount").from("trades")
     .where({ marketId, outcome, orderId, blockNumber });
   if (!tradesRow) throw new Error(`trade not found, orderId: ${orderId}`);
   let amount = tradesRow.amount!;

src/blockchain/rebuild-volume-for-all-markets.ts

@@ -0,0 +1,107 @@
+import BigNumber from "bignumber.js";
+import * as Knex from "knex";
+import { postProcessDatabaseResults } from "../server/post-process-database-results";
+import { Address, MarketsRow, TradesRow } from "../types";
+import { volumeForTrade } from "./log-processors/order-filled/update-volumetrics";
+
+export interface RebuildVolumeOpts {
+  manualPostProcessDatabaseResults: boolean; // iff true, data load will run postProcessDatabaseResults() "manually", instead of relying on Knex to have postProcessResponse set to augur-node's postProcessDatabaseResults.
+}
+
+export interface RebuildVolumeResult {
+  volumeInEthByMarketId: Map<Address, BigNumber>;
+  volumeInEthByOutcomeByMarketId: Map<Address, Map<number, BigNumber>>;
+}
+
+type RebuildVolumeMarketsRow = Pick<MarketsRow<BigNumber>, "marketId" | "minPrice" | "maxPrice">;
+type RebuildVolumeTradesRow = Pick<TradesRow<BigNumber>, "marketId" | "outcome" | "numCreatorTokens" | "numCreatorShares" | "numFillerTokens" | "numFillerShares">;
+
+// rebuildVolumeForAllMarkets will update all markets.volume
+// and outcomes.volume by computing the volume from scratch
+// (non-incrementally) using the log data stored in trades table.
+export async function rebuildVolumeForAllMarkets(db: Knex, opts: RebuildVolumeOpts): Promise<void> {
+  const params = await getDataForCalculateVolumeForAllMarkets(db, opts);
+  const newVolumeOrErr = calculateVolumeForAllMarkets(params.allMarkets, params.allTrades);
+  if (newVolumeOrErr instanceof Error) {
+    throw newVolumeOrErr;
+  }
+  await updateDBWithVolumeForAllMarkets(db, newVolumeOrErr);
+}
+
+// calculateVolumeForAllMarkets calculates each market and outcome's volume
+// from scratch (non-incrementally) for the passed markets and trades. Typically
+// a client would just use rebuildVolumeForAllMarkets() which wraps this. But
+// you may want to use this eg. for unit testing, to show that `incremental
+// volume calculation for N trades` == `non-incremental volume calculation`.
+function calculateVolumeForAllMarkets(allMarkets: Array<RebuildVolumeMarketsRow>, allTrades: Array<RebuildVolumeTradesRow>): RebuildVolumeResult | Error {
+  const marketsById: Map<Address, RebuildVolumeMarketsRow> = new Map();
+  allMarkets.forEach((m) => marketsById.set(m.marketId, m));
+  const volumeInEthByMarketId: Map<Address, BigNumber> = new Map();
+  const volumeInEthByOutcomeByMarketId: Map<Address, Map<number, BigNumber>> = new Map();
+  for (const tradesRow of allTrades) {
+    const marketsRow = marketsById.get(tradesRow.marketId);
+    if (marketsRow === undefined) {
+      return new Error(`marketId not found ${tradesRow.marketId}`);
+    }
+
+    const volumeFromThisTrade = volumeForTrade({
+      marketMinPrice: marketsRow.minPrice,
+      marketMaxPrice: marketsRow.maxPrice,
+      numCreatorTokens: tradesRow.numCreatorTokens,
+      numCreatorShares: tradesRow.numCreatorShares,
+      numFillerTokens: tradesRow.numFillerTokens,
+      numFillerShares: tradesRow.numFillerShares,
+    });
+
+    // 1. Update markets.volume with volume from this trade
+    const marketVolume = volumeInEthByMarketId.get(tradesRow.marketId);
+    if (marketVolume === undefined) {
+      volumeInEthByMarketId.set(tradesRow.marketId, volumeFromThisTrade);
+    } else {
+      volumeInEthByMarketId.set(tradesRow.marketId, marketVolume.plus(volumeFromThisTrade));
+    }
+
+    // 2. Update outcomes.volume with volume from this trade
+    let volumeInEthByOutcome = volumeInEthByOutcomeByMarketId.get(tradesRow.marketId);
+    if (volumeInEthByOutcome === undefined) {
+      volumeInEthByOutcome = new Map();
+      volumeInEthByOutcomeByMarketId.set(tradesRow.marketId, volumeInEthByOutcome);
+    }
+    const outcomeVolume = volumeInEthByOutcome.get(tradesRow.outcome);
+    if (outcomeVolume === undefined) {
+      volumeInEthByOutcome.set(tradesRow.outcome, volumeFromThisTrade);
+    } else {
+      volumeInEthByOutcome.set(tradesRow.outcome, outcomeVolume.plus(volumeFromThisTrade));
+    }
+  }
+
+  return {
+    volumeInEthByMarketId,
+    volumeInEthByOutcomeByMarketId,
+  };
+}
+
+async function getDataForCalculateVolumeForAllMarkets(db: Knex, opts: RebuildVolumeOpts): Promise<{
+  allMarkets: Array<RebuildVolumeMarketsRow>,
+  allTrades: Array<RebuildVolumeTradesRow>,
+}> {
+  // Why opts.manualPostProcessDatabaseResults? our DB layer converts whitelisted
+  // fields automatically via postProcessDatabaseResults(), but Knex may be
+  // passed from a context where this isn't setup (such as in a DB migration).
+  const allMarkets: Array<RebuildVolumeMarketsRow> = await db.select("marketId", "minPrice", "maxPrice").from("markets").then((result) => opts.manualPostProcessDatabaseResults ? postProcessDatabaseResults(result) : result);
+  const allTrades: Array<RebuildVolumeTradesRow> = await db.select("marketId", "outcome", "numCreatorTokens", "numCreatorShares", "numFillerTokens", "numFillerShares").from("trades").then((result) => opts.manualPostProcessDatabaseResults ? postProcessDatabaseResults(result) : result);
+  return {
+    allMarkets,
+    allTrades,
+  };
+}
+
+// updateDBWithVolumeForAllMarkets updates passed
+// db with result of calculateVolumeForAllMarkets().
+async function updateDBWithVolumeForAllMarkets(db: Knex, v: RebuildVolumeResult): Promise<void> {
+  v.volumeInEthByMarketId.forEach(async (volumeInEth: BigNumber, marketId: Address) => await db("markets").update("volume", volumeInEth.toString()).where({ marketId }));
+
+  v.volumeInEthByOutcomeByMarketId.forEach(async (volumeInEthByOutcome: Map<number, BigNumber>, marketId: Address) => {
+    volumeInEthByOutcome.forEach(async (volumeInEth: BigNumber, outcome: number) => await db("outcomes").update("volume", volumeInEth.toString()).where({ marketId, outcome }));
+  });
+}

src/migrations/20190201161645_rebuild_volume_for_all_markets.ts

@@ -0,0 +1,13 @@
+import * as Knex from "knex";
+import { rebuildVolumeForAllMarkets } from "../blockchain/rebuild-volume-for-all-markets";
+
+// A recent fix to volume requires a full DB rebuild, so instead
+// we're doing a migration to rebuild volume for all markets.
+
+exports.up = async (db: Knex): Promise<any> => {
+  await rebuildVolumeForAllMarkets(db, { manualPostProcessDatabaseResults: true }); // manualPostProcessDatabaseResults==true because Knex invoked via migration doesn't have postProcessDatabaseResults setup
+};
+
+exports.down = async (db: Knex): Promise<any> => {
+
+};