Integrate one-time dump and API for nonfungible.com (#1603)

* Add script for pulling NFT trade data from nonfungible.com

* corrections for current state of API

* change data model to match data source

* change primary key

* pull data from initial dump first, then API

* pull all supported NFT's, not just cryptokitties

* disable problematic data sources

* rename function to satisfy linter

* Rename table to nonfungible_dot_com_trades

* rename parser module to nonfungible_dot_com from non_fungible_dot_com, for consistency

* correct mistaken reference to Bloxy

* rename NonfungibleDotComTrade to ...TradeResponse

* `NftTrade` -> `NonfungibleDotComTrade`

* rename files to match prior object renaming

* use fetchAsync instead of axios

* improve fetchAsync error message: include URL

* avoid non-null contraints in API trades too, not just for trades from the one-time dump

* disable mythereum publisher

packages/pipeline/migrations/1543540108767-CreateNftTrades.ts

@@ -0,0 +1,31 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const nftTrades = new Table({
+    name: 'raw.nonfungible_dot_com_trades',
+    columns: [
+        { name: 'publisher', type: 'varchar', isPrimary: true },
+        { name: 'transaction_hash', type: 'varchar', isPrimary: true },
+        { name: 'asset_id', type: 'varchar', isPrimary: true },
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+        { name: 'log_index', type: 'integer', isPrimary: true },
+
+        { name: 'block_timestamp', type: 'bigint' },
+        { name: 'asset_descriptor', type: 'varchar' },
+        { name: 'market_address', type: 'varchar(42)' },
+        { name: 'total_price', type: 'numeric' },
+        { name: 'usd_price', type: 'numeric' },
+        { name: 'buyer_address', type: 'varchar(42)' },
+        { name: 'seller_address', type: 'varchar(42)' },
+        { name: 'meta', type: 'jsonb' },
+    ],
+});
+
+export class CreateNftTrades1543540108767 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.createTable(nftTrades);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(nftTrades);
+    }
+}

packages/pipeline/src/data_sources/nonfungible_dot_com/index.ts

