## ----include = FALSE----------------------------------------------------------
knitr::opts_chunk$set(
  collapse = TRUE,
  comment = "#>"
)

database_packages <- c("DBI", "dbplyr", "dplyr", "duckdb")
duckdb_available <- all(vapply(
  database_packages,
  requireNamespace,
  quietly = TRUE,
  FUN.VALUE = logical(1)
))

## -----------------------------------------------------------------------------
library(insurancerating)

## ----eval = FALSE-------------------------------------------------------------
# bytes_per_row <- as.numeric(object.size(portfolio_sample)) /
#   nrow(portfolio_sample)
# 
# estimated_object_gb <- bytes_per_row * expected_rows / 1024^3

## -----------------------------------------------------------------------------
periods <- data.frame(
  policy_id = c("P001", "P001", "P002"),
  coverage = c("Fire", "Fire", "Fire"),
  period_start = as.Date(c("2025-01-01", "2025-07-01", "2025-01-01")),
  period_end = as.Date(c("2025-06-30", "2025-12-31", "2025-12-31")),
  earned_exposure = c(0.5, 0.5, 1)
)

merge_date_ranges(
  periods,
  period_start = "period_start",
  period_end = "period_end",
  group_by = c("policy_id", "coverage"),
  aggregate_cols = "earned_exposure"
)

## -----------------------------------------------------------------------------
set.seed(2026)
local_portfolio <- data.frame(
  policy_id = seq_len(100000),
  sector = sample(c("Industry", "Retail", "Services"), 100000, TRUE),
  region = sample(c("North", "South", "West"), 100000, TRUE),
  reconstruction_value = sample(seq(100000, 2000000, by = 500), 100000, TRUE),
  earned_exposure = runif(100000, 0.25, 1),
  earned_premium = runif(100000, 100, 2500)
)

local_portfolio$reconstruction_value_1000 <-
  round(local_portfolio$reconstruction_value / 1000) * 1000

frequency <- with(
  local_portfolio,
  exp(
    -3 +
      0.20 * (sector == "Industry") -
      0.10 * (region == "North") +
      0.0000001 * reconstruction_value_1000
  )
)
local_portfolio$claim_count <- rpois(
  nrow(local_portfolio),
  lambda = local_portfolio$earned_exposure * frequency
)
local_portfolio$claim_amount <- 0
has_claims <- local_portfolio$claim_count > 0
local_portfolio$claim_amount[has_claims] <- rgamma(
  sum(has_claims),
  shape = 2 * local_portfolio$claim_count[has_claims],
  scale = 3000
)

local_grid <- rating_grid(
  local_portfolio,
  group_by = c("sector", "region", "reconstruction_value_1000"),
  exposure = "earned_exposure",
  aggregate_cols = c("claim_count", "claim_amount", "earned_premium")
)

data.frame(
  stage = c("Portfolio", "Rating grid"),
  rows = c(nrow(local_portfolio), nrow(local_grid))
)

head(local_grid)

## -----------------------------------------------------------------------------
grid_frequency_model <- glm(
  claim_count ~ sector + region + reconstruction_value_1000 +
    offset(log(earned_exposure)),
  family = poisson(link = "log"),
  data = local_grid
)

policy_frequency_model <- glm(
  claim_count ~ sector + region + reconstruction_value_1000 +
    offset(log(earned_exposure)),
  family = poisson(link = "log"),
  data = local_portfolio
)

all.equal(
  unname(coef(grid_frequency_model)),
  unname(coef(policy_frequency_model)),
  tolerance = 1e-8
)

## ----duckdb-memory-workflow, eval = duckdb_available--------------------------
library(DBI)
library(dbplyr)
library(dplyr)
library(duckdb)

con <- dbConnect(duckdb())

dbWriteTable(
  con,
  "portfolio",
  local_portfolio,
  overwrite = TRUE
)

portfolio_db <- tbl(con, "portfolio")

grid_db <- portfolio_db |>
  mutate(
    reconstruction_value_1000 =
      round(reconstruction_value / 1000) * 1000
  ) |>
  rating_grid_db(
    group_by = c("sector", "region", "reconstruction_value_1000"),
    exposure = "earned_exposure",
    aggregate_cols = c("claim_count", "claim_amount", "earned_premium")
  )

# Inspect the SQL without collecting the row-level portfolio.
sql_render(grid_db)

database_row_counts <- data.frame(
  source_rows = portfolio_db |>
    summarise(n = n()) |>
    collect() |>
    pull(n),
  reduced_rows = grid_db |>
    summarise(n = n()) |>
    collect() |>
    pull(n)
)

database_grid <- collect(grid_db)
database_row_counts
head(database_grid)

dbDisconnect(con, shutdown = TRUE)

## ----eval = FALSE-------------------------------------------------------------
# con <- dbConnect(duckdb(), dbdir = "portfolio.duckdb")

## ----eval = FALSE-------------------------------------------------------------
# library(DBI)
# library(dbplyr)
# library(dplyr)
# library(duckdb)
# 
# con <- dbConnect(duckdb())
# 
# dbExecute(con, "
#   CREATE VIEW portfolio AS
#   SELECT * FROM read_parquet('portfolio/*.parquet')
# ")
# portfolio_db <- tbl(con, "portfolio")

