track idex orderbook snapshots (#1397)

* Track Idex and Oasis Orderbook Snapshots

14 files changed, 690 insertions, 34 deletions

diff --git a/packages/pipeline/migrations/1544131658904-TokenOrderbookSnapshotAddOrderType.ts b/packages/pipeline/migrations/1544131658904-TokenOrderbookSnapshotAddOrderType.ts
new file mode 100644
index 000000000..a501ec6d8
--- /dev/null
+++ b/packages/pipeline/migrations/1544131658904-TokenOrderbookSnapshotAddOrderType.ts
@@ -0,0 +1,33 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class TokenOrderbookSnapshotAddOrderType1544131658904 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.token_orderbook_snapshots
+                DROP CONSTRAINT "PK_8a16487e7cb6862ec5a84ed3495",
+                ADD PRIMARY KEY (observed_timestamp, source, order_type, price, base_asset_symbol, quote_asset_symbol);
+            `,
+        );
+        await queryRunner.query(
+            `ALTER TABLE raw.token_orderbook_snapshots
+                ALTER COLUMN quote_asset_address DROP NOT NULL,
+                ALTER COLUMN base_asset_address DROP NOT NULL;
+            `,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.token_orderbook_snapshots
+                ALTER COLUMN quote_asset_address SET NOT NULL,
+                ALTER COLUMN base_asset_address SET NOT NULL;
+            `,
+        );
+        await queryRunner.query(
+            `ALTER TABLE raw.token_orderbook_snapshots
+                DROP CONSTRAINT token_orderbook_snapshots_pkey,
+                ADD CONSTRAINT "PK_8a16487e7cb6862ec5a84ed3495" PRIMARY KEY (observed_timestamp, source, price, base_asset_symbol, quote_asset_symbol);
+            `,
+        );
+    }
+}
diff --git a/packages/pipeline/src/data_sources/idex/index.ts b/packages/pipeline/src/data_sources/idex/index.ts
new file mode 100644
index 000000000..c1e53c08d
--- /dev/null
+++ b/packages/pipeline/src/data_sources/idex/index.ts
@@ -0,0 +1,82 @@
+import { fetchAsync } from '@0x/utils';
+
+const IDEX_BASE_URL = 'https://api.idex.market';
+const MARKETS_URL = `${IDEX_BASE_URL}/returnTicker`;
+const ORDERBOOK_URL = `${IDEX_BASE_URL}/returnOrderBook`;
+const MAX_ORDER_COUNT = 100; // Maximum based on https://github.com/AuroraDAO/idex-api-docs#returnorderbook
+export const IDEX_SOURCE = 'idex';
+
+export interface IdexMarketsResponse {
+    [marketName: string]: IdexMarket;
+}
+
+export interface IdexMarket {
+    last: string;
+    high: string;
+    low: string;
+    lowestAsk: string;
+    highestBid: string;
+    percentChange: string;
+    baseVolume: string;
+    quoteVolume: string;
+}
+
+export interface IdexOrderbook {
+    asks: IdexOrder[];
+    bids: IdexOrder[];
+}
+
+export interface IdexOrder {
+    price: string;
+    amount: string;
+    total: string;
+    orderHash: string;
+    params: IdexOrderParam;
+}
+
+export interface IdexOrderParam {
+    tokenBuy: string;
+    buySymbol: string;
+    buyPrecision: number;
+    amountBuy: string;
+    tokenSell: string;
+    sellSymbol: string;
+    sellPrecision: number;
+    amountSell: string;
+    expires: number;
+    nonce: number;
+    user: string;
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class IdexSource {
+    /**
+     * Call Idex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getMarketsAsync(): Promise<string[]> {
+        const params = { method: 'POST' };
+        const resp = await fetchAsync(MARKETS_URL, params);
+        const respJson: IdexMarketsResponse = await resp.json();
+        const markets: string[] = Object.keys(respJson);
+        return markets;
+    }
+
+    /**
+     * Retrieve orderbook from Idex API for a given market.
+     * @param marketId String identifying the market we want data for. Eg. 'REP_AUG'
+     */
+    public async getMarketOrderbookAsync(marketId: string): Promise<IdexOrderbook> {
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({
+                market: marketId,
+                count: MAX_ORDER_COUNT,
+            }),
+        };
+        const resp = await fetchAsync(ORDERBOOK_URL, params);
+        const respJson: IdexOrderbook = await resp.json();
+        return respJson;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/oasis/index.ts b/packages/pipeline/src/data_sources/oasis/index.ts
new file mode 100644
index 000000000..3b30e9dfd
--- /dev/null
+++ b/packages/pipeline/src/data_sources/oasis/index.ts
@@ -0,0 +1,103 @@
+import { fetchAsync } from '@0x/utils';
+
+const OASIS_BASE_URL = 'https://data.makerdao.com/v1';
+const OASIS_MARKET_QUERY = `query {
+  oasisMarkets(period: "1 week") {
+    nodes {
+      id
+      base
+      quote
+      buyVol
+      sellVol
+      price
+      high
+      low
+    }
+  }
+}`;
+const OASIS_ORDERBOOK_QUERY = `query ($market: String!) {
+  allOasisOrders(condition: { market: $market }) {
+    totalCount
+    nodes {
+      market
+      offerId
+      price
+      amount
+      act
+    }
+  }
+}`;
+export const OASIS_SOURCE = 'oasis';
+
+export interface OasisMarket {
+    id: string; // market symbol e.g MKRDAI
+    base: string; // base symbol   e.g MKR
+    quote: string; // quote symbol  e.g DAI
+    buyVol: number; // total buy volume (base)
+    sellVol: number; // total sell volume (base)
+    price: number; // volume weighted price (quote)
+    high: number; // max sell price
+    low: number; // min buy price
+}
+
+export interface OasisMarketResponse {
+    data: {
+        oasisMarkets: {
+            nodes: OasisMarket[];
+        };
+    };
+}
+
+export interface OasisOrder {
+    offerId: number; // Offer Id
+    market: string; // Market symbol (base/quote)
+    price: string; // Offer price (quote)
+    amount: string; // Offer amount (base)
+    act: string; // Action (ask|bid)
+}
+
+export interface OasisOrderbookResponse {
+    data: {
+        allOasisOrders: {
+            totalCount: number;
+            nodes: OasisOrder[];
+        };
+    };
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class OasisSource {
+    /**
+     * Call Ddex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getActiveMarketsAsync(): Promise<OasisMarket[]> {
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ query: OASIS_MARKET_QUERY }),
+        };
+        const resp = await fetchAsync(OASIS_BASE_URL, params);
+        const respJson: OasisMarketResponse = await resp.json();
+        const markets = respJson.data.oasisMarkets.nodes;
+        return markets;
+    }
+
+    /**
+     * Retrieve orderbook from Oasis API for a given market.
+     * @param marketId String identifying the market we want data for. Eg. 'REPAUG'.
+     */
+    public async getMarketOrderbookAsync(marketId: string): Promise<OasisOrder[]> {
+        const input = {
+            market: marketId,
+        };
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ query: OASIS_ORDERBOOK_QUERY, variables: input }),
+        };
+        const resp = await fetchAsync(OASIS_BASE_URL, params);
+        const respJson: OasisOrderbookResponse = await resp.json();
+        return respJson.data.allOasisOrders.nodes;
+    }
+}
diff --git a/packages/pipeline/src/entities/token_order.ts b/packages/pipeline/src/entities/token_order.ts
index 557705767..4b8f0abc3 100644
--- a/packages/pipeline/src/entities/token_order.ts
+++ b/packages/pipeline/src/entities/token_order.ts
@@ -10,20 +10,20 @@ export class TokenOrderbookSnapshot {
     public observedTimestamp!: number;
     @PrimaryColumn({ name: 'source' })
     public source!: string;
-    @Column({ name: 'order_type' })
+    @PrimaryColumn({ name: 'order_type' })
     public orderType!: OrderType;
     @PrimaryColumn({ name: 'price', type: 'numeric', transformer: bigNumberTransformer })
     public price!: BigNumber;
     @PrimaryColumn({ name: 'base_asset_symbol' })
     public baseAssetSymbol!: string;
-    @Column({ name: 'base_asset_address' })
-    public baseAssetAddress!: string;
+    @Column({ nullable: true, type: String, name: 'base_asset_address' })
+    public baseAssetAddress!: string | null;
     @Column({ name: 'base_volume', type: 'numeric', transformer: bigNumberTransformer })
     public baseVolume!: BigNumber;
     @PrimaryColumn({ name: 'quote_asset_symbol' })
     public quoteAssetSymbol!: string;
-    @Column({ name: 'quote_asset_address' })
-    public quoteAssetAddress!: string;
+    @Column({ nullable: true, type: String, name: 'quote_asset_address' })
+    public quoteAssetAddress!: string | null;
     @Column({ name: 'quote_volume', type: 'numeric', transformer: bigNumberTransformer })
     public quoteVolume!: BigNumber;
 }