@@ -0,0 +1,220 @@
+import { stringify } from 'querystring';
+
+import { logUtils } from '@0x/utils';
+
+import { fetchSuccessfullyOrThrowAsync } from '../../utils';
+
+// URL to use for getting nft trades from nonfungible.com.
+export const NONFUNGIBLE_DOT_COM_URL = 'https://nonfungible.com/api/v1';
+// Number of trades to get at once. This is a hard limit enforced by the API.
+const MAX_TRADES_PER_QUERY = 100;
+
+// Note(albrow): For now this will have to be manually updated by checking
+// https://nonfungible.com/
+export const knownPublishers = [
+    'axieinfinity',
+    // 'cryptokitties', // disabled until we get updated initial dump that isn't truncated
+    'cryptopunks',
+    'cryptovoxels',
+    'decentraland',
+    'decentraland_estate',
+    'etherbots',
+    'etheremon',
+    'ethtown',
+    // 'knownorigin', // disabled because of null characters in data being rejected by postgres
+    // 'mythereum', // worked at one time, but now seems dead
+    'superrare',
+];
+
+export interface NonfungibleDotComHistoryResponse {
+    data: NonfungibleDotComTradeResponse[];
+}
+
+export interface NonfungibleDotComTradeResponse {
+    _id: string;
+    transactionHash: string;
+    blockNumber: number;
+    logIndex: number;
+    blockTimestamp: string;
+    assetId: string;
+    assetDescriptor: string;
+    nftAddress: string;
+    marketAddress: string;
+    tokenTicker: string;
+    totalDecimalPrice: number;
+    totalPrice: string;
+    usdPrice: number;
+    currencyTransfer: object;
+    buyer: string;
+    seller: string;
+    meta: object;
+    image: string;
+    composedOf: string;
+    asset_link: string;
+    seller_address_link: string;
+    buyer_address_link: string;
+}
+
+/**
+ * Gets and returns all trades for the given publisher, starting at the given block number.
+ * Automatically handles pagination.
+ * @param publisher A valid "publisher" for the nonfungible.com API. (e.g. "cryptokitties")
+ * @param blockNumberStart The block number to start querying from.
+ */
+export async function getTradesAsync(
+    publisher: string,
+    blockNumberStart: number,
+): Promise<NonfungibleDotComTradeResponse[]> {
+    const allTrades: NonfungibleDotComTradeResponse[] = [];
+
+    /**
+     * due to high data volumes and rate limiting, we procured an initial data
+     * dump from nonfungible.com.  If the requested starting block number is
+     * contained in that initial dump, then pull relevant trades from there
+     * first.  Later (below) we'll get the more recent trades from the API itself.
+     */
+
+    if (blockNumberStart < highestBlockNumbersInIntialDump[publisher]) {
+        logUtils.log('getting trades from one-time dump');
+        // caller needs trades that are in the initial data dump, so get them
+        // from there, then later go to the API for the rest.
+        const initialDumpResponse: NonfungibleDotComHistoryResponse = await fetchSuccessfullyOrThrowAsync(
+            getInitialDumpUrl(publisher),
+        );
+        const initialDumpTrades = initialDumpResponse.data;
+        for (const initialDumpTrade of initialDumpTrades) {
+            if (!shouldProcessTrade(initialDumpTrade, allTrades)) {
+                continue;
+            }
+
+            ensureNonNull(initialDumpTrade);
+
+            allTrades.push(initialDumpTrade);
+        }
+        logUtils.log(`got ${allTrades.length} from one-time dump`);
+    }
+
+    const fullUrl = getFullUrlForPublisher(publisher);
+
+    /**
+     * API returns trades in reverse chronological order, so highest block
+     * numbers first.  The `start` query parameter indicates how far back in
+     * time (in number of trades) the results should start.  Here we iterate
+     * over both start parameter values and block numbers simultaneously.
+     * Start parameter values count up from zero.  Block numbers count down
+     * until reaching the highest block number in the initial dump.
+     */
+
+    const blockNumberStop = Math.max(highestBlockNumbersInIntialDump[publisher] + 1, blockNumberStart);
+    for (
+        let startParam = 0, blockNumber = Number.MAX_SAFE_INTEGER;
+        blockNumber > blockNumberStop;
+        startParam += MAX_TRADES_PER_QUERY
+    ) {
+        const response = await _getTradesWithOffsetAsync(fullUrl, publisher, startParam);
+        const tradesFromApi = response.data;
+        logUtils.log(
+            `got ${
+                tradesFromApi.length
+            } trades from API. blockNumber=${blockNumber}. blockNumberStop=${blockNumberStop}`,
+        );
+        for (const tradeFromApi of tradesFromApi) {
+            if (tradeFromApi.blockNumber <= blockNumberStop) {
+                blockNumber = blockNumberStop;
+                break;
+            }
+            if (!shouldProcessTrade(tradeFromApi, allTrades)) {
+                continue;
+            }
+            ensureNonNull(tradeFromApi);
+            allTrades.push(tradeFromApi);
+            blockNumber = tradeFromApi.blockNumber;
+        }
+    }
+
+    return allTrades;
+}
+
+function shouldProcessTrade(
+    trade: NonfungibleDotComTradeResponse,
+    existingTrades: NonfungibleDotComTradeResponse[],
+): boolean {
+    // check to see if this trade is already in existingTrades
+    const existingTradeIndex = existingTrades.findIndex(
+        // HACK! making assumptions about composition of primary key
+        e =>
+            e.transactionHash === trade.transactionHash &&
+            e.logIndex === trade.logIndex &&
+            e.blockNumber === trade.blockNumber,
+    );
+    if (existingTradeIndex !== -1) {
+        logUtils.log("we've already captured this trade. deciding whether to use the existing record or this one.");
+        if (trade.blockNumber > existingTrades[existingTradeIndex].blockNumber) {
+            logUtils.log('throwing out existing trade');
+            existingTrades.splice(existingTradeIndex, 1);
+        } else {
+            logUtils.log('letting existing trade stand, and skipping processing of this trade');
+            return false;
+        }
+    }
+    return true;
+}
+
+const highestBlockNumbersInIntialDump: { [publisher: string]: number } = {
+    axieinfinity: 7065913,
+    cryptokitties: 4658171,
+    cryptopunks: 7058897,
+    cryptovoxels: 7060783,
+    decentraland_estate: 7065181,
+    decentraland: 6938962,
+    etherbots: 5204980,
+    etheremon: 7065370,
+    ethtown: 7064126,
+    knownorigin: 7065160,
+    mythereum: 7065311,
+    superrare: 7065955,
+};
+
+async function _getTradesWithOffsetAsync(
+    url: string,
+    publisher: string,
+    offset: number,
+): Promise<NonfungibleDotComHistoryResponse> {
+    const resp: NonfungibleDotComHistoryResponse = await fetchSuccessfullyOrThrowAsync(
+        `${url}?${stringify({
+            publisher,
+            start: offset,
+            length: MAX_TRADES_PER_QUERY,
+        })}`,
+    );
+    return resp;
+}
+
+function getFullUrlForPublisher(publisher: string): string {
+    return `${NONFUNGIBLE_DOT_COM_URL}/market/${publisher}/history`;
+}
+
+function getInitialDumpUrl(publisher: string): string {
+    return `https://nonfungible-dot-com-one-time-data-dump.s3.amazonaws.com/sales_summary_${publisher}.json`;
+}
+
+function ensureNonNull(trade: NonfungibleDotComTradeResponse): void {
+    // these fields need to be set in order to avoid non-null
+    // constraint exceptions upon database insertion.
+    if (trade.logIndex === undefined) {
+        // for cryptopunks
+        trade.logIndex = 0;
+    }
+    if (trade.assetDescriptor === undefined) {
+        // for cryptopunks
+        trade.assetDescriptor = '';
+    }
+    if (trade.meta === undefined) {
+        // for cryptopunks
+        trade.meta = {};
+    }
+    if (trade.marketAddress === null) {
+        // for decentraland_estate
+        trade.marketAddress = '';
+    }
+}