## ----eval = FALSE-------------------------------------------------------------
# large_database_path <- "large_portfolio.duckdb"
# con <- dbConnect(duckdb(), dbdir = large_database_path)
# 
# dbExecute(con, "
#   CREATE TABLE portfolio_10m AS
#   SELECT
#     i AS policy_id,
#     'Sector ' || CAST(i % 20 AS VARCHAR) AS sector,
#     'Region ' || CAST(FLOOR(i / 20) % 10 AS VARCHAR) AS region,
#     100000 + (FLOOR(i / 200) % 100) * 1000 AS reconstruction_value,
#     0.5 + (i % 50) / 100.0 AS earned_exposure,
#     CASE WHEN i % 17 = 0 THEN 1 + CAST(i % 3 AS INTEGER) ELSE 0 END AS claim_count,
#     CASE WHEN i % 17 = 0 THEN 5000 + CAST(i % 50000 AS DOUBLE) ELSE 0 END AS claim_amount,
#     100 + (i % 2000) AS earned_premium
#   FROM range(10000000) AS portfolio(i)
# ")
# 
# portfolio_10m <- tbl(con, "portfolio_10m")
# 
# grid_10m_db <- portfolio_10m |>
#   mutate(
#     reconstruction_value_1000 =
#       round(reconstruction_value / 1000) * 1000
#   ) |>
#   rating_grid_db(
#     group_by = c("sector", "region", "reconstruction_value_1000"),
#     exposure = "earned_exposure",
#     aggregate_cols = c("claim_count", "claim_amount", "earned_premium")
#   )
# 
# sql_render(grid_10m_db)
# 
# row_comparison_10m <- data.frame(
#   source_rows = portfolio_10m |>
#     summarise(n = n()) |>
#     collect() |>
#     pull(n),
#   reduced_rows = grid_10m_db |>
#     summarise(n = n()) |>
#     collect() |>
#     pull(n)
# )
# row_comparison_10m$reduction <-
#   1 - row_comparison_10m$reduced_rows / row_comparison_10m$source_rows
# 
# row_comparison_10m
# grid_10m <- collect(grid_10m_db)

## ----eval = FALSE-------------------------------------------------------------
# dbExecute(con, "
#   CREATE TABLE portfolio_50m AS
#   SELECT
#     i AS policy_id,
#     'Sector ' || CAST(i % 20 AS VARCHAR) AS sector,
#     'Region ' || CAST(FLOOR(i / 20) % 10 AS VARCHAR) AS region,
#     100000 + (FLOOR(i / 200) % 100) * 1000 AS reconstruction_value,
#     0.5 + (i % 50) / 100.0 AS earned_exposure,
#     CASE WHEN i % 17 = 0 THEN 1 + CAST(i % 3 AS INTEGER) ELSE 0 END AS claim_count,
#     CASE WHEN i % 17 = 0 THEN 5000 + CAST(i % 50000 AS DOUBLE) ELSE 0 END AS claim_amount,
#     100 + (i % 2000) AS earned_premium
#   FROM range(50000000) AS portfolio(i)
# ")
# 
# portfolio_50m <- tbl(con, "portfolio_50m")
# 
# grid_50m_db <- portfolio_50m |>
#   mutate(
#     reconstruction_value_1000 =
#       round(reconstruction_value / 1000) * 1000
#   ) |>
#   rating_grid_db(
#     group_by = c("sector", "region", "reconstruction_value_1000"),
#     exposure = "earned_exposure",
#     aggregate_cols = c("claim_count", "claim_amount", "earned_premium")
#   )
# 
# row_comparison_50m <- data.frame(
#   source_rows = portfolio_50m |>
#     summarise(n = n()) |>
#     collect() |>
#     pull(n),
#   reduced_rows = grid_50m_db |>
#     summarise(n = n()) |>
#     collect() |>
#     pull(n)
# )
# row_comparison_50m$reduction <-
#   1 - row_comparison_50m$reduced_rows / row_comparison_50m$source_rows
# 
# row_comparison_50m
# grid_50m <- collect(grid_50m_db)

## ----eval = FALSE-------------------------------------------------------------
# dbExecute(con, "
#   CREATE TABLE portfolio_periods AS
#   SELECT * FROM (VALUES
#     ('P001', 'Fire', 'Industry', DATE '2025-01-01', DATE '2025-06-30', 0.5, 600.0),
#     ('P001', 'Fire', 'Industry', DATE '2025-07-01', DATE '2025-12-31', 0.5, 650.0),
#     ('P002', 'Fire', 'Retail',   DATE '2025-01-01', DATE '2025-12-31', 1.0, 900.0)
#   ) AS periods(
#     policy_id,
#     coverage,
#     sector,
#     period_start,
#     period_end,
#     earned_exposure,
#     earned_premium
#   )
# ")
# 
# periods_db <- tbl(con, "portfolio_periods")
# 
# merged_periods_db <- merge_date_ranges_db(
#   periods_db,
#   period_start = "period_start",
#   period_end = "period_end",
#   group_by = c("policy_id", "coverage", "sector"),
#   aggregate_cols = c("earned_exposure", "earned_premium"),
#   merge_gap_days = 1
# )
# 
# merged_periods <- collect(merged_periods_db)
# dbDisconnect(con, shutdown = TRUE)
# unlink(large_database_path)
# unlink(paste0(large_database_path, ".wal"))