9 files changed, 562 insertions, 0 deletions
diff --git a/packages/pipeline/src/parsers/bloxy/index.ts b/packages/pipeline/src/parsers/bloxy/index.ts
new file mode 100644
index 000000000..caa55d289
--- /dev/null
+++ b/packages/pipeline/src/parsers/bloxy/index.ts
@@ -0,0 +1,53 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { BLOXY_DEX_TRADES_URL, BloxyTrade } from '../../data_sources/bloxy';
+import { DexTrade } from '../../entities';
+
+/**
+ * Parses a raw trades response from the Bloxy Dex API and returns an array of
+ * DexTrade entities.
+ * @param rawTrades A raw order response from an SRA endpoint.
+ */
+export function parseBloxyTrades(rawTrades: BloxyTrade[]): DexTrade[] {
+    return R.map(_parseBloxyTrade, rawTrades);
+}
+
+/**
+ * Converts a single Bloxy trade into a DexTrade entity.
+ * @param rawTrade A single trade from the response from the Bloxy API.
+ */
+export function _parseBloxyTrade(rawTrade: BloxyTrade): DexTrade {
+    const dexTrade = new DexTrade();
+    dexTrade.sourceUrl = BLOXY_DEX_TRADES_URL;
+    dexTrade.txHash = rawTrade.tx_hash;
+    dexTrade.txTimestamp = new Date(rawTrade.tx_time).getTime();
+    dexTrade.txDate = rawTrade.tx_date;
+    dexTrade.txSender = rawTrade.tx_sender;
+    dexTrade.smartContractId = rawTrade.smart_contract_id;
+    dexTrade.smartContractAddress = rawTrade.smart_contract_address;
+    dexTrade.contractType = rawTrade.contract_type;
+    dexTrade.maker = rawTrade.maker;
+    dexTrade.taker = rawTrade.taker;
+    // TODO(albrow): The Bloxy API returns amounts and fees as a `number` type
+    // but some of their values have too many significant digits to be
+    // represented that way. Ideally they will switch to using strings and then
+    // we can update this code.
+    dexTrade.amountBuy = new BigNumber(rawTrade.amountBuy.toString());
+    dexTrade.makerFeeAmount = new BigNumber(rawTrade.makerFee.toString());
+    dexTrade.buyCurrencyId = rawTrade.buyCurrencyId;
+    dexTrade.buySymbol = filterNullCharacters(rawTrade.buySymbol);
+    dexTrade.amountSell = new BigNumber(rawTrade.amountSell.toString());
+    dexTrade.takerFeeAmount = new BigNumber(rawTrade.takerFee.toString());
+    dexTrade.sellCurrencyId = rawTrade.sellCurrencyId;
+    dexTrade.sellSymbol = filterNullCharacters(rawTrade.sellSymbol);
+    dexTrade.makerAnnotation = rawTrade.maker_annotation;
+    dexTrade.takerAnnotation = rawTrade.taker_annotation;
+    dexTrade.protocol = rawTrade.protocol;
+    dexTrade.buyAddress = rawTrade.buyAddress;
+    dexTrade.sellAddress = rawTrade.sellAddress;
+    return dexTrade;
+}
+
+// Works with any form of escaped null character (e.g., '\0' and '\u0000').
+const filterNullCharacters = R.replace(/\0/g, '');
diff --git a/packages/pipeline/src/parsers/ddex_orders/index.ts b/packages/pipeline/src/parsers/ddex_orders/index.ts
new file mode 100644
index 000000000..81132e8f0
--- /dev/null
+++ b/packages/pipeline/src/parsers/ddex_orders/index.ts
@@ -0,0 +1,77 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { DdexMarket, DdexOrder, DdexOrderbook } from '../../data_sources/ddex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook,
+ * other information attached.
+ * @param ddexOrderbook A raw orderbook that we pull from the Ddex API.
+ * @param ddexMarket 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 'ddex'.
+ */
+export function parseDdexOrders(
+    ddexOrderbook: DdexOrderbook,
+    ddexMarket: DdexMarket,
+    observedTimestamp: number,
+    source: string,
+): TokenOrder[] {
+    const aggregatedBids = aggregateOrders(ddexOrderbook.bids);
+    const aggregatedAsks = aggregateOrders(ddexOrderbook.asks);
+    const parsedBids = aggregatedBids.map(order => parseDdexOrder(ddexMarket, observedTimestamp, 'bid', source, order));
+    const parsedAsks = aggregatedAsks.map(order => parseDdexOrder(ddexMarket, observedTimestamp, 'ask', source, order));
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * 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
+ * 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 ddexOrder A <price, amount> tuple which we will convert to volume-basis.
+ */
+export function parseDdexOrder(
+    ddexMarket: DdexMarket,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    ddexOrder: [string, BigNumber],
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(ddexOrder[0]);
+    const amount = ddexOrder[1];
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+
+    tokenOrder.baseAssetSymbol = ddexMarket.baseToken;
+    tokenOrder.baseAssetAddress = ddexMarket.baseTokenAddress;
+    tokenOrder.baseVolume = price.times(amount);
+
+    tokenOrder.quoteAssetSymbol = ddexMarket.quoteToken;
+    tokenOrder.quoteAssetAddress = ddexMarket.quoteTokenAddress;
+    tokenOrder.quoteVolume = amount;
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/events/index.ts b/packages/pipeline/src/parsers/events/index.ts
new file mode 100644
index 000000000..e18106c75
--- /dev/null
+++ b/packages/pipeline/src/parsers/events/index.ts
@@ -0,0 +1,133 @@
+import { ExchangeCancelEventArgs, ExchangeCancelUpToEventArgs, ExchangeFillEventArgs } from '@0x/contract-wrappers';
+import { assetDataUtils } from '@0x/order-utils';
+import { AssetProxyId, ERC721AssetData } from '@0x/types';
+import { LogWithDecodedArgs } from 'ethereum-types';
+import * as R from 'ramda';
+
+import { ExchangeCancelEvent, ExchangeCancelUpToEvent, ExchangeFillEvent } from '../../entities';
+import { bigNumbertoStringOrNull } from '../../utils';
+
+/**
+ * Parses raw event logs for a fill event and returns an array of
+ * ExchangeFillEvent entities.
+ * @param eventLogs Raw event logs (e.g. returned from contract-wrappers).
+ */
+export const parseExchangeFillEvents: (
+    eventLogs: Array<LogWithDecodedArgs<ExchangeFillEventArgs>>,
+) => ExchangeFillEvent[] = R.map(_convertToExchangeFillEvent);
+
+/**
+ * Parses raw event logs for a cancel event and returns an array of
+ * ExchangeCancelEvent entities.
+ * @param eventLogs Raw event logs (e.g. returned from contract-wrappers).
+ */
+export const parseExchangeCancelEvents: (
+    eventLogs: Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>,
+) => ExchangeCancelEvent[] = R.map(_convertToExchangeCancelEvent);
+
+/**
+ * Parses raw event logs for a CancelUpTo event and returns an array of
+ * ExchangeCancelUpToEvent entities.
+ * @param eventLogs Raw event logs (e.g. returned from contract-wrappers).
+ */
+export const parseExchangeCancelUpToEvents: (
+    eventLogs: Array<LogWithDecodedArgs<ExchangeCancelUpToEventArgs>>,
+) => ExchangeCancelUpToEvent[] = R.map(_convertToExchangeCancelUpToEvent);
+
+/**
+ * Converts a raw event log for a fill event into an ExchangeFillEvent entity.
+ * @param eventLog Raw event log (e.g. returned from contract-wrappers).
+ */
+export function _convertToExchangeFillEvent(eventLog: LogWithDecodedArgs<ExchangeFillEventArgs>): ExchangeFillEvent {
+    const makerAssetData = assetDataUtils.decodeAssetDataOrThrow(eventLog.args.makerAssetData);
+    const makerAssetType = makerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+    const takerAssetData = assetDataUtils.decodeAssetDataOrThrow(eventLog.args.takerAssetData);
+    const takerAssetType = takerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+    const exchangeFillEvent = new ExchangeFillEvent();
+    exchangeFillEvent.contractAddress = eventLog.address as string;
+    exchangeFillEvent.blockNumber = eventLog.blockNumber as number;
+    exchangeFillEvent.logIndex = eventLog.logIndex as number;
+    exchangeFillEvent.rawData = eventLog.data as string;
+    exchangeFillEvent.transactionHash = eventLog.transactionHash;
+    exchangeFillEvent.makerAddress = eventLog.args.makerAddress;
+    exchangeFillEvent.takerAddress = eventLog.args.takerAddress;
+    exchangeFillEvent.feeRecipientAddress = eventLog.args.feeRecipientAddress;
+    exchangeFillEvent.senderAddress = eventLog.args.senderAddress;
+    exchangeFillEvent.makerAssetFilledAmount = eventLog.args.makerAssetFilledAmount;
+    exchangeFillEvent.takerAssetFilledAmount = eventLog.args.takerAssetFilledAmount;
+    exchangeFillEvent.makerFeePaid = eventLog.args.makerFeePaid;
+    exchangeFillEvent.takerFeePaid = eventLog.args.takerFeePaid;
+    exchangeFillEvent.orderHash = eventLog.args.orderHash;
+    exchangeFillEvent.rawMakerAssetData = eventLog.args.makerAssetData;
+    exchangeFillEvent.makerAssetType = makerAssetType;
+    exchangeFillEvent.makerAssetProxyId = makerAssetData.assetProxyId;
+    exchangeFillEvent.makerTokenAddress = makerAssetData.tokenAddress;
+    // tslint has a false positive here. Type assertion is required.
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    exchangeFillEvent.makerTokenId = bigNumbertoStringOrNull((makerAssetData as ERC721AssetData).tokenId);
+    exchangeFillEvent.rawTakerAssetData = eventLog.args.takerAssetData;
+    exchangeFillEvent.takerAssetType = takerAssetType;
+    exchangeFillEvent.takerAssetProxyId = takerAssetData.assetProxyId;
+    exchangeFillEvent.takerTokenAddress = takerAssetData.tokenAddress;
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    exchangeFillEvent.takerTokenId = bigNumbertoStringOrNull((takerAssetData as ERC721AssetData).tokenId);
+    return exchangeFillEvent;
+}
+
+/**
+ * Converts a raw event log for a cancel event into an ExchangeCancelEvent
+ * entity.
+ * @param eventLog Raw event log (e.g. returned from contract-wrappers).
+ */
+export function _convertToExchangeCancelEvent(
+    eventLog: LogWithDecodedArgs<ExchangeCancelEventArgs>,
+): ExchangeCancelEvent {
+    const makerAssetData = assetDataUtils.decodeAssetDataOrThrow(eventLog.args.makerAssetData);
+    const makerAssetType = makerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+    const takerAssetData = assetDataUtils.decodeAssetDataOrThrow(eventLog.args.takerAssetData);
+    const takerAssetType = takerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+    const exchangeCancelEvent = new ExchangeCancelEvent();
+    exchangeCancelEvent.contractAddress = eventLog.address as string;
+    exchangeCancelEvent.blockNumber = eventLog.blockNumber as number;
+    exchangeCancelEvent.logIndex = eventLog.logIndex as number;
+    exchangeCancelEvent.rawData = eventLog.data as string;
+    exchangeCancelEvent.transactionHash = eventLog.transactionHash;
+    exchangeCancelEvent.makerAddress = eventLog.args.makerAddress;
+    exchangeCancelEvent.takerAddress = eventLog.args.takerAddress;
+    exchangeCancelEvent.feeRecipientAddress = eventLog.args.feeRecipientAddress;
+    exchangeCancelEvent.senderAddress = eventLog.args.senderAddress;
+    exchangeCancelEvent.orderHash = eventLog.args.orderHash;
+    exchangeCancelEvent.rawMakerAssetData = eventLog.args.makerAssetData;
+    exchangeCancelEvent.makerAssetType = makerAssetType;
+    exchangeCancelEvent.makerAssetProxyId = makerAssetData.assetProxyId;
+    exchangeCancelEvent.makerTokenAddress = makerAssetData.tokenAddress;
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    exchangeCancelEvent.makerTokenId = bigNumbertoStringOrNull((makerAssetData as ERC721AssetData).tokenId);
+    exchangeCancelEvent.rawTakerAssetData = eventLog.args.takerAssetData;
+    exchangeCancelEvent.takerAssetType = takerAssetType;
+    exchangeCancelEvent.takerAssetProxyId = takerAssetData.assetProxyId;
+    exchangeCancelEvent.takerTokenAddress = takerAssetData.tokenAddress;
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    exchangeCancelEvent.takerTokenId = bigNumbertoStringOrNull((takerAssetData as ERC721AssetData).tokenId);
+    return exchangeCancelEvent;
+}
+
+/**
+ * Converts a raw event log for a cancelUpTo event into an
+ * ExchangeCancelUpToEvent entity.
+ * @param eventLog Raw event log (e.g. returned from contract-wrappers).
+ */
+export function _convertToExchangeCancelUpToEvent(
+    eventLog: LogWithDecodedArgs<ExchangeCancelUpToEventArgs>,
+): ExchangeCancelUpToEvent {
+    const exchangeCancelUpToEvent = new ExchangeCancelUpToEvent();
+    exchangeCancelUpToEvent.contractAddress = eventLog.address as string;
+    exchangeCancelUpToEvent.blockNumber = eventLog.blockNumber as number;
+    exchangeCancelUpToEvent.logIndex = eventLog.logIndex as number;
+    exchangeCancelUpToEvent.rawData = eventLog.data as string;
+    exchangeCancelUpToEvent.transactionHash = eventLog.transactionHash;
+    exchangeCancelUpToEvent.makerAddress = eventLog.args.makerAddress;
+    exchangeCancelUpToEvent.senderAddress = eventLog.args.senderAddress;
+    exchangeCancelUpToEvent.orderEpoch = eventLog.args.orderEpoch;
+    return exchangeCancelUpToEvent;
+}
diff --git a/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts b/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts
new file mode 100644
index 000000000..3efb90384
--- /dev/null
+++ b/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts
@@ -0,0 +1,38 @@
+import { CryptoCompareOHLCVRecord } from '../../data_sources/ohlcv_external/crypto_compare';
+import { OHLCVExternal } from '../../entities';
+
+const ONE_SECOND = 1000; // Crypto Compare uses timestamps in seconds instead of milliseconds
+
+export interface OHLCVMetadata {
+    exchange: string;
+    fromSymbol: string;
+    toSymbol: string;
+    source: string;
+    observedTimestamp: number;
+    interval: number;
+}
+/**
+ * Parses OHLCV records from Crypto Compare into an array of OHLCVExternal entities
+ * @param rawRecords an array of OHLCV records from Crypto Compare (not the full response)
+ */
+export function parseRecords(rawRecords: CryptoCompareOHLCVRecord[], metadata: OHLCVMetadata): OHLCVExternal[] {
+    return rawRecords.map(rec => {
+        const ohlcvRecord = new OHLCVExternal();
+        ohlcvRecord.exchange = metadata.exchange;
+        ohlcvRecord.fromSymbol = metadata.fromSymbol;
+        ohlcvRecord.toSymbol = metadata.toSymbol;
+        ohlcvRecord.startTime = rec.time * ONE_SECOND - metadata.interval;
+        ohlcvRecord.endTime = rec.time * ONE_SECOND;
+
+        ohlcvRecord.open = rec.open;
+        ohlcvRecord.close = rec.close;
+        ohlcvRecord.low = rec.low;
+        ohlcvRecord.high = rec.high;
+        ohlcvRecord.volumeFrom = rec.volumefrom;
+        ohlcvRecord.volumeTo = rec.volumeto;
+
+        ohlcvRecord.source = metadata.source;
+        ohlcvRecord.observedTimestamp = metadata.observedTimestamp;
+        return ohlcvRecord;
+    });
+}
diff --git a/packages/pipeline/src/parsers/paradex_orders/index.ts b/packages/pipeline/src/parsers/paradex_orders/index.ts
new file mode 100644
index 000000000..7966658a7
--- /dev/null
+++ b/packages/pipeline/src/parsers/paradex_orders/index.ts
@@ -0,0 +1,66 @@
+import { BigNumber } from '@0x/utils';
+
+import { ParadexMarket, ParadexOrder, ParadexOrderbookResponse } from '../../data_sources/paradex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook (orders are already aggregated by price point),
+ * 2) For each aggregated order, forms a TokenOrder entity with market data and
+ * other information attached.
+ * @param paradexOrderbookResponse An orderbook response from the Paradex API.
+ * @param paradexMarket 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 'paradex'.
+ */
+export function parseParadexOrders(
+    paradexOrderbookResponse: ParadexOrderbookResponse,
+    paradexMarket: ParadexMarket,
+    observedTimestamp: number,
+    source: string,
+): TokenOrder[] {
+    const parsedBids = paradexOrderbookResponse.bids.map(order =>
+        parseParadexOrder(paradexMarket, observedTimestamp, 'bid', source, order),
+    );
+    const parsedAsks = paradexOrderbookResponse.asks.map(order =>
+        parseParadexOrder(paradexMarket, observedTimestamp, 'ask', source, order),
+    );
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * Parse a single aggregated Ddex order in order to form a tokenOrder entity
+ * which can be saved into the database.
+ * @param paradexMarket An object containing information about the market where these
+ * orders 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 paradexOrder A ParadexOrder object; basically price, amount tuple.
+ */
+export function parseParadexOrder(
+    paradexMarket: ParadexMarket,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    paradexOrder: ParadexOrder,
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(paradexOrder.price);
+    const amount = new BigNumber(paradexOrder.amount);
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+
+    tokenOrder.baseAssetSymbol = paradexMarket.baseToken;
+    tokenOrder.baseAssetAddress = paradexMarket.baseTokenAddress as string;
+    tokenOrder.baseVolume = price.times(amount);
+
+    tokenOrder.quoteAssetSymbol = paradexMarket.quoteToken;
+    tokenOrder.quoteAssetAddress = paradexMarket.quoteTokenAddress as string;
+    tokenOrder.quoteVolume = amount;
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/relayer_registry/index.ts b/packages/pipeline/src/parsers/relayer_registry/index.ts
new file mode 100644
index 000000000..9723880a4
--- /dev/null
+++ b/packages/pipeline/src/parsers/relayer_registry/index.ts
@@ -0,0 +1,37 @@
+import * as R from 'ramda';
+
+import { RelayerResponse, RelayerResponseNetwork } from '../../data_sources/relayer-registry';
+import { Relayer } from '../../entities';
+
+/**
+ * Parses a raw relayer registry response into an array of Relayer entities.
+ * @param rawResp raw response from the relayer-registry json file.
+ */
+export function parseRelayers(rawResp: Map<string, RelayerResponse>): Relayer[] {
+    const parsedAsObject = R.mapObjIndexed(parseRelayer, rawResp);
+    return R.values(parsedAsObject);
+}
+
+function parseRelayer(relayerResp: RelayerResponse, uuid: string): Relayer {
+    const relayer = new Relayer();
+    relayer.uuid = uuid;
+    relayer.name = relayerResp.name;
+    relayer.homepageUrl = relayerResp.homepage_url;
+    relayer.appUrl = relayerResp.app_url;
+    const mainNetworkRelayerInfo = getMainNetwork(relayerResp);
+    if (mainNetworkRelayerInfo !== undefined) {
+        relayer.sraHttpEndpoint = mainNetworkRelayerInfo.sra_http_endpoint || null;
+        relayer.sraWsEndpoint = mainNetworkRelayerInfo.sra_ws_endpoint || null;
+        relayer.feeRecipientAddresses =
+            R.path(['static_order_fields', 'fee_recipient_addresses'], mainNetworkRelayerInfo) || [];
+        relayer.takerAddresses = R.path(['static_order_fields', 'taker_addresses'], mainNetworkRelayerInfo) || [];
+    } else {
+        relayer.feeRecipientAddresses = [];
+        relayer.takerAddresses = [];
+    }
+    return relayer;
+}
+
+function getMainNetwork(relayerResp: RelayerResponse): RelayerResponseNetwork | undefined {
+    return R.find(network => network.networkId === 1, relayerResp.networks);
+}
diff --git a/packages/pipeline/src/parsers/sra_orders/index.ts b/packages/pipeline/src/parsers/sra_orders/index.ts
new file mode 100644
index 000000000..ef8901e40
--- /dev/null
+++ b/packages/pipeline/src/parsers/sra_orders/index.ts
@@ -0,0 +1,62 @@
+import { APIOrder, OrdersResponse } from '@0x/connect';
+import { assetDataUtils, orderHashUtils } from '@0x/order-utils';
+import { AssetProxyId, ERC721AssetData } from '@0x/types';
+import * as R from 'ramda';
+
+import { SraOrder } from '../../entities';
+import { bigNumbertoStringOrNull } from '../../utils';
+
+/**
+ * Parses a raw order response from an SRA endpoint and returns an array of
+ * SraOrder entities.
+ * @param rawOrdersResponse A raw order response from an SRA endpoint.
+ */
+export function parseSraOrders(rawOrdersResponse: OrdersResponse): SraOrder[] {
+    return R.map(_convertToEntity, rawOrdersResponse.records);
+}
+
+/**
+ * Converts a single APIOrder into an SraOrder entity.
+ * @param apiOrder A single order from the response from an SRA endpoint.
+ */
+export function _convertToEntity(apiOrder: APIOrder): SraOrder {
+    // TODO(albrow): refactor out common asset data decoding code.
+    const makerAssetData = assetDataUtils.decodeAssetDataOrThrow(apiOrder.order.makerAssetData);
+    const makerAssetType = makerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+    const takerAssetData = assetDataUtils.decodeAssetDataOrThrow(apiOrder.order.takerAssetData);
+    const takerAssetType = takerAssetData.assetProxyId === AssetProxyId.ERC20 ? 'erc20' : 'erc721';
+
+    const sraOrder = new SraOrder();
+    sraOrder.exchangeAddress = apiOrder.order.exchangeAddress;
+    sraOrder.orderHashHex = orderHashUtils.getOrderHashHex(apiOrder.order);
+
+    sraOrder.makerAddress = apiOrder.order.makerAddress;
+    sraOrder.takerAddress = apiOrder.order.takerAddress;
+    sraOrder.feeRecipientAddress = apiOrder.order.feeRecipientAddress;
+    sraOrder.senderAddress = apiOrder.order.senderAddress;
+    sraOrder.makerAssetAmount = apiOrder.order.makerAssetAmount;
+    sraOrder.takerAssetAmount = apiOrder.order.takerAssetAmount;
+    sraOrder.makerFee = apiOrder.order.makerFee;
+    sraOrder.takerFee = apiOrder.order.takerFee;
+    sraOrder.expirationTimeSeconds = apiOrder.order.expirationTimeSeconds;
+    sraOrder.salt = apiOrder.order.salt;
+    sraOrder.signature = apiOrder.order.signature;
+
+    sraOrder.rawMakerAssetData = apiOrder.order.makerAssetData;
+    sraOrder.makerAssetType = makerAssetType;
+    sraOrder.makerAssetProxyId = makerAssetData.assetProxyId;
+    sraOrder.makerTokenAddress = makerAssetData.tokenAddress;
+    // tslint has a false positive here. Type assertion is required.
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    sraOrder.makerTokenId = bigNumbertoStringOrNull((makerAssetData as ERC721AssetData).tokenId);
+    sraOrder.rawTakerAssetData = apiOrder.order.takerAssetData;
+    sraOrder.takerAssetType = takerAssetType;
+    sraOrder.takerAssetProxyId = takerAssetData.assetProxyId;
+    sraOrder.takerTokenAddress = takerAssetData.tokenAddress;
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    sraOrder.takerTokenId = bigNumbertoStringOrNull((takerAssetData as ERC721AssetData).tokenId);
+
+    sraOrder.metadataJson = JSON.stringify(apiOrder.metaData);
+
+    return sraOrder;
+}
diff --git a/packages/pipeline/src/parsers/token_metadata/index.ts b/packages/pipeline/src/parsers/token_metadata/index.ts
new file mode 100644
index 000000000..f258af063
--- /dev/null
+++ b/packages/pipeline/src/parsers/token_metadata/index.ts
@@ -0,0 +1,47 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { MetamaskTrustedTokenMeta, ZeroExTrustedTokenMeta } from '../../data_sources/trusted_tokens';
+import { TokenMetadata } from '../../entities';
+import {} from '../../utils';
+
+/**
+ * Parses Metamask's trusted tokens list.
+ * @param rawResp raw response from the metamask json file.
+ */
+export function parseMetamaskTrustedTokens(rawResp: Map<string, MetamaskTrustedTokenMeta>): TokenMetadata[] {
+    const parsedAsObject = R.mapObjIndexed(parseMetamaskTrustedToken, rawResp);
+    return R.values(parsedAsObject);
+}
+
+/**
+ * Parses 0x's trusted tokens list.
+ * @param rawResp raw response from the 0x trusted tokens file.
+ */
+export function parseZeroExTrustedTokens(rawResp: ZeroExTrustedTokenMeta[]): TokenMetadata[] {
+    return R.map(parseZeroExTrustedToken, rawResp);
+}
+
+function parseMetamaskTrustedToken(resp: MetamaskTrustedTokenMeta, address: string): TokenMetadata {
+    const trustedToken = new TokenMetadata();
+
+    trustedToken.address = address;
+    trustedToken.decimals = new BigNumber(resp.decimals);
+    trustedToken.symbol = resp.symbol;
+    trustedToken.name = resp.name;
+    trustedToken.authority = 'metamask';
+
+    return trustedToken;
+}
+
+function parseZeroExTrustedToken(resp: ZeroExTrustedTokenMeta): TokenMetadata {
+    const trustedToken = new TokenMetadata();
+
+    trustedToken.address = resp.address;
+    trustedToken.decimals = resp.decimals ? new BigNumber(resp.decimals) : null;
+    trustedToken.symbol = resp.symbol;
+    trustedToken.name = resp.name;
+    trustedToken.authority = '0x';
+
+    return trustedToken;
+}
diff --git a/packages/pipeline/src/parsers/web3/index.ts b/packages/pipeline/src/parsers/web3/index.ts
new file mode 100644
index 000000000..f986efc59
--- /dev/null
+++ b/packages/pipeline/src/parsers/web3/index.ts
@@ -0,0 +1,49 @@
+import { BigNumber } from '@0x/utils';
+import { BlockWithoutTransactionData, Transaction as EthTransaction } from 'ethereum-types';
+
+import { Block, Transaction } from '../../entities';
+
+const MILLISECONDS_PER_SECOND = 1000;
+
+/**
+ * Parses a raw block and returns a Block entity.
+ * @param rawBlock a raw block (e.g. returned from web3-wrapper).
+ */
+export function parseBlock(rawBlock: BlockWithoutTransactionData): Block {
+    if (rawBlock.hash == null) {
+        throw new Error('Tried to parse raw block but hash was null');
+    }
+    if (rawBlock.number == null) {
+        throw new Error('Tried to parse raw block but number was null');
+    }
+
+    const block = new Block();
+    block.hash = rawBlock.hash;
+    block.number = rawBlock.number;
+    // Block timestamps are in seconds, but we use milliseconds everywhere else.
+    block.timestamp = rawBlock.timestamp * MILLISECONDS_PER_SECOND;
+    return block;
+}
+
+/**
+ * Parses a raw transaction and returns a Transaction entity.
+ * @param rawBlock a raw transaction (e.g. returned from web3-wrapper).
+ */
+export function parseTransaction(rawTransaction: EthTransaction): Transaction {
+    if (rawTransaction.blockHash == null) {
+        throw new Error('Tried to parse raw transaction but blockHash was null');
+    }
+    if (rawTransaction.blockNumber == null) {
+        throw new Error('Tried to parse raw transaction but blockNumber was null');
+    }
+
+    const tx = new Transaction();
+    tx.transactionHash = rawTransaction.hash;
+    tx.blockHash = rawTransaction.blockHash;
+    tx.blockNumber = rawTransaction.blockNumber;
+
+    tx.gasUsed = new BigNumber(rawTransaction.gas);
+    tx.gasPrice = rawTransaction.gasPrice;
+
+    return tx;
+}