packages/pipeline/src/entities/index.ts

@@ -7,6 +7,7 @@ export { DexTrade } from './dex_trade';
 export { ExchangeCancelEvent } from './exchange_cancel_event';
 export { ExchangeCancelUpToEvent } from './exchange_cancel_up_to_event';
 export { ExchangeFillEvent } from './exchange_fill_event';
+export { NonfungibleDotComTrade } from './nonfungible_dot_com_trade';
 export { OHLCVExternal } from './ohlcv_external';
 export { Relayer } from './relayer';
 export { SraOrder } from './sra_order';

packages/pipeline/src/entities/nonfungible_dot_com_trade.ts

@@ -0,0 +1,35 @@
+import { BigNumber } from '@0x/utils';
+import { Column, Entity, PrimaryColumn } from 'typeorm';
+
+import { bigNumberTransformer, numberToBigIntTransformer } from '../utils';
+
+@Entity({ name: 'nonfungible_dot_com_trades', schema: 'raw' })
+export class NonfungibleDotComTrade {
+    @PrimaryColumn({ name: 'transaction_hash' })
+    public transactionHash!: string;
+    @PrimaryColumn({ name: 'publisher' })
+    public publisher!: string;
+    @PrimaryColumn({ name: 'block_number', type: 'bigint', transformer: numberToBigIntTransformer })
+    public blockNumber!: number;
+    @PrimaryColumn({ name: 'log_index' })
+    public logIndex!: number;
+    @PrimaryColumn({ name: 'asset_id' })
+    public assetId!: string;
+
+    @Column({ name: 'block_timestamp', type: 'bigint', transformer: numberToBigIntTransformer })
+    public blockTimestamp!: number;
+    @Column({ name: 'asset_descriptor' })
+    public assetDescriptor!: string;
+    @Column({ name: 'market_address' })
+    public marketAddress!: string;
+    @Column({ name: 'total_price', type: 'numeric', transformer: bigNumberTransformer })
+    public totalPrice!: BigNumber;
+    @Column({ name: 'usd_price', type: 'numeric', transformer: bigNumberTransformer })
+    public usdPrice!: BigNumber;
+    @Column({ name: 'buyer_address' })
+    public buyerAddress!: string;
+    @Column({ name: 'seller_address' })
+    public sellerAddress!: string;
+    @Column({ type: 'jsonb' })
+    public meta!: object;
+}

