R CODE
--------------------------------------------------------------------------------------------------

#! /nix/var/nix/profiles/dsd-tools/bin/Rscript
#SBATCH --job-name=<JOB_NAME>
#SBATCH --partition=normalbatch
#SBATCH --mem <MEMORY_SESSION>M
#SBATCH --output=<PATH_CLIENT>/output_file_algos_<JOB_NAME>.out
#SBATCH --err=<PATH_CLIENT>/error_file_algos_<JOB_NAME>.err

## Libraries
library(dplyr)
library(dbplyr)
library(galaxy)
library(arrow)
library(readr)
#library(purrr)
library(stringr)
library(sparklyr)
library(lubridate)
library(DBI)
library(cli)

## Overview

# This document will be used to produce outputs for the TV products backtests for algos (retros) in **Brazil**.
# It covers the process from joining client's file with Alt data environment to retrieve the bureau subject id, preparing raw data, applying GCDF functions, calculating filters, and producing output file with the actual attributes and algorithms to the client.
# NOTE: This uses some functions that are not part of the installed galaxy package. These are stored in `bu_br/persist/` so you will need access to run those.

##--##--##--##--##--##--#-#
# Always run this part ####
##--##--##--##--##--##--#-#

##--##--##--##--##-##
## Manual Inputs ####
##--##--##--##--##-##

### Inputs and constants ----

# Input file from the client
input_path = "file:///shared/shape_tier3/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/p4/subject_sample"
delimiter = ","
output_path = "/shared/shape_tier3/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/retro/nubank_100/output_algos"

# Subject type (consumer or business) - set this to either "PF" or "PJ"
subj_type = "PF"

# CNPJ - set to a vector with all the cnpjs of the client, empty if not informed.
cnpj = c()

# the cpf and date input variables from the clients
input_cpf_var = "nr_cpf"
input_date_var = "data_referencia"

# Which template
## Set the template parameter to either "algoritmos", "indicadores" or "ambos" (for both algorithms & indicators). We might add a look-up for more templates later on, as the solutions team create them.
## The mode can be set either to "bkt" to output only attributes already deployed in Adviser, or "dsa" for the complete galaxy output capable for BRZ. Indicators are only available for "bkt" output, since they are created as we deploy TV in Adviser.
## Please consider the table below to choose template/mode:
## 	  	| algoritmos |indicadores|   ambos   |
## 	  	+------------+-----------+-----------+
## 	bkt	|    Yes     |    Yes    |    Yes    |
## 	dsa |    Yes     |    No     |    No     |
## 	  	+------------+-----------+-----------+
template = "algoritmos"
mode = "dsa"

# Select if an output is desired, either "ON" (consider only trades from the CNPJ), "OFF" (consider only trades not from the CNPJ) or "COMBINED" (consider all trades, no filter by CNPJ), can be multiple.
output_on   = "N"       ## 'Y' for yes or 'N' for no
output_off  = "N"      ## 'Y' for yes or 'N' for no
output_comb = "Y" ## 'Y' for yes or 'N' for no

# Table naming data
eid <- Sys.info()[['user']]  # Table names will default to your eID
prj <- "_retro_MM_p1p2p3"			 # Name for project to put in table names
prj2 <- paste0(prj, "_attrs") # This will be used to calculate GCDF for attributes

### Checking in case the template was not defined right ----

if(!exists("template")) {
  stop("The 'template' parameter hasn't been defined.")
} else if(!template %in% c("algoritmos", "indicadores", "ambos")) {
  stop("This template '", template, "' is not available.")
} else if(!exists("subj_type")) {
  stop("The 'subj_type' parameter hasn't been defined.")
} else if(!subj_type %in% c("PF", "PJ")) {
  stop("You must set 'subj_type' parameter to either 'PF', for consumers, or 'PJ', for businesses.")
} else if(!exists("mode")) {
  stop("The 'mode' parameter hasn't been defined.")
} else if(!mode %in% c("bkt", "dsa")) {
  stop("You must set 'mode' parameter to either 'bkt', for backtest, or 'dsa', for modelling.")
}

### Spark connection ----

# Most of the work is done with a Spark connection on the Yarn cluster. You should be able to use the same connection throughout the process.

# NOTE: The Java functions are not currently being used in this process, but are put here to show how they will be used in the future. The functions listed here have known differences from the GCDF definitions we want to use.
Sys.setenv(LD_LIBRARY_PATH = "")
spark_disconnect_all()
conf <- spark_config()
# NOTE: Change the parallelism and sql.shuffle.partitions accordingly to the size of the file you want to process, you can go up to 1600, but if your dataset is small don't raise the setup, it may actually worsen the job execution
conf$`spark.default.parallelism` <- "400"
conf$`spark.sql.shuffle.partitions` <- "400"
conf$`spark.network.timeout` <- '3600s'
conf$`spark.yarn.am.memory` <- '4G'
conf$`spark.executor.memoryOverhead` <- "25000" # May help with error: Container killed by YARN for exceeding memory limits. 19GB of 19GB physical memory used. https://spark.apache.org/docs/latest/configuration.html
# conf$`spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation` <- TRUE # Stopped working in Spark > 3.0 Will help with the error which happens when a table cannot be managed or overwritten.
# Below are a few configuration to avoid jobs being aborted due to timestamps/dates.
conf$spark.sql.parquet.datetimeRebaseModeInRead <- 'LEGACY'
conf$spark.sql.parquet.datetimeRebaseModeInWrite <- 'CORRECTED'
# conf$`spark.yarn.queue` <- "sha9brgp" # IF YOU ARE DELIVERY USE THIS QUEUE
conf$`spark.yarn.queue` <- "sha9brbureaugp" # IF YOU ARE DSA USE THIS QUEUE
cn <- spark_connect(master = "yarn", config = conf, app_name = "BKT_TV_ALGOS")

##--##--##--##--##--##--##--##--##--##--##--##--##--##--##-##
## Don't need to change this part, but run it every time ####
##--##--##--##--##--##--##--##--##--##--##--##--##--##--##-##

# File for logging, for tracking while the loop runs        
ln <- paste0(eid, prj, "_tv_algos_backtest_output_", format(Sys.time(), '%Y%m%d_%H:%M'), ".log")  

# This file contains the galaxy functions with any adjustments needed for Brazil conditions & also custom functions
source("/shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM/gal_funs_bra_algos.R")


# global constants for algorithms
set_galaxy_const(
  tvpa_tvua_min_msrd_window = 1,
  tvpa_tvua_max_msrd_window = 30,
  tvpa_tvua_inc_dec_thd_val = 0.1,   # INC_DEC_THD # matches with the spec doc
  tvpa_tvua_min_pay_thd_val = 0.965, # MIN_PAY_THD # matches with the spec doc
  tvpa_tvua_tran_thd_val = 1.15,     # TRAN_THD # matches with the spec doc
  tvpa_tvua_bal_thd_val = 1000,      # CUR_CC_CUST_THD ## on 2023-06-09 it was 2000 in the specs # On 2023-06-13, me and Daniel thought that 1000 would be fine to Brazil, this is for us to "remove" instability because people would just always pay it off at once if value is too low.
  tvpa_tvua_bc_thd_val = 2500,       # CUR_CC_ACT_THD ## on 2023-06-09 it was 2500 in the specs
  tvpa_tvua_paymnt_buffer_val = 10,  # CUR_MINPAY_THD ## on 2023-06-09 it was 10 in the specs, which seems to be the right value indeed, so changed here on the codes. # ON 2023-06-13: It should be 100, because people could round up to the next hundred when paying the card. ## On 2023-06-15 I had a second thought and asked Daniel if we shouldn't leave it as 10. I think no one would round up to 1,600 a payment that is originally 1,550. Specially nowadays in which payments are pretty much made digitally and folks hardly ever have to manually populate the payment value...
  tvpa_tvua_small_bal_val = 500,     # TMP_SMALL_BAL # matches with the spec doc. We had changed this one already. ## on 2023-06-13, we thought that 500 would be good. This is to tell that, if your total balance amount is lower than that, then things like "number of payments", or payment ration, would not be that significant.
  
  # global constants for inquiry based attributes.
  # There are some attributes, treated like algos (or the other way around), so that they are computed in some tvea function, because of that we are creating some global constants required in those.
  algo_inq_credit_shopping = 14L,
  algo_inq_double_inquiry = 1L,
  algo_inq_double_mortage = 45L,
  
  # global constants for Trades
  tvea_pb_value = 248000, # Premium Bankcard threshold
  tvea_apd_level_CreditCard = 200, # GTQ
  tvea_apd_level_BankRevolving = 200, # GTQ
  grouping_list = c('consno'),
  months_between_calc = 'run_date')

local_rate <- 8
max_delq_rte_cde <- 0

# NNote in 2024/06/28 we are not applying the suppression list anymore
# Suppression list (based on the decisions made during the TVLink AM/NA)
# library(readxl)
# suppressItens <- read_excel(
#   "/shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/TV_backtest_scripts/suppression_list.xlsx",
#   col_types = c("text", "text", "text", 
#                 "text", "text", "text")) %>% 
#   select(concat_line_lob) %>% 
#   unname() %>% unlist() %>% unique()

### data_date in Shape to use for retrieving only the sj_subject_id from the subject table

aux_data <- as.character(today() - days(153))

# IT IS NOT RECOMENDED TO USE DATA_DATES OLDER THAN 2024-06-30, BECAUSE THE TRADE_IDs WILL NOT HAVE THE ADJUSTMENT, DUE TO THE SOFT DELETE THING, WHICH WAS REFLECTED IN SHAPE FOR THE FIRST TIMEIN JUNE-2024.
# db_data_date <- 
#   tbl(cn, "brazilbureau.bu_br_trade_history") %>% 
#   sdf_distinct(data_date) %>% 
#   filter(data_date > aux_data) %>% 
#   summarise(max_data_date = max(as.character(data_date))) %>% 
#   collect() %>% unlist() %>% unname()

# db_data_date <- '2024-06-30' # TEMPORARY DATA_DATE LOCK DUE TO AUGUST PARTIAL LOADS IN SHAPE, RESUME REGULAR CODE IN SEPTEMBER.
db_data_date <- "2023-12-31"

##--##--##--##--##--##--##--##--##--##--##--#-#
# Select what to run from after this point ####
##--##--##--##--##--##--##--##--##--##--##--#-#


## Initial data preparation ----

### The Input file from the client ----

# IF YOUR INPUT_FILE HAS THE SJ_SUBJECT_ID IT WILL BREAK IN THE OUTPUT_FILE JOIN AT THE END OF THE SCRIPT, ALWAYS MAKE SURE YOU TAKE IT OFF
# tbl(cn, paste0("bu_brbureau.<client_dataset>")) %>%
#   select(-any_of("sj_subject_id")) %>% 
#   mutate(
#     cpf_TU = !!sym(input_cpf_var),
#     date_TU = !!sym(input_date_var)
#   ) %>%
#   spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_input_file"), mode = "overwrite")

spark_read_csv(cn, path = input_path, delimiter = delimiter)  %>% colnames()

spark_read_csv(cn, path = input_path, delimiter = delimiter) %>%
  mutate(
    monit_date =  as.Date('2023-06-30')
  ) %>%
  select(sj_subject_id, monit_date, isi_idbase_id=sj_cust_id) %>% 
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_input_file"), mode = "overwrite")

if("sj_subject_id" %in% colnames(tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_input_file")))) {
  stop("You must remove the column sj_subject_id from the input_file to avoid errors.")
}

### client's data treatment ----
# Here we are:
# Making data unique ate consumer+date level
# joining with alt data to retrieve isi_idbase_id, nr_anoobito and dt_nascimento to create a few flags to identify minor and those with deceased information
# Joining with bureau data to retrieve sj_subject_id
# joining with subject optlog to exclude those marked as opt-out
# writing the subject sample in disc

input_data <-
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_input_file")) %>% 
  mutate(cpf_TU = sj_subject_id)
  # %>% distinct(cpf_TU, date_TU)


input_data %>% 
  left_join(tbl(cn, "brazilbureau.bu_br_subject") %>% 
              filter(data_date == db_data_date) %>% 
              select(isi_idbase_id, sj_sub_type, data_date)
            , by = "isi_idbase_id") %>% 
  left_join(
    tbl(cn, "brazilbureau.bu_br_subject_optlog") %>%
      filter(data_date == db_data_date) %>%
      select(sj_subject_id, so_request_type, so_request_start_date) %>%
      group_by(sj_subject_id) %>%
      window_order(desc(so_request_start_date)) %>%
      filter(row_number() == 1)
    , by = "sj_subject_id"
  ) %>%
  # Excluding optouts
  filter(coalesce(so_request_type, 'OK') != 'C') %>% 
  mutate(
    consno = paste0(sj_subject_id,"||",monit_date)
  ) %>% 
  select(-so_request_type, -so_request_start_date) %>%
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_subject_sample"), mode = "overwrite")


# if(subj_type == "PF"){
#   
#   db_data_date_alt <- 
#     tbl(cn, "brazil.br_pf_dado_cadastral") %>% 
#     filter(data_date > aux_data) %>% 
#     summarise(max_data_date = max(as.character(data_date))) %>% 
#     collect() %>% unlist() %>% unname()
#   
#   input_data %>% 
#     left_join(
#       tbl(cn, "brazil.br_pf_dado_cadastral") %>% 
#         filter(data_date == db_data_date_alt) %>% 
#         select(nr_cpf, isi_idbase_id, nr_anoobito, dt_nascimento) %>% 
#         sdf_distinct()
#       , by = c("cpf_TU" = "nr_cpf")
#     ) %>% 
#     left_join(
#       tbl(cn, "brazilbureau.bu_br_subject") %>% 
#         filter(data_date == db_data_date) %>% 
#         select(isi_idbase_id, sj_subject_id, sj_sub_type, data_date)
#       , by = "isi_idbase_id"
#     ) %>% 
#     left_join(
#       tbl(cn, "brazilbureau.bu_br_subject_optlog") %>%
#         filter(data_date == db_data_date) %>%
#         select(sj_subject_id, so_request_type, so_request_start_date) %>%
#         group_by(sj_subject_id) %>%
#         window_order(desc(so_request_start_date)) %>%
#         filter(row_number() == 1)
#       , by = "sj_subject_id"
#     ) %>%
#     # Excluding request type = C (Cancelled) which indicates a opt-out
#     filter(coalesce(so_request_type, 'OK') != 'C') %>% 
#     mutate(
#       monit_date = as.Date(date_TU)
#       , consno = paste0(sj_subject_id,"||",monit_date)
#     ) %>% 
#     mutate(flg_deceased = case_when(is.na(nr_anoobito) ~ 0,
#                                     nr_anoobito < year(date_TU) ~ 1,
#                                     T ~ 0),
#            flg_minor    = case_when(is.na(dt_nascimento) ~ 1,
#                                     floor(months_between(date_TU, dt_nascimento)/12) < 18 ~ 1,
#                                     T ~ 0)) %>% 
#     select(-nr_anoobito, -dt_nascimento, -so_request_type, -so_request_start_date) %>%
#     spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_subject_sample"), mode = "overwrite")
#   
# } else if(subj_type == "PJ"){
#   
#   db_data_date_alt <- 
#     tbl(cn, "brazil.br_pj_dado_cadastral") %>% 
#     filter(data_date > aux_data) %>% 
#     summarise(max_data_date = max(as.character(data_date))) %>% 
#     collect() %>% unlist() %>% unname()
#   
#   input_data %>% 
#     left_join(tbl(cn, "brazil.br_pj_dado_cadastral") %>% 
#                 filter(data_date == db_data_date_alt) %>% 
#                 select(nr_cnpj, isi_idbase_id) %>% 
#                 sdf_distinct()
#               , by = c("cpf_TU" = "nr_cnpj")) %>% 
#     
#     
#     left_join(tbl(cn, "brazilbureau.bu_br_subject") %>% 
#                 filter(data_date == db_data_date) %>% 
#                 select(isi_idbase_id, sj_subject_id, sj_sub_type, data_date)
#               , by = "isi_idbase_id") %>% 
#     
#     left_join(
#       tbl(cn, "brazilbureau.bu_br_subject_optlog") %>%
#         filter(data_date == db_data_date) %>%
#         select(sj_subject_id, so_request_type, so_request_start_date) %>%
#         group_by(sj_subject_id) %>%
#         window_order(desc(so_request_start_date)) %>%
#         filter(row_number() == 1)
#       , by = "sj_subject_id"
#     ) %>%
#     # Excluding optouts
#     filter(coalesce(so_request_type, 'OK') != 'C') %>% 
#     mutate(
#       monit_date = as.Date(date_TU)
#       , consno = paste0(sj_subject_id,"||",monit_date)
#     ) %>% 
#     mutate(flg_deceased = 0, ## Just so we don't have problems in the output
#            flg_minor = 0) %>%
#     select(-so_request_type, -so_request_start_date) %>%
#     spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_subject_sample"), mode = "overwrite")
#   
# }

# At this point, the subject sample will have these fields below, and will be at subject + date_TU level
# cpf_TU, date_TU, isi_idbase_id, sj_subject_id, sj_sub_type, monit_date, transf_date, consno, flg_minor, flg_deceased
### just in case we wanna check, here are some counts
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% head()
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% count()
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% distinct(cpf_TU) %>% count()
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% distinct(sj_subject_id) %>% count()

## Data samples from Core ----
# From here, we will create the samples from brazilbureau tables, which will be used in the rest of the script.

### Trade Master Run Fin ----
tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% 
  select(sj_subject_id, monit_date) %>% 
  ## Only the records from trade master which are active in the most recent load
  left_join(
    tbl(cn, "brazilbureau.bu_br_trade_master") %>% 
      filter(
        data_date == db_data_date
        , tm_active_flag == 1
      ) %>%
      select(sj_subject_id, tm_trade_id, tm_unique_no, tm_contract_date, tm_mode,
             tm_last_installment_duedate, tm_future_contracted_value, tm_number_installments,
             tm_fi_cnpj, tm_contract_fi_cnpj, tm_created_date) ## 2024-07-25 CHANGE: pulled tm_created_date from here, instead of from where we are picking historical data_dates for trade_master (a few lines below)
    , by = "sj_subject_id"
  ) %>% 
  filter(
    monit_date >= '2021-07-01' ## Part of the new requirement for retros
    , tm_contract_date < monit_date
  ) %>% 
  ## Removing inactive status records, selecting all data_dates so we are able to collect the appropriate registers later, which are related to 
  ## the proper retroactive periods
  inner_join(
    tbl(cn, "brazilbureau.bu_br_trade_master") %>% 
      select(-tm_created_by) %>% 
      # OBS: REMOVE depara for mismatch because we're using older ´db_data_date´
      filter(tm_active_flag == 1, data_date <= db_data_date) %>% # NNote: To avoid getting future loads 
      # left_join(tbl(cn, "bu_brbureau.bferrei_depara_trade_id") %>% select(tm_trade_id,tm_created_by), by = "tm_trade_id") %>% 
      # OBS: REMOVING DB_DATA_DATE CHECK BECAUSE ISSUES WITH RETRO MISMATCH
      # mutate(tm_trade_id = coalesce(tm_created_by, tm_trade_id)) %>% select(-tm_created_by) %>% 
      # mutate(tm_trade_id = case_when(db_data_date >= '2024-06-20' ~ coalesce(tm_created_by, tm_trade_id), T ~ tm_trade_id)) %>% select(-tm_created_by) %>% 
      select(tm_trade_id, sj_subject_id, tm_calculation_date, tm_restriction_type,
             tm_quota_status, tm_balance_owed, tm_max_balance_owed, tm_quota_payout_date,
             tm_modified_date, tm_internal_base_date, data_date)
    , by = c("tm_trade_id", "sj_subject_id")
  ) %>% 
  # Here we are collecting the industry classification for the CNPJs
  left_join(
    tbl(cn, "bu_brbureau.FI_MASTER_TEMP"), by = c("tm_fi_cnpj"="CNPJ")
  ) %>% 
  mutate(
    member_code = case_when(coalesce(tm_contract_fi_cnpj, 0) == 0 ~ tm_fi_cnpj,
                            TRUE ~ tm_contract_fi_cnpj),			   
    industry = coalesce(industry, 99),
    tm_mode = trim(tm_mode),
    tm_mode_grp = as.integer(case_when(
      tm_mode %in% c('D01') ~ 1,
      tm_mode %in% c('E01', 'E02') ~ 2,
      tm_mode %in% c('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03',
                     'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01') ~ 3,
      TRUE ~ 0)) ## tm_mode_grp will be user to partition data in tables, as a way to optimize processing time.
    , tm_created_date = as.Date(tm_created_date)
    , tm_modified_date = as.Date(tm_modified_date)
    , transf_date = date_add(monit_date, -10L) ## Define transf_date per segment (simulated data report delay).
    , flg_retro = if_else(monit_date > tm_created_date, 0, 1) ## Part of the new requirement for retros
    , flg_delay = if_else(monit_date < date_add(tm_contract_date, 10L), 0, 1) ## Part of the new requirement for retros
    , flg_restriction = if_else(is.na(tm_restriction_type), 0, 1)
  ) %>%
  filter(flg_delay == 1, flg_restriction == 0) %>% ## Already filtering out trades opened in the last 60 days from the monit_date (new account report delay).
  select(-tm_contract_fi_cnpj, -tm_fi_cnpj) %>% ## We already have member_code.
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin"), mode = "overwrite"
                    , partition_by = "tm_mode_grp" 
  )

  

