# This code computes fundamental variables from Compustat

#.rs.restartR()
#(.packages())
# rm(list = ls())

library(RPostgreSQL)
library(dplyr, warn.conflicts = FALSE)

pg <- dbConnect(PostgreSQL())
dbExecute(pg, "SET work_mem TO '10GB'") # Speeds-up creation of temporary tables

# Define PG data ----

# Project schema
rs <- dbGetQuery(pg, "SET search_path TO nonans_product, public")

# Outside of project schema
industry_49 <- tbl(pg, sql("SELECT * FROM fama_french.industry_49"))
co_hgic <- tbl(pg, sql("SELECT * FROM comp.co_hgic")) # Historical GICS industry classification
funda <- tbl(pg, sql("SELECT * FROM comp.funda"))
sec_history <- tbl(pg, sql("SELECT * FROM comp.sec_history")) # Historical exchange membership

# Collect relevant data ----

start_date <- '1997-01-01'

# Acct. values are in millions
cols <- c("gvkey", "datadate", "fyear",
          "at", # Assets - Total 
          "sale", # Sales/Turnover (Net)
          "xrd", # Research and Development Expense
          "ib", # Income Before Extraordinary Items
          "seq", # SEQ -- Stockholders' Equity - Total
          "txditc", #TXDITC -- Deferred Taxes and Investment Tax Credit
          "pstk", # PSTK -- Preferred/Preference Stock (Capital) - Total
          "pstkrv", # PSTKRV -- Preferred Stock Redemption Value
          "pstkl", # PSTKL -- Preferred Stock Liquidating Value
          "ppent", # PPENT -- Property, Plant and Equipment - Total (Net)
          "ppegt", # PPEGT -- Property, Plant and Equipment - Total (Gross)
          "prcc_f", # Price Close - Annual - Fiscal
          "adjex_f", # Cumulative Adjustment Factor by Ex-Date - Fiscal (adjust PRC by dividing by CFACPR )
          "csho", # Common Shares Outstanding (in millions of shares)
          "cogs", # Cost of Goods Sold
          "fic", # Current ISO Country Code - Incorporation  
          "sich", # Standard Industrial Classification - Historical
          "sstk", # SSTK -- Sale of Common and Preferred Stock---WARNING: includes option exercises
          "dltis", # DLTIS -- Long-Term Debt Issuance
          "dd1", # DD1 -- Long-Term Debt Due in One Year
          "dltt", # DLTT	NUM	DLTT -- Long-Term Debt - Total
          "che" # CHE -- Cash and Short-Term Investments
          )

# Annual fundamentals
funda_alt <- 
    funda %>%
    # Keep standard reporting
    filter(indfmt == 'INDL' & datafmt == 'STD' & popsrc == 'D' & consol == 'C' & !is.na(fyear)) %>%   
    filter(datadate > start_date) %>%
    select(one_of(cols))

# Historical exchange membership
sec_history <- 
    sec_history %>%
    filter(item == 'EXCHG')

# Compute variabes--scaled and growth ----

dbGetQuery(pg, "DROP TABLE IF EXISTS fund")