packages/pipeline/src/ormconfig.ts

@@ -12,6 +12,7 @@ import {
     ExchangeCancelEvent,
     ExchangeCancelUpToEvent,
     ExchangeFillEvent,
+    NonfungibleDotComTrade,
     OHLCVExternal,
     Relayer,
     SraOrder,
@@ -33,6 +34,7 @@ const entities = [
     ExchangeCancelUpToEvent,
     ExchangeFillEvent,
     ERC20ApprovalEvent,
+    NonfungibleDotComTrade,
     OHLCVExternal,
     Relayer,
     SraOrder,

packages/pipeline/src/parsers/nonfungible_dot_com/index.ts

@@ -0,0 +1,42 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { NonfungibleDotComTradeResponse } from '../../data_sources/nonfungible_dot_com';
+import { NonfungibleDotComTrade } from '../../entities';
+
+/**
+ * Parses a raw trades from the nonfungible.com API and returns an array of
+ * NonfungibleDotComTrade entities.
+ * @param rawTrades A raw order response from an SRA endpoint.
+ */
+export function parseNonFungibleDotComTrades(
+    rawTrades: NonfungibleDotComTradeResponse[],
+    publisher: string,
+): NonfungibleDotComTrade[] {
+    return R.map(_parseNonFungibleDotComTrade.bind(null, publisher), rawTrades);
+}
+
+/**
+ * Converts a single trade from nonfungible.com into an NonfungibleDotComTrade entity.
+ * @param rawTrade A single trade from the response from the nonfungible.com API.
+ */
+export function _parseNonFungibleDotComTrade(
+    publisher: string,
+    rawTrade: NonfungibleDotComTradeResponse,
+): NonfungibleDotComTrade {
+    const nonfungibleDotComTrade = new NonfungibleDotComTrade();
+    nonfungibleDotComTrade.assetDescriptor = rawTrade.assetDescriptor;
+    nonfungibleDotComTrade.assetId = rawTrade.assetId;
+    nonfungibleDotComTrade.blockNumber = rawTrade.blockNumber;
+    nonfungibleDotComTrade.blockTimestamp = new Date(rawTrade.blockTimestamp).getTime();
+    nonfungibleDotComTrade.buyerAddress = rawTrade.buyer;
+    nonfungibleDotComTrade.logIndex = rawTrade.logIndex;
+    nonfungibleDotComTrade.marketAddress = rawTrade.marketAddress;
+    nonfungibleDotComTrade.meta = rawTrade.meta;
+    nonfungibleDotComTrade.sellerAddress = rawTrade.seller;
+    nonfungibleDotComTrade.totalPrice = new BigNumber(rawTrade.totalPrice);
+    nonfungibleDotComTrade.transactionHash = rawTrade.transactionHash;
+    nonfungibleDotComTrade.usdPrice = new BigNumber(rawTrade.usdPrice);
+    nonfungibleDotComTrade.publisher = publisher;
+    return nonfungibleDotComTrade;
+}

packages/pipeline/src/scripts/pull_nonfungible_dot_com_trades.ts

@@ -0,0 +1,43 @@
+// tslint:disable:no-console
+import 'reflect-metadata';
+import { Connection, ConnectionOptions, createConnection } from 'typeorm';
+
+import { getTradesAsync, knownPublishers } from '../data_sources/nonfungible_dot_com';
+import { NonfungibleDotComTrade } from '../entities';
+import * as ormConfig from '../ormconfig';
+import { parseNonFungibleDotComTrades } from '../parsers/nonfungible_dot_com';
+import { handleError } from '../utils';
+
+// Number of trades to save at once.
+const BATCH_SAVE_SIZE = 1000;
+
+let connection: Connection;
+
+(async () => {
+    connection = await createConnection(ormConfig as ConnectionOptions);
+    await getAndSaveTradesAsync();
+    process.exit(0);
+})().catch(handleError);
+
+async function getAndSaveTradesAsync(): Promise<void> {
+    const tradesRepository = connection.getRepository(NonfungibleDotComTrade);
+
+    for (const publisher of knownPublishers) {
+        console.log(`Getting latest trades for NFT ${publisher}...`);
+        const tradeWithHighestBlockNumber = await tradesRepository
+            .createQueryBuilder('nonfungible_dot_com_trades')
+            .where('nonfungible_dot_com_trades.publisher = :publisher', { publisher })
+            .orderBy({ 'nonfungible_dot_com_trades.block_number': 'DESC' })
+            .getOne();
+        const highestExistingBlockNumber =
+            tradeWithHighestBlockNumber === undefined ? 0 : tradeWithHighestBlockNumber.blockNumber;
+        console.log(`Highest block number in existing trades: ${highestExistingBlockNumber}`);
+        const rawTrades = await getTradesAsync(publisher, highestExistingBlockNumber);
+        console.log(`Parsing ${rawTrades.length} trades...`);
+        const trades = parseNonFungibleDotComTrades(rawTrades, publisher);
+        console.log(`Saving ${rawTrades.length} trades...`);
+        await tradesRepository.save(trades, { chunk: Math.ceil(trades.length / BATCH_SAVE_SIZE) });
+    }
+    const newTotalTrades = await tradesRepository.count();
+    console.log(`Done saving trades. There are now ${newTotalTrades} total NFT trades.`);
+}

packages/pipeline/src/utils/index.ts

@@ -1,4 +1,4 @@
-import { BigNumber } from '@0x/utils';
+import { BigNumber, fetchAsync } from '@0x/utils';
 export * from './transformers';
 export * from './constants';
 
@@ -51,3 +51,16 @@ export function handleError(e: any): void {
     }
     process.exit(1);
 }
+
+/**
+ * Does fetchAsync(), and checks the status code, throwing if it doesn't indicate success.
+ */
+export async function fetchSuccessfullyOrThrowAsync(url: string): Promise<any> {
+    const response = await fetchAsync(url);
+    if (!response.ok) {
+        throw new Error(
+            `Failed to fetch URL ${url}.  Unsuccessful HTTP status code (${response.status}): ${response.statusText}`,
+        );
+    }
+    return response.json();
+}

packages/pipeline/test/entities/nft_trades_test.ts

@@ -0,0 +1,48 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import 'reflect-metadata';
+
+import { NonfungibleDotComTrade } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseTrade: NonfungibleDotComTrade = {
+    assetDescriptor: 'Kitty #1002',
+    assetId: '1002',
+    blockNumber: 4608542,
+    blockTimestamp: 1543544083704,
+    buyerAddress: '0x316c55d1895a085c4b39a98ecb563f509301aaf7',
+    logIndex: 28,
+    marketAddress: '0xb1690C08E213a35Ed9bAb7B318DE14420FB57d8C',
+    meta: {
+        cattribute_body: 'munchkin',
+        cattribute_coloreyes: 'mintgreen',
+        cattribute_colorprimary: 'orangesoda',
+        cattribute_colorsecondary: 'coffee',
+        cattribute_colortertiary: 'kittencream',
+        cattribute_eyes: 'thicccbrowz',
+        cattribute_mouth: 'soserious',
+        cattribute_pattern: 'totesbasic',
+        generation: '0',
+        is_exclusive: false,
+        is_fancy: false,
+    },
+    sellerAddress: '0xba52c75764d6f594735dc735be7f1830cdf58ddf',
+    totalPrice: new BigNumber('9751388888888889'),
+    transactionHash: '0x468168419be7e442d5ff32d264fab24087b744bc2e37fdbac7024e1e74f4c6c8',
+    usdPrice: new BigNumber('3.71957'),
+    publisher: 'cryptokitties',
+};
+
+// tslint:disable:custom-no-magic-numbers
+describe('NonfungibleDotComTrade entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const tradesRepository = connection.getRepository(NonfungibleDotComTrade);
+        await testSaveAndFindEntityAsync(tradesRepository, baseTrade);
+    });
+});

