Graph dev

compare_data/README.md

@@ -57,7 +57,7 @@ of textile materials (other than wool or fine animal hair, cotton
 or man-made fibres), knitted or crocheted"""
 ```
 
-### Hitachi Dataset
+### Hitachi Dataset (index_hs6)
 For Hitachi, one would run an analagous script, although the `year` is provided differently (since the loader gives data from all years simultaneously). Note that both datasets may have certain products or entities missing.
 
 ```python
@@ -76,4 +76,23 @@ print(trade_flows_2020[(exporter_country,product_code)])
 """output: {'weight': 16.330946196791054, 'currency': 150.03981445684792}"""
 ```
 
-The above retrieves the values for flexible iron or steel tubing (hs6 code = `830710`) exported by the United States in `2020`. 
+The above retrieves the values for flexible iron or steel tubing (hs6 code = `830710`) exported by the United States in `2020`. 
+
+# Hitachi Dataset (logistic_data)
+This gathers data from transactions in `logistic_data` rather than `index_hs6`, as conducted in the previous section. First, run the following preprocessing script to extract relevant mapping tables. 
+```zsh
+python  ../temporal_graph/extract_tables.py --dir ./ --rs_login <Redshift username> <Redshift password>
+```
+This will save out three dictionaries (`./hitachi_{company, country, product}_mappers.json`). Below is a sample script for obtaining aggregated values.
+```python
+import read_logistic
+
+rs = RedshiftClass(args.rs_login[0], args.rs_login[1])
+product_map, trade_flow_map = get_Hitachi_data(rs, "exporter_product", hs_digits = 6, year = 2021, maps_dir = "./")
+
+#exports of refined lead products from the United States 
+print(trade_flow_map[('USA', '780110')])
+
+"""output: {'bill_count': 2, 'currency': 90210.9296875, 'quantity': 40742.0, 'weight': 40749.0} """
+```
+Note this may take slightly longer than the above acquisitions (i.e. with BACI and `index_hs6`).

compare_data/constants.py

@@ -15,12 +15,7 @@
  'BRAZIL': 76,
  'CANADA': 124,
  'CHILE': 152,
-  #this is an accordance with the dataset convention, and not 
-  #an indication of my personal views, similar to any
-  #other omissions and labellings throughout 
- 'CHINA HONGKONG': 156,
  'CHINA MAINLAND': 156,
- 'CHINA TAIWAN': 156,
  'COLOMBIA': 170,
  'COSTA RICA': 188,
  'CZECHIA': 203,

compare_data/read_logistic.py

@@ -0,0 +1,161 @@
+"""
+this file contains utility functions for reading the Hitachi logistic_data, 
+and aggregating quantities across products, countries, etc. for global, bilateral trade flows 
+** note: excludes transactions that constitute domestic trade
+"""
+
+import os 
+import glob
+import json
+import argparse
+import warnings
+import sys
+sys.path.append("/opt/libs")
+from crystal_api.apiclass import APIClass, RedshiftClass
+from crystal_api.apikeyclass import APIkeyClass
+from dotenv import load_dotenv
+import pandas as pd
+import time
+import datetime
+import numpy as np
+
+def get_days_between(date_1, date_2, date_format = "%Y-%m-%d"):
+    """
+    calculates the number of days between two dates (default format YY-MM-DD)
+    """
+    date_1_time = datetime.datetime.strptime(date_1, date_format)
+    date_2_time = datetime.datetime.strptime(date_2, date_format)
+    return (date_2_time - date_1_time).days
+
+def retrieve_Hitachi_table(name = "company", dir = "."):
+    """
+    Retrieves the Hitachi tables saved as .json files by the script extract_tables.py (see code for details)
+
+    Args:
+        name (str): The specific reference table to retrieve, must be out of {company, country, product}
+        dir (str): The path to the directory the table is stored in 
+
+    Returns:
+        dict: The retrieved reference table. Note that the company table comprises both the forward (id2company)
+        and inverse (company2id) mappings. 
+    """
+
+    with open(os.path.join(dir, f"hitachi_{name}_mappers.json"),"r") as file:
+        table = json.load(file) 
+    return table
+
+def get_aggregation_key(aggregation_type):
+    #orig_country and dest_country represent the exporter and importer of a transaction, respectively
+    entity2key = {"exporter": "orig_country", "importer": "dest_country", "product": "product"}
+    valid_entities = set(entity2key.keys())
+    entities = aggregation_type.split("_")
+    for entity in entities: 
+        if entity not in valid_entities: raise ValueError(f"{entity} not valid. must be among {valid_entities}")
+
+    return ",".join(entity2key[entity] for entity in entities)
+
+def aggregate_logistic(rs, aggregation_type = "product", hs_level = 6, start_date = "2019-01-01", 
+                       end_date = "2019-12-31", maps_dir = "./", verbose = False):
+    """
+    Args:
+        rs (RedshiftClass):  An instance of Redshift Class 
+        aggregation_type (str): specifies the entity combinations for which we calculate trade flows.
+                                Should take the form [entity1]_[entity2]_ ..., where each entity is among 
+                               ["exporter","importer","product"] and ordered as such (e.g. no product_importer)
+        hs_level (int), maps_dir (str): see get_Hitachi_date() below
+        start_date (str): the start date from which transactions are aggregated, inclusive
+        end_date (str): the end date from which transacted are aggregated, inclusive 
+        verbose (bool): Whether to print out status updates (to console) from the retrieval 
+
+    Returns:
+        dict: See (2) in the Returns documentation for get_Hitachi_data()
+    """
+    num_days_between = get_days_between(start_date, end_date)
+    country_map = retrieve_Hitachi_table(name = "country", dir = maps_dir)
+
+    #deduplicate transactions
+    PRIMARY_KEY = 'date, supplier_id, buyer_id, quantity, weight, price, amount, hs_code'
+    AGGREGATION_KEY = get_aggregation_key(aggregation_type)
+    product_filter = "product not like '% %' AND product not like '00%' AND len(product) = 6 \
+                    AND product not like '%,%'" #selecting valid HS6 codes
+    country_filter = "orig_country != dest_country AND orig_country != '' AND dest_country != ''"
+
+    #restrict the transactions to the specified time period, and deduplicate 
+    query = f"select {PRIMARY_KEY}, SUBSTRING(hs_code, 1, 6) as product, COUNT(*) as count, max(orig_country) as \
+    orig_country, max(dest_country) as dest_country, COUNT(DISTINCT id) as num_ids from logistic_data \
+    WHERE {product_filter} AND DATEDIFF(day, '{start_date}', date) BETWEEN 0 AND {num_days_between} \
+    AND {country_filter} GROUP BY {PRIMARY_KEY}, product"
+
+    #aggregate transactions under the desired entity combinations
+    query = f"select {AGGREGATION_KEY}, COUNT(*) as bill_count, SUM(quantity) as total_quantity, SUM(amount) as total_amount, \
+    SUM(weight) as total_weight from ({query}) GROUP BY {AGGREGATION_KEY}"
+
+    #query the RedShift API
+    if verbose == True: print("Querying logistic_data between {} and {}".format(start_date, end_date))
+    start_t = time.time()
+    df = rs.query_df(query).fillna(0)
+    end_t = time.time()
+    if verbose == True: print("Retrieved {} rows from logistic_data in {:.3f} seconds".format(len(df), end_t - start_t))
+
+    #process the returned dataframe
+    if ("orig_country" in df.columns):
+        df["orig_country"] = df["orig_country"].apply(lambda name: country_map[name])
+    if ("dest_country" in df.columns):
+        df["dest_country"] = df["dest_country"].apply(lambda name: country_map[name])
+    if ("product" in df.columns):
+        df["product"] = df["product"].apply(lambda code: code[:hs_level])
+        df = df.groupby(by = AGGREGATION_KEY.split(",")).sum(numeric_only = True).reset_index()
+
+    #transform the dataframe into a dictionary from entities to corresponding trade values
+    df["key"] = [",".join(entities) for entities in zip(*[list(df[key]) for key in AGGREGATION_KEY.split(",")])]
+    df_rows = [list(df[row]) for row in ["key","bill_count","total_quantity","total_amount","total_weight"]]
+    trade_flow_map = {}
+    for key, bill_count, quantity, amount, weight in zip(*df_rows):
+        entities = key.split(",")
+        metrics = {"bill_count": bill_count, "currency": amount, "quantity": quantity, "weight": weight}
+        trade_flow_map[tuple(entities) if len(entities) > 1 else entities[0]] = metrics
+
+    return trade_flow_map
+
+def get_Hitachi_data(rs, aggregation_type = "product", hs_level = 6, year = 2020, maps_dir = "./"):
+    """
+    reads the Hitachi logistic_data, aggregating transactions at the specified entity level
+
+    Args:
+        aggregation_type (str): species the entity for which we collate global trade flows (see aggregate_logistic)
+        hs_level (int): The granularity of HS products (whether to use first 2, 4, or 6 digits)
+        year (int): Year from which to collect data (from 2019 - 2023, inclusive)
+        maps_dir (str): path to the directory storing the Hitachi tables retrieved by ../temporal_graph/extract_tables.py
+
+    Returns:
+         tuple[dict]: Two dictionaries. (1) from product HS6 codes to
+                    Hitachi descriptions, (2) from entity (e.g. HS6 product) to aggregated amount (in USD)
+                    and weight (in tonnes) in global trade flows of that entity
+    """
+    assert year in list(range(2019,2023+1)), "year must be between 2019 and 2023, inclusive" 
+    product_map = retrieve_Hitachi_table("product", dir = maps_dir)
+    start_date, end_date = f"{year}-01-01", f"{year}-12-31"
+    trade_flow_map = aggregate_logistic(rs, aggregation_type, hs_level, start_date, end_date, maps_dir,
+                                       verbose = True)
+
+    return product_map, trade_flow_map 
+
+if __name__ == "__main__":
+    """
+    testing out this file's functionality in the command line
+    """
+    parser = argparse.ArgumentParser(description='Extracting graph data from the transactions in logistic_data')
+    parser.add_argument('--rs_login', nargs=2, help='Username and password for RedShift, in that order', default = None)
+    parser.add_argument('--hs_digits', nargs='?', help='Number of HS digits to group products by', default = 6,
+                       type = int)
+    parser.add_argument('--agg_type', nargs='?', help= 'entity level representations', default = "product")
+    parser.add_argument('--year', nargs='?', help='Year of data comparison', default = 2020, type = int)
+    args = parser.parse_args()
+
+    rs = RedshiftClass(args.rs_login[0], args.rs_login[1])
+    product_map, trade_flow_map = get_Hitachi_data(rs, args.agg_type, args.hs_digits, args.year, "./")
+    keys = list(trade_flow_map.keys())
+    sample_keys = np.random.choice(range(len(keys)), size = 10)
+    for key_id in sample_keys:
+        key = keys[key_id]
+        print(key, trade_flow_map[key])

temporal_graph/README.md

@@ -0,0 +1,44 @@
+# Temporal Graph Data
+
+This module is for transfiguring the `logistic_data` into a form usable for GNNs—that is, graphs with firms as <em>nodes</em> and time-stamped <em>edges</em> as (aggregated) transactions between firms.
+
+## Environment & Setup
+This should be run on the Hitachi `JupyterHub` server, interlaced with their remote `AWS ec2` data container and RedShift API. Install the following libraries.
+```zsh
+pip install pycountry-convert
+```
+To get vital tables for the products, countries, and companies that appear in the Hitachi data, run this:
+```zsh
+python extract_tables.py --rs_login <Redshift username> <Redshift password> --dir ./
+```
+It will save three dictionaries (`hitachi_{company, country, product}_mappers.json`) to this directory.
+
+## Acquire Graph Data
+For instance, to get time-stamped transactions starting from `2019-01-01` with `2` days aggregated per time stamp, and `10` time stamps worth of data (e.g. end date is `2019-01-20` with `20` total days), run the following script:
+```zsh
+python extract_graph_data.py --rs_login <Redshift username> <Redshift password> --start_date 2019-01-01 \
+--length_timestamps 2 --num_timestamps 10 --fname out.csv
+```
+This will save out the time-stamped edges as a spreadsheet to `out.csv`, where each row represents `{length_timestamps}` days worth of transactions of a particular HS6 product between two firms. An example is shown below.
+
+time_stamp  |  hs6  |  supplier_id           |  buyer_id   | total_amount | ...
+------------|-------------|---------------------|----------------- |---- | ---
+0      |  850760     |  company A  |  company B | 30 | ...
+1   |  850760     |  company A |  company C | 40 | ...
+2    |  850450  |  company B |  company A | 50 | ...
+
+The `time_stamp` column indicates the row includes transactions between `{time_stamp} * {length_timestamp}` and `({time_stamp}+1)* {length_timestamps} - 1` days <b>after</b> the `{start_date}`, inclusive. To alchemize the company IDs (e.g. supplier_id, buyer_id) into their company names, add the `--use_titles` flag to the above command. 
+
+## Transform into PyG Temporal Graph
+From the .csv file saved using `extract_graph_data.py` (<b>make sure</b> to have included the `--use_titles` flag), you'll want to use our `dataloading` module to transform it into a PyG graph. 
+```python
+from dataloading import SupplyChainDataset
+
+data = SupplyChainDataset("out.csv", start_date = "2022-01-01", length_timestamps = 2, metric = "total_amount")
+priorGraph, nextGraph = data.loadData(current_date = "2022-01-10", prior_days = 6, next_days = 4)
+
+for timestep, snapshot in enumerate(priorGraph): #iterate through temporal graph
+    print(type(snapshot)) #<class 'torch_geometric.data.hetero_data.HeteroData'>
+```
+
+This will load data from the last 6 days of `2022-01-10` (Jan 5th to Jan 10th) and the next 4 days (Jan 11th to 14th) into `priorGraph` and `nextGraph`, respectively. These are `DynamicHeteroGraphTemporalSignal` iterator objects from PyG Temporal, with firms as nodes and product-heterogeneous, dynamic edges that are time-stamped. Each time iteration corresponds to a PyG `HeteroData` graph. See the source code at `dataloading.py` for details.

temporal_graph/constants.py

@@ -0,0 +1,24 @@
+BATTERY_PARTS_DICT = {
+    'Aluminum foil': ['760410','760612','760719','761699'],
+    'Anode Paste': ['281122','282110','282300','281700'],
+    'Copper foil': ['740811','740822','740919','740921','740990','741011','741220','741533'],
+    'Cathode': ['810590','280430'],
+    'Cathode Paste': ['282200'],
+    'Electrolyte': ['382499','292390','284210','290123','280440','382490'],
+    'Battery Separator': ['390210','390230','401699'],
+    'Anode': ['790700','854519'],
+    'Cylinder Cell': ['731100'],
+    'Spacer': ['392310','392020'],
+    'Wire': ['831120','831190'],
+    'Cases': ['721240','722230','722699','730120','730690','730890','731816','731822','732599','732619','732620','391732','391910','391990','392051','392069','392099','392690','848049','860900','854720','420212'],
+    'Nickel Tab': ['750522','750610','750620'],
+    'Battery Cell': ['850640','850660','850680'],
+    'BMS':['850450','850730','850780','850790','853222','853223','853321','853340','853630','853641','853890','854190','854239','854290','854411','854442','854449'],
+    'BATTERY': ['850760']
+}
+
+BATTERY_PARTS_DICT_INV = {value.zfill(6):key for key in BATTERY_PARTS_DICT.keys() for value in BATTERY_PARTS_DICT[key]}
+BATTERY_RELATED_CODES = [code for row in list(BATTERY_PARTS_DICT.values()) for code in row]
+
+
+