Add two dimension binning and calculate binning result according to provided binning boundaries. (#88)

* Add two methods in Binning Runnable: two dimension binning and calcualte the binning result according to provided binning boundaries.

* Add more logs.

* Do some rephrase.

* Rename binning folder to bin to avoid the name conflict between binning folder and binning.py.

Author: brightcoder01

runnables/binning/__init__.py runnables/bin/__init__.pyrunnables/binning/__init__.py renamed to runnables/bin/__init__.py

@@ -57,7 +57,7 @@ def calc_binning_stats(
     bin_methods,
     bin_nums,
     cols_bin_boundaries,
-    reverse_cumsum=False):
+    reverse_cumsum):
     cols_bin_stats = []
     for i in range(len(sel_cols)):
         sel_col = sel_cols[i]
@@ -102,11 +102,46 @@ def calc_stats(
     bin_methods,
     bin_nums,
     cols_bin_boundaries,
-    reverse_cumsum=False):
+    reverse_cumsum):
     basic_stats_df = calc_basic_stats(in_md, sel_cols)
     cols_bin_stats_df = calc_binning_stats(in_md, sel_cols, bin_methods, bin_nums, cols_bin_boundaries, reverse_cumsum)
 
     stats_df = pd.merge(basic_stats_df, cols_bin_stats_df, how='inner', on='name')
 
     return stats_df
 
+
+def calc_two_dim_binning_stats(
+    in_md,
+    sel_col_1,
+    sel_col_2,
+    bin_method_1,
+    bin_method_2,
+    bin_num_1,
+    bin_num_2,
+    bin_boundaries_1,
+    bin_boundaries_2,
+    reverse_cumsum):
+    bin_o1, bins_1 = binning(in_md, sel_col_1, bin_method_1, bin_num_1, bin_boundaries_1)
+    bin_o2, bins_2 = binning(in_md, sel_col_2, bin_method_2, bin_num_2, bin_boundaries_2)
+
+    bin_num_1 = len(bins_1) - 1
+    bin_num_2 = len(bins_2) - 1
+
+    bin_o = bin_o1 * bin_num_2 + bin_o2
+    bin_prob_df = bin_o.value_counts(normalize=True).to_pandas().to_frame()
+    bin_prob_df = bin_prob_df.reindex(range(bin_num_1 * bin_num_2), fill_value=0)
+    two_dim_bin_prob_np = bin_prob_df.to_numpy().reshape((bin_num_1, bin_num_2))
+    two_dim_bin_cumsum_prob_np = cumsum(two_dim_bin_prob_np, reverse_cumsum)
+
+    return pd.DataFrame(two_dim_bin_prob_np), pd.DataFrame(two_dim_bin_cumsum_prob_np)
+
+
+def get_cols_bin_boundaries(stats_df):
+    col_boundaries = {}
+    for _, row in stats_df.iterrows():
+        col_name = row['name']
+        boundaries = [float(item) for item in row['bin_boundaries'].split(',')]
+        col_boundaries[col_name] = boundaries
+
+    return col_boundaries

runnables/binning/binning.py runnables/bin/binning_calculator.pyrunnables/binning/binning.py renamed to runnables/bin/binning_calculator.py

@@ -1,7 +1,8 @@
 import argparse
 import mars.dataframe as md
 import os
-from binning.binning import calc_stats
+import pandas as pd
+from bin.binning_calculator import calc_stats, calc_two_dim_binning_stats, get_cols_bin_boundaries
 from run_io.db_adapter import convertDSNToRfc1738
 from sqlalchemy import create_engine
 
@@ -12,7 +13,9 @@ def build_argument_parser():
     parser.add_argument("--columns", type=str, required=True)
     parser.add_argument("--bin_methods", type=str, required=False)
     parser.add_argument("--bin_nums", type=str, required=False)