packages/pipeline/test/parsers/nonfungible_dot_com/index_test.ts

@@ -0,0 +1,86 @@
+// tslint:disable:custom-no-magic-numbers
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { NonfungibleDotComTradeResponse } from '../../../src/data_sources/nonfungible_dot_com';
+import { NonfungibleDotComTrade } from '../../../src/entities';
+import { _parseNonFungibleDotComTrade } from '../../../src/parsers/nonfungible_dot_com';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+const input: NonfungibleDotComTradeResponse = {
+    _id: '5b4cd04244abdb5ac3a8063f',
+    assetDescriptor: 'Kitty #1002',
+    assetId: '1002',
+    blockNumber: 4608542,
+    blockTimestamp: '2017-11-23T18:50:19.000Z',
+    buyer: '0x316c55d1895a085c4b39a98ecb563f509301aaf7',
+    logIndex: 28,
+    nftAddress: '0xb1690C08E213a35Ed9bAb7B318DE14420FB57d8C',
+    marketAddress: '0xb1690C08E213a35Ed9bAb7B318DE14420FB57d8C',
+    tokenTicker: 'eth',
+    meta: {
+        cattribute_body: 'munchkin',
+        cattribute_coloreyes: 'mintgreen',
+        cattribute_colorprimary: 'orangesoda',
+        cattribute_colorsecondary: 'coffee',
+        cattribute_colortertiary: 'kittencream',
+        cattribute_eyes: 'thicccbrowz',
+        cattribute_mouth: 'soserious',
+        cattribute_pattern: 'totesbasic',
+        generation: '0',
+        is_exclusive: false,
+        is_fancy: false,
+    },
+    seller: '0xba52c75764d6f594735dc735be7f1830cdf58ddf',
+    totalDecimalPrice: 0.00975138888888889,
+    totalPrice: '9751388888888889',
+    transactionHash: '0x468168419be7e442d5ff32d264fab24087b744bc2e37fdbac7024e1e74f4c6c8',
+    usdPrice: 3.71957,
+    currencyTransfer: {},
+    image: '',
+    composedOf: '',
+    asset_link: '',
+    seller_address_link: '',
+    buyer_address_link: '',
+};
+
+const expected: NonfungibleDotComTrade = {
+    assetDescriptor: 'Kitty #1002',
+    assetId: '1002',
+    blockNumber: 4608542,
+    blockTimestamp: 1511463019000,
+    buyerAddress: '0x316c55d1895a085c4b39a98ecb563f509301aaf7',
+    logIndex: 28,
+    marketAddress: '0xb1690C08E213a35Ed9bAb7B318DE14420FB57d8C',
+    meta: {
+        cattribute_body: 'munchkin',
+        cattribute_coloreyes: 'mintgreen',
+        cattribute_colorprimary: 'orangesoda',
+        cattribute_colorsecondary: 'coffee',
+        cattribute_colortertiary: 'kittencream',
+        cattribute_eyes: 'thicccbrowz',
+        cattribute_mouth: 'soserious',
+        cattribute_pattern: 'totesbasic',
+        generation: '0',
+        is_exclusive: false,
+        is_fancy: false,
+    },
+    sellerAddress: '0xba52c75764d6f594735dc735be7f1830cdf58ddf',
+    totalPrice: new BigNumber('9751388888888889'),
+    transactionHash: '0x468168419be7e442d5ff32d264fab24087b744bc2e37fdbac7024e1e74f4c6c8',
+    usdPrice: new BigNumber('3.71957'),
+    publisher: 'cryptokitties',
+};
+
+describe('nonfungible.com', () => {
+    describe('_parseNonFungibleDotComTrade', () => {
+        it(`converts NonfungibleDotComTradeResponse to NonfungibleDotComTrade entity`, () => {
+            const actual = _parseNonFungibleDotComTrade(expected.publisher, input);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});