fund <- 
    funda_alt %>%
    # Set non-positive {at, sale, xrd} values to NA
    # WARNING: do NOT filter not to introduce gaps in time-series
    mutate_at(.vars = vars(at, sale, xrd), 
              .funs = funs(if_else(. <= 0, NA, .))) %>%
    # Replace missing values with 0s 
    # WARNING: for {xrd} missing and non-positive values are considered identical
    mutate_at(.vars = vars(xrd, txditc, sstk, dltis, dd1),
              .funs = funs(coalesce(., 0))) %>%
    mutate(pstk = sql(COALESCE(pstk, pstkrv, pstkl, 0))) %>% # Replace missing           
    # Compute new variables
    mutate(# Split-adjust prices, e.g, Microsoft
        #  4 012141 2002-06-30  2002  54.70       2
        #  5 012141 2003-06-30  2003  25.64       1
        prcc = if_else(adjex_f != 0, prcc_f/adjex_f, NA),
        market_to_book = (prcc_f*csho + at - (seq + txditc - pstk))/at,
        dt = dd1 + dltt, # Total debt
        lev = (dd1 + dltt)/at, 
        lev1 = dd1/at,
        dd1_to_che = dd1/NULLIF(che, 0),
        fai_fr = coalesce(ppent, ppegt)/at,
        mcap = prcc_f*csho, 
        loss = if_else(ib < 0, 1, 0)) %>%
    # Partition by gvkey and order by datadate
    group_by(gvkey) %>%
    arrange(datadate) %>%
    # Lag at and price variables
    mutate_at(.vars = vars(at, prcc, mcap),
              .funs = funs(lag = sql(lag(.)))) %>%
    # Compute variables based on lags
    mutate(ret_fr = if_else(prcc_lag != 0, prcc/prcc_lag - 1, NA), 
           sstk_sc = sstk/NULLIF(mcap_lag, 0), 
           dltis_sc = dltis/NULLIF(mcap_lag, 0)) %>%
    # Compute scaled flow variables as a fraction of assets total
    mutate_at(.vars = vars(sale, ib), 
              .funs = funs(to_at = ./at_lag)) %>%
    # Compute scaled flow variables as a fraction of sales
    mutate_at(.vars = vars(ib, xrd), 
              .funs = funs(to_sale = ./sale)) %>%
    # Convert size vars to billions
    mutate_at(.vars = vars(mcap, at, sale), 
              .funs = funs(bn = ./(10**3))) %>%
    # Compute growth rates as in Stephen Terry's JMP
    # WARNING: cannot send . and ._lag to sql
    # Growth based on raw variables
    mutate_at(.vars = vars(at, sale),
              .funs = funs(gr1 = if_else(. == 0 & sql(lag(.)) == 0, 0, 
                                         2*(. - sql(lag(.)))/
                                             (sql(NULLIF(abs(1.0*.) + lag(abs(1.0*.)), 0)))))) %>%
    # Long-term growth --- 3-year lag
    # Growth based on raw variables
    mutate_at(.vars = vars(at, sale),
              .funs = funs(gr3 = if_else(. == 0 & sql(lag(., 3L)) == 0, 0, 
                                         2*(. - sql(lag(., 3L)))/
                                             (sql(NULLIF(abs(1.0*.) + lag(abs(1.0*.), 3L), 0)))))) %>%
    # Clean up all variables that use wrong lag, i.e., fyear - fyear_lag != 1, ---very few cases
    mutate_at(.vars = vars(matches("lag|gr1|to_at"), ret_fr), 
              .funs = funs(if_else(fyear - lag(fyear) != 1, NA, .))) %>%
    # Clean up all variables that use wrong lag, i.e., fyear - fyear_lag != 1, ---very few cases
    mutate_at(.vars = vars(matches("gr3")), 
              .funs = funs(if_else(fyear - lag(fyear, 3L) != 3, NA, .))) %>%
    # Forward growth -- 1-year lead 
    mutate_at(.vars = vars(at, sale), 
              .funs = funs(fgr1 = if_else(. == 0 & sql(lead(.)) == 0, 0, 
                                          2*(sql(lead(.) - .))/
                                              (sql(NULLIF(abs(1.0*.) + lead(abs(1.0*.)), 0)))))) %>%
    # Forward growth -- 3-year lead
    mutate_at(.vars = vars(at, sale), 
              .funs = funs(fgr3 = if_else(. == 0 & sql(lead(., 3L)) == 0, 0, 
                                          2*(sql(lead(., 3L) - .))/
                                              (sql(NULLIF(abs(1.0*.) + lead(abs(1.0*.), 3L), 0)))))) %>%
    # Clean up all variables that use wrong lag, i.e., fyear - fyear_lag != 1, ---very few cases
    mutate_at(.vars = vars(matches("fgr1")), 
              .funs = funs(if_else(lead(fyear) - fyear != 1, NA, .))) %>%
    mutate_at(.vars = vars(matches("fgr3")), 
              .funs = funs(if_else(lead(fyear, 3L) - fyear != 3, NA, .))) %>%
    ungroup() %>%
    # Extrapolate industry sich
    # Extrapolate backwards within gvkey
    mutate(partition = sql("sum(CASE WHEN sich IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY gvkey ORDER BY datadate DESC)")) %>%
    mutate(sich = sql("first_value(sich) OVER (PARTITION BY gvkey, partition ORDER BY datadate DESC)")) %>%
    # Extrapolate forwards within gvkey
    mutate(partition = sql("sum(CASE WHEN sich IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY gvkey ORDER BY datadate)")) %>%
    mutate(sich = sql("first_value(sich) OVER (PARTITION BY gvkey, partition ORDER BY datadate)")) %>%
    select(-partition) %>%
    # Extract 3-digit and 4-digit SIC
    mutate(sic4 = sql("LPAD(sich::text, 4, '0')")) %>%
    mutate(sic3 = sql("substring(sic4 from 1 for 3)")) %>%
    # Extract year from datadate for Hoberg-Phillips data
    mutate(year = date_part('year', datadate)) %>%
    compute(name = "fund", temporary = TRUE, 
            indexes = list(c("gvkey", "datadate")))