### Trade Master Run Tel ----
tbl(cn, "brazilbureau.bu_br_telco_trade_master") %>% 
  filter(data_date == db_data_date,
         ttm_active_flag == 1) %>% ## Already pulling most recent data, and removing inactive status records
  select(ttm_trade_id, sj_subject_id, ttm_contract_date, ttm_created_date, ttm_restriction_type
         , ttm_contract_fi_cnpj, ttm_fi_cnpj_id, data_date) %>% ## Select only relevant fields, to optimize processing time.
  inner_join(
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% ## Only the records from the subjects of the sample.
      select(sj_subject_id, monit_date)
    , by = "sj_subject_id"
  ) %>% 
  filter(monit_date >= '2021-07-01') %>% ## Part of the new requirement for retros
  mutate(
    member_code = case_when(coalesce(ttm_contract_fi_cnpj, 0) == 0 ~ ttm_fi_cnpj_id,
                            TRUE ~ ttm_contract_fi_cnpj)
    # Industry for TelCo is 8
    , industry = 8
    , ttm_created_date = as.Date(ttm_created_date)
    , transf_date = date_add(monit_date, -10L) ## Define transf_date per segment (Pending data report delay study for telco).
    , flg_retro = if_else(monit_date > ttm_created_date, 0, 1) ## Part of the new requirement for retros
    , flg_delay = if_else(monit_date < date_add(ttm_contract_date, 10L), 0, 1) ## Part of the new requirement for retros
    , flg_restriction = if_else(is.na(ttm_restriction_type), 0, 1)
  ) %>%
  filter(flg_delay == 1, flg_restriction == 0) %>% ## Already filtering out trades opened in the last 60 days from the monit_date (new account report delay).
  select(-ttm_contract_fi_cnpj, -ttm_fi_cnpj_id) %>% ## We already have member_code.
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_tel"), mode = "overwrite")

# NOTE: Need telco trade history to supply calculation date for trade master
tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_tel")) %>%
  left_join(
    tbl(cn, paste0("brazilbureau.bu_br_telco_trade_history")) %>% 
      filter(
        data_date == db_data_date,
        tth_active_flag == 1
        , tth_invoice_status != "X" 
      ) %>%
      mutate(ttm_calculation_date = as.Date(tth_calculation_date)) %>% ## Picking the calculation_date of each history record, if null, we'll leave it.
      select(tth_trade_id, ttm_trade_id, tth_due_date, tth_value = tth_invoice_value,
             tth_created_date, ttm_calculation_date, tth_active_flag, 
             tth_calculation_date, tth_invoice_status),
    by = "ttm_trade_id") %>%
  # NOTE: Field names have 'ttm' prefix vs 'tm' for financial trades
  rename_with(~gsub("^tt", "t", .x)) %>%
  # We will summarize by th_due_date, because for TelCo it's common to have many invoices due at the same day (11.2% based in July/24 load), and we want the max to consider those groupings
  group_by(tm_trade_id, monit_date, th_due_date) %>%
  mutate(
    total_balance_owed = sum(th_value)
  ) %>% 
  # Need to summarize trade history fields to get some trade master ones
  group_by(tm_trade_id, monit_date) %>%
  mutate(
    tm_calculation_date = if_else(
      flg_retro == 1, min(tm_calculation_date), ## If it's a retro monit_date, we will use the minimun calculation_date.
      max(if_else(tm_calculation_date <= as.Date(monit_date), tm_calculation_date, ''))) ## If it's not a retro monit_date, we will use the maximun calculation_date, before the monit_date.
    , tm_max_balance_owed= max(if_else(th_due_date <= as.Date(monit_date), total_balance_owed, 0))
    , rnk = row_number(coalesce(th_due_date, "1900-01-01"))
  ) %>%
  ungroup() %>%
  select(-total_balance_owed) %>% 
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_telco_trade_hist_tmp"), mode = "overwrite")

# OBS: Using old energy table name since we're using older data_date for mismatch
### Trade Master Run Energy ----
tbl(cn, "brazilbureau.bu_br_energy_trade_master") %>% 
  filter(data_date == db_data_date,
         etm_active_flag == 1) %>% ## Already pulling most recent data, and removing inactive status records
  select(etm_trade_id, sj_subject_id, etm_contract_date, etm_created_date, etm_restriction_type, 
         etm_contract_datasource_cnpj, datasource_cnpj, data_date) %>% ## Select only relevant fields, to optimize processing time.
  inner_join(
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% ## Only the records from the subjects of the sample.
      select(sj_subject_id, monit_date)
    , by = "sj_subject_id"
  ) %>% 
  filter(etm_contract_date < monit_date,
         monit_date >= '2021-07-01') %>% ## Part of the new requirement for retros
  mutate(
    # NNOTE ! Given the lack of information in the contract's cnpj field, use source (datasource_cnpj)
    member_code = case_when(coalesce(etm_contract_datasource_cnpj, 0) == 0 ~ datasource_cnpj,
                            TRUE ~ etm_contract_datasource_cnpj)
    # Industry for Utilities is 29
    , industry = 29
    , etm_created_date = as.Date(etm_created_date)
    , transf_date = date_add(monit_date, -10L) # Temporarily setting 10 days until we make proper study for Telco and Utilities.
    # , transf_date = monit_date ## Define transf_date per segment (Pending data report delay study for telco).
    , flg_retro = if_else(monit_date > etm_created_date, 0, 1) ## Part of the new requirement for retros
    , flg_delay = if_else(monit_date <= date_add(etm_contract_date,10L), 0, 1) ## Part of the new requirement for retros
    , flg_restriction = if_else(is.na(etm_restriction_type), 0, 1)
  ) %>%
  filter(flg_delay == 1, 
         flg_restriction == 0) %>% ## Already filtering out trades opened in the last 10 days from the monit_date (new account report delay).
  select(-etm_contract_datasource_cnpj, -datasource_cnpj) %>% ## We already have member_code.
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_energy"), mode = "overwrite"
  )


# NOTE: Need electricity trade history to supply calculation date for trade master
tbl(cn, paste0("bu_brbureau.", eid, prj, "_gcdf_run_trade_master_energy")) %>%
  left_join(
    tbl(cn, paste0("brazilbureau.bu_br_energy_trade_history")) %>% 
      filter(
        data_date == db_data_date,
        eth_active_flag == 1
        , eth_invoice_status != "X" 
      ) %>%
      mutate(etm_calculation_date = as.Date(eth_calculation_date)) %>% ## Picking the calculation_date of each history record, if null, we'll leave it.
      select(eth_trade_id, etm_trade_id, eth_due_date, eth_value = eth_invoice_value,
             eth_created_date, etm_calculation_date, eth_active_flag, 
             eth_calculation_date, eth_invoice_status),
    by = "etm_trade_id") %>%
  # NOTE: Field names have 'etm' prefix vs 'tm' for financial trades
  rename_with(~gsub("^et", "t", .x)) %>%
  # We will summarize by th_due_date, even though for Energy it isn't as common to have many invoices due at the same day (3.3% based in July/24 load), but even for those we want the max to consider those groupings
  group_by(tm_trade_id, monit_date, th_due_date) %>%
  mutate(
    total_balance_owed = sum(th_value)
  ) %>% 
  # Need to summarize trade history fields to get some trade master ones
  group_by(tm_trade_id, monit_date) %>%
  mutate(
    monit_date = as.Date(monit_date),
    tm_calculation_date_temp = if_else(
      is.na(th_calculation_date), th_calculation_date,
      max(if_else(
        th_calculation_date <= monit_date, th_calculation_date, as.Date('2199-12-31')))),
    tm_calculation_date = if_else(tm_calculation_date_temp <= monit_date, tm_calculation_date_temp, monit_date),
    tm_max_balance_owed = max(if_else(th_due_date <= as.Date(monit_date), th_value, 0))
    , rnk = row_number(coalesce(th_due_date, "1900-01-01"))
  ) %>%
  ungroup() %>%
  select(-tm_calculation_date_temp) %>% 
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_energy_trade_hist_tmp"), mode = "overwrite")

# ### Trade Master Run Electricity ----
# tbl(cn, "brazilbureau.bu_br_pub_serv_trade_master") %>%
#   filter(data_date == db_data_date,
#          pstm_active_flag == 1) %>% ## Already pulling most recent data, and removing inactive status records
#   select(pstm_trade_id, sj_subject_id, pstm_contract_date, pstm_created_date, pstm_restriction_type,
#          pstm_contract_datasource_cnpj, datasource_cnpj, data_date) %>% ## Select only relevant fields, to optimize processing time.
#   inner_join(
#     tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% ## Only the records from the subjects of the sample.
#       select(sj_subject_id, monit_date)
#     , by = "sj_subject_id"
#   ) %>%
#   filter(pstm_contract_date < monit_date,
#          monit_date >= '2021-07-01') %>% ## Part of the new requirement for retros
#   mutate(
#     # NNOTE ! Given the lack of information in the contract's cnpj field, use source (datasource_cnpj)
#     member_code = case_when(coalesce(pstm_contract_datasource_cnpj, 0) == 0 ~ datasource_cnpj,
#                             TRUE ~ pstm_contract_datasource_cnpj)
#     # Industry for Utilities is 29
#     , industry = 29
#     , pstm_created_date = as.Date(pstm_created_date)
#     , transf_date = date_add(monit_date, -10L) ## Define transf_date per segment (Pending data report delay study for telco).
#     , flg_retro = if_else(monit_date > pstm_created_date, 0, 1) ## Part of the new requirement for retros
#     , flg_delay = if_else(monit_date < date_add(pstm_contract_date, 10L), 0, 1) ## Part of the new requirement for retros
#   ) %>%
#   filter(flg_delay == 1) %>% ## Already filtering out trades opened in the last 60 days from the monit_date (new account report delay).
#   select(-pstm_contract_datasource_cnpj, -datasource_cnpj) %>% ## We already have member_code.
#   spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_energy"), mode = "overwrite")
# 
# # NOTE: Need electricity trade history to supply calculation date for trade master
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_energy")) %>%
#   left_join(
#     tbl(cn, paste0("brazilbureau.bu_br_pub_serv_trade_history")) %>%
#       filter(
#         data_date == db_data_date,
#         psth_active_flag == 1
#         , psth_invoice_status != "X"
#       ) %>%
#       mutate(pstm_calculation_date = as.Date(psth_calculation_date)) %>% ## Picking the calculation_date of each history record, if null, we'll leave it.
#       select(psth_trade_id, pstm_trade_id, psth_due_date, psth_value = psth_invoice_value,
#              psth_created_date, pstm_calculation_date, psth_active_flag,
#              psth_calculation_date, psth_invoice_status),
#     by = "pstm_trade_id") %>%
#   # NOTE: Field names have 'pstm' prefix vs 'tm' for financial trades
#   rename_with(~gsub("^pst", "t", .x)) %>%
#   # We will summarize by th_due_date, even though for Energy it isn't as common to have many invoices due at the same day (3.3% based in July/24 load), but even for those we want the max to consider those groupings
#   group_by(tm_trade_id, monit_date, th_due_date) %>%
#   mutate(
#     total_balance_owed = sum(th_value)
#   ) %>% 
#   # Need to summarize trade history fields to get some trade master ones
#   group_by(tm_trade_id, monit_date) %>%
#   mutate(
#     tm_calculation_date = if_else(
#       flg_retro == 1, min(tm_calculation_date), ## If it's a retro monit_date, we will use the minimun calculation_date.
#       max(if_else(tm_calculation_date <= as.Date(monit_date), tm_calculation_date, ''))) ## If it's not a retro monit_date, we will use the maximun calculation_date, before the monit_date.
#     , tm_max_balance_owed= max(if_else(th_due_date <= as.Date(monit_date), total_balance_owed, 0))
#     , rnk = row_number(coalesce(th_due_date, "1900-01-01"))
#   ) %>%
#   ungroup() %>%
#   select(-total_balance_owed) %>% 
#   spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_energy_trade_hist_tmp"), mode = "overwrite")

### FS Trade Retro ----
tm_retro_temp <- tbl(cn, "brazilbureau.bu_br_trade_retro") %>%
  mutate(tm_created_date = as.Date(tm_created_date), tm_modified_date = as.Date(tm_modified_date)) %>% 
  filter(data_date <= db_data_date) %>% 
  # we are adding this left_join below to edit the tm_trade_id so to have it as the value in the field tm_created_by, from this lookup table, in case the original "tm_trade_id" is in the lookup table.
  # This is a fix for the fact that, due to the soft delete from the FIs, we were attaching a new trade id to the same trade, when it was "soft deleted" and started being reported again..
  # Also, this fix is meant only for the data_dates from june/2024 on.
  
  # OBS: REMOVE depara for mismatch because we're using older ´db_data_date´
  # left_join(tbl(cn, "bu_brbureau.bferrei_depara_trade_id") %>% select(tm_trade_id,tm_created_by), by = "tm_trade_id") %>% 
  # OBS: REMOVING DB_DATA_DATE CHECK BECAUSE ISSUES WITH RETRO MISMATCH
  # mutate(tm_trade_id = coalesce(tm_created_by, tm_trade_id)) %>% select(-tm_created_by) %>% 
  # mutate(tm_trade_id = case_when(db_data_date >= '2024-06-20' ~ coalesce(tm_created_by, tm_trade_id), T ~ tm_trade_id)) %>% select(-tm_created_by) %>% ## 2024-07-25 CHANGE: added the case_when condition for when db_data_date >= '2024-06-20'
  # We'll make this join considering the closest data_date to the actual monitoring_date, that way we'll have the appropriate registers to select from
  # the first load in trade_retro was in September/23 and the table doesn't stack, due to that if the monit_date is until that point we'll end up collecting all
  # history instead of only the most recent one we want.
  inner_join(tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin")) %>% 
               mutate(
                 data_date_aux = if_else(monit_date <= "2023-09-30", "2023-09-30", last_day(monit_date))
               ) %>% 
               select(tm_trade_id, monit_date, data_date_aux)
             , by = c("tm_trade_id"="tm_trade_id", "data_date"="data_date_aux")
  ) %>%
  distinct(
    tm_trade_id,
    sj_subject_id,
    tm_calculation_date,
    tm_restriction_type,
    tm_quota_status,
    tm_balance_owed,
    tm_max_balance_owed,
    tm_quota_payout_date,
    tm_created_date,
    tm_modified_date,
    tm_internal_base_date 
  ) %>% 
  compute()

### FS Trade Master plus Retro ----
# BERNARDO: Here we are binding the whole trade_master with whatever is in the trade_retro, enriched by the fields in trade_master
# This is what we'll need for them let's say, "finished trade_retro" in the scripts for the algorithms, because it has all records in the trade_retro, plus the most recent one, which is originally only in the trade_master.
# So, in here we'll de-duplicate to pick only the most recent record that is older than the monit_date (by it's modified_date).
tm_plus_retro <-
  sdf_bind_rows(
    
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin")) %>% 
      mutate(flg_master = 1),
    
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin")) %>% 
      select( # only the fields that we don't have in the trade_retro
        tm_trade_id, monit_date, transf_date, sj_subject_id, tm_unique_no, tm_contract_date, tm_mode,
        tm_last_installment_duedate, tm_future_contracted_value, 
        tm_number_installments,  
        member_code, industry, tm_mode_grp, flg_retro
      ) %>% 
      inner_join(
        tm_retro_temp %>% 
          select(
            tm_trade_id,
            tm_calculation_date,
            tm_restriction_type,
            tm_quota_status,
            tm_balance_owed,
            tm_max_balance_owed,
            tm_quota_payout_date,
            tm_created_date,
            tm_modified_date,
            tm_internal_base_date
          ),
        by = "tm_trade_id"
      ) %>% 
      mutate(flg_master = 0)
    
  ) %>% 
  filter((tm_created_date <= monit_date & tm_internal_base_date <= monit_date & flg_retro == 0) | flg_retro == 1) %>% ## Filter only records available before the monit_date for non-retro.
  group_by(tm_trade_id, monit_date) %>%
  # after ordering by the created_date, we need to order by the modified date, 
  # and that's actually what is going to make a difference. 
  # We are ordering by created_date first just in case, and also by the flg_master, 
  # just to avoid ties between the recent registers from master and retro, which shouldn't happen
  window_order(desc(tm_internal_base_date), desc(data_date), desc(flg_master)) %>% 
  mutate(
    desc_order = row_number(),
    max_row = n()
  ) %>%
  mutate(
    use_row = case_when(
      # For retro period, use oldest date
      flg_retro == 1 & desc_order == max_row ~ 1,
      # Otherwise, use newest
      flg_retro == 0 & desc_order == 1 ~ 1,
      TRUE ~ 0)
  ) %>% 
  filter(use_row == 1) %>% 
  compute()