+    parser.add_argument("--bin_input_table", type=str, required=False)
     parser.add_argument("--reverse_cumsum", type=bool, default=False)
+    parser.add_argument("--two_dim_bin_cols", type=str, required=False)
 
     return parser
 
@@ -23,29 +26,80 @@ def build_argument_parser():
     columns = args.columns.split(',')
     bin_methods = args.bin_methods.split(',') if args.bin_methods else None
     bin_nums = [int(item) for item in args.bin_nums.split(',')] if args.bin_nums else None
+    two_dim_bin_cols = args.two_dim_bin_cols.split(',') if args.two_dim_bin_cols else None
 
     select_input = os.getenv("SQLFLOW_TO_RUN_SELECT")
     output = os.getenv("SQLFLOW_TO_RUN_INTO")
+    output_tables = output.split(',')
     datasource = os.getenv("SQLFLOW_DATASOURCE")
 
+    # Check arguments
+    if two_dim_bin_cols:
+        assert(len(two_dim_bin_cols) == 2)
+        assert(len(output_tables) == 3)
+
     url = convertDSNToRfc1738(datasource, args.dbname)
     engine = create_engine(url)
     input_md = md.read_sql(
         sql=select_input,
         con=engine)
     input_md.execute()
 
+    cols_bin_boundaries = {}
+    if args.bin_input_table:
+        print("Get provided bin boundaries from table {}".format(args.bin_input_table))
+        bin_input_df = pd.read_sql_table(
+            table_name=args.bin_input_table,
+            con=engine)
+        cols_bin_boundaries = get_cols_bin_boundaries(bin_input_df)
+
+        if set(columns) > cols_bin_boundaries.keys():
+            raise ValueError("The provided bin boundaries contains keys: {}. But they cannot cover all the \
+                input columns: {}".format(cols_bin_boundaries.keys(), columns))
+
+        print("Ignore the bin_nums and bin_methods arguments")
+        bin_nums = [None for i in range(len(columns))]
+        bin_methods = [None for i in range(len(columns))]
+    
+    print("Calculate the statistics result for columns: {}".format(columns))
     stats_df = calc_stats(
         input_md,
         columns,
         bin_methods,
         bin_nums,
-        {},
+        cols_bin_boundaries,
         args.reverse_cumsum)
 
-    print("Persist the statistics result into the table {}".format(output))
+    print("Persist the statistics result into the table {}".format(output_tables[0]))
     stats_df.to_sql(
-        name=output,
+        name=output_tables[0],
         con=engine,
         index=False
     )
+
+    if args.two_dim_bin_cols:
+        print("Calculate two dimension binning result for columns: {}".format(columns))
+        bin_prob_df, bin_cumsum_prob_df = calc_two_dim_binning_stats(
+            input_md,
+            columns[0],
+            columns[1],
+            bin_methods[0],
+            bin_methods[1],
+            bin_nums[0],
+            bin_nums[1],
+            cols_bin_boundaries.get(columns[0], None),
+            cols_bin_boundaries.get(columns[1], None),
+            args.reverse_cumsum)
+
+        print("Persist the binning probabilities into table {}".format(output_tables[1]))
+        bin_prob_df.to_sql(
+            name=output_tables[1],
+            con=engine,
+            index=False
+        )
+        print("Persist the binning accumulated probabilities into table {}".format(output_tables[2]))
+        bin_cumsum_prob_df.to_sql(
+            name=output_tables[2],
+            con=engine,
+            index=False
+        )

runnables/binning/psi.py runnables/bin/psi_calculator.pyrunnables/binning/psi.py renamed to runnables/bin/psi_calculator.py

@@ -1,7 +1,7 @@
 import argparse
 import os
 import pandas as pd
-from binning.psi import calc_psi, get_cols_bin_probs
+from bin.psi_calculator import calc_psi, get_cols_bin_probs
 from run_io.db_adapter import convertDSNToRfc1738
 from sqlalchemy import create_engine