From 80da73ad845c277bf2984e2babe7bc9f550e1ac8 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 1 Jul 2023 01:51:19 +0000 Subject: [PATCH 01/14] added preprocessing code --- temporal_graph/README.md | 21 ++++ temporal_graph/extract_tables.py | 179 +++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 temporal_graph/README.md create mode 100644 temporal_graph/extract_tables.py diff --git a/temporal_graph/README.md b/temporal_graph/README.md new file mode 100644 index 0000000..4c27462 --- /dev/null +++ b/temporal_graph/README.md @@ -0,0 +1,21 @@ +# Temporal Graph Data + +This module is for transfiguring the `logistic_data` into a form usable for GNNs—that is, graphs with firms as nodes and time-stamped edges 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 --dir ./ +``` +It will save three dictionaries (`hitachi_{company, country, product}_mappers.json`) to this directory (`./temporal_graph`). + +## Acquiring Graph Data +TODO + + + + diff --git a/temporal_graph/extract_tables.py b/temporal_graph/extract_tables.py new file mode 100644 index 0000000..bd3194f --- /dev/null +++ b/temporal_graph/extract_tables.py @@ -0,0 +1,179 @@ +""" +this file extracts mapping tables (for companies, products, countries) for the entities +that appear in the Hitachi logistic_data +""" + +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 +from pycountry_convert import country_name_to_country_alpha3 + +def get_country_table(redshift_instance): + """ + Generates a map from country names as they appear in logistic_data to country + ISO alpha-3 codes, which are the international convention. + + Args: + redshift_instance (RedshiftClass): An instance of Redshift Class, + + Returns: + dict[str:str]: A dictionary from country names (key) to ISO alpha-3 codes (value). + """ + + #query for countries that appear as importers and exporters for transactions + df_importer = rs.query_df("select dest_country as country from logistic_data GROUP BY country") + df_exporter = rs.query_df("select orig_country as country from logistic_data GROUP BY country") + + #iterate through all countries and see whether they appear in the pycountry Python library + all_countries = set(df_importer["country"]).union(set(df_exporter["country"])) + country_to_iso3 = {} + for name in all_countries: + try: + country_name_tokens = [word.lower().capitalize() if word != "AND" else word.lower() for word in name.split(" ")] + country_name = " ".join(country_name_tokens) + iso3 = country_name_to_country_alpha3(country_name, cn_name_format="default") + country_to_iso3[name] = iso3 + except: + continue + + #manual labelling of countries whose names cannot be readily mapped. + country_to_iso3[""] = "" #for error avoidance as countries can appear blank for some transactions + country_to_iso3["KOSOVO"] = "XKX" + country_to_iso3["GUINEA-BISSAU"] = "GNB" + country_to_iso3["CHINA MAINLAND"] = "CHN" + country_to_iso3["COCOS (KEELING) ISLANDS"] = "CCK" + country_to_iso3["SAINT HELENA, ASCENSION AND TR"] = "SHN" + country_to_iso3["CONGO (KINSHASA)"] = "COD" + country_to_iso3["CHINA TAIWAN"] = "TWN" + country_to_iso3["CONGO (BRAZZAVILLE)"] = "COG" + country_to_iso3["REUNION"] = "REU" + country_to_iso3["NETHERLANDS ANTILLES"] = "NLD" + country_to_iso3["CHINA MACAO"] = "MAC" + country_to_iso3["CÔTE D'IVOIRE"] = "CIV" + country_to_iso3["SAINT BARTHELEMY"] = "BLM" + country_to_iso3["SAINT VINCENT AND THE GRENADIN"] = "VCT" + country_to_iso3["CHINA HONGKONG"] = "HKG" + country_to_iso3["VIRGIN ISLANDS (BRITISH)"] = "VGB" + + assert set(all_countries) == set(country_to_iso3.keys()) + return country_to_iso3 + +def get_company_table(redshift_instance): + """ + Generates a map from company IDs in logistic_data to their corresponding names / titles. + + Args: + redshift_instance (RedshiftClass): An instance of Redshift Class + + Returns: + company2id (dict[str:str]): A dictionary from company titles to IDs + id2company (dict[str:str]): A dictionary from company IDs to titles + """ + + #query for all companies that appear as either buyers or sellers + query = f"select COUNT(*) as count, supplier_id, supplier_t from logistic_data GROUP BY supplier_id, supplier_t" + df_supplier = rs.query_df(query) + query = f"select COUNT(*) as count, buyer_id, buyer_t from logistic_data GROUP BY buyer_id, buyer_t" + df_buyer = rs.query_df(query) + + #concatenate into one table, remove duplicates + df_combined = pd.concat([df_supplier.rename(columns = {"supplier_id":"company_id", "supplier_t":"company_t"}), + df_buyer.rename(columns = {"buyer_id":"company_id", "buyer_t":"company_t"})]) + df_combined = df_combined.groupby(by=["company_id","company_t"]).sum().reset_index() + + #generate {ID -> company} map and inverse {company -> ID} map + id2company_preliminary = {} #will need post-processing (see below code) + company2id = {} + rows = [list(df_combined[row_name]) for row_name in ["company_id", "company_t", "count"]] + + for company_id, company_t, count in zip(*rows): + company2id[company_t] = company_id #many titles can map to the same ID value + if (company_id in id2company_preliminary): + id2company_preliminary[company_id].append((company_t, count)) #keep track of all exitant titles + else: + id2company_preliminary[company_id] = [(company_t,count)] + + #for each ID, select company title (e.g. samsung vs SAMSUNG) appearing most often among the transactions + id2company = {} + for key in list(id2company_preliminary.keys()): + id2company[key] = sorted(id2company_preliminary[key], key = lambda item: item[1], reverse = True)[0][0] + + return company2id, id2company + +def get_product_table(redshift_instance): + """ + Generates a map from Harmonized System (HS) product codes (at the 2, 4, and 6 digit levels) + to text descriptions. Here, 2, 4, and 6 digits correspond to chapters, headings, and subheadings, + with an increasing level of granularity. + + Args: + redshift_instance (RedshiftClass): An instance of Redshift Class + + Returns: + dict[str:str]: A dictionary mapping between HS product codes (in string form) and + corresponding text descriptions. The codes should be zero-padded + at the front, with a total length of either 2,4, or 6. + + """ + + #query for products that appear in the Hitachi dataset and their descriptions + query = f"select * from hs_category_description" + df_products = redshift_instance.query_df(query) + row_names = ["hs6","category","sub_category","description"] + rows = [list(df_products[row_name]) for row_name in row_names] + + #iterate through all listed products extracted from the query + hscode_to_product = {} + for hs6, category, sub_category, description in zip(*rows): + hs6_code = hs6.zfill(6) #all six digits + hs4_code = str(int(hs6) // 10**2).zfill(4) #first four digits + hs2_code = str(int(hs6) // 10**4).zfill(2) #first two digits + + if (hs6_code not in hscode_to_product): + hscode_to_product[hs6_code] = description + if (hs4_code not in hscode_to_product): + hscode_to_product[hs4_code] = sub_category + if (hs2_code not in hscode_to_product): + hscode_to_product[hs2_code] = category + + #manual labellings + hscode_to_product["77"] = "Reserved for possible future use" + hscode_to_product["98"] = "Special Classification Provisions" + + return hscode_to_product + +if __name__ == "__main__": + + parser = argparse.ArgumentParser(description='Extracting tables for companies, countries, and products in logistic_data') + parser.add_argument('--dir', nargs='?', help='Where to store the company, country, and product tables', default = "./") + parser.add_argument('--rs_login', nargs=2, help='Username and password for RedShift, in that order', default = None) + args = parser.parse_args() + + #create the RedShift class instance based on the user-provided login, and retrieve tables + rs = RedshiftClass(args.rs_login[0], args.rs_login[1]) + product_table = get_product_table(rs) + company2id, id2company = get_company_table(rs) + company_table = {"company2id": company2id, "id2company": id2company} + country_table = get_country_table(rs) + + #save out the tables + with open(os.path.join(args.dir, "hitachi_product_mappers.json"),"w") as file: + json.dump(product_table, file, indent = 4) + with open(os.path.join(args.dir, "hitachi_company_mappers.json"),"w") as file: + json.dump(company_table, file, indent = 4) + with open(os.path.join(args.dir, "hitachi_country_mappers.json"),"w") as file: + json.dump(country_table, file, indent = 4) + + print("Saved out the tables to {}".format(args.dir)) + + + + \ No newline at end of file From dc6fd109f5081297ad56384827ec45faa67a56f5 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 1 Jul 2023 05:55:11 +0000 Subject: [PATCH 02/14] added edge dataset extraction code --- temporal_graph/README.md | 17 +++++-- temporal_graph/constants.py | 24 ++++++++++ temporal_graph/extract_graph_data.py | 71 ++++++++++++++++++++++++++++ temporal_graph/extract_tables.py | 28 +++++------ 4 files changed, 120 insertions(+), 20 deletions(-) create mode 100644 temporal_graph/constants.py create mode 100644 temporal_graph/extract_graph_data.py diff --git a/temporal_graph/README.md b/temporal_graph/README.md index 4c27462..bb9d193 100644 --- a/temporal_graph/README.md +++ b/temporal_graph/README.md @@ -11,11 +11,20 @@ To get vital tables for the products, countries, and companies that appear in th ```zsh python extract_tables.py --rs_login --dir ./ ``` -It will save three dictionaries (`hitachi_{company, country, product}_mappers.json`) to this directory (`./temporal_graph`). +It will save three dictionaries (`hitachi_{company, country, product}_mappers.json`) to this directory. ## Acquiring Graph Data -TODO - - +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 --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 | ... +------------|-------------|---------------------|----------------- |---- | --- +2.0 | 850760 | company A | company B | 30 | ... +4.0 | 850760 | company A | company C | 40 | ... +6.0 | 850450 | company B | company A | 50 | ... +The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. \ No newline at end of file diff --git a/temporal_graph/constants.py b/temporal_graph/constants.py new file mode 100644 index 0000000..cb9b38d --- /dev/null +++ b/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] + + + diff --git a/temporal_graph/extract_graph_data.py b/temporal_graph/extract_graph_data.py new file mode 100644 index 0000000..a01caab --- /dev/null +++ b/temporal_graph/extract_graph_data.py @@ -0,0 +1,71 @@ +""" +This file is for querying logistic_data and turning the result into an edge spreadsheet +(where each row represents a time-stamped, aggregated transaction between two firms or nodes). +Run python extract_graph_data.py -h to see details on argument passing + +dev note: limited to a small subset of products for now (e.g. battery-related codes) due to scale +""" + +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 +from constants import BATTERY_RELATED_CODES + +def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timestamps = 1, num_timestamps = 1, + hs6_codes = [850760], verbose = True): + + PRIMARY_KEY = 'date, supplier_id, buyer_id, quantity, weight, price, amount, hs_code' + SECONDARY_KEY = 'time_stamp, hs6, supplier_id, buyer_id' + K = float(length_timestamps) + max_day = length_timestamps * num_timestamps + + #restrict transactions to a certain list of product codes + product_condition = "" + for code in hs6_codes: product_condition += f"hs_code like '{code}%' OR " + product_condition = product_condition[:-3] #get rid of last OR at the tail + + #restrict the transactions to the specified time period, and deduplicate + query = f"select {PRIMARY_KEY}, DATEDIFF(day, '{start_date}', date) as time_interval, COUNT(*) as count, \ + COUNT(DISTINCT id) as num_ids from logistic_data WHERE ({product_condition}) AND time_interval \ + BETWEEN 0 AND {max_day-1} GROUP BY {PRIMARY_KEY}, time_interval" + + #aggregate based on the cadence specified by length_timestamps (number of days between consecutive time stamps) + query = f"select CEILING((time_interval + 1) / {K}) * {K} as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ + supplier_id, buyer_id, COUNT(*) as bill_count, SUM(quantity) as total_quantity, SUM(amount) as total_amount,\ + SUM(weight) as total_weight from ({query}) GROUP BY {SECONDARY_KEY} ORDER BY time_stamp" + + date_format = '%Y-%m-%d' + final_date = datetime.datetime.strptime(start_date, date_format) + datetime.timedelta(days = int(max_day) - 1) + final_date = final_date.strftime(date_format) + if verbose == True: print("Querying logistic_data between {} and {}".format(start_date, final_date)) + start_t = time.time() + df = rs.query_df(query) + end_t = time.time() + if verbose == True: print("Retrieved {} rows from logistic_data in {:.3f} seconds".format(len(df), end_t - start_t)) + return df + +if __name__ == "__main__": + + 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('--start_date', help='Starting day of form YY-MM-DD from which to collect data', default = "2019-01-01") + parser.add_argument('--length_timestamps', help='Number of days to aggregate per time stamp', default = 1, type = int) + parser.add_argument('--num_timestamps', help='Number of time stamps to retrieve', default = 1, type = int) + parser.add_argument('--fname', help='Path to the .csv file for storing the resulting data', default = None) + args = parser.parse_args() + + rs = RedshiftClass(args.rs_login[0], args.rs_login[1]) + df = retrieve_timestamped_data(rs, args.start_date, args.length_timestamps, args.num_timestamps, + hs6_codes = BATTERY_RELATED_CODES) + print(df.head(10)) + df.to_csv(args.fname, index = False) \ No newline at end of file diff --git a/temporal_graph/extract_tables.py b/temporal_graph/extract_tables.py index bd3194f..985ba72 100644 --- a/temporal_graph/extract_tables.py +++ b/temporal_graph/extract_tables.py @@ -16,21 +16,21 @@ import pandas as pd from pycountry_convert import country_name_to_country_alpha3 -def get_country_table(redshift_instance): +def get_country_table(redshift): """ Generates a map from country names as they appear in logistic_data to country ISO alpha-3 codes, which are the international convention. Args: - redshift_instance (RedshiftClass): An instance of Redshift Class, + redshift (RedshiftClass): An instance of Redshift Class, Returns: dict[str:str]: A dictionary from country names (key) to ISO alpha-3 codes (value). """ #query for countries that appear as importers and exporters for transactions - df_importer = rs.query_df("select dest_country as country from logistic_data GROUP BY country") - df_exporter = rs.query_df("select orig_country as country from logistic_data GROUP BY country") + df_importer = redshift.query_df("select dest_country as country from logistic_data GROUP BY country") + df_exporter = redshift.query_df("select orig_country as country from logistic_data GROUP BY country") #iterate through all countries and see whether they appear in the pycountry Python library all_countries = set(df_importer["country"]).union(set(df_exporter["country"])) @@ -66,12 +66,12 @@ def get_country_table(redshift_instance): assert set(all_countries) == set(country_to_iso3.keys()) return country_to_iso3 -def get_company_table(redshift_instance): +def get_company_table(redshift): """ Generates a map from company IDs in logistic_data to their corresponding names / titles. Args: - redshift_instance (RedshiftClass): An instance of Redshift Class + redshift (RedshiftClass): An instance of Redshift Class Returns: company2id (dict[str:str]): A dictionary from company titles to IDs @@ -80,9 +80,9 @@ def get_company_table(redshift_instance): #query for all companies that appear as either buyers or sellers query = f"select COUNT(*) as count, supplier_id, supplier_t from logistic_data GROUP BY supplier_id, supplier_t" - df_supplier = rs.query_df(query) + df_supplier = redshift.query_df(query) query = f"select COUNT(*) as count, buyer_id, buyer_t from logistic_data GROUP BY buyer_id, buyer_t" - df_buyer = rs.query_df(query) + df_buyer = redshift.query_df(query) #concatenate into one table, remove duplicates df_combined = pd.concat([df_supplier.rename(columns = {"supplier_id":"company_id", "supplier_t":"company_t"}), @@ -108,14 +108,14 @@ def get_company_table(redshift_instance): return company2id, id2company -def get_product_table(redshift_instance): +def get_product_table(redshift): """ Generates a map from Harmonized System (HS) product codes (at the 2, 4, and 6 digit levels) to text descriptions. Here, 2, 4, and 6 digits correspond to chapters, headings, and subheadings, with an increasing level of granularity. Args: - redshift_instance (RedshiftClass): An instance of Redshift Class + redshift (RedshiftClass): An instance of Redshift Class Returns: dict[str:str]: A dictionary mapping between HS product codes (in string form) and @@ -126,7 +126,7 @@ def get_product_table(redshift_instance): #query for products that appear in the Hitachi dataset and their descriptions query = f"select * from hs_category_description" - df_products = redshift_instance.query_df(query) + df_products = redshift.query_df(query) row_names = ["hs6","category","sub_category","description"] rows = [list(df_products[row_name]) for row_name in row_names] @@ -172,8 +172,4 @@ def get_product_table(redshift_instance): with open(os.path.join(args.dir, "hitachi_country_mappers.json"),"w") as file: json.dump(country_table, file, indent = 4) - print("Saved out the tables to {}".format(args.dir)) - - - - \ No newline at end of file + print("Saved out the tables to {}".format(args.dir)) \ No newline at end of file From 8a07fb7f895014f42392f5f8cbbfea91be0ff7bf Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 1 Jul 2023 18:30:45 +0000 Subject: [PATCH 03/14] added table extraction code --- temporal_graph/README.md | 2 +- temporal_graph/extract_graph_data.py | 53 ++++++++++++++++++++++++++-- 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/temporal_graph/README.md b/temporal_graph/README.md index bb9d193..a812409 100644 --- a/temporal_graph/README.md +++ b/temporal_graph/README.md @@ -27,4 +27,4 @@ time_stamp | hs6 | supplier_id | buyer_id | total_amount | ... 4.0 | 850760 | company A | company C | 40 | ... 6.0 | 850450 | company B | company A | 50 | ... -The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. \ No newline at end of file +The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. To alchemize the company IDs (e.g. supplier_id, buyer_id) into their company names, add the `--use_titles` flag to the above command. \ No newline at end of file diff --git a/temporal_graph/extract_graph_data.py b/temporal_graph/extract_graph_data.py index a01caab..50f360d 100644 --- a/temporal_graph/extract_graph_data.py +++ b/temporal_graph/extract_graph_data.py @@ -21,11 +21,44 @@ import datetime from constants import BATTERY_RELATED_CODES +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 retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timestamps = 1, num_timestamps = 1, hs6_codes = [850760], verbose = True): + """ + This will obtain a dataframe of time-stamped edges, which represent aggregated transactions between firms + + Args: + redshift (RedshiftClass): An instance of Redshift Class + start_date (str): The earliest date (YY-MM-DD) from which to retrieve transactions + length_timestamps (int): The number of days to aggregate per time stamp + num_timestamps (int): The number of time stamps to retrieve from logistic_data (i.e. the last day transactions + will be retrieved from is length_timestamps * num_timestamps - 1 days after start_date) + hs6_codes (List[int]): A list of HS6 products to narrow the transactions search + verbose (bool): Whether to print out status updates (to console) from the retrieval + + Returns: + pd.Dataframe: A dataframe where each row is an aggregated, time-stamped transactions between a supplier and buyer, + with details such as total_amount, total_weight, etc. + """ PRIMARY_KEY = 'date, supplier_id, buyer_id, quantity, weight, price, amount, hs_code' - SECONDARY_KEY = 'time_stamp, hs6, supplier_id, buyer_id' + AGGREGATION_KEY = 'time_stamp, hs6, supplier_id, buyer_id' K = float(length_timestamps) max_day = length_timestamps * num_timestamps @@ -42,7 +75,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest #aggregate based on the cadence specified by length_timestamps (number of days between consecutive time stamps) query = f"select CEILING((time_interval + 1) / {K}) * {K} as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ supplier_id, buyer_id, COUNT(*) as bill_count, SUM(quantity) as total_quantity, SUM(amount) as total_amount,\ - SUM(weight) as total_weight from ({query}) GROUP BY {SECONDARY_KEY} ORDER BY time_stamp" + SUM(weight) as total_weight from ({query}) GROUP BY {AGGREGATION_KEY} ORDER BY time_stamp" date_format = '%Y-%m-%d' final_date = datetime.datetime.strptime(start_date, date_format) + datetime.timedelta(days = int(max_day) - 1) @@ -62,10 +95,26 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest parser.add_argument('--length_timestamps', help='Number of days to aggregate per time stamp', default = 1, type = int) parser.add_argument('--num_timestamps', help='Number of time stamps to retrieve', default = 1, type = int) parser.add_argument('--fname', help='Path to the .csv file for storing the resulting data', default = None) + parser.add_argument('--use_titles', help = 'if provided, data uses company titles instead of IDs', action='store_true') args = parser.parse_args() rs = RedshiftClass(args.rs_login[0], args.rs_login[1]) df = retrieve_timestamped_data(rs, args.start_date, args.length_timestamps, args.num_timestamps, hs6_codes = BATTERY_RELATED_CODES) + + #create the company ID -> name mapper, and replace the IDs in the dataframe with company titles + if (args.use_titles == True): + id2company = retrieve_Hitachi_table(name = "company")["id2company"] + company_ids = list(id2company.keys()) + df_companies = pd.DataFrame.from_dict({"company_id": company_ids, + "company_t": [id2company[id] for id in company_ids]}) + + #replace the company supplier IDs + df = pd.merge(df, df_companies, left_on = "supplier_id", right_on = "company_id", how = "left") + df = df.rename(columns = {"company_t": "supplier_t"}).drop(columns = {"company_id","supplier_id"}) + #replace the company buyer IDs + df = pd.merge(df, df_companies, left_on = "buyer_id", right_on = "company_id", how = "left") + df = df.rename(columns = {"company_t": "buyer_t"}).drop(columns = {"company_id","buyer_id"}) + print(df.head(10)) df.to_csv(args.fname, index = False) \ No newline at end of file From ee2618045864fda835336f363b5282216100692f Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 1 Jul 2023 22:58:54 +0000 Subject: [PATCH 04/14] fixed table bug --- temporal_graph/extract_graph_data.py | 5 +++-- temporal_graph/extract_tables.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/temporal_graph/extract_graph_data.py b/temporal_graph/extract_graph_data.py index 50f360d..1802c81 100644 --- a/temporal_graph/extract_graph_data.py +++ b/temporal_graph/extract_graph_data.py @@ -65,11 +65,11 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest #restrict transactions to a certain list of product codes product_condition = "" for code in hs6_codes: product_condition += f"hs_code like '{code}%' OR " - product_condition = product_condition[:-3] #get rid of last OR at the tail + product_condition = "({}) AND".format(product_condition[:-3]) #get rid of last OR at the tail #restrict the transactions to the specified time period, and deduplicate query = f"select {PRIMARY_KEY}, DATEDIFF(day, '{start_date}', date) as time_interval, COUNT(*) as count, \ - COUNT(DISTINCT id) as num_ids from logistic_data WHERE ({product_condition}) AND time_interval \ + COUNT(DISTINCT id) as num_ids from logistic_data WHERE {product_condition} time_interval \ BETWEEN 0 AND {max_day-1} GROUP BY {PRIMARY_KEY}, time_interval" #aggregate based on the cadence specified by length_timestamps (number of days between consecutive time stamps) @@ -101,6 +101,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest rs = RedshiftClass(args.rs_login[0], args.rs_login[1]) df = retrieve_timestamped_data(rs, args.start_date, args.length_timestamps, args.num_timestamps, hs6_codes = BATTERY_RELATED_CODES) + df = df[df["hs6"].str.match('^(?!00)[0-9]{6}')] #check for valid HS6 product codes via regex #create the company ID -> name mapper, and replace the IDs in the dataframe with company titles if (args.use_titles == True): diff --git a/temporal_graph/extract_tables.py b/temporal_graph/extract_tables.py index 985ba72..f71f355 100644 --- a/temporal_graph/extract_tables.py +++ b/temporal_graph/extract_tables.py @@ -133,7 +133,7 @@ def get_product_table(redshift): #iterate through all listed products extracted from the query hscode_to_product = {} for hs6, category, sub_category, description in zip(*rows): - hs6_code = hs6.zfill(6) #all six digits + hs6_code = str(int(hs6)).zfill(6) #all six digits hs4_code = str(int(hs6) // 10**2).zfill(4) #first four digits hs2_code = str(int(hs6) // 10**4).zfill(2) #first two digits From 998bf18b42217aaa16e095a50dd08e5198b06b34 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Mon, 3 Jul 2023 23:16:33 +0000 Subject: [PATCH 05/14] added data loading with PyG Temporal --- temporal_graph/README.md | 15 ++- temporal_graph/dataloading.py | 190 ++++++++++++++++++++++++++++++++++ 2 files changed, 203 insertions(+), 2 deletions(-) create mode 100644 temporal_graph/dataloading.py diff --git a/temporal_graph/README.md b/temporal_graph/README.md index a812409..e468cf0 100644 --- a/temporal_graph/README.md +++ b/temporal_graph/README.md @@ -13,7 +13,7 @@ python extract_tables.py --rs_login --di ``` It will save three dictionaries (`hitachi_{company, country, product}_mappers.json`) to this directory. -## Acquiring Graph Data +## 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 --start_date 2019-01-01 \ @@ -27,4 +27,15 @@ time_stamp | hs6 | supplier_id | buyer_id | total_amount | ... 4.0 | 850760 | company A | company C | 40 | ... 6.0 | 850450 | company B | company A | 50 | ... -The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. To alchemize the company IDs (e.g. supplier_id, buyer_id) into their company names, add the `--use_titles` flag to the above command. \ No newline at end of file +The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. 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`, 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) +``` + +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` objects from PyG Temporal, with firms as nodes and product-heterogeneous, dynamic edges that are time-stamped. See the source code at `dataloading.py` for details. \ No newline at end of file diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py new file mode 100644 index 0000000..a0eb492 --- /dev/null +++ b/temporal_graph/dataloading.py @@ -0,0 +1,190 @@ +""" +This file is for alchemizing the .csv file of edges produced in extract_graph_data.py into Temporal +Graph data structures from the PyTorch Geometric library (see https://pytorch-geometric-temporal.readthedocs.io/en/latest/modules/signal.html) +""" + +import datetime +import numpy as np +from torch_geometric_temporal import DynamicHeteroGraphTemporalSignal +import pandas as pd +import networkx as nx + +def build_transaction_network(): + """ + TODO: create a NetworkX MultiDiGraph for graph parsing and general exploration / visualisation + """ + pass + +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 get_forward_date(date_1, num_days_ahead, date_format = "%Y-%m-%d"): + """ + retrieves the date (default format YY-MM-DD) days ahead of date_1. If + is negative, then it will retrieve a date before date_1. + """ + date_2 = datetime.datetime.strptime(date_1, date_format) + datetime.timedelta(days = num_days_ahead) + return date_2.strftime(date_format) + +class SupplyChainDataset(object): + + def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = "total_amount"): + """ + Constructor method for the temporal Supply Chain dataset, which formats the .csv from + extract_graph_data.py into graph-based data structures from torch_geometric_temporal + + Args: + edges_csv_file (str): the path to the CSV file storing transaction edges + start_date (str): start date from data was gathered (date corresponding to a time_stamp of 0) + length_timestamps (int): number of days aggregated per time stamp + metric (str): the metric (out of total_amount, total_quantity, bill_count, total_weight) + to assign to each edge as its "weight" + + Returns: + None + """ + all_metrics = {"total_amount", "total_quantity", "bill_count", "total_weight"} + if metric not in all_metrics: + raise ValueError(f"Metric must be {all_metrics}") + self.start_date = start_date + self.length_ts = length_timestamps + + #read the dataframe and excise missing supplier & buyer nodes, as well as values + df = pd.read_csv(edges_csv_file) + row_names = ["time_stamp","supplier_t","buyer_t","hs6", metric] + df = df[row_names][(~df[metric].isna()) & (df[metric] > 0) & (df["supplier_t"] != "") & (df["buyer_t"] != "")] + self.num_edges_total = len(df) + companies = list(set(df["supplier_t"]).union(set(df["buyer_t"]))) + self.num_nodes_total = len(companies) + + #establish nodes (using an ID dictionary) and heterogeneous edge information (using two parallel + #dictionaries storing the edge indices and metric / weights, respectively) + self.company_to_nodeID = {company_t: node_id for node_id, company_t in enumerate(companies)} + self.nodeID_to_company = {value:key for key,value in self.company_to_nodeID.items()} + self.edge_index_dict = {} + self.edge_weight_dict = {} + rows = [list(df[name]) for name in row_names] #for index, row in tqdm(df.iterrows()): + for time_stamp, supplier, buyer, product, amount in zip(*rows): + time_stamp = int(time_stamp) + product = str(int(product)).zfill(6) + #directed edge from supplier to buyer + material_edge = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] + #cash_edge = [self.company_to_nodeID[buyer], self.company_to_nodeID[supplier]] + if (time_stamp not in self.edge_index_dict): + self.edge_index_dict[time_stamp] = {product: [material_edge]} + self.edge_weight_dict[time_stamp] = {product: [amount]} + + elif (product not in self.edge_index_dict[time_stamp]): + self.edge_index_dict[time_stamp][product] = [material_edge] + self.edge_weight_dict[time_stamp][product] = [amount] + + else: + self.edge_index_dict[time_stamp][product].append(material_edge) + self.edge_weight_dict[time_stamp][product].append(amount) + + self.prevalent_timestamps = set(self.edge_index_dict.keys()) + + #get total input/output for each node at each timestep + #input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product + input_entity = "supplier_t" if metric == "total_amount" else "buyer_t" + output_entity = "buyer_t" if metric == "total_amount" else "supplier_t" + df_node_input = df.groupby(by = ["time_stamp", input_entity]).sum(numeric_only = True).reset_index() + df_node_output = df.groupby(by = ["time_stamp", output_entity]).sum(numeric_only = True).reset_index() + + node_input_rows = [list(df_node_input[row]) for row in ["time_stamp",input_entity,metric]] + node_output_rows = [list(df_node_output[row]) for row in ["time_stamp",output_entity,metric]] + self.node_features = {timestamp: np.zeros(shape = (self.num_nodes_total, 2)) for timestamp in self.prevalent_timestamps} + for time_stamp, firm, metric in zip(*node_input_rows): + self.node_features[time_stamp][self.company_to_nodeID[firm]][0] = metric + for time_stamp, firm, metric in zip(*node_output_rows): + self.node_features[time_stamp][self.company_to_nodeID[firm]][1] = metric + #dev note: change to sparse matrix implementation for larger data + + def get_edge_index_dict(self,time_stamp): + if (time_stamp not in self.prevalent_timestamps): + return {"firm": None} + edges_dict = self.edge_index_dict[time_stamp] + unique_products = list(edges_dict.keys()) #different edge relation types + return {("firm",product,"firm"): np.transpose(np.array(edges_dict[product])) for product in unique_products} + + def get_edge_weight_dict(self,time_stamp): + if (time_stamp not in self.prevalent_timestamps): + return {"firm": None} + weight_dict = self.edge_weight_dict[time_stamp] + unique_products = list(weight_dict.keys()) #different edge relation types + return {("firm",product,"firm"): np.array(weight_dict[product]) for product in unique_products} + + def get_date_range(self,time_stamp): + lower_date = get_forward_date(self.start_date, time_stamp - self.length_ts) + upper_date = get_forward_date(self.start_date, time_stamp - 1) + return [lower_date, upper_date] + + def getTemporalGraph(self, time_stamps): + """ + given a list of time_stamps, produces a PyG temporal graph covering time-stamped transaction edges + between node firms. Here, edge weights are the metric (e.g. total_amount, bill_count) given + to the constructor method, and node targets are the total input/output of each firm at each time. + The different edge relations correspond to different HS6 products (see get_edge_index_dict above), + and date_ranges corresponds to the interval of dates aggregated for each time_stamp. Use + self.company_to_nodeID and self.nodeID_to_company to convert between node indices and firm names. + """ + edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] + edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] + feature_dicts = [{"firm": None} for _ in range(len(time_stamps))] #not clear what node / firm features should be atm + target_dicts = [{"firm":self.node_features[ts].copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] + date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] + + tempGraph = DynamicHeteroGraphTemporalSignal(edge_index_dicts = edge_index_dicts, + edge_weight_dicts = edge_weight_dicts, + feature_dicts = feature_dicts, + target_dicts = target_dicts, + date_ranges = date_ranges) + return tempGraph + + def loadData(self, current_date, prior_days, next_days): + """ + This data function will provide two temporal graphs, the first as past input to a forecasting model, + and the second as the ground truth corresponding to future predictions (see getTemporalGraph above for graph info) + + note: for well-defined behavior, you'll want the difference (in days) between self.start_date + and current_date to be one less than a multiple of self.length_ts (length of timestamps) + i.e. -1 modulo self.length_ts. Same goes for the integer values of prior_days and next_days. + + Args: + current_date (str): The current_date from which we aim to predict the next days directly after + prior_days (int): The number of past days to extract, including current_date + next_days (int): The number of days after current_date that we want to predict + + Returns: + priorGraph (DynamicHeteroGraphTemporalSignal): a temporal Graph of product-stratified edges between firm nodes + for transactions occurring up to (prior_days - 1) days before the current_date (i.e. current_date is included) + nextGraph (DynamicHeteroGraphTemporalSignal): a temporal Graph of product-stratified edges between firm nodes + for transactions occurring up to next_days after the current_date (i.e. current_date is NOT included) + """ + days_after_start = get_days_between(self.start_date, current_date) + prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) * self.length_ts + prior_latest_ts = (days_after_start + 1) // self.length_ts * self.length_ts + prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, self.length_ts)) + + next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) * self.length_ts + next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts * self.length_ts + next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, self.length_ts)) + + priorGraph = self.getTemporalGraph(prior_timestamps) + nextGraph = self.getTemporalGraph(next_timestamps) + + return priorGraph, nextGraph + +if __name__ == "__main__": + + obj = SupplyChainDataset("out.csv", "2022-01-01", 1, "total_amount") + print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Edges: {obj.num_edges_total}") + priorGraph, nextGraph = obj.loadData(current_date = "2022-01-10", prior_days = 2, next_days = 3) + + + From b119834b24f319fd1ec9e9d7c70a4fd26671a1e5 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Thu, 6 Jul 2023 05:58:11 +0000 Subject: [PATCH 06/14] scaled to all product transactions --- temporal_graph/README.md | 9 ++++++--- temporal_graph/extract_graph_data.py | 22 +++++++++++----------- 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/temporal_graph/README.md b/temporal_graph/README.md index e468cf0..e9cfe67 100644 --- a/temporal_graph/README.md +++ b/temporal_graph/README.md @@ -27,15 +27,18 @@ time_stamp | hs6 | supplier_id | buyer_id | total_amount | ... 4.0 | 850760 | company A | company C | 40 | ... 6.0 | 850450 | company B | company A | 50 | ... -The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after the `{start_date}`, inclusive. As a note, for now, the script only gathers data for battery-related products due to scale. To alchemize the company IDs (e.g. supplier_id, buyer_id) into their company names, add the `--use_titles` flag to the above command. +The `time_stamp` column indicates the row includes transactions between `{time_stamp} - {length_timestamps}` and `{time_stamp}-1` days after 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`, you'll want to use our `dataloading` module to transform it into a PyG graph. +From the .csv file saved using `extract_graph_data.py` (make sure 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)) # ``` -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` objects from PyG Temporal, with firms as nodes and product-heterogeneous, dynamic edges that are time-stamped. See the source code at `dataloading.py` for details. \ No newline at end of file +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. \ No newline at end of file diff --git a/temporal_graph/extract_graph_data.py b/temporal_graph/extract_graph_data.py index 1802c81..bc7c6a1 100644 --- a/temporal_graph/extract_graph_data.py +++ b/temporal_graph/extract_graph_data.py @@ -2,8 +2,6 @@ This file is for querying logistic_data and turning the result into an edge spreadsheet (where each row represents a time-stamped, aggregated transaction between two firms or nodes). Run python extract_graph_data.py -h to see details on argument passing - -dev note: limited to a small subset of products for now (e.g. battery-related codes) due to scale """ import os @@ -39,7 +37,7 @@ def retrieve_Hitachi_table(name = "company", dir = "."): return table def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timestamps = 1, num_timestamps = 1, - hs6_codes = [850760], verbose = True): + hs6_restriction = [850760], verbose = True): """ This will obtain a dataframe of time-stamped edges, which represent aggregated transactions between firms @@ -49,7 +47,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest length_timestamps (int): The number of days to aggregate per time stamp num_timestamps (int): The number of time stamps to retrieve from logistic_data (i.e. the last day transactions will be retrieved from is length_timestamps * num_timestamps - 1 days after start_date) - hs6_codes (List[int]): A list of HS6 products to narrow the transactions search + hs6_restriction (List[int]): A list of HS6 products to narrow the transactions search. If None, no restrictions. verbose (bool): Whether to print out status updates (to console) from the retrieval Returns: @@ -59,13 +57,14 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest PRIMARY_KEY = 'date, supplier_id, buyer_id, quantity, weight, price, amount, hs_code' AGGREGATION_KEY = 'time_stamp, hs6, supplier_id, buyer_id' - K = float(length_timestamps) + K = float(length_timestamps) # floating point version to enable float (not integer) SQL division max_day = length_timestamps * num_timestamps - #restrict transactions to a certain list of product codes + #restrict transactions to a certain list of product codes if requested product_condition = "" - for code in hs6_codes: product_condition += f"hs_code like '{code}%' OR " - product_condition = "({}) AND".format(product_condition[:-3]) #get rid of last OR at the tail + if (hs6_restriction != None): + for code in hs6_restriction: product_condition += f"hs_code like '{code}%' OR " + product_condition = "({}) AND".format(product_condition[:-3]) #get rid of last OR at the tail #restrict the transactions to the specified time period, and deduplicate query = f"select {PRIMARY_KEY}, DATEDIFF(day, '{start_date}', date) as time_interval, COUNT(*) as count, \ @@ -73,7 +72,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest BETWEEN 0 AND {max_day-1} GROUP BY {PRIMARY_KEY}, time_interval" #aggregate based on the cadence specified by length_timestamps (number of days between consecutive time stamps) - query = f"select CEILING((time_interval + 1) / {K}) * {K} as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ + query = f"select CEILING((time_interval + 1) / {K}) * {length_timestamps} as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ supplier_id, buyer_id, 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} ORDER BY time_stamp" @@ -100,7 +99,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest rs = RedshiftClass(args.rs_login[0], args.rs_login[1]) df = retrieve_timestamped_data(rs, args.start_date, args.length_timestamps, args.num_timestamps, - hs6_codes = BATTERY_RELATED_CODES) + hs6_restriction = None) df = df[df["hs6"].str.match('^(?!00)[0-9]{6}')] #check for valid HS6 product codes via regex #create the company ID -> name mapper, and replace the IDs in the dataframe with company titles @@ -117,5 +116,6 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest df = pd.merge(df, df_companies, left_on = "buyer_id", right_on = "company_id", how = "left") df = df.rename(columns = {"company_t": "buyer_t"}).drop(columns = {"company_id","buyer_id"}) - print(df.head(10)) + print(df.head(5)) + print(df.tail(5)) df.to_csv(args.fname, index = False) \ No newline at end of file From d6ed8b1bfe40b8aa76671f14c96242d83993b815 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Thu, 6 Jul 2023 21:21:06 +0000 Subject: [PATCH 07/14] used sparse matrices for node info --- temporal_graph/dataloading.py | 25 ++++++++++--------------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py index a0eb492..e0eaba3 100644 --- a/temporal_graph/dataloading.py +++ b/temporal_graph/dataloading.py @@ -8,12 +8,7 @@ from torch_geometric_temporal import DynamicHeteroGraphTemporalSignal import pandas as pd import networkx as nx - -def build_transaction_network(): - """ - TODO: create a NetworkX MultiDiGraph for graph parsing and general exploration / visualisation - """ - pass +from scipy.sparse import csr_matrix def get_days_between(date_1, date_2, date_format = "%Y-%m-%d"): """ @@ -72,9 +67,9 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " for time_stamp, supplier, buyer, product, amount in zip(*rows): time_stamp = int(time_stamp) product = str(int(product)).zfill(6) - #directed edge from supplier to buyer + #directed edge from supplier to buyer (flow of material), will create the + #edge from buyer to supplier (flow of cash) on the spot in getTemporalGraph() material_edge = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] - #cash_edge = [self.company_to_nodeID[buyer], self.company_to_nodeID[supplier]] if (time_stamp not in self.edge_index_dict): self.edge_index_dict[time_stamp] = {product: [material_edge]} self.edge_weight_dict[time_stamp] = {product: [amount]} @@ -98,13 +93,13 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " node_input_rows = [list(df_node_input[row]) for row in ["time_stamp",input_entity,metric]] node_output_rows = [list(df_node_output[row]) for row in ["time_stamp",output_entity,metric]] - self.node_features = {timestamp: np.zeros(shape = (self.num_nodes_total, 2)) for timestamp in self.prevalent_timestamps} + self.node_targets = {timestamp: np.zeros((self.num_nodes_total, 2)) for timestamp in self.prevalent_timestamps} for time_stamp, firm, metric in zip(*node_input_rows): - self.node_features[time_stamp][self.company_to_nodeID[firm]][0] = metric + self.node_targets[time_stamp][self.company_to_nodeID[firm],0] = metric for time_stamp, firm, metric in zip(*node_output_rows): - self.node_features[time_stamp][self.company_to_nodeID[firm]][1] = metric - #dev note: change to sparse matrix implementation for larger data - + self.node_targets[time_stamp][self.company_to_nodeID[firm],1] = metric + self.node_targets = {timestamp: csr_matrix(self.node_targets[timestamp]) for timestamp in self.prevalent_timestamps} + def get_edge_index_dict(self,time_stamp): if (time_stamp not in self.prevalent_timestamps): return {"firm": None} @@ -136,7 +131,7 @@ def getTemporalGraph(self, time_stamps): edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] feature_dicts = [{"firm": None} for _ in range(len(time_stamps))] #not clear what node / firm features should be atm - target_dicts = [{"firm":self.node_features[ts].copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] + target_dicts = [{"firm":self.node_targets[ts].todense().copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] tempGraph = DynamicHeteroGraphTemporalSignal(edge_index_dicts = edge_index_dicts, @@ -184,7 +179,7 @@ def loadData(self, current_date, prior_days, next_days): obj = SupplyChainDataset("out.csv", "2022-01-01", 1, "total_amount") print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Edges: {obj.num_edges_total}") - priorGraph, nextGraph = obj.loadData(current_date = "2022-01-10", prior_days = 2, next_days = 3) + priorGraph, nextGraph = obj.loadData(current_date = "2022-01-10", prior_days = 5, next_days = 10) From 6e160b09638a3ac7aa9c037f77fc9cccef1c52bf Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Fri, 7 Jul 2023 00:00:55 +0000 Subject: [PATCH 08/14] added note features & dual edges --- temporal_graph/dataloading.py | 68 ++++++++++++++++++++++++++++------- 1 file changed, 56 insertions(+), 12 deletions(-) diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py index e0eaba3..46d7312 100644 --- a/temporal_graph/dataloading.py +++ b/temporal_graph/dataloading.py @@ -28,7 +28,8 @@ def get_forward_date(date_1, num_days_ahead, date_format = "%Y-%m-%d"): class SupplyChainDataset(object): - def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = "total_amount"): + def __init__(self, edges_csv_file, start_date, length_timestamps = 1, + metric = "total_amount", dual_edge = False, context_size = 5): """ Constructor method for the temporal Supply Chain dataset, which formats the .csv from extract_graph_data.py into graph-based data structures from torch_geometric_temporal @@ -39,6 +40,10 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " length_timestamps (int): number of days aggregated per time stamp metric (str): the metric (out of total_amount, total_quantity, bill_count, total_weight) to assign to each edge as its "weight" + dual_edge (bool): By default, edges are drawn from supplier -> buyer (flow of material). If True, + this will also draw reverse edges from buyer -> supplier (flow of cash). + context_size (int): Number of past days to average firm inputs/outputs over to create node features + for each time_stamp. Returns: None @@ -48,6 +53,8 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " raise ValueError(f"Metric must be {all_metrics}") self.start_date = start_date self.length_ts = length_timestamps + self.dual_edge = dual_edge + self.context_size = context_size #read the dataframe and excise missing supplier & buyer nodes, as well as values df = pd.read_csv(edges_csv_file) @@ -67,8 +74,7 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " for time_stamp, supplier, buyer, product, amount in zip(*rows): time_stamp = int(time_stamp) product = str(int(product)).zfill(6) - #directed edge from supplier to buyer (flow of material), will create the - #edge from buyer to supplier (flow of cash) on the spot in getTemporalGraph() + #directed edge from supplier to buyer (flow of material) material_edge = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] if (time_stamp not in self.edge_index_dict): self.edge_index_dict[time_stamp] = {product: [material_edge]} @@ -88,6 +94,7 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " #input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product input_entity = "supplier_t" if metric == "total_amount" else "buyer_t" output_entity = "buyer_t" if metric == "total_amount" else "supplier_t" + df_node_input = df.groupby(by = ["time_stamp", input_entity]).sum(numeric_only = True).reset_index() df_node_output = df.groupby(by = ["time_stamp", output_entity]).sum(numeric_only = True).reset_index() @@ -99,38 +106,73 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, metric = " for time_stamp, firm, metric in zip(*node_output_rows): self.node_targets[time_stamp][self.company_to_nodeID[firm],1] = metric self.node_targets = {timestamp: csr_matrix(self.node_targets[timestamp]) for timestamp in self.prevalent_timestamps} + #obtain the node features using a past sliding window of firm inputs/outputs + self.assemble_sliding_window() def get_edge_index_dict(self,time_stamp): if (time_stamp not in self.prevalent_timestamps): return {"firm": None} edges_dict = self.edge_index_dict[time_stamp] - unique_products = list(edges_dict.keys()) #different edge relation types - return {("firm",product,"firm"): np.transpose(np.array(edges_dict[product])) for product in unique_products} + unique_products = list(edges_dict.keys()) #different products that delineate edge relations + #constructs forward edges from supplier to buyer (flow of material) + index_dict = {("firm",(product,"material"),"firm"): np.transpose( + np.array(edges_dict[product])) for product in unique_products} + #constructs reverse edges from buyer to supplier (flow of cash) + if self.dual_edge == True: + dual_index_dict = {("firm",(product, "cash"), "firm"): np.flip(np.transpose( + np.array(edges_dict[product])), axis = 0).copy() for product in unique_products} + index_dict.update(dual_index_dict) + return index_dict def get_edge_weight_dict(self,time_stamp): if (time_stamp not in self.prevalent_timestamps): return {"firm": None} weight_dict = self.edge_weight_dict[time_stamp] - unique_products = list(weight_dict.keys()) #different edge relation types - return {("firm",product,"firm"): np.array(weight_dict[product]) for product in unique_products} - + unique_products = list(weight_dict.keys()) #different products that delineate edge relations + edge_weight_dict = {("firm",(product,"material"),"firm"): np.array( + weight_dict[product]) for product in unique_products} + if self.dual_edge == True: + dual_index_dict = {("firm",(product, "cash"), "firm"): np.array( + weight_dict[product]) for product in unique_products} + edge_weight_dict.update(dual_index_dict) + return edge_weight_dict + def get_date_range(self,time_stamp): lower_date = get_forward_date(self.start_date, time_stamp - self.length_ts) upper_date = get_forward_date(self.start_date, time_stamp - 1) return [lower_date, upper_date] + def assemble_sliding_window(self): + #calculates sliding window of past days, averaging input/output for each firm + #stores the averaged values in np.array self.node_IO_table, accessed by time_stamp indices + K = self.context_size + self.node_IO_table = np.zeros(shape = (max(self.prevalent_timestamps), self.num_nodes_total, 2)) + + rolling_sum = np.zeros(shape = (self.num_nodes_total, 2)) + for timestamp in range(1, max(self.prevalent_timestamps) + 1): + #for timestamp t, average each firm's values between timestamps t - K and t - 1 + if (timestamp - 1 in self.prevalent_timestamps): + rolling_sum += self.node_targets[timestamp - 1].todense() + if (timestamp - K - 1 in self.prevalent_timestamps): + rolling_sum -= self.node_targets[timestamp - K - 1].todense() + + #divide by the number of days the amounts are averaged over + self.node_IO_table[timestamp - 1] = rolling_sum / min(K, timestamp - 1) if timestamp > 1 else 0 + def getTemporalGraph(self, time_stamps): """ given a list of time_stamps, produces a PyG temporal graph covering time-stamped transaction edges between node firms. Here, edge weights are the metric (e.g. total_amount, bill_count) given - to the constructor method, and node targets are the total input/output of each firm at each time. + to the constructor method, node targets are the total input/output of each firm at each time. For now, + node features are the average input/output of the past days of each firm. + The different edge relations correspond to different HS6 products (see get_edge_index_dict above), and date_ranges corresponds to the interval of dates aggregated for each time_stamp. Use self.company_to_nodeID and self.nodeID_to_company to convert between node indices and firm names. """ edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] - feature_dicts = [{"firm": None} for _ in range(len(time_stamps))] #not clear what node / firm features should be atm + feature_dicts = [{"firm":self.node_IO_table[ts - 1].copy()} if (ts > 1 and ts <= max(self.prevalent_timestamps)) else None for ts in time_stamps] target_dicts = [{"firm":self.node_targets[ts].todense().copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] @@ -160,6 +202,9 @@ def loadData(self, current_date, prior_days, next_days): for transactions occurring up to (prior_days - 1) days before the current_date (i.e. current_date is included) nextGraph (DynamicHeteroGraphTemporalSignal): a temporal Graph of product-stratified edges between firm nodes for transactions occurring up to next_days after the current_date (i.e. current_date is NOT included) + + **Note that the edge keys are a tuple of the form ("firm", (product, flow direction), "firm"), where flow direction + is "material" for supplier -> buyer and "cash" for buyer -> supplier """ days_after_start = get_days_between(self.start_date, current_date) prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) * self.length_ts @@ -178,8 +223,7 @@ def loadData(self, current_date, prior_days, next_days): if __name__ == "__main__": obj = SupplyChainDataset("out.csv", "2022-01-01", 1, "total_amount") - print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Edges: {obj.num_edges_total}") + print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Supplier \u2192 Buyer Edges: {obj.num_edges_total}") priorGraph, nextGraph = obj.loadData(current_date = "2022-01-10", prior_days = 5, next_days = 10) - From 4ecd11de2db4270ba56ddba3d786f6845d068ba4 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 8 Jul 2023 05:36:40 +0000 Subject: [PATCH 09/14] added utility functions to SC Dataset --- temporal_graph/dataloading.py | 181 +++++++++++++++++++++++++--------- 1 file changed, 136 insertions(+), 45 deletions(-) diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py index 46d7312..229c90a 100644 --- a/temporal_graph/dataloading.py +++ b/temporal_graph/dataloading.py @@ -8,7 +8,7 @@ from torch_geometric_temporal import DynamicHeteroGraphTemporalSignal import pandas as pd import networkx as nx -from scipy.sparse import csr_matrix +from scipy.sparse import csr_matrix, lil_matrix def get_days_between(date_1, date_2, date_format = "%Y-%m-%d"): """ @@ -26,10 +26,25 @@ def get_forward_date(date_1, num_days_ahead, date_format = "%Y-%m-%d"): date_2 = datetime.datetime.strptime(date_1, date_format) + datetime.timedelta(days = num_days_ahead) return date_2.strftime(date_format) +def node_normalize(temporal_node_data, norms = None, epsilon = 10**(-10)): + """ + *temporal_node_data is of the shape [some number of timestamps, num_nodes, 2] + or [num_nodes, 2]. Norms is a dictionary with keys {"mean","std"} and the + corresponding values each have shape [num_nodes, 2] + """ + if (norms == None): return temporal_node_data + return (temporal_node_data - norms["mean"]) / (norms["std"] + epsilon) + +def node_unnormalize(normalized_node_data, norms = None, epsilon = 10**(-10)): + """ analagous parameter recommendations as normalize() above """ + if (norms == None): return normalized_node_data + return normalized_node_data * (norms["std"] + epsilon) + norms["mean"] + + class SupplyChainDataset(object): - def __init__(self, edges_csv_file, start_date, length_timestamps = 1, - metric = "total_amount", dual_edge = False, context_size = 5): + def __init__(self, edges_csv_file: str, start_date: str, length_timestamps: int = 1, + metric: str = "total_amount", dual_edge: bool = False, lags: int = 5): """ Constructor method for the temporal Supply Chain dataset, which formats the .csv from extract_graph_data.py into graph-based data structures from torch_geometric_temporal @@ -42,9 +57,8 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, to assign to each edge as its "weight" dual_edge (bool): By default, edges are drawn from supplier -> buyer (flow of material). If True, this will also draw reverse edges from buyer -> supplier (flow of cash). - context_size (int): Number of past days to average firm inputs/outputs over to create node features - for each time_stamp. - + lags (int): Number of past time_stamps to use for predicting the next time_stamp + Returns: None """ @@ -54,7 +68,7 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, self.start_date = start_date self.length_ts = length_timestamps self.dual_edge = dual_edge - self.context_size = context_size + self.lags = lags #read the dataframe and excise missing supplier & buyer nodes, as well as values df = pd.read_csv(edges_csv_file) @@ -63,6 +77,7 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, self.num_edges_total = len(df) companies = list(set(df["supplier_t"]).union(set(df["buyer_t"]))) self.num_nodes_total = len(companies) + self.num_products_total = len(set(df["hs6"])) #establish nodes (using an ID dictionary) and heterogeneous edge information (using two parallel #dictionaries storing the edge indices and metric / weights, respectively) @@ -78,17 +93,18 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, material_edge = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] if (time_stamp not in self.edge_index_dict): self.edge_index_dict[time_stamp] = {product: [material_edge]} - self.edge_weight_dict[time_stamp] = {product: [amount]} - + self.edge_weight_dict[time_stamp] = {product: [amount]} + elif (product not in self.edge_index_dict[time_stamp]): self.edge_index_dict[time_stamp][product] = [material_edge] self.edge_weight_dict[time_stamp][product] = [amount] - + else: self.edge_index_dict[time_stamp][product].append(material_edge) self.edge_weight_dict[time_stamp][product].append(amount) self.prevalent_timestamps = set(self.edge_index_dict.keys()) + self.max_ts = max(self.prevalent_timestamps) #latest time_stamp with non-empty data #get total input/output for each node at each timestep #input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product @@ -100,14 +116,16 @@ def __init__(self, edges_csv_file, start_date, length_timestamps = 1, node_input_rows = [list(df_node_input[row]) for row in ["time_stamp",input_entity,metric]] node_output_rows = [list(df_node_output[row]) for row in ["time_stamp",output_entity,metric]] + self.node_targets = {timestamp: np.zeros((self.num_nodes_total, 2)) for timestamp in self.prevalent_timestamps} for time_stamp, firm, metric in zip(*node_input_rows): self.node_targets[time_stamp][self.company_to_nodeID[firm],0] = metric for time_stamp, firm, metric in zip(*node_output_rows): self.node_targets[time_stamp][self.company_to_nodeID[firm],1] = metric self.node_targets = {timestamp: csr_matrix(self.node_targets[timestamp]) for timestamp in self.prevalent_timestamps} + #obtain the node features using a past sliding window of firm inputs/outputs - self.assemble_sliding_window() + self.assemble_node_features() def get_edge_index_dict(self,time_stamp): if (time_stamp not in self.prevalent_timestamps): @@ -142,37 +160,42 @@ def get_date_range(self,time_stamp): upper_date = get_forward_date(self.start_date, time_stamp - 1) return [lower_date, upper_date] - def assemble_sliding_window(self): - #calculates sliding window of past days, averaging input/output for each firm - #stores the averaged values in np.array self.node_IO_table, accessed by time_stamp indices - K = self.context_size - self.node_IO_table = np.zeros(shape = (max(self.prevalent_timestamps), self.num_nodes_total, 2)) + def assemble_node_features(self): + """ + calculates sliding window of past time_stamps, averaging input/output for each firm + stores the averaged values in np.array self.node_IO_table, accessed by time_stamp indices + """ + K = self.lags + self.node_IO_table = np.zeros(shape = (max(self.prevalent_timestamps) // self.length_ts, self.num_nodes_total, 2)) rolling_sum = np.zeros(shape = (self.num_nodes_total, 2)) - for timestamp in range(1, max(self.prevalent_timestamps) + 1): - #for timestamp t, average each firm's values between timestamps t - K and t - 1 - if (timestamp - 1 in self.prevalent_timestamps): - rolling_sum += self.node_targets[timestamp - 1].todense() - if (timestamp - K - 1 in self.prevalent_timestamps): - rolling_sum -= self.node_targets[timestamp - K - 1].todense() + for timestamp in range(self.length_ts, max(self.prevalent_timestamps) + 1, self.length_ts): + #for the (t)th timestamp, average each firm's values between the (t - K)th and (t - 1)th timestamps + if (timestamp - self.length_ts in self.prevalent_timestamps): + rolling_sum += self.node_targets[timestamp - self.length_ts].todense() + if (timestamp - (K + 1) * self.length_ts in self.prevalent_timestamps): + rolling_sum -= self.node_targets[timestamp - (K + 1) * self.length_ts].todense() - #divide by the number of days the amounts are averaged over - self.node_IO_table[timestamp - 1] = rolling_sum / min(K, timestamp - 1) if timestamp > 1 else 0 + #divide by the number of time_stamps the amounts are averaged over + self.node_IO_table[timestamp // self.length_ts - 1] = rolling_sum / min( + K, timestamp // self.length_ts - 1) if timestamp > self.length_ts else 0 def getTemporalGraph(self, time_stamps): """ given a list of time_stamps, produces a PyG temporal graph covering time-stamped transaction edges between node firms. Here, edge weights are the metric (e.g. total_amount, bill_count) given to the constructor method, node targets are the total input/output of each firm at each time. For now, - node features are the average input/output of the past days of each firm. + node features are the average input/output of the past days of each firm. - The different edge relations correspond to different HS6 products (see get_edge_index_dict above), - and date_ranges corresponds to the interval of dates aggregated for each time_stamp. Use - self.company_to_nodeID and self.nodeID_to_company to convert between node indices and firm names. + The edge relations are tuples of the form ("firm", (product, flow direction), "firm"), where flow + direction is "material" for supplier -> buyer and "cash" for buyer -> supplier. Date_ranges corresponds + to the interval of dates aggregated for each time_stamp. Use self.company_to_nodeID and + self.nodeID_to_company to convert between node indices and firm names. """ edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] - feature_dicts = [{"firm":self.node_IO_table[ts - 1].copy()} if (ts > 1 and ts <= max(self.prevalent_timestamps)) else None for ts in time_stamps] + feature_dicts = [{"firm":self.node_IO_table[ts // self.length_ts - 1].copy()} if ( + ts > self.length_ts and ts <= max(self.prevalent_timestamps)) else None for ts in time_stamps] target_dicts = [{"firm":self.node_targets[ts].todense().copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] @@ -182,6 +205,21 @@ def getTemporalGraph(self, time_stamps): target_dicts = target_dicts, date_ranges = date_ranges) return tempGraph + + def segment_timestamps(self, current_date, prior_days, next_days): + """ + retrieve two lists of time_stamps corresponding to the days before , + and the following days of the dataset, respectively. + """ + days_after_start = get_days_between(self.start_date, current_date) + prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) * self.length_ts + prior_latest_ts = (days_after_start + 1) // self.length_ts * self.length_ts + prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, self.length_ts)) + + next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) * self.length_ts + next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts * self.length_ts + next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, self.length_ts)) + return prior_timestamps, next_timestamps def loadData(self, current_date, prior_days, next_days): """ @@ -202,28 +240,81 @@ def loadData(self, current_date, prior_days, next_days): for transactions occurring up to (prior_days - 1) days before the current_date (i.e. current_date is included) nextGraph (DynamicHeteroGraphTemporalSignal): a temporal Graph of product-stratified edges between firm nodes for transactions occurring up to next_days after the current_date (i.e. current_date is NOT included) - - **Note that the edge keys are a tuple of the form ("firm", (product, flow direction), "firm"), where flow direction - is "material" for supplier -> buyer and "cash" for buyer -> supplier """ - days_after_start = get_days_between(self.start_date, current_date) - prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) * self.length_ts - prior_latest_ts = (days_after_start + 1) // self.length_ts * self.length_ts - prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, self.length_ts)) - - next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) * self.length_ts - next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts * self.length_ts - next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, self.length_ts)) - + prior_timestamps, next_timestamps = self.segment_timestamps(current_date, prior_days, next_days) priorGraph = self.getTemporalGraph(prior_timestamps) nextGraph = self.getTemporalGraph(next_timestamps) return priorGraph, nextGraph +class SupplyChainDatasetPredictive(SupplyChainDataset): + """ + time-lagged version of the Supply Chain Dataset, where node targets are set + time_stamps into the future, rather than the present values for the corresponding time_stamp + + TODO: alternatively, one can still use present values by setting use_present_labels == True, + and the node features instead will be comprised of the past days + """ + def __init__(self, *args, **kwargs): + super(SupplyChainDatasetPredictive,self).__init__(*args, **kwargs) + + def ts2index(self, time_stamp): #converts from a time_stamp to a place in a storage arraay + return time_stamp // self.length_ts - 1 + + # override function in the parent class + def assemble_node_features(self): + #storing the input/outputs of each firm at each timestamp in a sparse matrix + self.node_IO_table = lil_matrix((self.max_ts // self.length_ts, self.num_nodes_total * 2), dtype = np.float32) + for timestamp in range(self.length_ts, max(self.prevalent_timestamps) + 1, self.length_ts): + if (timestamp in self.prevalent_timestamps): + self.node_IO_table[self.ts2index(timestamp)] = self.node_targets[timestamp].reshape(1,-1) + else: + self.node_IO_table[self.ts2index(timestamp)] = 0 #missing data for that timestamp + + def get_node_normalizations(self, timestamps: list): + if (len(timestamps) <= 1): return None + time_indices = [self.ts2index(ts) for ts in timestamps] + node_IO_selected = np.array(self.node_IO_table[time_indices].todense()).reshape(-1, self.num_nodes_total, 2) + mean, std = np.mean(node_IO_selected, axis = 0), np.std(node_IO_selected, axis = 0) + return {"mean": mean, "std": std} + + # override function in the parent class + def getTemporalGraph(self, time_stamps, norms): + edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] + edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] + + #set the features to be a panorama of the next - 1 time_stamps, including the current one + feature_dicts = [{"firm": node_normalize(np.array(self.node_IO_table[self.ts2index(ts): self.ts2index(ts) + self.lags,:].todense()).reshape( + -1, self.num_nodes_total, 2), norms)} if (self.lags <= self.ts2index(ts) + self.lags <= self.ts2index(self.max_ts) + 1) else None for ts in time_stamps] + + #set the targets to be node inputs/outputs time_stamps into the future + target_dicts = [{"firm": node_normalize(np.array(self.node_IO_table[self.ts2index(ts) + self.lags,:].todense()).reshape(-1,2), norms)} if ( + 0 <= self.ts2index(ts) + self.lags <= self.ts2index(self.max_ts)) else None for ts in time_stamps] + + date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] + tempGraph = DynamicHeteroGraphTemporalSignal(edge_index_dicts = edge_index_dicts, + edge_weight_dicts = edge_weight_dicts, + feature_dicts = feature_dicts, + target_dicts = target_dicts, + date_ranges = date_ranges) + return tempGraph + + # override function in the parent class + def loadData(self, current_date, prior_days, next_days, normalize = True): + prior_timestamps, next_timestamps = self.segment_timestamps(current_date, prior_days, next_days) + # set normalization values based on valid prior_timestamps (using for training) + norms = self.get_node_normalizations([ts for ts in prior_timestamps if 0 <= self.ts2index(ts) <= self.ts2index(self.max_ts)]) + priorGraph = self.getTemporalGraph(prior_timestamps, norms if normalize == True else None) + nextGraph = self.getTemporalGraph(next_timestamps, norms if normalize == True else None) + + return priorGraph, nextGraph, norms + if __name__ == "__main__": - obj = SupplyChainDataset("out.csv", "2022-01-01", 1, "total_amount") + obj = SupplyChainDatasetPredictive("daily_transactions_2019.csv", "2019-01-01", 1, + metric = "total_amount", lags = 5) print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Supplier \u2192 Buyer Edges: {obj.num_edges_total}") - priorGraph, nextGraph = obj.loadData(current_date = "2022-01-10", prior_days = 5, next_days = 10) - - + priorGraph, nextGraph, norms = obj.loadData(current_date = "2019-03-10", prior_days = 50, next_days = 10) + for timestep, snapshot in enumerate(priorGraph): + print(snapshot) + break From c87085c77a7f9fbfedccd8ee27e0b1e2b63b21d0 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 8 Jul 2023 06:22:44 +0000 Subject: [PATCH 10/14] update README --- compare_data/README.md | 14 ++++++++++++-- compare_data/constants.py | 5 ----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/compare_data/README.md b/compare_data/README.md index f7aa0d0..46aceba 100644 --- a/compare_data/README.md +++ b/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,14 @@ 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`. \ No newline at end of file +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) +To acquire Hitachi values from the `logistic_data` (instead of `index_hs6` above), one can run: +```python +import read_logistic + + +``` + +address scale limitations for dyad aggregations \ No newline at end of file diff --git a/compare_data/constants.py b/compare_data/constants.py index 057353f..13ba097 100644 --- a/compare_data/constants.py +++ b/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, From 22921210496eaa793ae47884edabef6ead3f090b Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sat, 8 Jul 2023 06:41:49 +0000 Subject: [PATCH 11/14] aligned time_stamps with array indices --- temporal_graph/README.md | 8 ++++---- temporal_graph/extract_graph_data.py | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/temporal_graph/README.md b/temporal_graph/README.md index e9cfe67..98cf7cf 100644 --- a/temporal_graph/README.md +++ b/temporal_graph/README.md @@ -23,11 +23,11 @@ This will save out the time-stamped edges as a spreadsheet to `out.csv`, where e time_stamp | hs6 | supplier_id | buyer_id | total_amount | ... ------------|-------------|---------------------|----------------- |---- | --- -2.0 | 850760 | company A | company B | 30 | ... -4.0 | 850760 | company A | company C | 40 | ... -6.0 | 850450 | company B | company A | 50 | ... +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_timestamps}` and `{time_stamp}-1` days after 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. +The `time_stamp` column indicates the row includes transactions between `{time_stamp} * {length_timestamp}` and `({time_stamp}+1)* {length_timestamps} - 1` days after 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` (make sure to have included the `--use_titles` flag), you'll want to use our `dataloading` module to transform it into a PyG graph. diff --git a/temporal_graph/extract_graph_data.py b/temporal_graph/extract_graph_data.py index bc7c6a1..2b3563a 100644 --- a/temporal_graph/extract_graph_data.py +++ b/temporal_graph/extract_graph_data.py @@ -72,7 +72,7 @@ def retrieve_timestamped_data(redshift, start_date = "2019-01-01", length_timest BETWEEN 0 AND {max_day-1} GROUP BY {PRIMARY_KEY}, time_interval" #aggregate based on the cadence specified by length_timestamps (number of days between consecutive time stamps) - query = f"select CEILING((time_interval + 1) / {K}) * {length_timestamps} as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ + query = f"select CEILING((time_interval + 1) / {K}) - 1 as time_stamp, SUBSTRING(hs_code, 1, 6) as hs6,\ supplier_id, buyer_id, 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} ORDER BY time_stamp" From d2870f35def0a858a4f53b1825acc0654af94a9b Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Sun, 9 Jul 2023 09:12:48 +0000 Subject: [PATCH 12/14] added logistic data aggregations --- compare_data/README.md | 15 +++- compare_data/read_logistic.py | 161 ++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 3 deletions(-) create mode 100644 compare_data/read_logistic.py diff --git a/compare_data/README.md b/compare_data/README.md index 46aceba..f981d7c 100644 --- a/compare_data/README.md +++ b/compare_data/README.md @@ -79,11 +79,20 @@ print(trade_flows_2020[(exporter_country,product_code)]) 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) -To acquire Hitachi values from the `logistic_data` (instead of `index_hs6` above), one can run: +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 +``` +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')]) -address scale limitations for dyad aggregations \ No newline at end of file +"""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`). \ No newline at end of file diff --git a/compare_data/read_logistic.py b/compare_data/read_logistic.py new file mode 100644 index 0000000..813e5f8 --- /dev/null +++ b/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]) \ No newline at end of file From a9d5e1f051fb93425c808acfdc8491805155b27b Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Tue, 11 Jul 2023 05:26:13 +0000 Subject: [PATCH 13/14] refactored dataloader --- temporal_graph/dataloading.py | 227 ++++++++++++---------------------- 1 file changed, 82 insertions(+), 145 deletions(-) diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py index 229c90a..fd61b0d 100644 --- a/temporal_graph/dataloading.py +++ b/temporal_graph/dataloading.py @@ -9,6 +9,7 @@ import pandas as pd import networkx as nx from scipy.sparse import csr_matrix, lil_matrix +import time def get_days_between(date_1, date_2, date_format = "%Y-%m-%d"): """ @@ -43,8 +44,8 @@ def node_unnormalize(normalized_node_data, norms = None, epsilon = 10**(-10)): class SupplyChainDataset(object): - def __init__(self, edges_csv_file: str, start_date: str, length_timestamps: int = 1, - metric: str = "total_amount", dual_edge: bool = False, lags: int = 5): + def __init__(self, csv_file, start_date, length_timestamps = 1, + metric = "total_amount", dual_edge = False, lags = 5): """ Constructor method for the temporal Supply Chain dataset, which formats the .csv from extract_graph_data.py into graph-based data structures from torch_geometric_temporal @@ -69,15 +70,15 @@ def __init__(self, edges_csv_file: str, start_date: str, length_timestamps: int self.length_ts = length_timestamps self.dual_edge = dual_edge self.lags = lags - + #read the dataframe and excise missing supplier & buyer nodes, as well as values - df = pd.read_csv(edges_csv_file) + df = pd.read_csv(csv_file) row_names = ["time_stamp","supplier_t","buyer_t","hs6", metric] df = df[row_names][(~df[metric].isna()) & (df[metric] > 0) & (df["supplier_t"] != "") & (df["buyer_t"] != "")] - self.num_edges_total = len(df) companies = list(set(df["supplier_t"]).union(set(df["buyer_t"]))) - self.num_nodes_total = len(companies) - self.num_products_total = len(set(df["hs6"])) + self.num_nodes, self.num_edges, self.num_products = len(companies), len(df), len(set(df["hs6"])) + self.prevalent_timestamps = set(df["time_stamp"]) + self.max_ts = max(self.prevalent_timestamps) #establish nodes (using an ID dictionary) and heterogeneous edge information (using two parallel #dictionaries storing the edge indices and metric / weights, respectively) @@ -103,9 +104,6 @@ def __init__(self, edges_csv_file: str, start_date: str, length_timestamps: int self.edge_index_dict[time_stamp][product].append(material_edge) self.edge_weight_dict[time_stamp][product].append(amount) - self.prevalent_timestamps = set(self.edge_index_dict.keys()) - self.max_ts = max(self.prevalent_timestamps) #latest time_stamp with non-empty data - #get total input/output for each node at each timestep #input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product input_entity = "supplier_t" if metric == "total_amount" else "buyer_t" @@ -113,115 +111,104 @@ def __init__(self, edges_csv_file: str, start_date: str, length_timestamps: int df_node_input = df.groupby(by = ["time_stamp", input_entity]).sum(numeric_only = True).reset_index() df_node_output = df.groupby(by = ["time_stamp", output_entity]).sum(numeric_only = True).reset_index() - node_input_rows = [list(df_node_input[row]) for row in ["time_stamp",input_entity,metric]] node_output_rows = [list(df_node_output[row]) for row in ["time_stamp",output_entity,metric]] - self.node_targets = {timestamp: np.zeros((self.num_nodes_total, 2)) for timestamp in self.prevalent_timestamps} + self.node_features = np.zeros(shape = (self.max_ts + 1, self.num_nodes, 2), dtype = np.float32) for time_stamp, firm, metric in zip(*node_input_rows): - self.node_targets[time_stamp][self.company_to_nodeID[firm],0] = metric + self.node_features[int(time_stamp), self.company_to_nodeID[firm], 0] = metric for time_stamp, firm, metric in zip(*node_output_rows): - self.node_targets[time_stamp][self.company_to_nodeID[firm],1] = metric - self.node_targets = {timestamp: csr_matrix(self.node_targets[timestamp]) for timestamp in self.prevalent_timestamps} - - #obtain the node features using a past sliding window of firm inputs/outputs - self.assemble_node_features() - - def get_edge_index_dict(self,time_stamp): + self.node_features[int(time_stamp), self.company_to_nodeID[firm], 1] = metric + + def get_edge_index_dict(self, time_stamp): if (time_stamp not in self.prevalent_timestamps): - return {"firm": None} + return None edges_dict = self.edge_index_dict[time_stamp] - unique_products = list(edges_dict.keys()) #different products that delineate edge relations + unique_products = list(edges_dict.keys()) #constructs forward edges from supplier to buyer (flow of material) index_dict = {("firm",(product,"material"),"firm"): np.transpose( np.array(edges_dict[product])) for product in unique_products} + #constructs reverse edges from buyer to supplier (flow of cash) if self.dual_edge == True: dual_index_dict = {("firm",(product, "cash"), "firm"): np.flip(np.transpose( np.array(edges_dict[product])), axis = 0).copy() for product in unique_products} index_dict.update(dual_index_dict) return index_dict - - def get_edge_weight_dict(self,time_stamp): + + def get_edge_weight_dict(self, time_stamp): if (time_stamp not in self.prevalent_timestamps): - return {"firm": None} + return None weight_dict = self.edge_weight_dict[time_stamp] unique_products = list(weight_dict.keys()) #different products that delineate edge relations edge_weight_dict = {("firm",(product,"material"),"firm"): np.array( weight_dict[product]) for product in unique_products} + if self.dual_edge == True: dual_index_dict = {("firm",(product, "cash"), "firm"): np.array( weight_dict[product]) for product in unique_products} edge_weight_dict.update(dual_index_dict) return edge_weight_dict - + def get_date_range(self,time_stamp): - lower_date = get_forward_date(self.start_date, time_stamp - self.length_ts) - upper_date = get_forward_date(self.start_date, time_stamp - 1) + lower_date = get_forward_date(self.start_date, self.length_ts * time_stamp) + upper_date = get_forward_date(self.start_date, self.length_ts * (time_stamp + 1) - 1) return [lower_date, upper_date] - def assemble_node_features(self): + def segment_timestamps(self, current_date, prior_days, next_days): """ - calculates sliding window of past time_stamps, averaging input/output for each firm - stores the averaged values in np.array self.node_IO_table, accessed by time_stamp indices + retrieve two lists of time_stamps corresponding to the days before , + and the following days of the dataset, respectively. """ - K = self.lags - self.node_IO_table = np.zeros(shape = (max(self.prevalent_timestamps) // self.length_ts, self.num_nodes_total, 2)) + days_after_start = get_days_between(self.start_date, current_date) + prior_earliest_ts = np.ceil((days_after_start - prior_days + 1) / self.length_ts) + #prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) - 1 + prior_latest_ts = (days_after_start + 1) // self.length_ts - 1 + prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, 1)) - rolling_sum = np.zeros(shape = (self.num_nodes_total, 2)) - for timestamp in range(self.length_ts, max(self.prevalent_timestamps) + 1, self.length_ts): - #for the (t)th timestamp, average each firm's values between the (t - K)th and (t - 1)th timestamps - if (timestamp - self.length_ts in self.prevalent_timestamps): - rolling_sum += self.node_targets[timestamp - self.length_ts].todense() - if (timestamp - (K + 1) * self.length_ts in self.prevalent_timestamps): - rolling_sum -= self.node_targets[timestamp - (K + 1) * self.length_ts].todense() - - #divide by the number of time_stamps the amounts are averaged over - self.node_IO_table[timestamp // self.length_ts - 1] = rolling_sum / min( - K, timestamp // self.length_ts - 1) if timestamp > self.length_ts else 0 - - def getTemporalGraph(self, time_stamps): + #next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) - 1 + next_earliest_ts = np.ceil((days_after_start + 1) / self.length_ts) + next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts - 1 + next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, 1)) + return prior_timestamps, next_timestamps + + def get_node_norms(self, time_stamps): + if (len(time_stamps) <= 1): + return None + node_features = self.node_features[time_stamps] + mean, std = np.mean(node_features, axis = 0), np.std(node_features, axis = 0) + return {"mean": mean, "std": std} + + def getTemporalGraph(self, time_stamps, norms = None): """ given a list of time_stamps, produces a PyG temporal graph covering time-stamped transaction edges between node firms. Here, edge weights are the metric (e.g. total_amount, bill_count) given to the constructor method, node targets are the total input/output of each firm at each time. For now, - node features are the average input/output of the past days of each firm. + node features are the concatenated input/output of the past days of each firm. The edge relations are tuples of the form ("firm", (product, flow direction), "firm"), where flow direction is "material" for supplier -> buyer and "cash" for buyer -> supplier. Date_ranges corresponds - to the interval of dates aggregated for each time_stamp. Use self.company_to_nodeID and + to the interval of dates aggregated for each time_stamp. Use self.company_to_nodeID and self.nodeID_to_company to convert between node indices and firm names. """ - edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] - edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] - feature_dicts = [{"firm":self.node_IO_table[ts // self.length_ts - 1].copy()} if ( - ts > self.length_ts and ts <= max(self.prevalent_timestamps)) else None for ts in time_stamps] - target_dicts = [{"firm":self.node_targets[ts].todense().copy()} if ts in self.prevalent_timestamps else None for ts in time_stamps ] date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] + edge_index_dicts = [self.get_edge_index_dict(ts - self.lags) for ts in time_stamps] + edge_weight_dicts = [self.get_edge_weight_dict(ts - self.lags) for ts in time_stamps] + + #TODO: rework these to accomodate backward lag and normalizations + feature_dicts = [{"firm": node_normalize(self.node_features[ts - self.lags: ts], norms) if ( + self.lags <= ts <= self.max_ts) else None} for ts in time_stamps] + target_dicts = [{"firm": node_normalize(self.node_features[ts], norms) if ( + 0 <= ts <= self.max_ts) else None} for ts in time_stamps] tempGraph = DynamicHeteroGraphTemporalSignal(edge_index_dicts = edge_index_dicts, - edge_weight_dicts = edge_weight_dicts, - feature_dicts = feature_dicts, - target_dicts = target_dicts, - date_ranges = date_ranges) + edge_weight_dicts = edge_weight_dicts, + feature_dicts = feature_dicts, + target_dicts = target_dicts, + date_ranges = date_ranges) return tempGraph - - def segment_timestamps(self, current_date, prior_days, next_days): - """ - retrieve two lists of time_stamps corresponding to the days before , - and the following days of the dataset, respectively. - """ - days_after_start = get_days_between(self.start_date, current_date) - prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) * self.length_ts - prior_latest_ts = (days_after_start + 1) // self.length_ts * self.length_ts - prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, self.length_ts)) - - next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) * self.length_ts - next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts * self.length_ts - next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, self.length_ts)) - return prior_timestamps, next_timestamps - - def loadData(self, current_date, prior_days, next_days): + + def loadData(self, current_date, prior_days, next_days, normalize = True ): """ This data function will provide two temporal graphs, the first as past input to a forecasting model, and the second as the ground truth corresponding to future predictions (see getTemporalGraph above for graph info) @@ -242,79 +229,29 @@ def loadData(self, current_date, prior_days, next_days): for transactions occurring up to next_days after the current_date (i.e. current_date is NOT included) """ prior_timestamps, next_timestamps = self.segment_timestamps(current_date, prior_days, next_days) - priorGraph = self.getTemporalGraph(prior_timestamps) - nextGraph = self.getTemporalGraph(next_timestamps) - - return priorGraph, nextGraph - -class SupplyChainDatasetPredictive(SupplyChainDataset): - """ - time-lagged version of the Supply Chain Dataset, where node targets are set - time_stamps into the future, rather than the present values for the corresponding time_stamp - - TODO: alternatively, one can still use present values by setting use_present_labels == True, - and the node features instead will be comprised of the past days - """ - def __init__(self, *args, **kwargs): - super(SupplyChainDatasetPredictive,self).__init__(*args, **kwargs) - - def ts2index(self, time_stamp): #converts from a time_stamp to a place in a storage arraay - return time_stamp // self.length_ts - 1 - - # override function in the parent class - def assemble_node_features(self): - #storing the input/outputs of each firm at each timestamp in a sparse matrix - self.node_IO_table = lil_matrix((self.max_ts // self.length_ts, self.num_nodes_total * 2), dtype = np.float32) - for timestamp in range(self.length_ts, max(self.prevalent_timestamps) + 1, self.length_ts): - if (timestamp in self.prevalent_timestamps): - self.node_IO_table[self.ts2index(timestamp)] = self.node_targets[timestamp].reshape(1,-1) - else: - self.node_IO_table[self.ts2index(timestamp)] = 0 #missing data for that timestamp + #set normalization values based on lagged prior time_stamps (no intersection with next days) + lagged_priors = [ts - self.lags for ts in prior_timestamps if 0 <= ts - self.lags <= self.max_ts] + norms = self.get_node_norms(lagged_priors) if normalize == True else None - def get_node_normalizations(self, timestamps: list): - if (len(timestamps) <= 1): return None - time_indices = [self.ts2index(ts) for ts in timestamps] - node_IO_selected = np.array(self.node_IO_table[time_indices].todense()).reshape(-1, self.num_nodes_total, 2) - mean, std = np.mean(node_IO_selected, axis = 0), np.std(node_IO_selected, axis = 0) - return {"mean": mean, "std": std} - - # override function in the parent class - def getTemporalGraph(self, time_stamps, norms): - edge_index_dicts = [self.get_edge_index_dict(ts) for ts in time_stamps] - edge_weight_dicts = [self.get_edge_weight_dict(ts) for ts in time_stamps] - - #set the features to be a panorama of the next - 1 time_stamps, including the current one - feature_dicts = [{"firm": node_normalize(np.array(self.node_IO_table[self.ts2index(ts): self.ts2index(ts) + self.lags,:].todense()).reshape( - -1, self.num_nodes_total, 2), norms)} if (self.lags <= self.ts2index(ts) + self.lags <= self.ts2index(self.max_ts) + 1) else None for ts in time_stamps] - - #set the targets to be node inputs/outputs time_stamps into the future - target_dicts = [{"firm": node_normalize(np.array(self.node_IO_table[self.ts2index(ts) + self.lags,:].todense()).reshape(-1,2), norms)} if ( - 0 <= self.ts2index(ts) + self.lags <= self.ts2index(self.max_ts)) else None for ts in time_stamps] - - date_ranges = [{"firm": np.array(self.get_date_range(ts))} for ts in time_stamps] - tempGraph = DynamicHeteroGraphTemporalSignal(edge_index_dicts = edge_index_dicts, - edge_weight_dicts = edge_weight_dicts, - feature_dicts = feature_dicts, - target_dicts = target_dicts, - date_ranges = date_ranges) - return tempGraph - - # override function in the parent class - def loadData(self, current_date, prior_days, next_days, normalize = True): - prior_timestamps, next_timestamps = self.segment_timestamps(current_date, prior_days, next_days) - # set normalization values based on valid prior_timestamps (using for training) - norms = self.get_node_normalizations([ts for ts in prior_timestamps if 0 <= self.ts2index(ts) <= self.ts2index(self.max_ts)]) - priorGraph = self.getTemporalGraph(prior_timestamps, norms if normalize == True else None) - nextGraph = self.getTemporalGraph(next_timestamps, norms if normalize == True else None) - + priorGraph = self.getTemporalGraph(prior_timestamps, norms) + nextGraph = self.getTemporalGraph(next_timestamps, norms) return priorGraph, nextGraph, norms -if __name__ == "__main__": - obj = SupplyChainDatasetPredictive("daily_transactions_2019.csv", "2019-01-01", 1, - metric = "total_amount", lags = 5) - print(f"Total Firms: {obj.num_nodes_total}\nTotal Time-Stamped Supplier \u2192 Buyer Edges: {obj.num_edges_total}") - priorGraph, nextGraph, norms = obj.loadData(current_date = "2019-03-10", prior_days = 50, next_days = 10) +if __name__ == "__main__": + """command line testing""" + start_t = time.time() + fname = "./storage/daily_transactions_2021.csv" + sc = SupplyChainDataset(fname, start_date = "2021-01-01", dual_edge = False, lags = 5) + end_t = time.time() + print(f"Processed dataset at {fname} in {end_t - start_t:.2f} seconds.") + print(f"Total Firms: {sc.num_nodes}\nTotal Time-Stamped Supplier \u2192 Buyer Edges: {sc.num_edges}") + priorGraph, nextGraph, norms = sc.loadData(current_date = "2021-03-10", prior_days = 50, next_days = 10) + for timestep, snapshot in enumerate(priorGraph): - print(snapshot) - break + #print(snapshot["firm"].x.shape) + snapshot_map = snapshot.to_dict() + for key in list(snapshot_map.keys())[:4]: + print(key, {label: feature if label not in ["x","y"] else feature.shape for label, feature in snapshot_map[key].items()}) + print() + break \ No newline at end of file From 7c30dde218803c27d777d3ebe897a636562605a1 Mon Sep 17 00:00:00 2001 From: Ben Yan Date: Wed, 12 Jul 2023 18:59:39 +0000 Subject: [PATCH 14/14] added firm-level homogeneous network --- temporal_graph/dataloading.py | 172 ++++++++++++++++++++++++++-------- 1 file changed, 135 insertions(+), 37 deletions(-) diff --git a/temporal_graph/dataloading.py b/temporal_graph/dataloading.py index fd61b0d..1279361 100644 --- a/temporal_graph/dataloading.py +++ b/temporal_graph/dataloading.py @@ -5,11 +5,13 @@ import datetime import numpy as np -from torch_geometric_temporal import DynamicHeteroGraphTemporalSignal +from torch_geometric_temporal import DynamicHeteroGraphTemporalSignal, DynamicGraphTemporalSignal import pandas as pd import networkx as nx from scipy.sparse import csr_matrix, lil_matrix import time +import torch +import os def get_days_between(date_1, date_2, date_format = "%Y-%m-%d"): """ @@ -26,7 +28,7 @@ def get_forward_date(date_1, num_days_ahead, date_format = "%Y-%m-%d"): """ date_2 = datetime.datetime.strptime(date_1, date_format) + datetime.timedelta(days = num_days_ahead) return date_2.strftime(date_format) - + def node_normalize(temporal_node_data, norms = None, epsilon = 10**(-10)): """ *temporal_node_data is of the shape [some number of timestamps, num_nodes, 2] @@ -34,18 +36,20 @@ def node_normalize(temporal_node_data, norms = None, epsilon = 10**(-10)): corresponding values each have shape [num_nodes, 2] """ if (norms == None): return temporal_node_data + #winsorize the data + return (temporal_node_data - norms["mean"]) / (norms["std"] + epsilon) + #return (temporal_node_data - np.min(temporal_node_data, axis = (0,1))) / (np.max(temporal_node_data, axis = (0,1)) - np.min(temporal_node_data, axis = (0,1))) def node_unnormalize(normalized_node_data, norms = None, epsilon = 10**(-10)): """ analagous parameter recommendations as normalize() above """ if (norms == None): return normalized_node_data return normalized_node_data * (norms["std"] + epsilon) + norms["mean"] - class SupplyChainDataset(object): def __init__(self, csv_file, start_date, length_timestamps = 1, - metric = "total_amount", dual_edge = False, lags = 5): + metric = "total_amount", edge_type = "material", lags = 5): """ Constructor method for the temporal Supply Chain dataset, which formats the .csv from extract_graph_data.py into graph-based data structures from torch_geometric_temporal @@ -56,19 +60,22 @@ def __init__(self, csv_file, start_date, length_timestamps = 1, length_timestamps (int): number of days aggregated per time stamp metric (str): the metric (out of total_amount, total_quantity, bill_count, total_weight) to assign to each edge as its "weight" - dual_edge (bool): By default, edges are drawn from supplier -> buyer (flow of material). If True, - this will also draw reverse edges from buyer -> supplier (flow of cash). - lags (int): Number of past time_stamps to use for predicting the next time_stamp + edge_type (str): Out of {'material', 'cash', 'both'}. By default, edges are drawn from supplier -> buyer (flow of + material). Reverse edges from buyer -> supplier (flow of cash) can also be drawn. + + lags (int): Number of past time_stamps to use for predicting the next time_stamp Returns: None """ all_metrics = {"total_amount", "total_quantity", "bill_count", "total_weight"} if metric not in all_metrics: - raise ValueError(f"Metric must be {all_metrics}") + raise ValueError(f"metric must be in {all_metrics}") self.start_date = start_date self.length_ts = length_timestamps - self.dual_edge = dual_edge + if (edge_type not in ["material","cash","both"]): + raise ValueError("edge_type must be in {'material','cash','both'}") + self.edge_type = edge_type self.lags = lags #read the dataframe and excise missing supplier & buyer nodes, as well as values @@ -80,32 +87,39 @@ def __init__(self, csv_file, start_date, length_timestamps = 1, self.prevalent_timestamps = set(df["time_stamp"]) self.max_ts = max(self.prevalent_timestamps) - #establish nodes (using an ID dictionary) and heterogeneous edge information (using two parallel - #dictionaries storing the edge indices and metric / weights, respectively) + #establish node IDs by mapping firms to a unique indices, and obtain edge and node features self.company_to_nodeID = {company_t: node_id for node_id, company_t in enumerate(companies)} self.nodeID_to_company = {value:key for key,value in self.company_to_nodeID.items()} + self.create_edge_attributes(df, metric) + self.create_node_features(df, metric) + + def create_edge_attributes(self, df, metric): + """procure edge information by using two parallel dictionaries to store edge indices and weights""" self.edge_index_dict = {} self.edge_weight_dict = {} - rows = [list(df[name]) for name in row_names] #for index, row in tqdm(df.iterrows()): + row_names = ["time_stamp","supplier_t","buyer_t","hs6", metric] + rows = [list(df[name]) for name in row_names] for time_stamp, supplier, buyer, product, amount in zip(*rows): time_stamp = int(time_stamp) product = str(int(product)).zfill(6) + #directed edge from supplier to buyer (flow of material) - material_edge = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] + edge_material_flow = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] if (time_stamp not in self.edge_index_dict): - self.edge_index_dict[time_stamp] = {product: [material_edge]} + self.edge_index_dict[time_stamp] = {product: [edge_material_flow]} self.edge_weight_dict[time_stamp] = {product: [amount]} elif (product not in self.edge_index_dict[time_stamp]): - self.edge_index_dict[time_stamp][product] = [material_edge] + self.edge_index_dict[time_stamp][product] = [edge_material_flow] self.edge_weight_dict[time_stamp][product] = [amount] else: - self.edge_index_dict[time_stamp][product].append(material_edge) + self.edge_index_dict[time_stamp][product].append(edge_material_flow) self.edge_weight_dict[time_stamp][product].append(amount) - #get total input/output for each node at each timestep - #input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product + def create_node_features(self, df, metric): + """get total input/output for each node at each timestep + input: supplier gets amount, buyer gets product, output: supplier loses amount, buyer loses product""" input_entity = "supplier_t" if metric == "total_amount" else "buyer_t" output_entity = "buyer_t" if metric == "total_amount" else "supplier_t" @@ -119,21 +133,22 @@ def __init__(self, csv_file, start_date, length_timestamps = 1, self.node_features[int(time_stamp), self.company_to_nodeID[firm], 0] = metric for time_stamp, firm, metric in zip(*node_output_rows): self.node_features[int(time_stamp), self.company_to_nodeID[firm], 1] = metric - + def get_edge_index_dict(self, time_stamp): if (time_stamp not in self.prevalent_timestamps): return None edges_dict = self.edge_index_dict[time_stamp] unique_products = list(edges_dict.keys()) + index_dict = {} #constructs forward edges from supplier to buyer (flow of material) - index_dict = {("firm",(product,"material"),"firm"): np.transpose( - np.array(edges_dict[product])) for product in unique_products} + if (self.edge_type == "material" or self.edge_type == "both"): + index_dict.update({("firm",(product,"material"),"firm"): np.transpose( + np.array(edges_dict[product])) for product in unique_products}) #constructs reverse edges from buyer to supplier (flow of cash) - if self.dual_edge == True: - dual_index_dict = {("firm",(product, "cash"), "firm"): np.flip(np.transpose( - np.array(edges_dict[product])), axis = 0).copy() for product in unique_products} - index_dict.update(dual_index_dict) + if (self.edge_type == "cash" or self.edge_type == "both"): + index_dict.update({("firm",(product, "cash"), "firm"): np.flip(np.transpose( + np.array(edges_dict[product])), axis = 0).copy() for product in unique_products}) return index_dict def get_edge_weight_dict(self, time_stamp): @@ -141,13 +156,16 @@ def get_edge_weight_dict(self, time_stamp): return None weight_dict = self.edge_weight_dict[time_stamp] unique_products = list(weight_dict.keys()) #different products that delineate edge relations - edge_weight_dict = {("firm",(product,"material"),"firm"): np.array( - weight_dict[product]) for product in unique_products} - - if self.dual_edge == True: - dual_index_dict = {("firm",(product, "cash"), "firm"): np.array( - weight_dict[product]) for product in unique_products} - edge_weight_dict.update(dual_index_dict) + edge_weight_dict = {} + #assigns weight attributes to forward edges from supplier to buyer (flow of material) + if (self.edge_type == "material" or self.edge_type == "both"): + edge_weight_dict.update({("firm",(product,"material"),"firm"): np.array( + weight_dict[product]) for product in unique_products}) + + #assigns weight attributes to reverse edges from buyer to supplier (flow of cash) + if (self.edge_type == "cash" or self.edge_type == "both"): + edge_weight_dict.update({("firm",(product, "cash"), "firm"): np.array( + weight_dict[product]) for product in unique_products}) return edge_weight_dict def get_date_range(self,time_stamp): @@ -162,21 +180,20 @@ def segment_timestamps(self, current_date, prior_days, next_days): """ days_after_start = get_days_between(self.start_date, current_date) prior_earliest_ts = np.ceil((days_after_start - prior_days + 1) / self.length_ts) - #prior_earliest_ts = np.ceil((days_after_start - prior_days + 1 + self.length_ts) / self.length_ts) - 1 prior_latest_ts = (days_after_start + 1) // self.length_ts - 1 prior_timestamps = list(range(int(prior_earliest_ts), int(prior_latest_ts) + 1, 1)) - #next_earliest_ts = np.ceil((days_after_start + 1 + self.length_ts) / self.length_ts) - 1 next_earliest_ts = np.ceil((days_after_start + 1) / self.length_ts) next_latest_ts = (days_after_start + 1 + next_days) // self.length_ts - 1 next_timestamps = list(range(int(next_earliest_ts), int(next_latest_ts) + 1, 1)) return prior_timestamps, next_timestamps def get_node_norms(self, time_stamps): + """ obtain mean and standard deviation of raw node features over the time dimension """ if (len(time_stamps) <= 1): return None node_features = self.node_features[time_stamps] - mean, std = np.mean(node_features, axis = 0), np.std(node_features, axis = 0) + mean, std = np.mean(node_features), np.std(node_features) return {"mean": mean, "std": std} def getTemporalGraph(self, time_stamps, norms = None): @@ -195,7 +212,6 @@ def getTemporalGraph(self, time_stamps, norms = None): edge_index_dicts = [self.get_edge_index_dict(ts - self.lags) for ts in time_stamps] edge_weight_dicts = [self.get_edge_weight_dict(ts - self.lags) for ts in time_stamps] - #TODO: rework these to accomodate backward lag and normalizations feature_dicts = [{"firm": node_normalize(self.node_features[ts - self.lags: ts], norms) if ( self.lags <= ts <= self.max_ts) else None} for ts in time_stamps] target_dicts = [{"firm": node_normalize(self.node_features[ts], norms) if ( @@ -237,12 +253,94 @@ def loadData(self, current_date, prior_days, next_days, normalize = True ): nextGraph = self.getTemporalGraph(next_timestamps, norms) return priorGraph, nextGraph, norms +#revise based on refactored class (use homogenuous supplier -> buyer edges) +class FirmDataset(SupplyChainDataset): + def __init__(self, csv_file, start_date, length_timestamps = 1, + metric = "total_amount", lags = 5): + """ + same parameters as ancestor class (SupplyChainDataset). A modified version that aggregates + transactions between firms over all products, i.e. no product stratification for edges. + """ + all_metrics = {"total_amount", "total_quantity", "bill_count", "total_weight"} + if metric not in all_metrics: + raise ValueError(f"Metric must be {all_metrics}") + self.start_date = start_date + self.length_ts = length_timestamps + self.lags = lags + #for edge_type, if metric == total_amount, then draw buyer -> supplier, otherwise supplier -> buyer + self.edge_type = "cash" if metric == "total_amount" else "material" + + #read the dataframe and excise missing supplier & buyer nodes, as well as values + df = pd.read_csv(csv_file) + row_names = ["time_stamp","supplier_t","buyer_t", metric] + df = df[row_names][(~df[metric].isna()) & (df[metric] > 0) & (df["supplier_t"] != "") & (df["buyer_t"] != "")] + + #aggregate over all products + df = df.groupby(by = ["time_stamp","supplier_t","buyer_t"]).sum().reset_index() + companies = list(set(df["supplier_t"]).union(set(df["buyer_t"]))) + self.num_nodes = len(companies) + self.prevalent_timestamps = set(df["time_stamp"]) + self.max_ts = max(self.prevalent_timestamps) + self.company_to_nodeID = {company_t: node_id for node_id, company_t in enumerate(companies)} + self.nodeID_to_company = {value:key for key,value in self.company_to_nodeID.items()} + + self.create_edge_attributes(df, metric) + self.create_node_features(df, metric) + self.node_features = np.log(self.node_features + 1) + self.edge_weight_dict = {key: np.log(np.array(value) + 1) for key, value in self.edge_weight_dict.items()} + + def create_edge_attributes(self, df, metric): + self.edge_index_dict = {} + self.edge_weight_dict = {} + row_names = ["time_stamp","supplier_t","buyer_t", metric] + rows = [list(df[name]) for name in row_names] + for time_stamp, supplier, buyer, amount in zip(*rows): + time_stamp = int(time_stamp) + if (self.edge_type == "cash"): + edge_transactions = [self.company_to_nodeID[buyer], self.company_to_nodeID[supplier]] + else: + edge_transactions = [self.company_to_nodeID[supplier], self.company_to_nodeID[buyer]] + if (time_stamp not in self.edge_index_dict): + self.edge_index_dict[time_stamp] = [edge_transactions] + self.edge_weight_dict[time_stamp] = [amount] + else: + self.edge_index_dict[time_stamp].append(edge_transactions) + self.edge_weight_dict[time_stamp].append(amount) + + #override ancestor class to use homogeneous edges + def get_edge_index_dict(self, time_stamp): + if (time_stamp not in self.prevalent_timestamps): + return None + return np.array(self.edge_index_dict[time_stamp]).T + + #override ancestor class to use homogeneous edges + def get_edge_weight_dict(self, time_stamp): + if (time_stamp not in self.prevalent_timestamps): + return None + return np.array(self.edge_weight_dict[time_stamp]) + + #override ancestor class to use DynamicGraphTemporalSignal + def getTemporalGraph(self, time_stamps, norms = None): + date_ranges = [np.array(self.get_date_range(ts)) for ts in time_stamps] + edge_indices = [self.get_edge_index_dict(ts - self.lags) for ts in time_stamps] + edge_weights = [self.get_edge_weight_dict(ts - self.lags) for ts in time_stamps] + features = [node_normalize(self.node_features[ts - self.lags: ts,:], norms).transpose(1,0,2).reshape(-1,10) if ( + self.lags <= ts <= self.max_ts) else None for ts in time_stamps] + targets = [node_normalize(self.node_features[ts, :], norms) if ( + 0 <= ts <= self.max_ts) else None for ts in time_stamps] + + tempGraph = DynamicGraphTemporalSignal(edge_indices = edge_indices, + edge_weights = edge_weights, + features = features, + targets = targets, + date_ranges = date_ranges) + return tempGraph if __name__ == "__main__": """command line testing""" start_t = time.time() fname = "./storage/daily_transactions_2021.csv" - sc = SupplyChainDataset(fname, start_date = "2021-01-01", dual_edge = False, lags = 5) + sc = SupplyChainDataset(fname, start_date = "2021-01-01", edge_type = "material", lags = 5) end_t = time.time() print(f"Processed dataset at {fname} in {end_t - start_t:.2f} seconds.") print(f"Total Firms: {sc.num_nodes}\nTotal Time-Stamped Supplier \u2192 Buyer Edges: {sc.num_edges}")