### Trades Master ----
sdf_bind_rows(
  # Financial trades
  tm_plus_retro %>%
    select(
      tm_trade_id, monit_date, transf_date, sj_subject_id, tm_unique_no, tm_contract_date, tm_mode,
      tm_calculation_date, tm_last_installment_duedate, tm_future_contracted_value, 
      tm_number_installments, tm_quota_status, tm_balance_owed, 
      tm_quota_payout_date, tm_created_date, tm_max_balance_owed,
      member_code, industry, tm_mode_grp),
  # TelCo Trades
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_telco_trade_hist_tmp")) %>%
    filter(rnk == 1) %>% ## Selecting only 1 Telco trade master record (We had many due to the different calculation_date).
    # Need restriction type to identify TelCo fraud
    select(tm_trade_id, monit_date, transf_date, sj_subject_id, tm_contract_date, tm_calculation_date, tm_created_date,
           tm_max_balance_owed, tm_restriction_type, member_code, industry) %>%
    # Create LOB information for TelCo
    mutate(
      tm_mode = 'T01',
      tm_mode_grp = 4),
  # Energy Trades
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_energy_trade_hist_tmp")) %>%
    filter(rnk == 1) %>% ## Selecting only 1 Energy trade master record (We had many due to the different calculation_date).
    # Need restriction type to identify energy fraud
    select(tm_trade_id, monit_date, transf_date, sj_subject_id, tm_contract_date, tm_calculation_date, tm_created_date,
           tm_max_balance_owed, tm_restriction_type, member_code, industry) %>%
    # Create LOB information for Energy
    mutate(
      tm_mode = 'EL01',
      tm_mode_grp = 5)
) %>% 
  mutate(tm_calculation_date = as.Date(tm_calculation_date)) %>% 
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre"), 
                    mode = "overwrite", partition_by = "tm_mode_grp")

# check mismatch
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre")) %>% filter(sj_subject_id == 5136470) %>% collect() %>% View()

### Trade History Run ----
sdf_bind_rows(
  tbl(cn, "brazilbureau.bu_br_trade_history") %>%
    filter(
      data_date == db_data_date
      , th_active_flag == 1
    ) %>%
    select(th_trade_id, tm_trade_id, th_due_date, th_value, th_min_pay_value, th_created_date, th_active_flag, data_date) %>% 
    ## We are using the _gcdf_run_trade_master table as starting point because it already has all the trades and monitoring dates to which we need to compute GCDF metrics
    inner_join(
      tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre")) %>%
        filter(tm_mode_grp %in% c(1, 2, 3)) %>% ## Only Fin Trades
        select(tm_trade_id, monit_date, transf_date, tm_mode_grp, tm_calculation_date), 
      by = c("tm_trade_id")),
  ## Here we are collecting the Telco records, already selected only for the trades we are watching.
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_telco_trade_hist_tmp")) %>%
    select(th_trade_id, tm_trade_id, monit_date, transf_date, th_due_date, th_value, th_created_date,
           tm_calculation_date, th_calculation_date, th_invoice_status) %>%
    mutate(
      th_min_pay_value = th_value,
      tm_mode_grp = 4),
  ## Here we are collecting the Energy records, already selected only for the trades we are watching.
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_energy_trade_hist_tmp")) %>%
    select(th_trade_id, tm_trade_id, monit_date, transf_date, th_due_date, th_value, th_created_date,
           tm_calculation_date, th_calculation_date, th_invoice_status) %>%
    mutate(
      th_min_pay_value = th_value,
      tm_mode_grp = 5)
  
) %>%
  ## Create the filter if we should or not use the historic record.
  ## 2024-07-31 Adding tm_mode_grp in order to set min_create_dt correctly, once tm_trade_ids are not unique throught different segments
  group_by(tm_mode_grp, tm_trade_id, monit_date) %>% 
  mutate(min_created_dt = min(th_created_date)) %>% 
  ungroup() %>% 
  mutate(
    th_use = case_when(
      monit_date >= min_created_dt & monit_date >= as.Date(th_created_date) ~ 1, ## Live-data logic, the bill must be created before the monit_date.
      monit_date < min_created_dt & th_due_date <= transf_date ~ 1, ## Historic logic, the bill must be due before the monit_date.
      TRUE ~ 0)) %>%
  filter(th_use == 1) %>% ## Already filtering only the valid records
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history"), 
                    mode = "overwrite", partition_by = "tm_mode_grp")

### Trade History Payment Run ----
# This one takes a while to run because there are lots of th_trade_ids
sdf_bind_rows(
  # FS
  tbl(cn, "brazilbureau.bu_br_trade_history_payment") %>%
    filter(
      data_date == db_data_date
      , thp_active_flag == 1
    ) %>%
    select(th_trade_id, thp_payment_date, thp_payment_value, thp_created_date) %>% 
    inner_join(
      tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history")) %>%
        filter(tm_mode_grp %in% c(1, 2, 3)) %>%
        select(tm_trade_id, th_trade_id, tm_mode_grp, th_due_date, th_created_date, monit_date, transf_date), 
      by = "th_trade_id"),
  # TelCo
  tbl(cn, "brazilbureau.bu_br_telco_trade_history_payment") %>% 
    filter(
      data_date == db_data_date,
      tthp_active_flag == 1) %>%
    select(tth_trade_id, tthp_payment_date, tthp_payment_value, tthp_created_date) %>%
    rename_with(~gsub("^tt", "t", .x)) %>%
    inner_join(
      tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history")) %>%
        filter(tm_mode_grp == 4) %>%
        select(tm_trade_id, th_trade_id, tm_mode_grp, th_due_date, th_created_date, monit_date, transf_date), 
      by = "th_trade_id"),
  # Energy
  tbl(cn, "brazilbureau.bu_br_energy_trade_history_payment") %>%
    filter(
      data_date == db_data_date,
      ethp_active_flag == 1) %>%
    select(eth_trade_id, ethp_payment_date, ethp_payment_value, ethp_created_date) %>%
    rename_with(~gsub("^et", "t", .x)) %>%
    inner_join(
      tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history")) %>%
        filter(tm_mode_grp == 5) %>%
        select(tm_trade_id, th_trade_id, tm_mode_grp, th_due_date, th_created_date, monit_date, transf_date),
      by = "th_trade_id")
) %>% 
  ## 2024-07-31 Adding tm_mode_grp in order to set min_create_dt correctly, once tm_trade_ids are not unique throught different segments
  group_by(tm_mode_grp, tm_trade_id, monit_date) %>% 
  mutate(min_created_dt = min(thp_created_date)) %>% 
  ungroup() %>%
  mutate(
    thp_use = case_when(
      monit_date >= min_created_dt & monit_date >= as.Date(thp_created_date) ~ 1,
      monit_date < min_created_dt & thp_payment_date <= transf_date ~ 1,
      TRUE ~ 0)) %>%
  filter(thp_use == 1) %>%
  sdf_distinct() %>% # Is this really necessary?
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history_payment"), 
                    mode = "overwrite", partition_by = "tm_mode_grp")

# check mismatch
# tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history_payment")) %>% 
#   filter(tm_trade_id == 881429759) %>% collect() %>% View('check_algos_gcdf_run_trade_history_payment')

### Trade payment run ----
tbl(cn, "brazilbureau.bu_br_trade_payment") %>%
  filter(data_date == db_data_date,
         trim(coalesce(tp_payment_type, '99')) != 'L',
         tp_active_flag == 1) %>%
  inner_join(
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre")) %>%
      filter(tm_mode_grp %in% c(1, 2, 3)) %>%
      select(tm_trade_id, monit_date, tm_mode_grp, transf_date),
    by = "tm_trade_id") %>%
  group_by(tm_trade_id, monit_date) %>% 
  mutate(min_created_dt = min(tp_created_date)) %>% 
  ungroup() %>%
  mutate(
    tp_use = case_when(
      monit_date >= min_created_dt & monit_date >= as.Date(tp_created_date) ~ 1,
      monit_date < min_created_dt & tp_payment_date <= transf_date ~ 1,
      TRUE ~ 0)) %>%
  filter(tp_use == 1) %>%
  select(tm_trade_id, paymt_dt = tp_payment_date, paymt_value = tp_payment_value, monit_date, transf_date, tm_mode_grp, data_date) %>%
  sdf_distinct() %>% # Is this really necessary?
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_payment"), 
                    mode = "overwrite", partition_by = "tm_mode_grp")


### Trade future run ----
tbl(cn, "brazilbureau.bu_br_trade_future") %>%
  filter(data_date == db_data_date) %>%
  inner_join(
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre")) %>%
      filter(tm_mode_grp %in% c(1, 2, 3)) %>%
      select(tm_trade_id, monit_date, tm_mode_grp, transf_date),
    by = "tm_trade_id") %>%
  group_by(tm_trade_id, monit_date) %>% 
  mutate(min_created_dt = min(tf_created_date)) %>% 
  ungroup() %>%
  mutate(
    tf_use = case_when(
      monit_date >= min_created_dt & monit_date >= as.Date(tf_created_date) ~ 1,
      monit_date < min_created_dt ~ 1,
      TRUE ~ 0)) %>%
  filter(tf_use == 1) %>%
  select(
    tm_trade_id, tf_next_instal_due_date, tf_future_periodicity,
    tf_next_instal_value, tf_num_installments, tf_num_inst_payable, tf_min_pay_value,
    tf_created_date, data_date, tm_mode_grp, monit_date, transf_date
  ) %>%
  sdf_distinct() %>% 
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_future"), 
                    mode = "overwrite", partition_by = "tm_mode_grp")


### Disputes ----
tbl(cn, "brazilbureau.bu_br_dispute_details") %>% 
  filter(data_date == db_data_date, ## We are checking for current dispute status, so bkt results will be reflecting on that.
         # ,!trim(disp_status) %in% c('5', '7', '8') ## In theory, this is no longer needed, since we had disp_condition_1 and disp_condition_2.
         disp_active_flag == 1
  ) %>%
  select(disp_fi_member_cnpj, disp_fi_master_cnpj, sj_subject_id, disp_opr_unique_num,
         disp_status, disp_type, disp_reason_code, data_date) %>%
  # TODO - left join to gcdf_run_trade_master to propagate class 2 to member code
  inner_join(
    tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% 
      select(sj_subject_id, monit_date), 
    by = "sj_subject_id") %>%
  mutate(member_code = case_when(
    coalesce(disp_fi_member_cnpj, 0) == 0 ~ disp_fi_master_cnpj,
    TRUE ~ disp_fi_member_cnpj)) %>%
  group_by(member_code, sj_subject_id, monit_date) %>%
  mutate(
    disp_condition_2 = max(case_when(
      disp_type == 'C' & disp_status %in% c('I', 'S', 'A', 'E', '0') & disp_reason_code == 'X01' ~ 1,
      TRUE ~ 0)) ) %>%
  group_by(member_code, sj_subject_id, disp_opr_unique_num, monit_date) %>%
  summarize(
    disp_condition_1 = max(case_when(
      disp_type == 'O' & disp_status %in% c('I', 'S', 'A', 'E', '0') ~ 1,
      TRUE ~ 0)),
    disp_condition_2 = max(disp_condition_2) ) %>%
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_dispute_details_sample"), mode = "overwrite")

##--##--##--##--##--##--##--##--##--##--##
## filtering the raw data accordingly ----

## Now we want to create the sample for the so called trade_retro table. In the meanwhile, we are building it from the trade_master, looking at its many data_dates, but as soon as it's available in shape, then, of course, we'll build it straight from the trade_retro itself.

### Trades RETRO sampling ----
sdf_bind_rows(
  
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin")),
  
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_fin")) %>% 
    select( # only the fields that we don't have in the trade_retro
      tm_trade_id, monit_date, transf_date, sj_subject_id, tm_unique_no, tm_contract_date, tm_mode,
      tm_last_installment_duedate, tm_future_contracted_value, 
      tm_number_installments,  
      member_code, tm_mode_grp, flg_retro
    ) %>% 
    inner_join(
      tm_retro_temp %>% 
        select(
          tm_trade_id,
          tm_calculation_date,
          tm_restriction_type,
          tm_quota_status,
          tm_balance_owed,
          tm_max_balance_owed,
          tm_quota_payout_date,
          tm_created_date,
          tm_modified_date,
          tm_internal_base_date
        ),
      by = "tm_trade_id"
    )
  
) %>%
  spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_trade_retro_sample"), mode = "overwrite", partition_by = "tm_mode_grp")

### assigning each payment made to the appropriate month for paymt_amt_act filter ----