diff --git a/packages/pipeline/src/parsers/ddex_orders/index.ts b/packages/pipeline/src/parsers/ddex_orders/index.ts
index 81132e8f0..52a998f9f 100644
--- a/packages/pipeline/src/parsers/ddex_orders/index.ts
+++ b/packages/pipeline/src/parsers/ddex_orders/index.ts
@@ -1,7 +1,8 @@
 import { BigNumber } from '@0x/utils';
-import * as R from 'ramda';
 
-import { DdexMarket, DdexOrder, DdexOrderbook } from '../../data_sources/ddex';
+import { aggregateOrders } from '../utils';
+
+import { DdexMarket, DdexOrderbook } from '../../data_sources/ddex';
 import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
 import { OrderType } from '../../types';
 
@@ -28,19 +29,6 @@ export function parseDdexOrders(
 }
 
 /**
- * Aggregates orders by price point for consistency with other exchanges.
- * Querying the Ddex API at level 3 setting returns a breakdown of
- * individual orders at each price point. Other exchanges only give total amount
- * at each price point. Returns an array of <price, amount> tuples.
- * @param ddexOrders A list of Ddex orders awaiting aggregation.
- */
-export function aggregateOrders(ddexOrders: DdexOrder[]): Array<[string, BigNumber]> {
-    const sumAmount = (acc: BigNumber, order: DdexOrder): BigNumber => acc.plus(order.amount);
-    const aggregatedPricePoints = R.reduceBy(sumAmount, new BigNumber(0), R.prop('price'), ddexOrders);
-    return Object.entries(aggregatedPricePoints);
-}
-
-/**
  * Parse a single aggregated Ddex order in order to form a tokenOrder entity
  * which can be saved into the database.
  * @param ddexMarket An object containing information about the market where these
diff --git a/packages/pipeline/src/parsers/idex_orders/index.ts b/packages/pipeline/src/parsers/idex_orders/index.ts
new file mode 100644
index 000000000..dfe27455c
--- /dev/null
+++ b/packages/pipeline/src/parsers/idex_orders/index.ts
@@ -0,0 +1,77 @@
+import { BigNumber } from '@0x/utils';
+
+import { aggregateOrders } from '../utils';
+
+import { IdexOrder, IdexOrderbook, IdexOrderParam } from '../../data_sources/idex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook,
+ * 2) Aggregates them by price point,
+ * 3) Parses them into entities which are then saved into the database.
+ * @param idexOrderbook raw orderbook that we pull from the Idex API.
+ * @param observedTimestamp Time at which the orders for the market were pulled.
+ * @param source The exchange where these orders are placed. In this case 'idex'.
+ */
+export function parseIdexOrders(idexOrderbook: IdexOrderbook, observedTimestamp: number, source: string): TokenOrder[] {
+    const aggregatedBids = aggregateOrders(idexOrderbook.bids);
+    // Any of the bid orders' params will work
+    const idexBidOrder = idexOrderbook.bids[0];
+    const parsedBids =
+        aggregatedBids.length > 0
+            ? aggregatedBids.map(order => parseIdexOrder(idexBidOrder.params, observedTimestamp, 'bid', source, order))
+            : [];
+
+    const aggregatedAsks = aggregateOrders(idexOrderbook.asks);
+    // Any of the ask orders' params will work
+    const idexAskOrder = idexOrderbook.asks[0];
+    const parsedAsks =
+        aggregatedAsks.length > 0
+            ? aggregatedAsks.map(order => parseIdexOrder(idexAskOrder.params, observedTimestamp, 'ask', source, order))
+            : [];
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * Parse a single aggregated Idex order in order to form a tokenOrder entity
+ * which can be saved into the database.
+ * @param idexOrderParam An object containing information about the market where these
+ * trades have been placed.
+ * @param observedTimestamp The time when the API response returned back to us.
+ * @param orderType 'bid' or 'ask' enum.
+ * @param source Exchange where these orders were placed.
+ * @param idexOrder A <price, amount> tuple which we will convert to volume-basis.
+ */
+export function parseIdexOrder(
+    idexOrderParam: IdexOrderParam,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    idexOrder: [string, BigNumber],
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(idexOrder[0]);
+    const amount = idexOrder[1];
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+    tokenOrder.baseVolume = amount;
+    tokenOrder.quoteVolume = price.times(amount);
+
+    if (orderType === 'bid') {
+        tokenOrder.baseAssetSymbol = idexOrderParam.buySymbol;
+        tokenOrder.baseAssetAddress = idexOrderParam.tokenBuy;
+        tokenOrder.quoteAssetSymbol = idexOrderParam.sellSymbol;
+        tokenOrder.quoteAssetAddress = idexOrderParam.tokenSell;
+    } else {
+        tokenOrder.baseAssetSymbol = idexOrderParam.sellSymbol;
+        tokenOrder.baseAssetAddress = idexOrderParam.tokenSell;
+        tokenOrder.quoteAssetSymbol = idexOrderParam.buySymbol;
+        tokenOrder.quoteAssetAddress = idexOrderParam.tokenBuy;
+    }
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/oasis_orders/index.ts b/packages/pipeline/src/parsers/oasis_orders/index.ts
new file mode 100644
index 000000000..7aafbf460
--- /dev/null
+++ b/packages/pipeline/src/parsers/oasis_orders/index.ts
@@ -0,0 +1,71 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { aggregateOrders } from '../utils';
+
+import { OasisMarket, OasisOrder } from '../../data_sources/oasis';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook,
+ * 2) Aggregates them according to price point,
+ * 3) Builds TokenOrder entity with other information attached.
+ * @param oasisOrderbook A raw orderbook that we pull from the Oasis API.
+ * @param oasisMarket An object containing market data also directly from the API.
+ * @param observedTimestamp Time at which the orders for the market were pulled.
+ * @param source The exchange where these orders are placed. In this case 'oasis'.
+ */
+export function parseOasisOrders(
+    oasisOrderbook: OasisOrder[],
+    oasisMarket: OasisMarket,
+    observedTimestamp: number,
+    source: string,
+): TokenOrder[] {
+    const aggregatedBids = aggregateOrders(R.filter(R.propEq('act', 'bid'), oasisOrderbook));
+    const aggregatedAsks = aggregateOrders(R.filter(R.propEq('act', 'ask'), oasisOrderbook));
+    const parsedBids = aggregatedBids.map(order =>
+        parseOasisOrder(oasisMarket, observedTimestamp, 'bid', source, order),
+    );
+    const parsedAsks = aggregatedAsks.map(order =>
+        parseOasisOrder(oasisMarket, observedTimestamp, 'ask', source, order),
+    );
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * Parse a single aggregated Oasis order to form a tokenOrder entity
+ * which can be saved into the database.
+ * @param oasisMarket An object containing information about the market where these
+ * trades have been placed.
+ * @param observedTimestamp The time when the API response returned back to us.
+ * @param orderType 'bid' or 'ask' enum.
+ * @param source Exchange where these orders were placed.
+ * @param oasisOrder A <price, amount> tuple which we will convert to volume-basis.
+ */
+export function parseOasisOrder(
+    oasisMarket: OasisMarket,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    oasisOrder: [string, BigNumber],
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(oasisOrder[0]);
+    const amount = oasisOrder[1];
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+
+    tokenOrder.baseAssetSymbol = oasisMarket.base;
+    tokenOrder.baseAssetAddress = null; // Oasis doesn't provide address information
+    tokenOrder.baseVolume = price.times(amount);
+
+    tokenOrder.quoteAssetSymbol = oasisMarket.quote;
+    tokenOrder.quoteAssetAddress = null; // Oasis doesn't provide address information
+    tokenOrder.quoteVolume = amount;
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/utils.ts b/packages/pipeline/src/parsers/utils.ts
new file mode 100644
index 000000000..860729e9f
--- /dev/null
+++ b/packages/pipeline/src/parsers/utils.ts
@@ -0,0 +1,28 @@
+import { BigNumber } from '@0x/utils';
+
+export interface GenericRawOrder {
+    price: string;
+    amount: string;
+}
+
+/**
+ * Aggregates individual orders by price point. Filters zero amount orders.
+ * @param rawOrders An array of objects that have price and amount information.
+ */
+export function aggregateOrders(rawOrders: GenericRawOrder[]): Array<[string, BigNumber]> {
+    const aggregatedOrders = new Map<string, BigNumber>();
+    rawOrders.forEach(order => {
+        const amount = new BigNumber(order.amount);
+        if (amount.isZero()) {
+            return;
+        }
+        // Use string instead of BigNum to aggregate by value instead of variable.
+        // Convert to BigNumber first to consolidate different string
+        // representations of the same number. Eg. '0.0' and '0.00'.
+        const price = new BigNumber(order.price).toString();
+
+        const existingAmount = aggregatedOrders.get(price) || new BigNumber(0);
+        aggregatedOrders.set(price, amount.plus(existingAmount));
+    });
+    return Array.from(aggregatedOrders.entries());
+}
diff --git a/packages/pipeline/src/scripts/pull_idex_orderbook_snapshots.ts b/packages/pipeline/src/scripts/pull_idex_orderbook_snapshots.ts
new file mode 100644
index 000000000..d47c1dd3f
--- /dev/null
+++ b/packages/pipeline/src/scripts/pull_idex_orderbook_snapshots.ts
@@ -0,0 +1,63 @@
+import { logUtils } from '@0x/utils';
+import * as R from 'ramda';
+import { Connection, ConnectionOptions, createConnection } from 'typeorm';
+
+import { IDEX_SOURCE, IdexSource } from '../data_sources/idex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../entities';
+import * as ormConfig from '../ormconfig';
+import { parseIdexOrders } from '../parsers/idex_orders';
+import { handleError } from '../utils';
+
+// Number of orders to save at once.
+const BATCH_SAVE_SIZE = 1000;
+
+// Number of markets to retrieve orderbooks for at once.
+const MARKET_ORDERBOOK_REQUEST_BATCH_SIZE = 100;
+
+// Delay between market orderbook requests.
+const MILLISEC_MARKET_ORDERBOOK_REQUEST_DELAY = 2000;
+
+let connection: Connection;
+
+(async () => {
+    connection = await createConnection(ormConfig as ConnectionOptions);
+    const idexSource = new IdexSource();
+    logUtils.log('Getting all IDEX markets');
+    const markets = await idexSource.getMarketsAsync();
+    logUtils.log(`Got ${markets.length} markets.`);
+    for (const marketsChunk of R.splitEvery(MARKET_ORDERBOOK_REQUEST_BATCH_SIZE, markets)) {
+        await Promise.all(
+            marketsChunk.map(async (marketId: string) => getAndSaveMarketOrderbook(idexSource, marketId)),
+        );
+        await new Promise<void>(resolve => setTimeout(resolve, MILLISEC_MARKET_ORDERBOOK_REQUEST_DELAY));
+    }
+    process.exit(0);
+})().catch(handleError);
+
+/**
+ * Retrieve orderbook from Idex API for a given market. Parse orders and insert
+ * them into our database.
+ * @param idexSource Data source which can query Idex API.
+ * @param marketId String representing market of interest, eg. 'ETH_TIC'.
+ */
+async function getAndSaveMarketOrderbook(idexSource: IdexSource, marketId: string): Promise<void> {
+    logUtils.log(`${marketId}: Retrieving orderbook.`);
+    const orderBook = await idexSource.getMarketOrderbookAsync(marketId);
+    const observedTimestamp = Date.now();
+
+    if (!R.has('bids', orderBook) || !R.has('asks', orderBook)) {
+        logUtils.warn(`${marketId}: Orderbook faulty.`);
+        return;
+    }
+
+    logUtils.log(`${marketId}: Parsing orders.`);
+    const orders = parseIdexOrders(orderBook, observedTimestamp, IDEX_SOURCE);
+
+    if (orders.length > 0) {
+        logUtils.log(`${marketId}: Saving ${orders.length} orders.`);
+        const TokenOrderRepository = connection.getRepository(TokenOrder);
+        await TokenOrderRepository.save(orders, { chunk: Math.ceil(orders.length / BATCH_SAVE_SIZE) });
+    } else {
+        logUtils.log(`${marketId}: 0 orders to save.`);
+    }
+}
diff --git a/packages/pipeline/src/scripts/pull_oasis_orderbook_snapshots.ts b/packages/pipeline/src/scripts/pull_oasis_orderbook_snapshots.ts
new file mode 100644
index 000000000..0ffa5fd47
--- /dev/null
+++ b/packages/pipeline/src/scripts/pull_oasis_orderbook_snapshots.ts
@@ -0,0 +1,58 @@
+import { logUtils } from '@0x/utils';
+import * as R from 'ramda';
+import { Connection, ConnectionOptions, createConnection } from 'typeorm';
+
+import { OASIS_SOURCE, OasisMarket, OasisSource } from '../data_sources/oasis';
+import { TokenOrderbookSnapshot as TokenOrder } from '../entities';
+import * as ormConfig from '../ormconfig';
+import { parseOasisOrders } from '../parsers/oasis_orders';
+import { handleError } from '../utils';
+
+// Number of orders to save at once.
+const BATCH_SAVE_SIZE = 1000;
+
+// Number of markets to retrieve orderbooks for at once.
+const MARKET_ORDERBOOK_REQUEST_BATCH_SIZE = 50;
+
+// Delay between market orderbook requests.
+const MILLISEC_MARKET_ORDERBOOK_REQUEST_DELAY = 1000;
+
+let connection: Connection;
+
+(async () => {
+    connection = await createConnection(ormConfig as ConnectionOptions);
+    const oasisSource = new OasisSource();
+    logUtils.log('Getting all active Oasis markets');
+    const markets = await oasisSource.getActiveMarketsAsync();
+    logUtils.log(`Got ${markets.length} markets.`);
+    for (const marketsChunk of R.splitEvery(MARKET_ORDERBOOK_REQUEST_BATCH_SIZE, markets)) {
+        await Promise.all(
+            marketsChunk.map(async (market: OasisMarket) => getAndSaveMarketOrderbook(oasisSource, market)),
+        );
+        await new Promise<void>(resolve => setTimeout(resolve, MILLISEC_MARKET_ORDERBOOK_REQUEST_DELAY));
+    }
+    process.exit(0);
+})().catch(handleError);
+
+/**
+ * Retrieve orderbook from Oasis API for a given market. Parse orders and insert
+ * them into our database.
+ * @param oasisSource Data source which can query Oasis API.
+ * @param marketId String identifying market we want data for. eg. 'REPAUG'.
+ */
+async function getAndSaveMarketOrderbook(oasisSource: OasisSource, market: OasisMarket): Promise<void> {
+    logUtils.log(`${market.id}: Retrieving orderbook.`);
+    const orderBook = await oasisSource.getMarketOrderbookAsync(market.id);
+    const observedTimestamp = Date.now();
+
+    logUtils.log(`${market.id}: Parsing orders.`);
+    const orders = parseOasisOrders(orderBook, market, observedTimestamp, OASIS_SOURCE);
+
+    if (orders.length > 0) {
+        logUtils.log(`${market.id}: Saving ${orders.length} orders.`);
+        const TokenOrderRepository = connection.getRepository(TokenOrder);
+        await TokenOrderRepository.save(orders, { chunk: Math.ceil(orders.length / BATCH_SAVE_SIZE) });
+    } else {
+        logUtils.log(`${market.id}: 0 orders to save.`);
+    }
+}
diff --git a/packages/pipeline/test/parsers/ddex_orders/index_test.ts b/packages/pipeline/test/parsers/ddex_orders/index_test.ts
index 213100f44..9f4bfe7e3 100644
--- a/packages/pipeline/test/parsers/ddex_orders/index_test.ts
+++ b/packages/pipeline/test/parsers/ddex_orders/index_test.ts
@@ -4,7 +4,7 @@ import 'mocha';
 
 import { DdexMarket } from '../../../src/data_sources/ddex';
 import { TokenOrderbookSnapshot as TokenOrder } from '../../../src/entities';
-import { aggregateOrders, parseDdexOrder } from '../../../src/parsers/ddex_orders';
+import { parseDdexOrder } from '../../../src/parsers/ddex_orders';
 import { OrderType } from '../../../src/types';
 import { chaiSetup } from '../../utils/chai_setup';
 
@@ -13,19 +13,6 @@ const expect = chai.expect;
 
 // tslint:disable:custom-no-magic-numbers
 describe('ddex_orders', () => {
-    describe('aggregateOrders', () => {
-        it('aggregates orders by price point', () => {
-            const input = [
-                { price: '1', amount: '20', orderId: 'testtest' },
-                { price: '1', amount: '30', orderId: 'testone' },
-                { price: '2', amount: '100', orderId: 'testtwo' },
-            ];
-            const expected = [['1', new BigNumber(50)], ['2', new BigNumber(100)]];
-            const actual = aggregateOrders(input);
-            expect(actual).deep.equal(expected);
-        });
-    });
-
     describe('parseDdexOrder', () => {
         it('converts ddexOrder to TokenOrder entity', () => {
             const ddexOrder: [string, BigNumber] = ['0.5', new BigNumber(10)];
diff --git a/packages/pipeline/test/parsers/idex_orders/index_test.ts b/packages/pipeline/test/parsers/idex_orders/index_test.ts
new file mode 100644
index 000000000..d54ecb9a8
--- /dev/null
+++ b/packages/pipeline/test/parsers/idex_orders/index_test.ts
@@ -0,0 +1,87 @@
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { IdexOrderParam } from '../../../src/data_sources/idex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../../src/entities';
+import { parseIdexOrder } from '../../../src/parsers/idex_orders';
+import { OrderType } from '../../../src/types';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('idex_orders', () => {
+    describe('parseIdexOrder', () => {
+        // for market listed as 'DEF_ABC'.
+        it('correctly converts bid type idexOrder to TokenOrder entity', () => {
+            const idexOrder: [string, BigNumber] = ['0.5', new BigNumber(10)];
+            const idexOrderParam: IdexOrderParam = {
+                tokenBuy: '0x0000000000000000000000000000000000000000',
+                buySymbol: 'ABC',
+                buyPrecision: 2,
+                amountBuy: '10',
+                tokenSell: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+                sellSymbol: 'DEF',
+                sellPrecision: 2,
+                amountSell: '5',
+                expires: Date.now() + 100000,
+                nonce: 1,
+                user: '0x212345667543456435324564345643453453333',
+            };
+            const observedTimestamp: number = Date.now();
+            const orderType: OrderType = 'bid';
+            const source: string = 'idex';
+
+            const expected = new TokenOrder();
+            expected.source = 'idex';
+            expected.observedTimestamp = observedTimestamp;
+            expected.orderType = 'bid';
+            expected.price = new BigNumber(0.5);
+            expected.baseAssetSymbol = 'ABC';
+            expected.baseAssetAddress = '0x0000000000000000000000000000000000000000';
+            expected.baseVolume = new BigNumber(10);
+            expected.quoteAssetSymbol = 'DEF';
+            expected.quoteAssetAddress = '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81';
+            expected.quoteVolume = new BigNumber(5);
+
+            const actual = parseIdexOrder(idexOrderParam, observedTimestamp, orderType, source, idexOrder);
+            expect(actual).deep.equal(expected);
+        });
+        it('correctly converts ask type idexOrder to TokenOrder entity', () => {
+            const idexOrder: [string, BigNumber] = ['0.5', new BigNumber(10)];
+            const idexOrderParam: IdexOrderParam = {
+                tokenBuy: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+                buySymbol: 'DEF',
+                buyPrecision: 2,
+                amountBuy: '5',
+                tokenSell: '0x0000000000000000000000000000000000000000',
+                sellSymbol: 'ABC',
+                sellPrecision: 2,
+                amountSell: '10',
+                expires: Date.now() + 100000,
+                nonce: 1,
+                user: '0x212345667543456435324564345643453453333',
+            };
+            const observedTimestamp: number = Date.now();
+            const orderType: OrderType = 'ask';
+            const source: string = 'idex';
+
+            const expected = new TokenOrder();
+            expected.source = 'idex';
+            expected.observedTimestamp = observedTimestamp;
+            expected.orderType = 'ask';
+            expected.price = new BigNumber(0.5);
+            expected.baseAssetSymbol = 'ABC';
+            expected.baseAssetAddress = '0x0000000000000000000000000000000000000000';
+            expected.baseVolume = new BigNumber(10);
+            expected.quoteAssetSymbol = 'DEF';
+            expected.quoteAssetAddress = '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81';
+            expected.quoteVolume = new BigNumber(5);
+
+            const actual = parseIdexOrder(idexOrderParam, observedTimestamp, orderType, source, idexOrder);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/oasis_orders/index_test.ts b/packages/pipeline/test/parsers/oasis_orders/index_test.ts
new file mode 100644
index 000000000..9e8ba9a40
--- /dev/null
+++ b/packages/pipeline/test/parsers/oasis_orders/index_test.ts
@@ -0,0 +1,49 @@
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { OasisMarket } from '../../../src/data_sources/oasis';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../../src/entities';
+import { parseOasisOrder } from '../../../src/parsers/oasis_orders';
+import { OrderType } from '../../../src/types';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('oasis_orders', () => {
+    describe('parseOasisOrder', () => {
+        it('converts oasisOrder to TokenOrder entity', () => {
+            const oasisOrder: [string, BigNumber] = ['0.5', new BigNumber(10)];
+            const oasisMarket: OasisMarket = {
+                id: 'ABCDEF',
+                base: 'DEF',
+                quote: 'ABC',
+                buyVol: 100,
+                sellVol: 200,
+                price: 1,
+                high: 1,
+                low: 0,
+            };
+            const observedTimestamp: number = Date.now();
+            const orderType: OrderType = 'bid';
+            const source: string = 'oasis';
+
+            const expected = new TokenOrder();
+            expected.source = 'oasis';
+            expected.observedTimestamp = observedTimestamp;
+            expected.orderType = 'bid';
+            expected.price = new BigNumber(0.5);
+            expected.baseAssetSymbol = 'DEF';
+            expected.baseAssetAddress = null;
+            expected.baseVolume = new BigNumber(5);
+            expected.quoteAssetSymbol = 'ABC';
+            expected.quoteAssetAddress = null;
+            expected.quoteVolume = new BigNumber(10);
+
+            const actual = parseOasisOrder(oasisMarket, observedTimestamp, orderType, source, oasisOrder);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/utils/index_test.ts b/packages/pipeline/test/parsers/utils/index_test.ts
new file mode 100644
index 000000000..5a0d0f182
--- /dev/null
+++ b/packages/pipeline/test/parsers/utils/index_test.ts
@@ -0,0 +1,30 @@
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { aggregateOrders, GenericRawOrder } from '../../../src/parsers/utils';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('aggregateOrders', () => {
+    it('aggregates order by price point', () => {
+        const input = [
+            { price: '1', amount: '20', orderHash: 'testtest', total: '20' },
+            { price: '1', amount: '30', orderHash: 'testone', total: '30' },
+            { price: '2', amount: '100', orderHash: 'testtwo', total: '200' },
+        ];
+        const expected = [['1', new BigNumber(50)], ['2', new BigNumber(100)]];
+        const actual = aggregateOrders(input);
+        expect(actual).deep.equal(expected);
+    });
+
+    it('handles empty orders gracefully', () => {
+        const input: GenericRawOrder[] = [];
+        const expected: Array<[string, BigNumber]> = [];
+        const actual = aggregateOrders(input);
+        expect(actual).deep.equal(expected);
+    });
+});