# Compute sales-based HHI
hhi4 <- 
    fund %>%
    # Compute market shares by sic4
    group_by(sic4, fyear) %>%
    mutate(s2 = (sale/sum(sale))^2) %>%
    summarize(hhi4 = sum(s2)) %>%
    ungroup() %>%
    compute()

hhi3 <- 
    fund %>%
    # Compute market shares by sic3
    group_by(sic3, fyear) %>%
    mutate(s2 = (sale/sum(sale))^2) %>%
    summarize(hhi3 = sum(s2)) %>%
    ungroup() %>%
    compute()

# Compute 49 Fama-French industries
ff49 <- 
    fund %>%
    select(gvkey, datadate, sich) %>%
    # Force cartesian product
    mutate(dum = 1) %>%
    left_join(industry_49 %>%
                  mutate(dum = 1), by = "dum") %>%
    select(-dum) %>%
    filter(sql("sich BETWEEN sic_low AND sic_high")) %>%
    mutate(ff_industry_49 = ind_desc,
           ff_industry_49_num = ind_num) %>%
    compute(name = "ff49", temporary = TRUE, 
            indexes = list(c("gvkey", "datadate")))

# Merge GICS industries and historical exchanges ----

gics <- 
    fund %>%
    select(gvkey, datadate) %>%
    inner_join(co_hgic) %>%
    filter(indfrom <= datadate & (datadate <= indthru | is.na(indthru))) %>%
    select(gvkey, datadate, gsector, ggroup) %>%
    distinct() %>%
    compute(name = "gics", temporary = TRUE, 
            indexes = list(c("gvkey", "datadate")))

exch <- 
    fund %>%
    select(gvkey, datadate) %>%
    inner_join(sec_history) %>%
    filter(effdate <= datadate & (datadate <= thrudate | is.na(thrudate))) %>%
    rename(exchg = itemvalue) %>%
    select(gvkey, datadate, exchg) %>%
    # gvkey can have several securities listed and not listed
    group_by_(.dots = c("gvkey", "datadate")) %>%
    summarize(exchg = sql("array_agg(DISTINCT exchg ORDER BY exchg)")) %>%
    ungroup() %>%
    distinct() %>%
    compute(name = "exch", temporary = TRUE, 
            indexes = list(c("gvkey", "datadate")))

# Merge and write to PG ----

dbGetQuery(pg, "DROP TABLE IF EXISTS fundann")

fundann <- 
    fund %>%
    left_join(hhi4) %>%
    left_join(hhi3) %>%
    left_join(ff49) %>%
    left_join(gics) %>%
    left_join(exch) %>%
    # For multiple {gvkey, fyear}, keep max(datadate)
    group_by(gvkey, fyear) %>%
    filter(datadate == max(datadate)) %>%
    ungroup() %>%
    compute(name = "fundann", temporary=FALSE,
            indexes=list(c("gvkey", "datadate"), c("gvkey")))

db_comment <- paste0("CREATED USING azakolyukina/nonans_product/create_fundann.R ON ", Sys.time())
dbGetQuery(pg, paste0("COMMENT ON TABLE fundann IS '",
                      db_comment, "';"))

dbGetQuery(pg, "ALTER TABLE fundann OWNER TO nonans_product")
dbGetQuery(pg, "GRANT SELECT ON fundann TO nonans_product_access")
dbGetQuery(pg, "VACUUM ANALYZE fundann")

dbDisconnect(pg)