sdf_sql(cn, str_glue("
select
tm_trade_id,
monit_date,
tm_mode_grp,
assigned_paymt_month,
sum(thp_payment_value) as paymt_amt_act_assign
from
(
  select
  th.tm_trade_id,
  th.th_due_date,
  th.tm_mode_grp,
  thp.monit_date,
  --thp.thp_payment_id,
  thp.thp_payment_date,
  thp.thp_payment_value,
  case when thp.thp_payment_date >= date_add(th.th_due_date, -30) and
            thp.thp_payment_date <= least(last_day(th.th_due_date), thp.monit_date)
       then
            --- we had to add this extra condition here for the cases in which the monit_date was before the due month. Say the due date is march/10, monit date is feb/25 and payment date is feb/20, in this case we don't want to assign feb as the paymt_month but still march, so it couldn't just be the least between last_day(due_date) and the monit_date
            case when month(th.th_due_date) = month(thp.monit_date) and year(th.th_due_date) = year(thp.monit_date)
                 then least(last_day(th.th_due_date), thp.monit_date)
                 else last_day(th.th_due_date) end
       else least(last_day(thp.thp_payment_date), thp.monit_date)
       end as assigned_paymt_month
  from
  bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history th
  inner join
  bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment thp on
  th.th_trade_id = thp.th_trade_id
  and th.monit_date = thp.monit_date
  and th.tm_trade_id = thp.tm_trade_id
  and th.tm_mode_grp = thp.tm_mode_grp
  where
  thp.thp_payment_date <= thp.monit_date
) a
group by
tm_trade_id,
monit_date,
tm_mode_grp,
assigned_paymt_month")) %>% 
          spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_paymt_month_assignment"), mode = "overwrite",  partition_by = "tm_mode_grp")
        
        ##--##--##--##--##--##--##--##--##--##--##--##--##        
        ## Building GCDF dataset for algos calculations ----
        
        # In case you need to rerun pay attention to this point, the best is to start from scratch !
        dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_gcdf_trhi_sql_pre'))
        dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_gcdf_trhi_sql'))
        
        for (ref_id in 0:29) {
          
          cat(paste("L001 starting on:", ref_id, "at", Sys.time(), "\n"), file = ln, append = TRUE) 
          
          ### Trade master ----
          tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master_pre")) %>% 
            filter(tm_contract_date <= least(monit_date, ADD_MONTHS(LAST_DAY(monit_date), -ref_id))) %>% 
            # We'll be creating a monit_date_orig which will be the original date supplied and the one used in all the joins with the sample tables
            # and then the monit_date will be the date used for the calculations (from the original to 30 past positions, which we'll use the ref_id to create)
            mutate(
              monit_date_orig = monit_date
              , monit_date = least(monit_date, last_day(add_months(last_day(monit_date), -ref_id)))
              , data_join = least(last_day(tm_calculation_date),last_day(monit_date))
            ) %>% 
            spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master"), mode = "overwrite")
          
          # check mismatch
          # tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master")) %>% filter(tm_trade_id == 654712510)
          
          ### Trade Date V0 ----
          sdf_sql(cn, str_glue(
            "SELECT
        tdts.*,
        YEAR(tdts.ref_date) * 100 + MONTH(tdts.ref_date) AS ref_month
    FROM (
        SELECT
            tm.tm_trade_id,
            tm.member_code,
            tm.tm_mode_grp,
            tm.tm_calculation_date AS reported_dt,
			      tm.monit_date,
			      tm.monit_date_orig,
            cast(1 as int) as ref_date_id,
            cast(tm.monit_date as date) as ref_date
        FROM bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master AS tm
    ) AS tdts")) %>%
            spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_trade_date"), mode = "overwrite")
          
          # check mismatch
          # tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_trade_date")) %>% filter(tm_trade_id == 654712510) %>% collect() %>% View()
          
          ### CC MOP v0 ----
          sdf_sql(cn, str_glue(
            "select
cc.*,
row_number() over(partition by cc.tm_trade_id, cc.tm_mode_grp, cc.monit_date, cc.ref_date_id order by coalesce(cc.th_due_date, cast('1900-01-01' as date)) desc, cc.past_due_amt, cc.value_paid desc, cc.th_trade_id) as due_date_order, -- for us to pick the most recent due date
case when past_due_amt <= 0.000001 then 1 else 0 end as flg_good
from
(
  select
  cc2.*,
  case when cc2.tm_mode_grp = 1 AND cc2.th_value < 10 then 0 else greatest(cc2.th_min_pay_value - cc2.value_paid, 0) end as past_due_amt
  from
  (
    select
    cc1.tm_trade_id,
	  cc1.tm_mode_grp,
    cc1.monit_date,
	  cc1.monit_date_orig,
    cc1.ref_date_id,
    cc1.th_trade_id,
    FIRST(cc1.th_due_date) AS th_due_date,
    FIRST(cc1.dpd_cc) AS dpd_cc,
    FIRST(cc1.th_value) AS th_value,
    FIRST(cc1.th_min_pay_value) AS th_min_pay_value,
    sum(coalesce(pay.thp_payment_value, 0)) as value_paid
	from
    (
      select
      trade_date.tm_trade_id,
	    trade_date.tm_mode_grp,
      trade_date.monit_date,
	    trade_date.monit_date_orig,
      trade_date.ref_date_id,
      trade_date.ref_date,
      th.th_trade_id,
      th.th_due_date,
      datediff(least(trade_date.ref_date, trade_date.reported_dt), th.th_due_date) as dpd_cc,
      th.th_value,
      th.th_min_pay_value
      from
        (select 
            tm_trade_id, 
			tm_mode_grp,
			monit_date, 
			monit_date_orig, 
			ref_date_id, 
			ref_date, 
			reported_dt,
            CASE WHEN member_code = 360305000104 THEN DATE_ADD(reported_dt, -15) ELSE reported_dt END AS match_rept_dt
        from bu_brbureau.{eid}{prj}_algos_gcdf_trade_date where tm_mode_grp = 1) trade_date				
      left join
        (select
            tm_trade_id, 
			tm_mode_grp,
			monit_date, 
			th_trade_id, 
			th_due_date, 
			th_value, 
			th_min_pay_value
        from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history
        where tm_mode_grp = 1 AND th_use = 1
        ) th on
          trade_date.monit_date_orig = th.monit_date and
		      trade_date.tm_mode_grp  = th.tm_mode_grp and
          trade_date.tm_trade_id  = th.tm_trade_id and
          trade_date.ref_date    >= th.th_due_date and
          trade_date.match_rept_dt >= th.th_due_date
    ) cc1
    left join
    bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment pay on
	  cc1.tm_trade_id = pay.tm_trade_id and
      cc1.monit_date_orig = pay.monit_date and
      cc1.th_trade_id  = pay.th_trade_id      and
      cc1.ref_date    >= pay.thp_payment_date and
	  pay.tm_mode_grp = 1 AND pay.thp_use = 1
    group by
    cc1.tm_trade_id,
	  cc1.tm_mode_grp,
    cc1.monit_date,
	  cc1.monit_date_orig,
    cc1.ref_date_id,
    cc1.th_trade_id
  ) cc2
) cc")) %>%
              group_by(tm_trade_id, monit_date, monit_date_orig, ref_date_id, tm_mode_grp) %>%
              mutate(
                last_good = min(case_when(flg_good == 1 ~ due_date_order, TRUE ~ 999)), 
                last_obs  = n(),
                flg_history = max(case_when(!is.na(th_trade_id) ~ 1, TRUE ~ 0)),
                flg_good_history = max(case_when(flg_good == 1 ~ 1, TRUE ~ 0))
              ) %>%
              summarize(
                past_due_amt = max(case_when(due_date_order == 1 ~ past_due_amt, TRUE ~ NaN)),
                dpd = coalesce(
                  max(case_when(flg_history == 0 ~ -999, TRUE ~ NaN)),
                  max(case_when(due_date_order == 1 & flg_good == 1 ~ 0, TRUE ~ NaN)),
                  max(case_when(flg_good_history == 1 & due_date_order == last_good - 1 ~ dpd_cc, TRUE ~ NaN)),
                  max(case_when(due_date_order == last_obs ~ dpd_cc, TRUE ~ NaN))),
                # value_paid = max(case_when(due_date_order == 1 ~ value_paid, TRUE ~ NaN)),
                th_value = max(case_when(due_date_order == 1 & !is.na(th_value) ~ greatest(th_value, 0), TRUE ~ -999)),
                th_due_date = max(case_when(due_date_order == 1 ~ coalesce(th_due_date, as.Date('1900-01-01')), TRUE ~ as.Date('1900-01-01'))),
                .groups = "drop") %>%
              # This mutate can be dropped when we move this calculation to the API functions
              mutate(mop_v0 = case_when(
                dpd < 0 ~ 0,
                dpd == 0 ~ 1,
                dpd < 31 ~ 2,
                dpd < 62 ~ 3,
                dpd < 92 ~ 4,
                dpd < 123 ~ 5,
                dpd >= 123 ~ 6,
                TRUE ~ -999)) %>%
              group_by(tm_trade_id, monit_date, monit_date_orig, tm_mode_grp) %>%
              # Note: Above this point you could use for Trade History
              # Note: this whole summarize can be replaced by the array collection/row filter
              summarize(
                mop_v0_1 = as.integer(min(case_when(ref_date_id == 1 ~ mop_v0, TRUE ~ 998))),
                dpd_v0_1 = as.integer(max(case_when(ref_date_id == 1 ~ dpd, TRUE ~ -999))),
                past_due_amt_v0_1 = max(case_when(ref_date_id == 1 ~ past_due_amt, TRUE ~ 0)),
                # paymt_amt_act_v0_1 = max(case_when(ref_date_id == 1 ~ value_paid, TRUE ~ 0)),
                max_hist_bal_v0_1 = as.numeric(0),
                billed_balance_v0_1 = max(case_when(ref_date_id == 1 ~ th_value, TRUE ~ -999)),
                most_recent_due_date = max(case_when(ref_date_id == 1 ~ th_due_date, TRUE ~ as.Date('1900-01-01'))),
                remain_paymt = as.integer(NaN),
                remain_inst = as.integer(NaN),
                remain_paymt_bal = as.integer(NaN),
                .groups = "drop"
              ) %>%
              # mutate(
              #     mop_ary = sql("collect_list(dpd) over(partition by tm_trade_id order by ref_date_id range between current row and unbounded following)")
              # ) %>%
              # #filter(ref_date_id == 1) %>%
              # #select(-ref_date_id, -dpd) %>%
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_cc"), mode = "overwrite")
            
            
            ### TELCO & Energy MOP v0 ----
            sdf_sql(cn, str_glue(
              "SELECT
      tl.*,
			CASE WHEN past_due_amt <= 0.000001 THEN 1 ELSE 0 END AS flg_good
      FROM
      (
        SELECT
          tl2.*,
            CASE
              WHEN tl2.tm_mode_grp IN (5, 4)
                AND tl2.th_calculation_date <= tl2.ref_date 
                AND COALESCE(tl2.th_invoice_status, 'F') = 'A' 
                AND tl2.th_value >= 10 
                AND COALESCE(tl2.value_paid, 0) < tl2.th_min_pay_value 
                  THEN tl2.th_value
              WHEN tl2.tm_mode_grp IN (5, 4) 
                AND tl2.th_calculation_date > tl2.ref_date 
                AND tl2.th_value >= 10 
                AND COALESCE(tl2.value_paid, 0) < tl2.th_min_pay_value 
                  THEN tl2.th_value
              ELSE 0 END AS past_due_amt
        FROM
        (
          SELECT
            tl1.tm_trade_id,
            tl1.tm_mode_grp,
			      tl1.monit_date,
			      tl1.monit_date_orig,
            tl1.ref_date_id,
            tl1.ref_date,
            tl1.th_trade_id,
            FIRST(tl1.th_due_date) AS th_due_date,
            FIRST(tl1.dpd_elec) AS dpd_elec,
            FIRST(tl1.th_value) AS th_value,
            FIRST(tl1.th_min_pay_value) AS th_min_pay_value,
            FIRST(tl1.th_calculation_date) AS th_calculation_date,
            FIRST(tl1.th_invoice_status) AS th_invoice_status,
            sum(coalesce(pay.thp_payment_value, 0)) as value_paid
          FROM
          (
            SELECT
              trade_date.tm_trade_id,
              trade_date.tm_mode_grp,
			        trade_date.monit_date,
			        trade_date.monit_date_orig,
              trade_date.ref_date_id,
              trade_date.ref_date,
              th.th_trade_id,
              th.th_due_date,
              datediff(least(trade_date.ref_date, trade_date.reported_dt), th.th_due_date) as dpd_elec,
              th.th_value,
              th.th_min_pay_value,
              th.th_calculation_date,
              th.th_invoice_status
            FROM
            (
              SELECT
                tm_trade_id,
                tm_mode_grp,
				        monit_date,
				        monit_date_orig,
                ref_date_id,
                ref_date,
                reported_dt,
                CASE WHEN member_code = 360305000104 THEN DATE_ADD(reported_dt, -15) ELSE reported_dt END AS match_rept_dt
              FROM bu_brbureau.{eid}{prj}_algos_gcdf_trade_date where tm_mode_grp IN (5, 4)) trade_date
              LEFT JOIN
              (
                SELECT
                  tm_trade_id,
                  tm_mode_grp,
				          monit_date,
                  th_trade_id,
                  th_due_date,
                  th_value,
                  th_min_pay_value,
                  th_calculation_date,
                  th_invoice_status
                FROM bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history
                WHERE tm_mode_grp IN (5, 4) AND th_use = 1
              ) th ON
				        trade_date.monit_date_orig = th.monit_date and
                trade_date.tm_trade_id     = th.tm_trade_id  AND
                trade_date.tm_mode_grp     = th.tm_mode_grp  AND
                trade_date.ref_date       >= th.th_due_date AND
                trade_date.match_rept_dt  >= th.th_due_date
          ) tl1
          LEFT JOIN
          bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment pay ON
            tl1.tm_trade_id     = pay.tm_trade_id   AND
			      tl1.monit_date_orig = pay.monit_date    AND
            tl1.th_trade_id     = pay.th_trade_id   AND
            tl1.ref_date       >= pay.thp_payment_date AND
            tl1.tm_mode_grp     = pay.tm_mode_grp   AND
            pay.tm_mode_grp IN (5, 4)               AND
            pay.thp_use = 1
          GROUP BY
            tl1.tm_trade_id,
            tl1.tm_mode_grp,
			      tl1.monit_date,
			      tl1.monit_date_orig,
            tl1.ref_date_id,
            tl1.ref_date,
            tl1.th_trade_id
       ) tl2
     ) tl")) %>% 
              group_by(tm_trade_id, tm_mode_grp, monit_date, monit_date_orig, ref_date_id) %>%
              mutate(
                past_due_amt = sum(past_due_amt),
                last_obs  = n(),
                flg_history = max(case_when(!is.na(th_trade_id) ~ 1, TRUE ~ 0)),
                flg_good_history = max(case_when(flg_good == 1 ~ 1, TRUE ~ 0)),
                dpd_elec = max(case_when(flg_good == 0 ~ dpd_elec, TRUE ~ NaN)),
                missed_paymt = sum(case_when(flg_good == 0 ~ 1, TRUE ~ 0))
                # really calculating dpd, only considering invoices with missed_payment
              ) %>% 
              summarize(
                past_due_amt = max(case_when(ref_date_id == 1 ~ past_due_amt, TRUE ~ NaN)),
                dpd = coalesce(
                  max(case_when(flg_history == 0 ~ -999, TRUE ~ NaN)),
                  max(case_when(flg_good == 0 ~ dpd_elec, TRUE ~ NaN))),
                missed_paymt = max(missed_paymt),
                th_due_date = max(coalesce(th_due_date, as.Date('1900-01-01'))), # most recent due_date
                flg_history = max(flg_history), # NNote: we'll use this to verify if we actually have valid history in the trade
                .groups = "drop") %>% 
              # This mutate can be dropped when we move this calculation to the API functions
              mutate(
                # mop_v0 = case_when(missed_paymt >= 6 ~ 6, TRUE ~ missed_paymt + 1),
                mop_v0 = case_when(
                  # NNote: Adding this rule here, which will check that if there is no history (no invoices valid for the period), the MOP won't be calculated
                  flg_history == 0 ~ 0 
                  , missed_paymt >= 6 ~ 6
                  , TRUE ~ missed_paymt + 1
                )
              ) %>%
              # Note: we want to make sure the mop considered for each ref_date_id will actually
              # be the worst case scenario even when facing parcial payments in due_date_orders
              group_by(tm_trade_id, tm_mode_grp, monit_date, monit_date_orig, ref_date_id) %>%
              mutate(
                mop_v0 = max(mop_v0),
              ) %>%    
              group_by(tm_trade_id, monit_date, monit_date_orig, tm_mode_grp) %>%
              # Note: Above this point you could use for Trade History
              # Note: this whole summarize can be replaced by the array collection/row filter
              summarize(
                mop_v0_1 = as.integer(min(case_when(ref_date_id == 1 ~ mop_v0, TRUE ~ 998))),
                dpd_v0_1 = as.integer(max(case_when(ref_date_id == 1 ~ dpd, TRUE ~ -999))),
                past_due_amt_v0_1 = max(case_when(ref_date_id == 1 ~ past_due_amt, TRUE ~ 0)),
                max_hist_bal_v0_1 = as.numeric(0),
                billed_balance_v0_1 = -999,
                most_recent_due_date = max(case_when(ref_date_id == 1 ~ th_due_date, TRUE ~ as.Date('1900-01-01'))),
                remain_paymt = as.integer(NaN),
                remain_inst = as.integer(NaN),
                remain_paymt_bal = as.integer(NaN),
                .groups = "drop"
              ) %>% 
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_telco_utilities"), mode = "overwrite")
            
            
            ### REV MOP #1 ----
            sdf_sql(cn, str_glue(
              "SELECT
        trade_date.tm_trade_id,
	    	trade_date.monit_date,
		    trade_date.monit_date_orig,
        trade_date.ref_date_id,
        trade_date.ref_date,
        trade_date.reported_dt,
        CASE WHEN trade_date.reported_dt >= th.th_due_date THEN th.th_due_date ELSE NULL END  AS th_due_date,
        CASE WHEN trade_date.reported_dt >= th.th_due_date THEN LAST_DAY(th.th_due_date) ELSE NULL END AS due_month,
        CASE WHEN trade_date.reported_dt >= th.th_due_date THEN th.th_value ELSE NULL END AS th_value,
        MAX(COALESCE(th.th_value, 0)) OVER(PARTITION BY trade_date.tm_trade_id, trade_date.monit_date,  trade_date.ref_date_id) AS max_hist_bal
    FROM (
        SELECT *
        FROM bu_brbureau.{eid}{prj}_algos_gcdf_trade_date
        WHERE tm_mode_grp = 2
        ) AS trade_date
    LEFT JOIN (
        SELECT *
        FROM bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history
        WHERE tm_mode_grp = 2 AND th_use = 1
        ) AS th
    ON  trade_date.tm_trade_id = th.tm_trade_id AND
		    trade_date.monit_date_orig = th.monit_date AND
        trade_date.ref_date >= th.th_due_date and
    		trade_date.reported_dt >= th.th_due_date
")) %>%
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_rev1"), mode = "overwrite")
            
            # check mismatch
            # tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_rev1")) %>% filter(tm_trade_id == 654712510) %>% collect() %>% View('check_gcdf_mop_v0_rev1')
           
            
            ### REV MOP final ----
            sdf_sql(cn, str_glue(
              "SELECT
        tb2.tm_trade_id,
	    	tb2.monit_date,
		    tb2.monit_date_orig,
        tb2.ref_date_id,
        tb2.ref_date,
        LAST_DAY(td.ref_date) AS due_month,
        tb2.max_hist_bal,
        a.th_due_date,
        a.th_value,
        DATEDIFF(LEAST(a.ref_date, a.reported_dt), a.th_due_date) AS dpd_rev,
        ROW_NUMBER() OVER(PARTITION BY tb2.tm_trade_id, tb2.monit_date, tb2.ref_date_id
                          ORDER BY a.th_due_date DESC) AS due_date_order,
        CASE WHEN COALESCE(a.th_value, 0) <= 0.000001 THEN 1 ELSE 0 END AS flg_good
    FROM (
        SELECT
            tm_trade_id,
			      monit_date,
			      monit_date_orig,
            ref_date_id,
            ref_date,
            max_hist_bal,
            MIN(due_month) AS last_due_month,
            MAX(due_month) AS first_due_month
        FROM bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_rev1
        GROUP BY tm_trade_id, monit_date, monit_date_orig, ref_date_id, ref_date, max_hist_bal
    ) AS tb2
    LEFT JOIN bu_brbureau.{eid}{prj}_algos_gcdf_trade_date AS td
    ON  tb2.tm_trade_id = td.tm_trade_id AND
		    tb2.monit_date = td.monit_date_orig AND
        tb2.first_due_month >= LAST_DAY(td.ref_date) AND
        tb2.last_due_month <= LAST_DAY(td.ref_date) AND
		    td.tm_mode_grp = 2
    LEFT JOIN bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_rev1 AS a
    ON  tb2.tm_trade_id = a.tm_trade_id AND
		    tb2.monit_date = a.monit_date AND
        tb2.ref_date_id = a.ref_date_id --AND
        --LAST_DAY(td.ref_date) = a.due_month
")) %>%
              group_by(tm_trade_id, monit_date, monit_date_orig, ref_date_id) %>%
              mutate(
                last_good = min(case_when(flg_good == 1 ~ due_date_order, TRUE ~ 999)), 
                last_obs  = max(due_date_order),
                flg_history = max(case_when(!is.na(due_month) ~ 1, TRUE ~ 0)),
                flg_good_history = max(case_when(!is.na(due_month) & flg_good == 1 ~ 1, TRUE ~ 0))
              ) %>%
              summarize(
                past_due_amt = max(case_when(
                  due_date_order > 1 ~ NaN,
                  # flg_history == 0  ~ -999, # Not present in the last Algos code for REV, only for CC.
                  months_between(last_day(ref_date), last_day(th_due_date)) > 1 ~ 0,  # TODO DPD is 0 if observed after due date?
                  TRUE ~ coalesce(th_value, 0))),
                # dpd = coalesce(
                # max(case_when(flg_history == 0 ~ -999, TRUE ~ NaN)), 
                # max(case_when(
                # due_date_order == 1 & flg_good == 1 ~ 0,
                # due_date_order == 1 & datediff(ref_date, th_due_date) > 30 ~ 0,  # TODO interval different than for past due amt
                # TRUE ~ NaN)),
                # max(case_when(flg_good_history == 1 & due_date_order == last_good - 1 ~ dpd_rev + 1, TRUE ~ NaN)),
                # max(case_when(due_date_order == last_obs ~ dpd_rev + 1, TRUE ~ NaN))),
                # NOTE: This gets calculated by a separate process
                # value_paid = max(case_when(
                #     due_date_order > 1 ~ NaN,
                #     flg_history == 0 ~ -999,
                #     TRUE ~ 0)),
                # TODO - We have th_value 
                # th_value = max(case_when(due_date_order == 1 ~ th_value, TRUE ~ -999)),
                th_due_date = max(case_when(due_date_order == 1 ~ th_due_date), TRUE ~ NaN),
                max_hist_bal = max(case_when(due_date_order == 1 ~ max_hist_bal, TRUE ~ NaN)),
                .groups = "drop") %>%
              mutate(mop_v0 = 0,
                     dpd = 0) %>%
              group_by(tm_trade_id, monit_date, monit_date_orig) %>%
              # Note: Above this point you could use for Trade History
              # Note: this whole summarize can be replaced by the array collection/row filter
              summarize(
                tm_mode_grp = 2,
                mop_v0_1 = as.integer(min(case_when(ref_date_id == 1 ~ mop_v0, TRUE ~ 998))),
                dpd_v0_1 = as.integer(max(case_when(ref_date_id == 1 ~ dpd, TRUE ~ -999))),
                past_due_amt_v0_1 = max(case_when(ref_date_id == 1 ~ past_due_amt, TRUE ~ 0)),
                # paymt_amt_act_v0_1 = max(case_when(ref_date_id == 1 ~ value_paid, TRUE ~ 0)),
                max_hist_bal_v0_1 = max(case_when(ref_date_id == 1 ~ max_hist_bal, TRUE ~ 0)),
                billed_balance_v0_1 = max(case_when(ref_date_id == 1 ~ past_due_amt, TRUE ~ -999)),
                most_recent_due_date = max(case_when(ref_date_id == 1 ~ th_due_date, TRUE ~ as.Date('1900-01-01'))),
                remain_paymt = as.integer(NaN),
                remain_inst = as.integer(NaN),
                remain_paymt_bal = as.integer(NaN),
                .groups = "drop"
              ) %>% 
              # mutate(
              #     mop_ary = sql("collect_list(dpd) over(partition by tm_trade_id order by ref_date_id range between current row and unbounded following)")
              # ) %>%
              # #filter(ref_date_id == 1) %>%
              # #select(-ref_date_id, -dpd) %>%
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_rev"), mode = "overwrite")
            
            
        
            ### Retro Remaining Payments and installments ----
            sdf_sql(cn, str_glue(
              "select 
        a.tm_trade_id,
    		a.monit_date,
		    a.monit_date_orig,
        a.ref_date_id,
        FIRST(tf_num_inst_payable) + SUM(case when amt_ind = 1 and timing_ind = 1
                                         then 1 else 0 end) AS remain_paymt,
        FIRST(tf_num_inst_payable) + SUM(case when amt_ind = 1
                                         then 1 else 0 end) as remain_paymt_bal,
        FIRST(tf_num_installments + inbetween_inst) AS remain_inst
from
(
  select
      th.tm_trade_id,
	    th.monit_date,
	    th.monit_date_orig,
      th.ref_date_id,
      th.th_trade_id,
      FIRST(th.tf_num_inst_payable) AS tf_num_inst_payable,
      FIRST(th.tf_num_installments) AS tf_num_installments,
      CASE WHEN sum(thp.thp_payment_value) / FIRST(th.th_value) > 0.9 THEN 1 ELSE 0 END AS amt_ind,
      FIRST(CASE WHEN th.first_created_date >= th.th_due_date THEN 1 ELSE 0 END) AS timing_ind,
      MIN(th.inbetween_inst) AS inbetween_inst
  from
  (
    select
    a.*,
    th.th_trade_id,
    th.th_value,
    th.th_due_date,
    SUM(CASE WHEN a.ref_date <= th.th_due_date AND th.th_due_date < CAST(a.first_created_date AS DATE) THEN 1 ELSE 0 END) OVER(
        PARTITION BY a.tm_trade_id, a.monit_date, a.monit_date_orig, a.ref_date_id) AS inbetween_inst
    from (
        select
            b.tm_trade_id,
			      b.monit_date,
			      b.monit_date_orig,
            b.ref_date_id,
            b.ref_date,
            sum(b.tf_num_inst_payable) as tf_num_inst_payable,
            sum(b.tf_num_installments) as tf_num_installments,
            min(b.tf_created_date) as first_created_date
        from
        (
            select
                tf.tm_trade_id,
				        trade_date.monit_date,
				        trade_date.monit_date_orig,
                trade_date.ref_date_id,
                trade_date.ref_date,
                tf.tf_num_installments,
                tf.tf_num_inst_payable,
                tf.tf_created_date,
                row_number() over(partition by tf.tm_trade_id, trade_date.ref_date_id, tf.tf_future_periodicity 
                                  order by tf.tf_created_date, coalesce(tf.tf_next_instal_due_date, cast('2199-12-31' as date)), tf.tf_num_inst_payable asc) as order_id -- in this query. this ordering must be ascendent, so we pick the oldest record
            from
                bu_brbureau.{eid}{prj}_algos_gcdf_trade_date as trade_date
            inner join
                bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_future tf on
                trade_date.tm_trade_id = tf.tm_trade_id and
				        trade_date.monit_date_orig = tf.monit_date AND
                trade_date.ref_date < cast(tf.tf_created_date as date) -- the casting in the created_date here is for us to catch the records with created date at the monitoring date
            where
                trade_date.tm_mode_grp = 3 AND
                tf.tm_mode_grp = 3
            ) b
        WHERE b.order_id = 1
        group by
            b.tm_trade_id,
			      b.monit_date,
			      b.monit_date_orig,
            b.ref_date_id,
            b.ref_date
    ) AS a
    left join
    bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history th on
    a.tm_trade_id = th.tm_trade_id AND
	  a.monit_date_orig = th.monit_date
  ) th
  left join
  bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment thp on
  th.th_trade_id = thp.th_trade_id and
  th.monit_date_orig = thp.monit_date and
  th.ref_date < thp.thp_payment_date and 
  thp.thp_payment_date < cast(th.first_created_date as date) AND
  thp.tm_mode_grp = 3
  group by 
    th.tm_trade_id, 
  	th.monit_date,
	  th.monit_date_orig,
    th.ref_date_id, 
    th.th_trade_id
) a
group by
    a.tm_trade_id,
	  a.monit_date,
	  a.monit_date_orig,
    a.ref_date_id"
            )) %>% 
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_paymt_pre"), mode = "overwrite")
            
            
            ### putting the post and pre parts for the remain_paymt and remain_inst filters together ----
            sdf_sql(cn, str_glue(
              "select
  post.tm_trade_id,
  post.monit_date,
  post.monit_date_orig,
  post.ref_date_id,
  post.ref_date,
  post.reported_dt,
  case when post.in_future = 1 then post.remain_paymt else pre.remain_paymt end as remain_paymt,
  case when post.in_future = 1 then post.remain_paymt else pre.remain_paymt_bal end as remain_paymt_bal,
  case when post.in_future = 1 then post.remain_inst else pre.remain_inst end as remain_inst,
  CASE WHEN post.in_future = 1 
       THEN LEAST(6, 1 + GREATEST(post.remain_paymt - post.remain_inst, 0))
       ELSE LEAST(6, 1 + GREATEST(pre.remain_paymt - pre.remain_inst, 0))
       END AS paymt_mop
  from (
    select
            a.tm_trade_id,
		      	a.monit_date,
      			a.monit_date_orig,
            a.ref_date_id,
            FIRST(a.ref_date) AS ref_date,
            FIRST(a.reported_dt) AS reported_dt,
            MAX(in_future) AS in_future,
            sum(a.tf_num_installments) as remain_inst,
            sum(a.tf_num_inst_payable) as remain_paymt
    from
    (
      select
          trade_date.tm_trade_id,
		      trade_date.monit_date,
		      trade_date.monit_date_orig,
          trade_date.ref_date_id,
          trade_date.ref_date,
          CASE WHEN trade_date.member_code = 360305000104 THEN date_add(trade_date.reported_dt, -15) ELSE trade_date.reported_dt end as reported_dt,
          CASE WHEN tf.tm_trade_id IS NOT NULL THEN 1 ELSE 0 END AS in_future,
          tf.tf_num_installments,
          tf.tf_num_inst_payable,
          row_number() over(partition by trade_date.tm_trade_id, trade_date.ref_date_id, tf.tf_future_periodicity 
                            order by tf.tf_created_date desc, coalesce(tf.tf_next_instal_due_date, cast('2199-12-31' as date)) desc, tf.tf_num_inst_payable asc) as order_id
      from
      bu_brbureau.{eid}{prj}_algos_gcdf_trade_date as trade_date
      left join
      bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_future tf on
        trade_date.tm_trade_id = tf.tm_trade_id and
		    trade_date.monit_date_orig = tf.monit_date and
        trade_date.ref_date >= cast(tf.tf_created_date as date) AND
        tf.tm_mode_grp = 3
      where
        trade_date.tm_mode_grp = 3
    ) a -- This subquery will gather info for the ref_dates that are after the first created date in production, picking the most recent record to each trade and periodicity. This will also have the trades to which we'll use the remain_paym and remain_inst infor for MOP calculations.
    where
    a.order_id = 1
    group by
    a.tm_trade_id,
	  a.monit_date,
	  a.monit_date_orig,
    a.ref_date_id
  ) AS post
  left join
  bu_brbureau.{eid}{prj}_algos_gcdf_paymt_pre pre on
  post.tm_trade_id = pre.tm_trade_id and
  post.monit_date = pre.monit_date and
  post.ref_date_id = pre.ref_date_id"
            )) %>%
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_remain_paymt_inst"), mode = "overwrite")
            
            
            ### INST MOP (final) ----
            sdf_sql(cn, str_glue("
select
    b.tm_trade_id,
	  b.monit_date,
	  b.monit_date_orig,
    tm.tm_mode,
    b.ref_date,
    b.ref_date_id,
    CASE WHEN b.flag_history = 0 OR
              (tm.tm_mode = 'C01' AND (tm.tm_quota_status <> 2 OR tm.tm_quota_payout_date >= LEAST(b.reported_dt, b.ref_date)))
         THEN 0
         WHEN b.paymt_mop = 1
         THEN 1
         WHEN b.paymt_mop > 1 AND ((b.remain_paymt = 1 AND b.remain_inst = 0 AND tm.tm_number_installments > 1) OR (b.remain_paymt = 0))
         THEN 1
         WHEN b.paymt_mop > 1
         THEN CAST(LEAST(b.paymt_mop + 
                GREATEST(MONTHS_BETWEEN(LEAST(b.reported_dt, b.ref_date), COALESCE(tm.tm_last_installment_duedate, CAST('2199-12-31' AS DATE))), 0),
                6) AS INT)
         ELSE -9
         END AS mop_v0,
         CAST(NULL AS INT) AS dpd_v0,
         CASE WHEN b.flag_history = 0 OR
              (tm.tm_mode = 'C01' AND (tm.tm_quota_status <> 2 OR tm.tm_quota_payout_date >= LEAST(b.reported_dt, b.ref_date)))
         THEN 0
         WHEN b.paymt_mop = 1
         THEN 0
         WHEN b.paymt_mop > 1 AND ((b.remain_paymt = 1 AND b.remain_inst = 0 AND tm.tm_number_installments > 1) OR (b.remain_paymt = 0))
         THEN 0
         WHEN b.paymt_mop > 1
         THEN b.paymt_diff_v0
         ELSE 0
         END AS past_due_amt_v0,
    b.paymt_amt_act_v0,
    b.most_recent_th_value,
    b.most_recent_due_date,
    b.remain_paymt,
    b.remain_inst,
    b.remain_paymt_bal
from (
    select
        tm_trade_id,
		    monit_date,
		    monit_date_orig,
        ref_date_id,
        FIRST(ref_date) AS ref_date,
        FIRST(reported_dt) AS reported_dt,
        MAX(CASE WHEN a.th_trade_id is not null then 1 else 0 end) as flag_history,
        sum(greatest(a.th_value - a.value_paid, 0)) as paymt_diff_v0,
        sum(case when a.due_date_order_id = 1 then value_paid else 0 end) as paymt_amt_act_v0,
        sum(case when a.due_date_order_id = 1 then th_value else 0 end) as most_recent_th_value,
        max(case when a.due_date_order_id = 1 then th_due_date else cast('1900-01-01' as date) end) as most_recent_due_date,
        FIRST(remain_paymt) AS remain_paymt,
        FIRST(remain_inst) AS remain_inst,
        FIRST(remain_paymt_bal) AS remain_paymt_bal,
        FIRST(paymt_mop) AS paymt_mop
    from (
        select
            tr_h_in.tm_trade_id,
			      tr_h_in.monit_date,
			      tr_h_in.monit_date_orig,
            tr_h_in.ref_date_id,
            tr_h_in.th_trade_id,
            FIRST(tr_h_in.ref_date) AS ref_date,
            FIRST(tr_h_in.reported_dt) AS reported_dt,
            FIRST(tr_h_in.due_date_order_id) AS due_date_order_id,
            FIRST(tr_h_in.th_due_date) AS th_due_date,
            FIRST(tr_h_in.th_value) AS th_value,
            SUM(coalesce(pay2.thp_payment_value, 0)) as value_paid,
            FIRST(remain_paymt) AS remain_paymt,
            FIRST(remain_inst) AS remain_inst,
            FIRST(remain_paymt_bal) AS remain_paymt_bal,
            FIRST(paymt_mop) AS paymt_mop
        from (
            select
                fut.tm_trade_id,
				        fut.monit_date,
				        fut.monit_date_orig,
                fut.ref_date_id, 
                th.th_trade_id,
                fut.ref_date, 
                fut.reported_dt, 
                th.th_due_date,
                th.th_value,
                row_number() over(partition by fut.tm_trade_id, fut.monit_date, fut.monit_date_orig, fut.ref_date_id 
                               order by coalesce(th.th_due_date, cast('1900-01-01' as date)) desc, th.th_trade_id) as due_date_order_id,
                fut.remain_paymt,
                fut.remain_inst,
                fut.remain_paymt_bal,
                fut.paymt_mop
            FROM bu_brbureau.{eid}{prj}_algos_gcdf_remain_paymt_inst AS fut
            left join (
                select * 
                from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history
                WHERE tm_mode_grp = 3 and th_use = 1
                ) th
            on
                fut.tm_trade_id = th.tm_trade_id and
				        fut.monit_date_orig = th.monit_date and
                fut.ref_date >= th.th_due_date and
                fut.reported_dt > th.th_due_date
            ) tr_h_in
        left join bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment pay2 
        on
            tr_h_in.tm_trade_id = pay2.tm_trade_id AND
			      tr_h_in.monit_date_orig = pay2.monit_date AND
            tr_h_in.th_trade_id = pay2.th_trade_id and
            tr_h_in.ref_date   >= pay2.thp_payment_date AND
            pay2.tm_mode_grp = 3 AND
            pay2.thp_use = 1
        group by
            tr_h_in.tm_trade_id,
			      tr_h_in.monit_date,
			      monit_date_orig,
            tr_h_in.ref_date_id,
            tr_h_in.th_trade_id
        ) a
    group by
        tm_trade_id,
		    monit_date,
		    monit_date_orig,
        ref_date_id
    ) b
left join bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master AS tm
on
    b.tm_trade_id = tm.tm_trade_id AND
	  b.monit_date_orig  = tm.monit_date_orig AND
		tm.tm_mode_grp = 3")) %>%
              group_by(tm_trade_id, monit_date, monit_date_orig) %>%
              # Note: Above this point you could use for Trade History
              # Note: this whole summarize can be replaced by the array collection/row filter
              summarize(
                tm_mode_grp = 3,
                mop_v0_1 = as.integer(min(case_when(ref_date_id == 1 ~ mop_v0, TRUE ~ 998))),
                dpd_v0_1 = as.integer(max(case_when(ref_date_id == 1 ~ dpd_v0, TRUE ~ -999))),
                past_due_amt_v0_1 = max(case_when(ref_date_id == 1 ~ past_due_amt_v0, TRUE ~ 0)),
                # paymt_amt_act_v0_1 = max(case_when(ref_date_id == 1 ~ value_paid, TRUE ~ 0)),
                max_hist_bal_v0_1 = as.numeric(NaN),
                billed_balance_v0_1 = max(case_when(ref_date_id == 1 ~ most_recent_th_value, TRUE ~ -999)),
                most_recent_due_date = max(case_when(ref_date_id == 1 ~ most_recent_due_date, TRUE ~ as.Date('1900-01-01'))),
                remain_paymt = max(case_when(ref_date_id == 1 ~ remain_paymt)),
                remain_inst = max(case_when(ref_date_id == 1 ~ remain_inst)),
                remain_paymt_bal = max(case_when(ref_date_id == 1 ~ remain_paymt_bal)),
                .groups = "drop"
              ) %>% 
              spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_mop_v0_inst"), mode = "overwrite")
            
            ### Future Info ----
            sdf_sql(cn, str_glue(
              "select
               trade_date.tm_trade_id,
		           trade_date.monit_date,
		           trade_date.monit_date_orig,
		           trade_date.tm_mode_grp,
       case when e.tm_trade_id is not null and year(e.tf_next_instal_due_date) = year(ADD_MONTHS(least(trade_date.monit_date, trade_date.reported_dt), 1)) and month(e.tf_next_instal_due_date) = month(ADD_MONTHS(least(trade_date.monit_date, trade_date.reported_dt), 1)) --- after picking the record with the most recent created_date, we want to actually use the oldest between reported_dt and monit_date, because, if monit date is in may and the trade was last reported in march, then we want to see which payments were swcheduled to april..
     then 
          case when trim(trade_date.tm_mode) = 'D01' then e.tf_min_pay_value else e.tf_next_instal_value end
     else
          case when trim(trade_date.tm_mode) = 'D01'
               then case when coalesce(f.paid_value, 0) >= f.th_min_pay_value then null else f.th_min_pay_value end
               else case when coalesce(f.paid_value, 0) >= (f.th_value * 0.1) then null else f.th_value end
               end
     end as paymt_amt_sch
from
(
  select 
    tm_trade_id,
	monit_date,
	monit_date_orig,
	tm_calculation_date as reported_dt,
	tm_mode,
    tm_mode_grp
    --, LAST_DAY(ADD_MONTHS(LEAST(monit_date, tm_calculation_date), 1)) AS match_dt
  from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master
  WHERE tm_mode_grp IN (1, 3)
) trade_date
left join
(
	  select
	  tf2.*
	  from
	  (
    select
    tf.tm_trade_id,
	trade_date.monit_date,
    tf.tf_next_instal_due_date,
    tf.tf_next_instal_value,
    tf.tf_min_pay_value,
    row_number() over(partition by tf.tm_trade_id, tf.monit_date, tf.tm_mode_grp order by tf.tf_created_date desc, tf.tf_next_instal_due_date) as order_id -- amongst the most recent records, we want to see which has the first next due_date
    from
    bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_future tf
	inner join 
	bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master trade_date on
	tf.tm_trade_id = trade_date.tm_trade_id AND
	tf.monit_date = trade_date.monit_date_orig
    where
    cast(tf.tf_created_date as date) <= trade_date.monit_date AND
    tf.tm_mode_grp IN (1, 3)
	  ) tf2
  where
  order_id = 1
) e on 
trade_date.tm_trade_id = e.tm_trade_id AND
trade_date.monit_date = e.monit_date 
--AND LAST_DAY(e.tf_next_instal_due_date) = trade_date.match_dt

left join
(
  select
  f0.tm_trade_id,
  f0.monit_date,
  f0.monit_date_orig,
  f0.th_value,
  f0.th_min_pay_value,
  sum(thp.thp_payment_value) as paid_value
  from
  (
  select
      th.th_trade_id,
      th.tm_trade_id,
	  trade_date.monit_date,
	  trade_date.monit_date_orig,
      trade_date.tm_calculation_date,
      th.th_value,
      th.th_min_pay_value,
      row_number() over(partition by th.tm_trade_id, th.monit_date order by th.th_due_date desc, th.th_value desc) as order_id
  from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history th
  inner join
      (select 
        tm_trade_id,
		    monit_date,
		    monit_date_orig,
        tm_calculation_date,
		    tm_mode_grp,
        LAST_DAY(ADD_MONTHS(LEAST(monit_date, tm_calculation_date), 1)) AS match_dt 
      from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master
      WHERE tm_mode_grp IN (1, 3)
  ) trade_date on
        th.tm_trade_id = trade_date.tm_trade_id AND
		    th.monit_date = trade_date.monit_date_orig AND
        LAST_DAY(th.th_due_date) = match_dt AND
        th.tm_mode_grp IN (1, 3) AND
		    trade_date.tm_mode_grp IN (1, 3)
  ) f0
  LEFT JOIN
  (select thp.th_trade_id, thp.monit_date, thp.thp_payment_date, thp.thp_payment_value
   from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment  thp
   	inner join 
	(select trade_date.tm_trade_id, th.th_trade_id, trade_date.monit_date, trade_date.monit_date_orig
	from
	bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history th
	inner join
	bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master trade_date on
	th.tm_trade_id = trade_date.tm_trade_id AND
	th.monit_date = trade_date.monit_date_orig
	) thm on
	thm.th_trade_id = thp.th_trade_id AND
	thm.monit_date_orig = thp.monit_date
   where thp.thp_payment_date <= thm.monit_date AND
   thp.tm_mode_grp IN (1, 3)
  ) thp on
  f0.th_trade_id = thp.th_trade_id AND
  f0.monit_date_orig = thp.monit_date AND
  thp.thp_payment_date <= f0.tm_calculation_date -- this condition in the case when inside the summation is just to make sure we will not have payments hapenning after the reported_dt, which should not happen by definition, but we want to prevent due to data quality issues we see
  WHERE f0.order_id = 1
  group by
  f0.tm_trade_id,
  f0.monit_date,
  f0.monit_date_orig,
  f0.th_value,
  f0.th_min_pay_value
) f on
trade_date.tm_trade_id = f.tm_trade_id and
trade_date.monit_date = f.monit_date"
            )) %>%
    spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_future_info"), mode = "overwrite")  
  
  cat(paste("L002 Finish through MOPs at", Sys.time(), "\n"), file = ln, append = TRUE)
  
  ### Trade Retro Tmp for Credit Cards ----
  # adding a table here to retrieve the appropriate record for the balance and high_credit filters for credit cards
  # This appropriate record is the one from the data_date in the trade_master where we have the most recent reported_dt that is before or at the monit_dt
  
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_trade_retro_sample")) %>% 
    filter(tm_mode_grp == 1) %>% 
    select(-monit_date) %>% 
    # Left join to collect the original monit_date's and current observed monit_date for each trades which are available in the referenced month
    # This should duplicate the table since we may have the same trade/subject appearing multiple times in the run_trade_master
    left_join(tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_master")) %>% 
                select(tm_trade_id, monit_date)
              , by = "tm_trade_id") %>% 
    # Now we'll group by the main fields and apply the filters to collect 
    # the actual correspondent fields has they were in previous loads, 
    # using the monit_date observed and not the original
    group_by(sj_subject_id, tm_trade_id, monit_date) %>% 
    filter(
      tm_calculation_date <= monit_date,
      # these 2 conditions below is for us to get balance and high_credit missing if the trade had not been reported in the same month as the monit_dt
      year(tm_calculation_date) == year(monit_date),
      month(tm_calculation_date) == month(monit_date)
    ) %>% 
    arrange(desc(tm_calculation_date),desc(tm_modified_date)) %>%
    filter(row_number() == 1) %>% 
    ungroup() %>% 
    select(sj_subject_id, tm_trade_id, monit_date, tm_calculation_date, tm_balance_owed, tm_max_balance_owed, tm_mode_grp) %>% 
    mutate(tm_balance_owed = pmax(tm_balance_owed, 0),
           tm_max_balance_owed = pmax(tm_max_balance_owed, 0)) %>% 
    spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_retro_cc_tmp"), mode = "overwrite")
  
  ### High Credit TelCo & Utilities ----
  # TO DO: Collect this info in the mop queries like we are doing for Revolving
  # We are running this table here to compute the right maximum historical value of the invoice, for telco trades, to each monit_dt
  # We also have invoices which are due at the same day, to avoid not capturing that information we are first summing up the values per due date
  tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_gcdf_run_trade_history")) %>% 
    filter(monit_date <= th_due_date, tm_mode_grp %in% c(4, 5)) %>% 
    group_by(tm_mode_grp, tm_trade_id, monit_date, th_due_date) %>%
    mutate(total_balance_owed = sum(th_value)) %>% 
    group_by(tm_mode_grp, tm_trade_id, monit_date) %>% 
    summarise(tm_max_balance_owed = max(coalesce(total_balance_owed,0))) %>% 
    ungroup() %>% 
    spark_write_table(paste0("bu_brbureau.", eid, prj, "_algos_high_credit_telco_utilities"), mode = "overwrite")
  
  ### Output table ----
  post_gcdf_trhi = sdf_sql(cn, str_glue(
    "SELECT
CONCAT(CASE WHEN a.tm_mode_grp = 4 THEN 'TEL' ELSE 'FS' END,
a.tm_trade_id, '0', a.member_code, '0', a.sj_subject_id) as hashkey
, a.sj_subject_id AS consno_orig  
, sj.consno -- we had to add this concatenation so to be able to compute the attributes to the same subject at different monitoring dates. Galaxy functions don't allow for this. 
, CONCAT(CASE WHEN a.tm_mode_grp = 4 THEN 'TEL' WHEN a.tm_mode_grp = 5 THEN 'EL' ELSE 'FS' END, a.tm_trade_id) AS accid -- TU internal trade identifier
, a.member_code AS member_code
, cast(NULL AS STRING) AS member_name 
, a.tm_unique_no AS acc_no 
, a.tm_contract_date AS opened_dt
, ADD_MONTHS(LAST_DAY(a.tm_contract_date),-1) AS opened_dt_key
, a.tm_calculation_date AS reported_dt 
, cast(NULL AS DATE) AS last_act_dt 
, a.industry
, cast(NULL AS INT) AS fintech 
, case when trim(a.tm_mode) = 'A01' then 12
       when trim(a.tm_mode) = 'A02' then 50
       when trim(a.tm_mode) = 'A04' then 9
       when trim(a.tm_mode) = 'A05' then 13
       when trim(a.tm_mode) = 'A99' then 19
       when trim(a.tm_mode) = 'B01' then 23
       when trim(a.tm_mode) = 'B02' then 83
       when trim(a.tm_mode) = 'B03' then 6
       when trim(a.tm_mode) = 'B04' then 13
       when trim(a.tm_mode) = 'B05' then 3
       when trim(a.tm_mode) = 'B06' then 84
       when trim(a.tm_mode) = 'B07' then 62
       when trim(a.tm_mode) = 'B99' then 90
       when trim(a.tm_mode) = 'F01' then 89
       when trim(a.tm_mode) = 'G01' then 87
       when trim(a.tm_mode) = 'C01' then 85
       when trim(a.tm_mode) = 'D01' then 4
       when trim(a.tm_mode) = 'E01' then 36
       when trim(a.tm_mode) = 'E02' then 88
	     when trim(a.tm_mode) = 'T01' then 21
	     when trim(a.tm_mode) = 'EL01' then 101
       else 99 end as lob 
, case when b.tm_trade_id is null then null 
       when b.term is null then null
       else greatest(b.term, 1) end AS term 
, 'BRL' AS currency
, cast(1 AS decimal(10,3)) AS usd_rate 
, cast(1 AS decimal(10,3)) AS local_rate 

, case when trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01') then cast(a.tm_future_contracted_value as double) --- for installments and consortium, we can just pick this field.
	   else null end as credit_limit 

, case when trim(a.tm_mode) IN ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01') then cast(a.tm_future_contracted_value as double) --- for installments and consortium, we can just pick this field.
   	   when trim(a.tm_mode) IN ('D01') then tr.tm_max_balance_owed  -- we are retrieving the high_credit from the record we pulled from the trade_retro (tr). There we have already made the high_credit not negative (greatest(max_balance, 0))
       when trim(a.tm_mode) IN ('T01', 'EL01') then cast(greatest(hctu.tm_max_balance_owed, i.past_due_amt_v0_1) as double) -- for telco trades, we need to retrieve the maximum historical value of the invoice, for invoices with due_date before or at the monit_dt. For TelCo and Energy we have situations where Utilization went onver 100% because we ended up setting balance as past due amt, which could be way higher than the maximum invoice value because of the way we calculate delinquency
   	   when trim(a.tm_mode) IN ('E01', 'E02') then coalesce(i.max_hist_bal_v0_1,0)
	   else null end as high_credit 

, g.remain_paymt 

, g.remain_inst

, case when trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01')
     then 
          case when g.remain_paymt_bal is null then null
               when (g.remain_paymt_bal / greatest(a.tm_number_installments, b.term)) is null then null
               else cast(least(( g.remain_paymt_bal / greatest(a.tm_number_installments, b.term)) * a.tm_future_contracted_value, a.tm_future_contracted_value) as double)
               end --- for installments, we will pick the % of payments still remaining, and apply it to the contracted value, as a proxy.
	   when trim(a.tm_mode) = 'D01' then tr.tm_balance_owed -- we are retrieving the balance from the record we pulled from the trade_retro (tr). There we have already made the balance not negative (greatest(balance, 0)).
     when trim(a.tm_mode) in ('E01', 'E02', 'T01', 'EL01') then coalesce(i.past_due_amt_v0_1,0) 
	   else null end AS balance

, case when trim(a.tm_mode) in ('E01', 'E02') then 0
       when i.mop_v0_1 > 1 then greatest(0, i.past_due_amt_v0_1)
       else 0 end AS past_due_amt 

, case when trim(a.tm_mode) in ('B03', 'B05', 'B06', 'B07', 'B99', 'C01') then 1 else 0 end AS secured
, cast(NULL AS INT) AS active_temp

, case when trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01')
            and g.remain_paymt = 0
	   then 1
	   else 0 end as closed_ea

, case when trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01')
            and g.remain_paymt = 0
       then a.tm_calculation_date 
  	   else cast(null as date) end as closed_ea_dt

, cast(NULL AS INT) AS choff_temp
, cast(null as date) as choff_dt_temp
, cast(null as double) as choff_amt

, cast(NULL AS INT) AS bk_trade
, cast(NULL AS date) AS bk_trade_filing_dt
, cast(NULL AS INT) AS bk_pr
, cast(NULL AS date) AS bk_pr_filing_dt
, cast(NULL AS INT) AS bk_pr_discharge_dt
, cast(NULL AS INT) AS primary_flag_temp
, cast(NULL AS INT) AS suppress

, case when trim(a.tm_mode) = 'D01' or trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B03', 'B04', 'B05', 'B06', 'B07', 'B99', 'F01', 'G01', 'C01')
       then f.paymt_amt_sch
       else null end as paymt_amt_sch 

, coalesce(paymt_mo.paymt_amt_act_assign, 0) AS paymt_amt_act 
, last_pmt.paymt_value as last_paymt_amt
, last_pmt.paymt_dt as last_paymt_dt 
, coalesce(i.mop_v0_1,998) AS mop 

, cast(NULL AS date) AS paymt_pat_last_update
, cast('X' as string) as gcdf_pattern
, cast(NULL AS string) AS state
, cast(NULL AS bigint) AS zip
, cast(NULL AS bigint) AS zip4
, cast(NULL AS date) AS birth_dt
, cast(NULL AS INT) AS gender
, cast(NULL AS INT) AS deceased
, CASE WHEN a.tm_mode in ('T01', 'EL01') AND a.tm_restriction_type = 'F' THEN 1 ELSE NULL END AS fraud_ind
, sj.sj_sub_type AS own_ind
, cast(1 AS INT) AS own_struct 
, case when disp.disp_condition_1 = 1
		 or disp.disp_condition_2 = 1
	   then 1 else 0 end as disputed
, cast(NULL AS INT) AS deferral
, cast(NULL AS INT) AS repo_ind
, cast(NULL AS INT) AS foreclosed
, cast(NULL AS INT) AS lost_card
, cast(NULL AS INT) AS restructured
, case when trim(a.tm_mode) in ('A01', 'A02', 'A04', 'A05', 'A99', 'B01', 'B02', 'B04', 'B05', 'B07', 'B99', 'F01', 'G01', 'C01') -- we have removed the tm_modes regarding real state trades (B03 and B06)
            and (a.tm_number_installments > 1 or a.tm_number_installments is null)
       then 1 else 0 end as installment
, case when trim(a.tm_mode) in ('D01', 'E01', 'E02') then 1 else 0 end as revolving
, case when i.billed_balance_v0_1 = -999 then null else i.billed_balance_v0_1 end as gcdf_rc1
, coalesce(i.most_recent_due_date, cast('1900-01-01' as date)) as gcdf_rc1_frsrpt_dt
, 'BRA' AS country
, a.monit_date as data_date 
, {ref_id} as data_date_ref --  NNOTE ! We are adding the ref_id from the dates to have the reference of which is the first observation, we need that to create a few algos
, coalesce(i.dpd_v0_1,-999) as dpd 

FROM
bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master a
inner join
bu_brbureau.{eid}{prj}_algos_subject_sample sj on
a.sj_subject_id = sj.sj_subject_id and
a.monit_date_orig = sj.monit_date

left join
bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_retro_cc_tmp tr on
a.tm_trade_id     = tr.tm_trade_id and
a.sj_subject_id   = tr.sj_subject_id and
a.monit_date      = tr.monit_date and
a.tm_mode_grp     = tr.tm_mode_grp

left join bu_brbureau.{eid}{prj}_algos_dispute_details_sample disp on
a.monit_date_orig = disp.monit_date and
a.member_code =   disp.member_code and
a.sj_subject_id = disp.sj_subject_id and
a.tm_unique_no =  disp.disp_opr_unique_num

left join
bu_brbureau.{eid}{prj}_algos_high_credit_telco_utilities hctu on
a.tm_trade_id   = hctu.tm_trade_id and
a.monit_date    = hctu.monit_date and
a.tm_mode_grp   = hctu.tm_mode_grp

left join
(
	select
	a2.*,
	case when a2.tm_contract_date < '2021-04-01' or a2.min_due_date = '2199-12-31' 
		then cast(months_between(last_day(a2.tm_last_installment_duedate), last_day(a2.tm_contract_date)) as int)
		else cast(months_between(last_day(a2.tm_last_installment_duedate), last_day(a2.min_due_date)) as int) + 1 
		end as term
	from
	(
		SELECT
		a1.monit_date,
		a1.monit_date_orig,
		a1.tm_trade_id,
		a1.tm_mode_grp,
		-- a1.tm_number_installments,
		a1.tm_contract_date,
		a1.tm_last_installment_duedate,
		min(coalesce(b1.th_due_date, cast('2199-12-31' as date))) as min_due_date
		from
		bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_master a1
		left join
		bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history b1 on
		a1.monit_date_orig = b1.monit_date and
		a1.tm_trade_id = b1.tm_trade_id AND
		b1.tm_mode_grp = 3
		where
		a1.tm_mode_grp = 3
		group by
		a1.monit_date,
		a1.monit_date_orig,
		a1.tm_trade_id,
		a1.tm_mode_grp,
		-- a1.tm_number_installments,
		a1.tm_contract_date,
		a1.tm_last_installment_duedate
	) as a2
) b on 
a.monit_date_orig = b.monit_date_orig and
a.monit_date = b.monit_date and
a.tm_trade_id = b.tm_trade_id and
a.tm_mode_grp = b.tm_mode_grp

left join
(
select
*
from
  (
  select
  	    *,
  	    row_number() over(partition by lp01.tm_trade_id, lp01.monit_date, lp01.monit_date_orig order by lp01.paymt_dt desc, lp01.paymt_value desc) nd_id
  	from
  	(
  		select
  		    thp.tm_trade_id,
  			  td.monit_date,
  			  td.monit_date_orig,
  			  thp.thp_payment_date as paymt_dt,
  		    thp.thp_payment_value as paymt_value
  		from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_history_payment thp
  		inner join bu_brbureau.{eid}{prj}_algos_gcdf_trade_date td on
  		thp.tm_trade_id = td.tm_trade_id and
  		thp.monit_date = td.monit_date_orig
  		where thp.thp_payment_date <= td.monit_date 
  
  		union all
  		
  		select
  		    tp.tm_trade_id,
  			  td.monit_date,
  			  td.monit_date_orig,
  		    tp.paymt_dt,
  		    tp.paymt_value
  		from bu_brbureau.{eid}{prj}_algos_gcdf_run_trade_payment tp
  		inner join bu_brbureau.{eid}{prj}_algos_gcdf_trade_date td on
  		tp.tm_trade_id = td.tm_trade_id and
  		tp.monit_date = td.monit_date_orig
  		where	tp.paymt_dt <= td.monit_date
  	) lp01
  )
	where
	nd_id = 1
) last_pmt on
a.tm_trade_id = last_pmt.tm_trade_id and
a.monit_date = last_pmt.monit_date and
a.monit_date_orig = last_pmt.monit_date_orig

left join
bu_brbureau.{eid}{prj}_algos_gcdf_future_info f on 
a.monit_date = f.monit_date and
a.monit_date_orig = f.monit_date_orig and
a.tm_trade_id = f.tm_trade_id AND
a.tm_mode_grp = f.tm_mode_grp

left join
(select * from bu_brbureau.{eid}{prj}_algos_gcdf_remain_paymt_inst where ref_date_id = 1) g on 
a.monit_date = g.monit_date and
a.monit_date_orig = g.monit_date_orig and
a.tm_trade_id = g.tm_trade_id

left join
(
    
	(SELECT * FROM bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_cc)   
	  
	UNION ALL
	(SELECT * FROM bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_rev)
    
	UNION ALL
  (SELECT * FROM bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_inst)
	
	UNION ALL
	(SELECT * FROM bu_brbureau.{eid}{prj}_algos_gcdf_mop_v0_telco_utilities)

) i on
a.tm_trade_id     = i.tm_trade_id and
a.monit_date      = i.monit_date and
a.monit_date_orig = i.monit_date_orig AND
a.tm_mode_grp     = i.tm_mode_grp

LEFT JOIN bu_brbureau.{eid}{prj}_algos_paymt_month_assignment AS paymt_mo ON
a.tm_trade_id     = paymt_mo.tm_trade_id and
a.monit_date_orig = paymt_mo.monit_date and
a.data_join       = paymt_mo.assigned_paymt_month AND
a.tm_mode_grp     = paymt_mo.tm_mode_grp"
  )) %>% relocate(data_date, .after = last_col()) %>% compute()
  
  table_exists <- tryCatch(
    {tbl(cn, paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql")); TRUE},
    error = function(sc){return(FALSE)})
  if (table_exists) {
    post_gcdf_trhi %>%
      spark_write_table(paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql"), mode = "append")
  } else {
    post_gcdf_trhi %>%
      spark_write_table(paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql"), mode = "overwrite", partition_by = "data_date")
  }
  
  cat(paste("L003 Finished final gcdf trade history table (pre filters) at", Sys.time(), "\n"), file = ln, append = TRUE)
  
        }
        
        cat(paste("\n", "Finished all FOR loop at", Sys.time(), "\n"), file = ln, append = TRUE)
        
        # sdf_sql(cn, paste0("drop table bu_brbureau.", eid, prj, "_gcdf_trhi_sql")) %>% sdf_nrow()
        
        # # Review ----
        # tbl(cn, paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql"))
        #   group_by(data_date_ref) %>%
        #   summarize(
        #     rows = n(),
        #     trades = n_distinct(accid),
        #     subjects = n_distinct(consno),
        #     .groups = "drop") %>%
        #   arrange(desc(data_date_ref)) %>%
        #   collect() %>% as.data.frame()
        # 
          
          tbl(cn, paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql")) %>% sdf_distinct(data_date)
        
        # ## Cleanup ----
        # 
        # # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_subject_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_master_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_history_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_history_payment_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_payment_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_future_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_dispute_details_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_subject_optlog_sample'))
        # 
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trade_retro_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_master_pre'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_history'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_history_payment'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_future'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_payment'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_paymt_month_assignment'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_energy_trade_hist_tmp'))
        # 
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_master'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_master_tel'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_master_fin'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_master_energy'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_trade_date'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_cc'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_telco_utilities'))
        # 
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_rev1'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_inst'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_rev'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_bal_rev1_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_bal_rev2_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_paymt_post_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_paymt_pre'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_retro_remain_paymt_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_retro_remain_inst_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_paymt_pre_2_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_remain_paymt_inst'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_future_info'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_inst_prev_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_mop_v0_inst_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_gcdf_run_trade_retro_cc_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_telco_trade_hist_tmp'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_high_credit_telco_utilities'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_gcdf_trhi_sql_pre'))
        # 
        ##--##--##--##--##--##--##
        ## Preparing GCDF filters for algorithms computations ----
        
        cat(paste("L004 Starting algorithms at", Sys.time(), "\n"), file = ln, append = TRUE)
        
        
        # Script for calculating attributes gcdf to be used in algos script (calculate_gcdf_attrs function)
        # TEMP - Using custom version for mismatch
        source("/shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/retro/nubank_100/gcdf_attribs_for_algos.R")
        
        # Calculate GCDF for attributes on `data_date`
        
        cat(paste("L004.5 Calculating attributes GCDF at", Sys.time(), "\n"), file = ln, append = TRUE)
        
        gcdf_attributes_table_name <- calculate_gcdf_attrs(cn, prj2, input_data)
        cat(paste("L004.6 Output attributes GCDF table: ", gcdf_attributes_table_name, "\n"), file = ln, append = TRUE)
        
        if(output_comb == 'Y'){
          
          # # check mismatch
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_gcdf_trhi_sql')) %>% filter(grepl("812080", consno)) %>% 
          #   # select(all_trade,consno,accid,run_date_x, run_date_y,msrd,mop,mis,req,min_pay,closed,closed_dt,
          #   #        secured,opened_dt,msop,lob,balance,high_credit,
          #   #        past_due_amt,paymt_amt_act,credit_limit,disputed,
          #   #        all_trade,gcdf_rc1,gcdf_rc2,gcdf_rc3,gcdf_rc4) %>%
          #   collect() %>% View('check_gcdf_trhi_sql')
          # 
          
          ### Output COMB ----    
          #### Modifying GCDF ----
          
          # NOTE: This should ultimately be handled in GCDF step
          tbl(cn, paste0("bu_brbureau.", eid, prj, "_gcdf_trhi_sql")) %>%
            # mutate(suppressFI_TM = paste0(member_code,"_",lob)) %>% 
            # filter(!suppressFI_TM %in% suppressItens) %>%  
            # select(-suppressFI_TM) %>%
            filter(disputed == 0) %>%
            # first_date_out: this date will be used to make sure we will not exclude the subjects that only have trades before the last observed period,
            #                 we are using the minimum data_date per consno to avoid fixing a date which wouldn't cover the real observed scenario 
            #                 for each case
            # first_date_in: this date will be used to define the run_date for the filter calculation functions
            group_by(consno) %>% 
            mutate(first_date_out = as.Date(add_months(last_day(min(data_date)), -1))
                   # , first_date_in = as.Date(max(data_date))
                   , run_date = as.Date(max(data_date))) %>% 
            ungroup() %>% 
            filter(
              last_day(reported_dt) >= last_day(data_date) | reported_dt <= first_date_out
            ) %>% 
            ## we added this condition in 2023-06-16, to fix the curr flag thing for the kind of current balance, 
            ## or current apd, etc, filters. We think that this is going to do the job for the valid_mth 
            ## thing and we will not require this "compute_flag" being defined a few lines below, 
            ## neither the gcdf_rc3 column that we are using for this.
            #  we'll use this compute_flag vector to define the "valid_mth vector later on
            # 2024-03-08: we are commenting compute_flag and setting gcdf_rc3 to character (which is is the pattern for other countries)
            mutate(
              # compute_flag = (last_day(reported_dt) >= data_date),
              orig_reported_dt = reported_dt
            ) %>% 
            mutate(
              reported_dt = case_when(orig_reported_dt <= first_date_out ~ as.Date("1900-01-01"),
                                      TRUE ~ as.Date(data_date)) 
            ) %>%  
            mutate(
              # DD NOTE: Most of these are fixing type issues created in the original query
              secured = if_else(is.na(secured), FALSE, as.logical(secured)), 
              active = TRUE, 
              closed = if_else(is.na(closed_ea), FALSE, as.logical(closed_ea)), 
              choff = if_else(is.na(choff_temp), FALSE, as.logical(choff_temp)), 
              suppress = if_else(is.na(suppress), FALSE, as.logical(suppress)),
              disputed = if_else(is.na(disputed), FALSE, as.logical(disputed)),
              deferral = if_else(is.na(deferral), FALSE, as.logical(deferral)),
              installment = if_else(is.na(installment), NA, as.logical(installment)),
              revolving = if_else(is.na(revolving), NA, as.logical(revolving)),
              own_struct = 1,
              paymt_freq = "M" ,
              closed_dt = closed_ea_dt,
              suppress = FALSE,
              all_trade = TRUE,
              msat = as.integer(case_when(
                all_trade ~ as.integer(case_when(
                  !is.na(reported_dt) ~ as.integer(months_between(data_date, reported_dt)), 
                  T ~ as.integer(999))), 
                T ~ as.integer(999))),
              gcdf_rc2 = high_credit, # we are using this to filter high_credit by original value, not round.
              # gcdf_rc3 = compute_flag,
              gcdf_rc3 = "",
              gcdf_rc4 = "",
              gcdf_rc2_frsrpt_dt = as.Date("1900-01-01"),
              gcdf_rc3_frsrpt_dt = as.Date(orig_reported_dt),
              gcdf_rc4_frsrpt_dt = as.Date("1900-01-01"),
              paymt_pat_start_dt = as.Date("2020-03-05"),
              bk_trade = FALSE,
              bk_dt = as.Date("1900-01-01"),
              curr_T = FALSE # NN: Adding this to avoid errors, but it's only properly used by PHP
            ) %>%
            spark_write_table(paste0("bu_brbureau.", eid, prj, "_brz_trhi_tmp"), mode = "overwrite")
          cat(paste("  L005 Wrote brz_trhi at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          #### Trade History filters ----
          tbl(cn, paste0("bu_brbureau.", eid, prj, "_brz_trhi_tmp")) %>% 
            calc_final_filters_bra() %>% # changed the name, but that's basically calc_tvua_filters_bra/calc_tvpa_filters_bra
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_trhi_filt'), mode = "overwrite")
          cat(paste("  L006 Wrote _trhi_filt at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_brz_trhi_tmp'))
          
          # temporary mismatch: save filters for BRASA team. OBS: getting data for only 1k subjects because there are million rows in _trhi_filt 
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% sdf_nrow()
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% colnames()
          # 
          # adviser_tbl <- spark_read_csv(cn, "file:///shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/retro/nubank_100/adv_files/23-algos-10K-20240826-AdviserOutput-File.csv", delimiter=';') %>% 
          #               select(sj_subject_id = SubjectId)
          # adviser_tbl %>% sdf_nrow()
          # 
          # adviser_tbl %>% 
          #   sdf_sample(fraction = 0.1, replacement = FALSE, seed = 202408) %>% 
          #   left_join(tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% 
          #               separate(consno, c("consno"), sep="[||]") %>% 
          #               mutate(sj_subject_id = as.numeric(consno)),
          #               by = 'sj_subject_id') %>% 
          #   sdf_repartition(2) %>% 
          #   spark_write_csv(paste0('file:///shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/retro/nubank_100/filters/filters_23_algos_nubank_', format(Sys.time(), '%Y%m%d')), quote="", mode='overwrite')
          # 
          #   
          # # check mismatch
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% filter(grepl("7652570", consno)) %>% 
          #   select(all_trade,consno,accid,run_date_x, run_date_y,msrd,mop,mis,req,min_pay,closed,closed_dt,
          #          secured,opened_dt,msop,lob,balance,high_credit,
          #          past_due_amt,paymt_amt_act,credit_limit,disputed,
          #          all_trade,gcdf_rc1,gcdf_rc2,gcdf_rc3,gcdf_rc4) %>%
          #   collect() %>% View('check_trhi_filt')
          # 
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% colnames()
          # 
          #### Calculate TVPA sets ---- 
          cat(paste("  L007 Calculating TVPA ---------", "\n"), file = ln, append = TRUE)
            
          # OBS: ONLY CALCUTING SETS FOR 23 NUBANK ALGORITHMS
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            tvpa_grp7() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp07'), mode = "overwrite")
          cat(paste("    L014 Created grp7 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            tvpa_grp8() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp08'), mode = "overwrite")
          cat(paste("    L015 Created grp8 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # We are having an issue in the current snap for this function, so we created a custom with a correction for it to work, more details in gal_funs_bra_algos.R
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>% ## Do we need to establish a special value for RVLR14-RVLR28, instead of NA?
            # tvpa_grp9() %>%
            tvpa_grp9_c() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp09c'), mode = "overwrite")
          cat(paste("    L016 Created grp9 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # We are having an issue in the current snap for this function, so we created a custom with a correction for it to work, more details in gal_funs_bra_algos.R
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            # tvpa_grp11() %>%
            tvpa_grp11_c() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp11c'), mode = "overwrite")
          cat(paste("    L018 Created grp11 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            tvpa_grp12() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp12'), mode = "overwrite")
          cat(paste("    L019 Created grp12 at", Sys.time(), "\n"), file = ln, append = TRUE)
        
          # We are having an issue in the current snap for this function, so we created a custom with a correction for it to work, more details in gal_funs_bra_algos.R
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            # tvpa_grp17() %>%
            tvpa_grp17_c() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp17c'), mode = "overwrite")
          cat(paste("    L024 Created grp17c at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            tvpa_grp_custom17() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_custom17'), mode = "overwrite")
          cat(paste("    L026 Created custom17 at", Sys.time(), "\n"), file = ln, append = TRUE)
        
          # CADV Series is exclusive to UK, if you used any of them in a project please use AGG5 instead, its the same
          # if(mode == 'dsa'){
          #   ##--##--##--## DSA Only ##--##--##--##
          #   # We are having an issue in the current snap for this function, so we created a custom with a correction for it to work, more details in gal_funs_bra_algos.R
          #   tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
          #     # tvpa_grp20() %>%
          #     tvpa_grp20_c() %>%
          #     spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp20c'), mode = "overwrite")
          #   cat(paste("    L029 Created grp20c at", Sys.time(), "\n"), file = ln, append = TRUE)
          # }
          
          # We are having an issue in the current snap for this function, so we created a custom with a correction for it to work, more details in gal_funs_bra_algos.R
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%
            mutate(tot_cur_closed = 0) %>% # only needed for PHP
            # tvpa_grp21() %>%
            tvpa_grp21_c() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_grp21c'), mode = "overwrite")
          cat(paste("    L030 Created grp21c at", Sys.time(), "\n"), file = ln, append = TRUE)

          
          #### Algos from TVEA ----
          source("/shared/nas/shape/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM/gal_funs_bra.R")
          # Some algos that are actually being computed in tvea
          # Because of that, we'll use the "single" GCDF table here and use the tvea functions
          # the process below will match almost exactly with what is done in the attributes bkt script
          
          # Getting gcdf directly from attributes GCDF now
          
          # check mismatch
          # tbl(cn, gcdf_attributes_table_name) %>% filter(grepl("7652570", consno)) %>% collect() %>% View('check_gcdf_attributes_table_name')
          # gcdf_attributes_table_name <- "bu_brbureau.fespiri_retro_MM_p1p2p3_attrs_gcdf_trd_sql"
          
          ## Modify GCDF ##
          tbl(cn, gcdf_attributes_table_name) %>%
            # NN: Collecting the "single GCDF", which is the GCDF in the first observation period
            # filter(data_date_ref == 0) %>% 
            # mutate(suppressFI_TM = paste0(member_code,"_",lob)) %>% 
            # filter(!suppressFI_TM %in% suppressItens) %>%  
            # select(-suppressFI_TM) %>%
            # We don't want to include disputed trades to the calculations, we do have a disputed flag which is considered for many variables but not all, 
            # and since legally we can't use that information we are excluding it completely
            filter(disputed == 0) %>% 
            mutate(
              reported_dt = case_when(!is.na(reported_dt) ~ pmin(reported_dt, as.Date(data_date)), T ~ reported_dt)
            ) %>%  
            mutate(
              # DD NOTE: Most of these are fixing type issues created in the original query
              first_bureau_dt = as.Date(data_date)
              , secured = if_else(is.na(secured), FALSE, as.logical(secured))
              # ,active = if_else(is.na(active), TRUE, as.logical(active))
              , active = TRUE
              , closed = if_else(is.na(closed_ea), FALSE, as.logical(closed_ea))
              , choff = if_else(is.na(choff_temp), FALSE, as.logical(choff_temp))
              , bk = if_else(is.na(bk_pr), FALSE, as.logical(bk_pr))
              , primary_flag = if_else(is.na(primary_flag_temp), TRUE, as.logical(primary_flag_temp))
              , suppress = if_else(is.na(suppress), FALSE, as.logical(suppress))
              , deceased = if_else(is.na(deceased), FALSE, as.logical(deceased))
              , fraud_ind = if_else(is.na(fraud_ind), FALSE, as.logical(fraud_ind))
              , disputed = if_else(is.na(disputed), FALSE, as.logical(disputed))
              , deferral = if_else(is.na(deferral), FALSE, as.logical(deferral))
              #,repo_ind = if_else(is.na(repo_ind), FALSE, as.logical(repo_ind))
              , foreclosed = if_else(is.na(foreclosed), FALSE, as.logical(foreclosed))
              , lost_card = if_else(is.na(lost_card), FALSE, as.logical(lost_card))
              , restructured = if_else(is.na(restructured), FALSE, as.logical(restructured))
              , installment = if_else(is.na(installment), NA, as.logical(installment))
              , revolving = if_else(is.na(revolving), NA, as.logical(revolving))
              , disaster = FALSE
              , own_struct = 1
              , paymt_freq = "M" 
              , closed_dt = closed_ea_dt
              , suppress = FALSE
              , active = TRUE
              , score_curr_tier = 1
              , score_current = 600
              , paymt_pat = gcdf_pattern
              , choff_dt = choff_dt_temp
              , fintech = FALSE
              , all_trade = TRUE
            ) %>% mutate(
              msat = as.integer(case_when(
                all_trade ~ as.integer(case_when(
                  !is.na(reported_dt) ~ as.integer(months_between(data_date, reported_dt)), 
                  T ~ as.integer(999))), 
                T ~ as.integer(999)))
              , gcdf_rc2 = high_credit # we are using this to filter high_credit by original value, not round.
              , gcdf_rc3 = ""
              , gcdf_rc4 = ""
              , gcdf_rc2_frsrpt_dt = as.Date("1900-01-01")
              , gcdf_rc3_frsrpt_dt = as.Date("1900-01-01")
              , gcdf_rc4_frsrpt_dt = as.Date("1900-01-01")
              , paymt_pat_start_dt = as.Date("2020-03-05")
              , bk_trade = FALSE
              , bk_dt = as.Date("1900-01-01")
              , next_paymt_bacen = 0 # THIS IS JUST FOR BACEN, NOT USED HERE
            ) %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_algos_bra_trd_tmp'), mode = "overwrite")
          cat(paste("  L037 Created bra_trd at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # Create filters (now getting more normal)
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_bra_trd_tmp')) %>%
            currency_collapser %>%
            prepare_filters %>% 
            prepare_dpd_spark %>%
            mutate(high_credit = gcdf_rc2) %>% 
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_algos_bra_trd_filt'), mode = "overwrite")
          cat(paste("  L038 Created bra_trd_filt at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # This custom prepare_filters is to created a couple of columns needed for custom attributes
          # We are saving the filter table to avoid having to run again in case the session drops
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_bra_trd_filt')) %>% 
            prepare_filters_custom2() %>% 
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_algos_trd_filt'), mode = "overwrite")
          cat(paste("  L039 Wrote algos_trd_filt at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_bra_trd_tmp'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_bra_trd_filt'))
          # 
          cat(paste("  Creating algos from TVEA -------", "\n"), file = ln, append = TRUE)
          
          # # check mismatch
          # tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>% separate(consno, c("consno"), sep="[||]") %>% 
          #   filter(consno == 7652570) %>% select(all_trade,consno,accid,data_date,reported_dt,months_since_update,mop,closed,closed_dt,member_code,
          #                                                             member_name,active,first_bureau_dt,secured,opened_dt,msop,lob,CURR_ACT,balance,credit_line,high_credit,
          #                                                             past_due_amt,G201_T,paymt_amt_act,paymt_amt_sch,credit_limit,disputed,
          #                                                             all_trade,disaster,gcdf_pattern,gcdf_rc1,gcdf_rc2,gcdf_rc3,gcdf_rc4) %>% collect() %>% View("check_trd_filt")
          # 
          
          # We have a custom from tvea_grp8 because when mthly_paymt_amt_sch is used we want the values to be rounded, updated in gal_funs_bra_algos
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>%
            tvea_grp8_c() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvea_grp08c'), mode = "overwrite")
          
          cat(paste("    L040 Created grp8c at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>%
            mutate(deceased = FALSE) %>% 
            tvea_grp10() %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvea_grp10'), mode = "overwrite")
          cat(paste("    L041 Created grp10 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          # OBS: ALL BELLOW ADDED FOR NUBANK MISMATCH
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>% 
            tvea_grp3() %>%
            select(-AT28AF) %>% # THIS ATTRIBUTES HAS TO BE COMPUTED WITH CUSTOM7
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvea_grp03'), mode = "overwrite")
          cat(paste("    L113 Created grp3 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>% 
            tvea_grp5() %>%  
            select(-G216S, -G216SF) %>% ## These attributes are being computed in a separate custom function, we had to make changes in one of the condition in the codes to each of these.
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvea_grp05'), mode = "overwrite")
          cat(paste("    L214 Created grp5 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          tbl(cn, paste0('bu_brbureau.',eid, prj, '_algos_trd_filt')) %>% 
            tvea_grp_custom6() %>% 
            select(-data_date) %>%
            spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvea_gr06c'), mode = "overwrite")
          cat(paste("    L122 Created grp06c custom at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          ## Collect all ##
          
          if(mode == 'bkt'){
            tvpa_all <- 
              list(
                  # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp01')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp02')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp03')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp04')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp05')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp06')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp07')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp08')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp09c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp10c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp11c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp12')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp17c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_custom17')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp21c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp08c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp10'))
              ) %>%
              purrr::reduce(full_join, by = c("consno")) %>%
              spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_all'), mode="overwrite")
            
            cat(paste("  L042 bkt mode combine all at", Sys.time(), "\n"), file = ln, append = TRUE)
          } else if(mode == 'dsa'){
            tvpa_all <- 
              list(
                  # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp01')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp02')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp03')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp04')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp05')),
                  #  tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp06')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp07')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp08')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp09c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp11c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp12')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp17c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_custom17')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_grp21c')),
                   
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp01')),
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp02')),
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp05')),
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp07')),
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp09')),
                   # tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvua_grp10c')),
                   
                   # added for nubank mismatch
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp03')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp05')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_gr06c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp08c')),
                   tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvea_grp10'))
              ) %>%
              purrr::reduce(full_join, by = c("consno")) %>%
              spark_write_table(paste0('bu_brbureau.',eid, prj, '_tvpa_all'), mode="overwrite")
            cat(paste("  L042 dsa mode combine all at", Sys.time(), "\n"), file = ln, append = TRUE)
          }
          
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp01'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp02'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp03'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp04'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp05'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp06'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp07'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp08'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp09c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp10c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp11c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp12'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp13'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp14'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp15c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_custom16'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp17c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp16'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_custom17'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp18c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp19'))
          # # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp20c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvpa_grp21c'))
          # 
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp01'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp02'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp05'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp07'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp09'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvua_grp10c'))
          # 
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvea_grp08c'))
          # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.',eid, prj, '_tvea_grp10'))	
          
          # Here we'll add the original subject_id and monit date back to the table with the attributes
          
          tvpa_all <- tbl(cn, paste0('bu_brbureau.',eid, prj, '_tvpa_all')) %>%
            inner_join(
              tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% 
                select(consno, sj_subject_id, monit_date), 
              by = "consno") %>% 
            compute()
          cat(paste("  L043 Compute combine all at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          algos_28_nubank <- c("AT06SF","BALMAG01","BALMAG02","BALMAG03","BC34D","BCPMTNUM","BCPMTSTR","CV21","CV22","CV23","CV24","CV31","G230SF","PAYMNT11","PAYMNT52","PAYMNT53","PAYMNT58","PAYMNT62","PAYMNT67","PAYMNT68","PAYMNT73","PAYMNT77","RVLR05","RVLR09","RVLR30","RVLR31","RVLR32","TRV14")
          
          algos_p2 <- c("TRV03", "TRV04", "TRV05", "TRV06", "TRV07", "TRV08", "TRV09", "TRV10", "TRV11", "TRV12", "TRV13", "TRV14", "TRV18", "TRV23", "BALMAG01", "BALMAG02", "BALMAG03", "RVLR01", "RVLR03", "RVLR05", "RVLR07", "RVLR08", "RVLR09", "RVLR10", "RVLR29", "RVLR30", "RVLR31", "RVLR32", "RVLR33", "WALSHR01", "WALSHR02", "WALSHR03", "WALSHR04", "WALSHR05", "WALSHR06", "WALSHR07", "WALSHR08", "WALSHR09", "WALSHR10", "PAYMNT01", "PAYMNT02", "PAYMNT03", "PAYMNT04", "PAYMNT05", "PAYMNT06", "PAYMNT07", "PAYMNT08", "PAYMNT09", "PAYMNT10", "PAYMNT11", "PAYMNT50", "PAYMNT51", "PAYMNT52", "PAYMNT53", "PAYMNT55", "PAYMNT56", "PAYMNT57", "PAYMNT58", "PAYMNT60", "PAYMNT61", "PAYMNT62", "PAYMNT63", "PAYMNT65", "PAYMNT66", "PAYMNT67", "PAYMNT68", "PAYMNT70", "PAYMNT71", "PAYMNT72", "PAYMNT73", "PAYMNT75", "PAYMNT76", "PAYMNT77", "PAYMNT78", "WALSHRS01", "WALSHRS02", "CV13", "CV17", "CV18", "CV19", "CV20", "CV21", "CV22", "CV23", "CV24", "CV25", "CV26", "CV27", "CV28", "CV31", "BCPMTNUM")
          algos_p2_char <- c("BCPMTSTR")
          algos_p3 <- c("AGGS907","AGGS908","AGGS909","ALL231","ALL232","ALL233","ALL234","ALL235","ALL252","ALL253","ALL254","ALL255","BKC201","BKC202","BKC203","BKC204","BKC205","BKC222","BKC223","BKC224","BKC225","BKC231","BKC232","BKC233","BKC234","BKC235","BKC252","BKC253","BKC254","BKC255","PER231","PER232","PER233","PER234","PER235","PER252","PER253","PER254","PER255","AGG901","AGG902","AGG903","AGG904","MNPMAG01","MNPMAG02","MNPMAG03","PDMAG02","PDMAG03","PDMAG05","PDMAG01","PDMAG1A","AGG301","AGG302","AGG303","AGG304","AGG305","AGG306","AGG307","AGG308","AGG309","AGG310","AGG311","AGG312","AGG313","AGG314","AGG315","AGG316","AGG317","AGG318","AGG319","AGG320","AGG321","AGG322","AGG323","AGG324","AGG101","AGG102","AGG103","AGG104","AGG105","AGG106","AGG107","AGG108","AGG109","AGG110","AGG111","AGG112","AGG113","AGG114","AGG115","AGG116","AGG117","AGG118","AGG119","AGG120","AGG121","AGG122","AGG123","AGG124","AGG201","AGG202","AGG203","AGG204","AGG205","AGG206","AGG207","AGG208","AGG209","AGG210","AGG211","AGG212","AGG213","AGG214","AGG215","AGG216","AGG217","AGG218","AGG219","AGG220","AGG221","AGG222","AGG223","AGG224","AGG1001","AGG1002","AGG1003","AGG1004","AGG1005","AGG1006","AGG1007","AGG1008","AGG1009","AGG1010","AGG1011","AGG1012","AGG1013","AGG1014","AGG1015","AGG1016","AGG1017","AGG1018","AGG1019","AGG1020","AGG1021","AGG1022","AGG1023","AGG1024","AGG1101","AGG1102","AGG1103","AGG1104","AGG1105","AGG1106","AGG1107","AGG1108","AGG1109","AGG1110","AGG1111","AGG1112","AGG1113","AGG1114","AGG1115","AGG1116","AGG1117","AGG1118","AGG1119","AGG1120","AGG1121","AGG1122","AGG1123","AGG1124","AGG1201","AGG1202","AGG1203","AGG1204","AGG1205","AGG1206","AGG1207","AGG1208","AGG1209","AGG1210","AGG1211","AGG1212","AGG1213","AGG1214","AGG1215","AGG1216","AGG1217","AGG1218","AGG1219","AGG1220","AGG1221","AGG1222","AGG1223","AGG1224","AGG1301","AGG1302","AGG1303","AGG1304","AGG1305","AGG1306","AGG1307","AGG1308","AGG1309","AGG1310","AGG1311","AGG1312","AGG1313","AGG1314","AGG1315","AGG1316","AGG1317","AGG1318","AGG1319","AGG1320","AGG1321","AGG1322","AGG1323","AGG1324","AGG701","AGG702","AGG703","AGG704","AGG705","AGG706","AGG707","AGG708","AGG709","AGG710","AGG711","AGG712","AGG713","AGG714","AGG715","AGG716","AGG717","AGG718","AGG719","AGG720","AGG721","AGG722","AGG723","AGG724","AGG801","AGG802","AGG803","AGG804","AGG805","AGG806","AGG807","AGG808","AGG809","AGG810","AGG811","AGG812","AGG813","AGG814","AGG815","AGG816","AGG817","AGG818","AGG819","AGG820","AGG821","AGG822","AGG823","AGG824","AGG9101","AGG9102","AGG9103","AGG9104","AGG9105","AGG9106","AGG9107","AGG9108","AGG9109","AGG9110","AGG9111","AGG9112","AGG9113","AGG9114","AGG9115","AGG9116","AGG9117","AGG9118","AGG9119","AGG9120","AGG9121","AGG9122","AGG9123","AGG9124","AGG9301","AGG9302","AGG9303","AGG9304","AGG9305","AGG9306","AGG9307","AGG9308","AGG9309","AGG9310","AGG9311","AGG9312","AGG9313","AGG9314","AGG9315","AGG9316","AGG9317","AGG9318","AGG9319","AGG9320","AGG9321","AGG9322","AGG9323","AGG9324","AGG9401","AGG9402","AGG9403","AGG9404","AGG9405","AGG9406","AGG9407","AGG9408","AGG9409","AGG9410","AGG9411","AGG9412","AGG9413","AGG9414","AGG9415","AGG9416","AGG9417","AGG9418","AGG9419","AGG9420","AGG9421","AGG9422","AGG9423","AGG9424","AGG9501","AGG9502","AGG9503","AGG9504","AGG9505","AGG9506","AGG9507","AGG9508","AGG9509","AGG9510","AGG9511","AGG9512","AGG9513","AGG9514","AGG9515","AGG9516","AGG9517","AGG9518","AGG9519","AGG9520","AGG9521","AGG9522","AGG9523","AGG9524","AGG9701","AGG9702","AGG9703","AGG9704","AGG9705","AGG9706","AGG9707","AGG9708","AGG9709","AGG9710","AGG9711","AGG9712","AGG9713","AGG9714","AGG9715","AGG9716","AGG9717","AGG9718","AGG9719","AGG9720","AGG9721","AGG9722","AGG9723","AGG9724","TRANBAL01","TRANBAL02","TRANBAL03","TRANBAL04","TRANBAL05","TRANBAL06","TRANBAL07","TRANBAL08","TRANBAL09","TRANBAL10","TRANBAL11","TRANBAL12","TRANBAL13","TRANBAL14","TRANBAL15","TRANBAL16","TRANBAL17","TRANBAL18","TRANBAL19","TRANBAL20","TRANBAL21","TRANBAL22","TRANBAL23","TRANBAL24","INTDB")
          algos_p4 <- c("CVTDB","AGG401","AGG402","AGG403","AGG404","AGG405","AGG406","AGG407","AGG408","AGG409","AGG410","AGG411","AGG412","AGG413","AGG414","AGG415","AGG416","AGG417","AGG418","AGG419","AGG420","AGG421","AGG422","AGG423","AGG424","AGG501","AGG502","AGG503","AGG504","AGG505","AGG506","AGG507","AGG508","AGG509","AGG510","AGG511","AGG512","AGG513","AGG514","AGG515","AGG516","AGG517","AGG518","AGG519","AGG520","AGG521","AGG522","AGG523","AGG524","AGG9201","AGG9202","AGG9203","AGG9204","AGG9205","AGG9206","AGG9207","AGG9208","AGG9209","AGG9210","AGG9211","AGG9212","AGG9213","AGG9214","AGG9215","AGG9216","AGG9217","AGG9218","AGG9219","AGG9220","AGG9221","AGG9222","AGG9223","AGG9224","AGGS901","AGGS902","AGGS903","AGGS911","WALSRES02","REVS101","REVS102","REVS103","REVS104","REVS105","REVS106","REVS107","REVS108","REVS109","REVS110","REVS111","REVS112","REVS113","REVS114","REVS115","REVS116","REVS117","REVS118","REVS119","REVS120","REVS121","REVS122","REVS123","REVS124")
          
          algos_full <- names(tvpa_all %>% head() %>% collect()) %>% .[!. %in% c("consno","sj_subject_id","monit_date")] # Not sure why this stopped working
          algos_full <- setdiff(colnames(tvpa_all), c("consno","sj_subject_id","monit_date")) 
          
          ### Output data file ----
          
          # output_file <- tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_input_file")) %>% 
          #   left_join(
          #     tbl(cn, paste0("bu_brbureau.", eid, prj, "_algos_subject_sample")) %>% 
          #       select(cpf_TU, date_TU, sj_subject_id, flg_deceased, flg_minor)
          #     , by = c("cpf_TU" = "cpf_TU", "date_TU" = "date_TU")
          #   ) %>% 
          #   left_join(
          #     tvpa_all %>% 
          #       select(sj_subject_id, monit_date, all_of(c(algos_full)))
          #     , by = c("sj_subject_id" = "sj_subject_id", "date_TU" = "monit_date")
          #   ) %>% select(-sj_subject_id, -cpf_TU, -date_TU) %>% 
          #   compute()
          # cat(paste("L044 Compute output_file at", Sys.time(), "\n"), file = ln, append = TRUE)
          # 
          
          output_file <- tvpa_all %>% select(sj_subject_id, monit_date, algos_28_nubank) 
          output_file <-  output_file %>% mutate(
            across(any_of(c(algos_p2_char)), ~ if_else(is.na(.x), "Valor especial", .x)) ,
            # across(any_of(c(algos_full)), ~ if_else(is.na(.x) | flg_deceased == 1 | flg_minor == 1, -7, .x))
          ) %>% 
            # select(-flg_deceased, -flg_minor) %>% 
            compute()
          cat(paste("L045 Compute output_file #2 at", Sys.time(), "\n"), file = ln, append = TRUE)
          
          if(template %in% c("algoritmos","ambos")) {
            if(mode %in% c("bkt")) {
              # This will keep the outputs saved in bu_brbureau
              # temporary different output location for comparing purposes with old run of "_tv_backtest_output_algos".
              
              output_file %>% 
                select(-c(setdiff(algos_full, c(algos_p2,algos_p2_char,algos_p3,algos_p4)))) %>% 
                compute() %>%
                spark_write_table(paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos"), mode = "overwrite")
              cat(paste("L046 Write output algos at", Sys.time(), "\n"), file = ln, append = TRUE)
              
              
              # This will keep the outputs saved in the defined output_path
              # output_file_local_algos <- arrow::open_dataset(paste0("/shared/hdfs/bu_brbureau/db/", eid, str_to_lower(prj), "_tv_backtest_output_algos")) %>%
              # collect()
              # write.csv2(output_file_local_algos, file = output_path, row.names = FALSE, quote = FALSE, na = "")
              # write_csv_arrow(output_file_local_algos, file = paste0(output_path,"_full.csv"))
            }
            if(mode %in% c("dsa")) {
              output_file %>% 
                compute() %>%
                spark_write_table(paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos"), mode = "overwrite")
              cat(paste("L046 Write output algos at", Sys.time(), "\n"), file = ln, append = TRUE)
              
              # save CSV
              output_file %>% sdf_repartition(1) %>% compute() %>%
              spark_write_csv(paste0("file:///shared/shape_tier3/bu_br/persist/Produto_Analytics/TruVision/Mismatch/R_MM_Spark/using_git_scripts/retro/nubank_100/output_28_algos_nubank_", format(Sys.time(), '%Y%m%d')), quote="", null_value = NULL, mode='overwrite')
            }
          }
          
          if (template %in% c("indicadores","ambos")){
            if(mode %in% c("bkt")) {
              # This will keep the outputs saved in bu_brbureau
              output_file %>% 
                select(-c(setdiff(algos_full, c(algos_p2, algos_p2_char)))) %>% 
                indicators_spark_algos() %>% 
                select(setdiff(colnames(.), c(algos_p2, algos_p2_char))) %>% 
                spark_write_table(paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos_indic"), mode = "overwrite") 
              cat(paste("L047 Write output indic at", Sys.time(), "\n"), file = ln, append = TRUE)
              
              # This will keep the outputs saved in the defined output_path
              # output_file_local_indic <- arrow::open_dataset(paste0("/shared/hdfs/bu_brbureau/db/", eid, str_to_lower(prj), "_tv_backtest_output_algos_indic")) %>%
              # collect()
              # write.csv2(output_file_local_indic, file = output_path, row.names = FALSE, quote = FALSE, na = "")
              # write_csv_arrow(output_file_local_indic, file = paste0(output_path,".csv"))
              
            } 
          }
        }

        # Final review ----
        cat(paste("Final review at:", Sys.time(), "\n"), file = ln, append = TRUE)
        
        if(output_comb == 'Y'){
          
          if(template %in% c("algoritmos","ambos")) {
            # Checking volume and header from attributes
            
            print(" Count from tv_backtest_output_algos")
            tbl(cn, paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos")) %>%
              summarize(
                rows = n(),
                parties = n_distinct(!!sym(input_cpf_var)))
            
            # print(" Header from tv_backtest_output_algos")
            # tbl(cn, paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos")) %>%
            #   head(20)
            
          } 
          if (template %in% c("indicadores","ambos") & mode == 'bkt'){
            # Checking volume and header from attributes
            
            print(" Count from tv_backtest_output_algos_indic")
            tbl(cn, paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos_indic")) %>%
              summarize(
                rows = n(),
                parties = n_distinct(!!sym(input_cpf_var)))
            
            # print(" Header from tv_backtest_output_algos_indic")
            # tbl(cn, paste0("bu_brbureau.", eid, prj, "_tv_backtest_output_algos_indic")) %>%
            #   head(20)
            
          }
        }
        
        cat(paste("Script finished at:", Sys.time(), "\n"), file = ln, append = TRUE)
        cat(paste("Make sure to do your validations and go back for the final cleanup", "\n"), file = ln, append = TRUE)
        
        
        # # check mismatch
        # tbl(cn, paste0('bu_brbureau.',eid, prj, '_trhi_filt')) %>%  filter(grepl("5136470", consno)) %>%  
        #   separate(consno, c("consno"), sep="[||]") %>% 
        #   filter(consno == 5136470)  %>% collect() %>% View("check_trhi_filt")
        # 
        # select(all_trade,consno,accid,data_date,reported_dt,months_since_update,mop,closed,closed_dt,member_code,
        #        member_name,active,first_bureau_dt,secured,opened_dt,msop,lob,CURR_ACT,balance,credit_line,high_credit,
        #        past_due_amt,G201_T,paymt_amt_act,paymt_amt_sch,credit_limit,disputed,
        #        all_trade,disaster,gcdf_pattern,gcdf_rc1,gcdf_rc2,gcdf_rc3,gcdf_rc4)
        # 
        
        
        # Final Cleanup ----
        # Please execute this final cleanup manually after checking out that all the tables 
        # needed were properly created
        
        # Comb:
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_trhi_filt'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trd_filt'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tvpa_all'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos_indic'))
        
        # ON:
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_trhi_filt_on')) 
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trd_filt_on'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tvpa_all_on'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos_on'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos_indic_on'))
        
        # OFF:
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_trhi_filt_off'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_trd_filt_off'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tvpa_all_off'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos_off'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_tv_backtest_output_algos_indic_off'))
        
        # Other supports:
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_subject_sample'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_gcdf_trhi_sql'))
        # dbSendStatement(cn, paste0('DROP TABLE IF EXISTS bu_brbureau.', eid, prj, '_algos_input_file'))

--------------------------------------------------------------------------------------------------


Rewrite this R script that uses spkarlyr (and other libraries) into a python script that uses pyspark (and other libraries)