Merge development

14 files changed, 661 insertions, 0 deletions

diff --git a/packages/pipeline/migrations/1542070840010-InitialSchema.ts b/packages/pipeline/migrations/1542070840010-InitialSchema.ts
new file mode 100644
index 000000000..895f9e6c9
--- /dev/null
+++ b/packages/pipeline/migrations/1542070840010-InitialSchema.ts
@@ -0,0 +1,187 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const blocks = new Table({
+    name: 'raw.blocks',
+    columns: [
+        { name: 'number', type: 'bigint', isPrimary: true },
+        { name: 'hash', type: 'varchar', isPrimary: true },
+        { name: 'timestamp', type: 'bigint' },
+    ],
+});
+
+const exchange_cancel_events = new Table({
+    name: 'raw.exchange_cancel_events',
+    columns: [
+        { name: 'contract_address', type: 'char(42)', isPrimary: true },
+        { name: 'log_index', type: 'integer', isPrimary: true },
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+
+        { name: 'raw_data', type: 'varchar' },
+
+        { name: 'transaction_hash', type: 'varchar' },
+        { name: 'maker_address', type: 'char(42)' },
+        { name: 'taker_address', type: 'char(42)' },
+        { name: 'fee_recipient_address', type: 'char(42)' },
+        { name: 'sender_address', type: 'char(42)' },
+        { name: 'order_hash', type: 'varchar' },
+
+        { name: 'raw_maker_asset_data', type: 'varchar' },
+        { name: 'maker_asset_type', type: 'varchar' },
+        { name: 'maker_asset_proxy_id', type: 'varchar' },
+        { name: 'maker_token_address', type: 'char(42)' },
+        { name: 'maker_token_id', type: 'varchar', isNullable: true },
+        { name: 'raw_taker_asset_data', type: 'varchar' },
+        { name: 'taker_asset_type', type: 'varchar' },
+        { name: 'taker_asset_proxy_id', type: 'varchar' },
+        { name: 'taker_token_address', type: 'char(42)' },
+        { name: 'taker_token_id', type: 'varchar', isNullable: true },
+    ],
+});
+
+const exchange_cancel_up_to_events = new Table({
+    name: 'raw.exchange_cancel_up_to_events',
+    columns: [
+        { name: 'contract_address', type: 'char(42)', isPrimary: true },
+        { name: 'log_index', type: 'integer', isPrimary: true },
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+
+        { name: 'raw_data', type: 'varchar' },
+
+        { name: 'transaction_hash', type: 'varchar' },
+        { name: 'maker_address', type: 'char(42)' },
+        { name: 'sender_address', type: 'char(42)' },
+        { name: 'order_epoch', type: 'varchar' },
+    ],
+});
+
+const exchange_fill_events = new Table({
+    name: 'raw.exchange_fill_events',
+    columns: [
+        { name: 'contract_address', type: 'char(42)', isPrimary: true },
+        { name: 'log_index', type: 'integer', isPrimary: true },
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+
+        { name: 'raw_data', type: 'varchar' },
+
+        { name: 'transaction_hash', type: 'varchar' },
+        { name: 'maker_address', type: 'char(42)' },
+        { name: 'taker_address', type: 'char(42)' },
+        { name: 'fee_recipient_address', type: 'char(42)' },
+        { name: 'sender_address', type: 'char(42)' },
+        { name: 'maker_asset_filled_amount', type: 'varchar' },
+        { name: 'taker_asset_filled_amount', type: 'varchar' },
+        { name: 'maker_fee_paid', type: 'varchar' },
+        { name: 'taker_fee_paid', type: 'varchar' },
+        { name: 'order_hash', type: 'varchar' },
+
+        { name: 'raw_maker_asset_data', type: 'varchar' },
+        { name: 'maker_asset_type', type: 'varchar' },
+        { name: 'maker_asset_proxy_id', type: 'varchar' },
+        { name: 'maker_token_address', type: 'char(42)' },
+        { name: 'maker_token_id', type: 'varchar', isNullable: true },
+        { name: 'raw_taker_asset_data', type: 'varchar' },
+        { name: 'taker_asset_type', type: 'varchar' },
+        { name: 'taker_asset_proxy_id', type: 'varchar' },
+        { name: 'taker_token_address', type: 'char(42)' },
+        { name: 'taker_token_id', type: 'varchar', isNullable: true },
+    ],
+});
+
+const relayers = new Table({
+    name: 'raw.relayers',
+    columns: [
+        { name: 'uuid', type: 'varchar', isPrimary: true },
+        { name: 'name', type: 'varchar' },
+        { name: 'sra_http_endpoint', type: 'varchar', isNullable: true },
+        { name: 'sra_ws_endpoint', type: 'varchar', isNullable: true },
+        { name: 'app_url', type: 'varchar', isNullable: true },
+        { name: 'fee_recipient_addresses', type: 'char(42)', isArray: true },
+        { name: 'taker_addresses', type: 'char(42)', isArray: true },
+    ],
+});
+
+const sra_orders = new Table({
+    name: 'raw.sra_orders',
+    columns: [
+        { name: 'exchange_address', type: 'char(42)', isPrimary: true },
+        { name: 'order_hash_hex', type: 'varchar', isPrimary: true },
+
+        { name: 'source_url', type: 'varchar' },
+        { name: 'last_updated_timestamp', type: 'bigint' },
+        { name: 'first_seen_timestamp', type: 'bigint' },
+
+        { name: 'maker_address', type: 'char(42)' },
+        { name: 'taker_address', type: 'char(42)' },
+        { name: 'fee_recipient_address', type: 'char(42)' },
+        { name: 'sender_address', type: 'char(42)' },
+        { name: 'maker_asset_filled_amount', type: 'varchar' },
+        { name: 'taker_asset_filled_amount', type: 'varchar' },
+        { name: 'maker_fee', type: 'varchar' },
+        { name: 'taker_fee', type: 'varchar' },
+        { name: 'expiration_time_seconds', type: 'int' },
+        { name: 'salt', type: 'varchar' },
+        { name: 'signature', type: 'varchar' },
+
+        { name: 'raw_maker_asset_data', type: 'varchar' },
+        { name: 'maker_asset_type', type: 'varchar' },
+        { name: 'maker_asset_proxy_id', type: 'varchar' },
+        { name: 'maker_token_address', type: 'char(42)' },
+        { name: 'maker_token_id', type: 'varchar', isNullable: true },
+        { name: 'raw_taker_asset_data', type: 'varchar' },
+        { name: 'taker_asset_type', type: 'varchar' },
+        { name: 'taker_asset_proxy_id', type: 'varchar' },
+        { name: 'taker_token_address', type: 'char(42)' },
+        { name: 'taker_token_id', type: 'varchar', isNullable: true },
+
+        { name: 'metadata_json', type: 'varchar' },
+    ],
+});
+
+const token_on_chain_metadata = new Table({
+    name: 'raw.token_on_chain_metadata',
+    columns: [
+        { name: 'address', type: 'char(42)', isPrimary: true },
+        { name: 'decimals', type: 'integer' },
+        { name: 'symbol', type: 'varchar' },
+        { name: 'name', type: 'varchar' },
+    ],
+});
+
+const transactions = new Table({
+    name: 'raw.transactions',
+    columns: [
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+        { name: 'block_hash', type: 'varchar', isPrimary: true },
+        { name: 'transaction_hash', type: 'varchar', isPrimary: true },
+        { name: 'gas_used', type: 'bigint' },
+        { name: 'gas_price', type: 'bigint' },
+    ],
+});
+
+export class InitialSchema1542070840010 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.createSchema('raw');
+
+        await queryRunner.createTable(blocks);
+        await queryRunner.createTable(exchange_cancel_events);
+        await queryRunner.createTable(exchange_cancel_up_to_events);
+        await queryRunner.createTable(exchange_fill_events);
+        await queryRunner.createTable(relayers);
+        await queryRunner.createTable(sra_orders);
+        await queryRunner.createTable(token_on_chain_metadata);
+        await queryRunner.createTable(transactions);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(blocks.name);
+        await queryRunner.dropTable(exchange_cancel_events.name);
+        await queryRunner.dropTable(exchange_cancel_up_to_events.name);
+        await queryRunner.dropTable(exchange_fill_events.name);
+        await queryRunner.dropTable(relayers.name);
+        await queryRunner.dropTable(sra_orders.name);
+        await queryRunner.dropTable(token_on_chain_metadata.name);
+        await queryRunner.dropTable(transactions.name);
+
+        await queryRunner.dropSchema('raw');
+    }
+}
diff --git a/packages/pipeline/migrations/1542147915364-NewSraOrderTimestampFormat.ts b/packages/pipeline/migrations/1542147915364-NewSraOrderTimestampFormat.ts
new file mode 100644
index 000000000..5a8f3fec8
--- /dev/null
+++ b/packages/pipeline/migrations/1542147915364-NewSraOrderTimestampFormat.ts
@@ -0,0 +1,48 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+export class NewSraOrderTimestampFormat1542147915364 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                DROP CONSTRAINT "PK_09bfb9980715329563bd53d667e",
+                ADD PRIMARY KEY (order_hash_hex, exchange_address, source_url);
+            `,
+        );
+
+        await queryRunner.query(
+            `CREATE TABLE raw.sra_orders_observed_timestamps (
+                order_hash_hex varchar NOT NULL,
+                exchange_address varchar NOT NULL,
+                source_url varchar NOT NULL,
+                observed_timestamp bigint NOT NULL,
+                FOREIGN KEY
+                    (order_hash_hex, exchange_address, source_url)
+                    REFERENCES raw.sra_orders (order_hash_hex, exchange_address, source_url),
+                PRIMARY KEY (order_hash_hex, exchange_address, source_url, observed_timestamp)
+            );`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                DROP COLUMN last_updated_timestamp,
+                DROP COLUMN first_seen_timestamp;`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable('raw.sra_orders_observed_timestamps');
+
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                ADD COLUMN last_updated_timestamp bigint NOT NULL DEFAULT 0,
+                ADD COLUMN first_seen_timestamp bigint NOT NULL DEFAULT 0;`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                DROP CONSTRAINT sra_orders_pkey,
+                ADD CONSTRAINT "PK_09bfb9980715329563bd53d667e" PRIMARY KEY ("exchange_address", "order_hash_hex");
+            `,
+        );
+    }
+}
diff --git a/packages/pipeline/migrations/1542152278484-RenameSraOrdersFilledAmounts.ts b/packages/pipeline/migrations/1542152278484-RenameSraOrdersFilledAmounts.ts
new file mode 100644
index 000000000..a13e3efa5
--- /dev/null
+++ b/packages/pipeline/migrations/1542152278484-RenameSraOrdersFilledAmounts.ts
@@ -0,0 +1,13 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class RenameSraOrdersFilledAmounts1542152278484 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.renameColumn('raw.sra_orders', 'maker_asset_filled_amount', 'maker_asset_amount');
+        await queryRunner.renameColumn('raw.sra_orders', 'taker_asset_filled_amount', 'taker_asset_amount');
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.renameColumn('raw.sra_orders', 'maker_asset_amount', 'maker_asset_filled_amount');
+        await queryRunner.renameColumn('raw.sra_orders', 'taker_asset_amount', 'taker_asset_filled_amount');
+    }
+}
diff --git a/packages/pipeline/migrations/1542234704666-ConvertBigNumberToNumeric.ts b/packages/pipeline/migrations/1542234704666-ConvertBigNumberToNumeric.ts
new file mode 100644
index 000000000..5200ef7cc
--- /dev/null
+++ b/packages/pipeline/migrations/1542234704666-ConvertBigNumberToNumeric.ts
@@ -0,0 +1,53 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class ConvertBigNumberToNumeric1542234704666 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_fill_events
+                ALTER COLUMN maker_asset_filled_amount TYPE numeric USING maker_asset_filled_amount::numeric,
+                ALTER COLUMN taker_asset_filled_amount TYPE numeric USING taker_asset_filled_amount::numeric,
+                ALTER COLUMN maker_fee_paid TYPE numeric USING maker_fee_paid::numeric,
+                ALTER COLUMN taker_fee_paid TYPE numeric USING taker_fee_paid::numeric;`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_cancel_up_to_events
+                ALTER COLUMN order_epoch TYPE numeric USING order_epoch::numeric;`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                ALTER COLUMN maker_asset_amount TYPE numeric USING maker_asset_amount::numeric,
+                ALTER COLUMN taker_asset_amount TYPE numeric USING taker_asset_amount::numeric,
+                ALTER COLUMN maker_fee TYPE numeric USING maker_fee::numeric,
+                ALTER COLUMN taker_fee TYPE numeric USING taker_fee::numeric,
+                ALTER COLUMN expiration_time_seconds TYPE numeric USING expiration_time_seconds::numeric,
+                ALTER COLUMN salt TYPE numeric USING salt::numeric;`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.sra_orders
+                ALTER COLUMN maker_asset_amount TYPE varchar USING maker_asset_amount::varchar,
+                ALTER COLUMN taker_asset_amount TYPE varchar USING taker_asset_amount::varchar,
+                ALTER COLUMN maker_fee TYPE varchar USING maker_fee::varchar,
+                ALTER COLUMN taker_fee TYPE varchar USING taker_fee::varchar,
+                ALTER COLUMN expiration_time_seconds TYPE varchar USING expiration_time_seconds::varchar,
+                ALTER COLUMN salt TYPE varchar USING salt::varchar;`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_cancel_up_to_events
+                ALTER COLUMN order_epoch TYPE varchar USING order_epoch::varchar;`,
+        );
+
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_fill_events
+                ALTER COLUMN maker_asset_filled_amount TYPE varchar USING maker_asset_filled_amount::varchar,
+                ALTER COLUMN taker_asset_filled_amount TYPE varchar USING taker_asset_filled_amount::varchar,
+                ALTER COLUMN maker_fee_paid TYPE varchar USING maker_fee_paid::varchar,
+                ALTER COLUMN taker_fee_paid TYPE varchar USING taker_fee_paid::varchar;`,
+        );
+    }
+}
diff --git a/packages/pipeline/migrations/1542249766882-AddHomepageUrlToRelayers.ts b/packages/pipeline/migrations/1542249766882-AddHomepageUrlToRelayers.ts
new file mode 100644
index 000000000..9a4811ad5
--- /dev/null
+++ b/packages/pipeline/migrations/1542249766882-AddHomepageUrlToRelayers.ts
@@ -0,0 +1,14 @@
+import { MigrationInterface, QueryRunner, TableColumn } from 'typeorm';
+
+export class AddHomepageUrlToRelayers1542249766882 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.addColumn(
+            'raw.relayers',
+            new TableColumn({ name: 'homepage_url', type: 'varchar', default: `'unknown'` }),
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropColumn('raw.relayers', 'homepage_url');
+    }
+}
diff --git a/packages/pipeline/migrations/1542401122477-MakeTakerAddressNullable.ts b/packages/pipeline/migrations/1542401122477-MakeTakerAddressNullable.ts
new file mode 100644
index 000000000..957c85a36
--- /dev/null
+++ b/packages/pipeline/migrations/1542401122477-MakeTakerAddressNullable.ts
@@ -0,0 +1,17 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class MakeTakerAddressNullable1542401122477 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_cancel_events
+                ALTER COLUMN taker_address DROP NOT NULL;`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.exchange_cancel_events
+                ALTER COLUMN taker_address SET NOT NULL;`,
+        );
+    }
+}
diff --git a/packages/pipeline/migrations/1542655823221-NewMetadataAndOHLCVTables.ts b/packages/pipeline/migrations/1542655823221-NewMetadataAndOHLCVTables.ts
new file mode 100644
index 000000000..838f5ba9c
--- /dev/null
+++ b/packages/pipeline/migrations/1542655823221-NewMetadataAndOHLCVTables.ts
@@ -0,0 +1,60 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class NewMetadataAndOHLCVTables1542655823221 implements MigrationInterface {
+    // tslint:disable-next-line
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(`
+            CREATE TABLE raw.token_metadata (
+                address VARCHAR NOT NULL,
+                authority VARCHAR NOT NULL,
+                decimals INT NULL,
+                symbol VARCHAR NULL,
+                name VARCHAR NULL,
+
+                PRIMARY KEY (address, authority)
+            );
+        `);
+
+        await queryRunner.dropTable('raw.token_on_chain_metadata');
+
+        await queryRunner.query(`
+            CREATE TABLE raw.ohlcv_external (
+                exchange VARCHAR NOT NULL,
+                from_symbol VARCHAR NOT NULL,
+                to_symbol VARCHAR NOT NULL,
+                start_time BIGINT NOT NULL,
+                end_time BIGINT NOT NULL,
+
+                open DOUBLE PRECISION NOT NULL,
+                close DOUBLE PRECISION NOT NULL,
+                low DOUBLE PRECISION NOT NULL,
+                high DOUBLE PRECISION NOT NULL,
+                volume_from DOUBLE PRECISION NOT NULL,
+                volume_to DOUBLE PRECISION NOT NULL,
+
+                source VARCHAR NOT NULL,
+                observed_timestamp BIGINT NOT NULL,
+
+                PRIMARY KEY (exchange, from_symbol, to_symbol, start_time, end_time, source, observed_timestamp)
+            );
+        `);
+    }
+
+    // tslint:disable-next-line
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(`
+            CREATE TABLE raw.token_on_chain_metadata (
+                address VARCHAR NOT NULL,
+                decimals INT NULL,
+                symbol VARCHAR NULL,
+                name VARCHAR NULL,
+
+                PRIMARY KEY (address)
+            );
+        `);
+
+        await queryRunner.dropTable('raw.token_metadata');
+
+        await queryRunner.dropTable('raw.ohlcv_external');
+    }
+}
diff --git a/packages/pipeline/migrations/1543434472116-TokenOrderbookSnapshots.ts b/packages/pipeline/migrations/1543434472116-TokenOrderbookSnapshots.ts
new file mode 100644
index 000000000..a7117c753
--- /dev/null
+++ b/packages/pipeline/migrations/1543434472116-TokenOrderbookSnapshots.ts
@@ -0,0 +1,30 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const tokenOrderbookSnapshots = new Table({
+    name: 'raw.token_orderbook_snapshots',
+    columns: [
+        { name: 'observed_timestamp', type: 'bigint', isPrimary: true },
+        { name: 'source', type: 'varchar', isPrimary: true },
+        { name: 'order_type', type: 'order_t' },
+        { name: 'price', type: 'numeric', isPrimary: true },
+
+        { name: 'base_asset_symbol', type: 'varchar', isPrimary: true },
+        { name: 'base_asset_address', type: 'char(42)' },
+        { name: 'base_volume', type: 'numeric' },
+
+        { name: 'quote_asset_symbol', type: 'varchar', isPrimary: true },
+        { name: 'quote_asset_address', type: 'char(42)' },
+        { name: 'quote_volume', type: 'numeric' },
+    ],
+});
+
+export class TokenOrderbookSnapshots1543434472116 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(`CREATE TYPE order_t AS enum('bid', 'ask');`);
+        await queryRunner.createTable(tokenOrderbookSnapshots);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(tokenOrderbookSnapshots.name);
+    }
+}
diff --git a/packages/pipeline/migrations/1543446690436-CreateDexTrades.ts b/packages/pipeline/migrations/1543446690436-CreateDexTrades.ts
new file mode 100644
index 000000000..267cf144b
--- /dev/null
+++ b/packages/pipeline/migrations/1543446690436-CreateDexTrades.ts
@@ -0,0 +1,41 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const dexTrades = new Table({
+    name: 'raw.dex_trades',
+    columns: [
+        { name: 'source_url', type: 'varchar', isPrimary: true },
+        { name: 'tx_hash', type: 'varchar', isPrimary: true },
+
+        { name: 'tx_timestamp', type: 'bigint' },
+        { name: 'tx_date', type: 'varchar' },
+        { name: 'tx_sender', type: 'varchar(42)' },
+        { name: 'smart_contract_id', type: 'bigint' },
+        { name: 'smart_contract_address', type: 'varchar(42)' },
+        { name: 'contract_type', type: 'varchar' },
+        { name: 'maker', type: 'varchar(42)' },
+        { name: 'taker', type: 'varchar(42)' },
+        { name: 'amount_buy', type: 'numeric' },
+        { name: 'maker_fee_amount', type: 'numeric' },
+        { name: 'buy_currency_id', type: 'bigint' },
+        { name: 'buy_symbol', type: 'varchar' },
+        { name: 'amount_sell', type: 'numeric' },
+        { name: 'taker_fee_amount', type: 'numeric' },
+        { name: 'sell_currency_id', type: 'bigint' },
+        { name: 'sell_symbol', type: 'varchar' },
+        { name: 'maker_annotation', type: 'varchar' },
+        { name: 'taker_annotation', type: 'varchar' },
+        { name: 'protocol', type: 'varchar' },
+        { name: 'buy_address', type: 'varchar(42)', isNullable: true },
+        { name: 'sell_address', type: 'varchar(42)', isNullable: true },
+    ],
+});
+
+export class CreateDexTrades1543446690436 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.createTable(dexTrades);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(dexTrades);
+    }
+}
diff --git a/packages/pipeline/migrations/1543980079179-ConvertTokenMetadataDecimalsToBigNumber.ts b/packages/pipeline/migrations/1543980079179-ConvertTokenMetadataDecimalsToBigNumber.ts
new file mode 100644
index 000000000..351bc7eb8
--- /dev/null
+++ b/packages/pipeline/migrations/1543980079179-ConvertTokenMetadataDecimalsToBigNumber.ts
@@ -0,0 +1,17 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class ConvertTokenMetadataDecimalsToBigNumber1543980079179 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.token_metadata
+                ALTER COLUMN decimals TYPE numeric USING decimals::numeric;`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.token_metadata
+                ALTER COLUMN decimals TYPE numeric USING decimals::integer;`,
+        );
+    }
+}
diff --git a/packages/pipeline/migrations/1543983324954-ConvertTransactionGasPriceToBigNumber.ts b/packages/pipeline/migrations/1543983324954-ConvertTransactionGasPriceToBigNumber.ts
new file mode 100644
index 000000000..dcb0fd727
--- /dev/null
+++ b/packages/pipeline/migrations/1543983324954-ConvertTransactionGasPriceToBigNumber.ts
@@ -0,0 +1,19 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class ConvertTransactionGasPriceToBigNumber1543983324954 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.transactions
+                ALTER COLUMN gas_price TYPE numeric USING gas_price::numeric,
+                ALTER COLUMN gas_used TYPE numeric USING gas_used::numeric;`,
+        );
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.query(
+            `ALTER TABLE raw.transactions
+                ALTER COLUMN gas_price TYPE numeric USING gas_price::bigint,
+                ALTER COLUMN gas_used TYPE numeric USING gas_used::bigint;`,
+        );
+    }
+}
diff --git a/packages/pipeline/migrations/1544131464368-CreateERC20ApprovalEvents.ts b/packages/pipeline/migrations/1544131464368-CreateERC20ApprovalEvents.ts
new file mode 100644
index 000000000..2e84e0ec8
--- /dev/null
+++ b/packages/pipeline/migrations/1544131464368-CreateERC20ApprovalEvents.ts
@@ -0,0 +1,26 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const erc20ApprovalEvents = new Table({
+    name: 'raw.erc20_approval_events',
+    columns: [
+        { name: 'token_address', type: 'varchar(42)', isPrimary: true },
+        { name: 'log_index', type: 'integer', isPrimary: true },
+        { name: 'block_number', type: 'bigint', isPrimary: true },
+
+        { name: 'raw_data', type: 'varchar' },
+        { name: 'transaction_hash', type: 'varchar' },
+        { name: 'owner_address', type: 'varchar(42)' },
+        { name: 'spender_address', type: 'varchar(42)' },
+        { name: 'amount', type: 'numeric' },
+    ],
+});
+
+export class CreateERC20TokenApprovalEvents1544131464368 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.createTable(erc20ApprovalEvents);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(erc20ApprovalEvents);
+    }
+}
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/migrations/1545440485644-CreateCopperTables.ts b/packages/pipeline/migrations/1545440485644-CreateCopperTables.ts
new file mode 100644
index 000000000..64bf70af4
--- /dev/null
+++ b/packages/pipeline/migrations/1545440485644-CreateCopperTables.ts
@@ -0,0 +1,103 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+const leads = new Table({
+    name: 'raw.copper_leads',
+    columns: [
+        { name: 'id', type: 'bigint', isPrimary: true },
+        { name: 'name', type: 'varchar', isNullable: true },
+        { name: 'first_name', type: 'varchar', isNullable: true },
+        { name: 'last_name', type: 'varchar', isNullable: true },
+        { name: 'middle_name', type: 'varchar', isNullable: true },
+        { name: 'assignee_id', type: 'bigint', isNullable: true },
+        { name: 'company_name', type: 'varchar', isNullable: true },
+        { name: 'customer_source_id', type: 'bigint', isNullable: true },
+        { name: 'monetary_value', type: 'integer', isNullable: true },
+        { name: 'status', type: 'varchar' },
+        { name: 'status_id', type: 'bigint' },
+        { name: 'title', type: 'varchar', isNullable: true },
+        { name: 'date_created', type: 'bigint' },
+        { name: 'date_modified', type: 'bigint', isPrimary: true },
+    ],
+});
+const activities = new Table({
+    name: 'raw.copper_activities',
+    columns: [
+        { name: 'id', type: 'bigint', isPrimary: true },
+        { name: 'parent_id', type: 'bigint' },
+        { name: 'parent_type', type: 'varchar' },
+        { name: 'type_id', type: 'bigint' },
+        { name: 'type_category', type: 'varchar' },
+        { name: 'type_name', type: 'varchar', isNullable: true },
+        { name: 'user_id', type: 'bigint' },
+        { name: 'old_value_id', type: 'bigint', isNullable: true },
+        { name: 'old_value_name', type: 'varchar', isNullable: true },
+        { name: 'new_value_id', type: 'bigint', isNullable: true },
+        { name: 'new_value_name', type: 'varchar', isNullable: true },
+        { name: 'date_created', type: 'bigint' },
+        { name: 'date_modified', type: 'bigint', isPrimary: true },
+    ],
+});
+
+const opportunities = new Table({
+    name: 'raw.copper_opportunities',
+    columns: [
+        { name: 'id', type: 'bigint', isPrimary: true },
+        { name: 'name', type: 'varchar' },
+        { name: 'assignee_id', isNullable: true, type: 'bigint' },
+        { name: 'close_date', isNullable: true, type: 'varchar' },
+        { name: 'company_id', isNullable: true, type: 'bigint' },
+        { name: 'company_name', isNullable: true, type: 'varchar' },
+        { name: 'customer_source_id', isNullable: true, type: 'bigint' },
+        { name: 'loss_reason_id', isNullable: true, type: 'bigint' },
+        { name: 'pipeline_id', type: 'bigint' },
+        { name: 'pipeline_stage_id', type: 'bigint' },
+        { name: 'primary_contact_id', isNullable: true, type: 'bigint' },
+        { name: 'priority', isNullable: true, type: 'varchar' },
+        { name: 'status', type: 'varchar' },
+        { name: 'interaction_count', type: 'bigint' },
+        { name: 'monetary_value', isNullable: true, type: 'integer' },
+        { name: 'win_probability', isNullable: true, type: 'integer' },
+        { name: 'date_created', type: 'bigint' },
+        { name: 'date_modified', type: 'bigint', isPrimary: true },
+        { name: 'custom_fields', type: 'jsonb' },
+    ],
+});
+
+const activityTypes = new Table({
+    name: 'raw.copper_activity_types',
+    columns: [
+        { name: 'id', type: 'bigint', isPrimary: true },
+        { name: 'category', type: 'varchar' },
+        { name: 'name', type: 'varchar' },
+        { name: 'is_disabled', type: 'boolean', isNullable: true },
+        { name: 'count_as_interaction', type: 'boolean', isNullable: true },
+    ],
+});
+
+const customFields = new Table({
+    name: 'raw.copper_custom_fields',
+    columns: [
+        { name: 'id', type: 'bigint', isPrimary: true },
+        { name: 'name', type: 'varchar' },
+        { name: 'data_type', type: 'varchar' },
+        { name: 'field_type', type: 'varchar', isNullable: true },
+    ],
+});
+
+export class CreateCopperTables1544055699284 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.createTable(leads);
+        await queryRunner.createTable(activities);
+        await queryRunner.createTable(opportunities);
+        await queryRunner.createTable(activityTypes);
+        await queryRunner.createTable(customFields);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<any> {
+        await queryRunner.dropTable(leads.name);
+        await queryRunner.dropTable(activities.name);
+        await queryRunner.dropTable(opportunities.name);
+        await queryRunner.dropTable(activityTypes.name);
+        await queryRunner.dropTable(customFields.name);
+    }
+}