R/okx_candle_data.R

In investdatar: Investment Data Access and Preparation Toolkit

#' @import data.table

.okx_candle_timeframe_to <- function(tag, unit = c("seconds", "minutes", "hours")) {
  unit <- match.arg(unit)
  parsed <- .parse_frequency(tag)
  minutes <- parsed$seconds / 60

  switch(unit,
    seconds = minutes * 60,
    minutes = minutes,
    hours = minutes / 60
  )
}

#' Get Last Completed Candle Time
#'
#' Floor current time to the last completed candle start for the given bar.
#'
#' @param bar Character. OKX timeframe tag (e.g., `"1m"`, `"15m"`, `"4H"`, `"1D"`).
#' @param tz Character. IANA timezone.
#'
#' @return POSIXct.
#' @export
get_source_utime_okx_candle <- function(bar, tz = "UTC") {
  infer_source_utime_from_frequency(bar, reference_time = Sys.time(), tz = tz)
}

.normalize_okx_candles <- function(dt, inst_id = NULL, bar = NULL) {
  if (is.null(dt)) {
    return(NULL)
  }

  dt <- data.table::as.data.table(dt)
  if ("timestamp" %in% names(dt) && !"datetime" %in% names(dt)) {
    data.table::setnames(dt, "timestamp", "datetime")
  }

  keep <- intersect(
    c("datetime", "open", "high", "low", "close", "volume", "volCcy", "volCcyQuote", "confirm"),
    names(dt)
  )
  dt <- dt[, keep, with = FALSE]

  if ("confirm" %in% names(dt)) {
    dt <- dt[confirm == 1L]
  }

  numeric_cols <- intersect(c("open", "high", "low", "close", "volume", "volCcy", "volCcyQuote"), names(dt))
  dt[, (numeric_cols) := lapply(.SD, as.numeric), .SDcols = numeric_cols]
  dt[, datetime := as.POSIXct(datetime, tz = attr(datetime, "tzone") %||% "UTC")]

  if (!is.null(inst_id) && !"inst_id" %in% names(dt)) {
    dt[, inst_id := inst_id]
  }
  if (!is.null(bar) && !"bar" %in% names(dt)) {
    dt[, bar := bar]
  }

  .standardize_market_ohlcv(
    dt,
    source = "okx",
    symbol = inst_id,
    interval = bar,
    time_col = "datetime"
  )
}

`%||%` <- function(x, y) if (is.null(x)) y else x

#' Get Latest OKX Candle Data
#'
#' @param inst_id Instrument identifier.
#' @param bar Candle interval.
#' @param limit Integer page size.
#' @param config Optional OKX API config. If omitted, defaults from the
#'   package config and `OKX_API_KEY` / `OKX_SECRET_KEY` /
#'   `OKX_PASSPHRASE` environment variables are used.
#' @param tz Output time zone.
#'
#' @return `data.table` or `NULL`.
#' @export
get_source_data_okx_candle <- function(inst_id, bar, limit = 100L, config = NULL, tz = "UTC") {
  config <- .get_api_config("okx", config = config)
  .require_suggested_package("okxr", "to retrieve OKX candles.")
  dt <- okxr::get_market_candles(inst_id, bar, limit = limit, config = config, tz = tz)
  .normalize_okx_candles(dt, inst_id = inst_id, bar = bar)
}

#' Get Historical OKX Candle Data
#'
#' @param inst_id Instrument identifier.
#' @param bar Candle interval.
#' @param before Optional pagination cursor.
#' @param limit Integer page size.
#' @param config Optional OKX API config. If omitted, defaults from the
#'   package config and `OKX_API_KEY` / `OKX_SECRET_KEY` /
#'   `OKX_PASSPHRASE` environment variables are used.
#' @param tz Output time zone.
#'
#' @return `data.table` or `NULL`.
#' @export
get_source_hist_data_okx_candle <- function(inst_id, bar, before = NULL, limit = 100L, config = NULL, tz = "UTC") {
  config <- .get_api_config("okx", config = config)
  .require_suggested_package("okxr", "to retrieve OKX historical candles.")
  dt <- okxr::get_market_history_candles(inst_id, bar, before = before, limit = limit, config = config, tz = tz)
  .normalize_okx_candles(dt, inst_id = inst_id, bar = bar)
}

#' Get Local OKX Candle Data
#'
#' @param inst_id Instrument identifier.
#' @param bar Candle interval.
#' @param local_path Optional OKX storage path.
#' @param storage Local storage mode: monolithic `"single"` or monthly
#'   partitioned `"monthly"`.
#' @param from,to Optional bounds used to prune monthly partitions before read.
#'
#' @return `data.table` or `NULL`.
#' @export
get_local_okx_candle <- function(inst_id, bar, local_path = NULL,
                                 storage = c("single", "monthly"),
                                 from = NULL, to = NULL) {
  storage <- match.arg(storage)
  if (is.null(local_path)) {
    local_path <- get_source_data_path("crypto", subdir = "okx")
  }
  local_file <- file.path(local_path, sprintf("%s_%s.rds", inst_id, bar))
  if (storage == "monthly") {
    return(get_local_data_partitioned(local_file, "datetime", from = from, to = to, order_cols = "datetime"))
  }
  .read_local_data_table(local_file, sort_cols = "datetime")
}

#' Synchronize Local OKX Candle Data
#'
#' @param inst_id Instrument identifier.
#' @param bar Candle interval.
#' @param config Optional OKX API config. If omitted, defaults from the
#'   package config and `OKX_API_KEY` / `OKX_SECRET_KEY` /
#'   `OKX_PASSPHRASE` environment variables are used.
#' @param local_path Optional OKX storage path.
#' @param mode Either `"latest"` or `"history"`.
#' @param before Optional history cursor.
#' @param limit Integer page size.
#' @param tz Output time zone.
#' @param storage Local storage mode: monolithic `"single"` or monthly
#'   partitioned `"monthly"`.
#'
#' @return A sync result list.
#' @export
sync_local_okx_candle <- function(inst_id, bar, config = NULL, local_path = NULL,
                                  mode = c("latest", "history"), before = NULL,
                                  limit = 100L, tz = "UTC",
                                  storage = c("single", "monthly")) {
  mode <- match.arg(mode)
  storage <- match.arg(storage)
  config <- .get_api_config("okx", config = config)
  if (is.null(local_path)) {
    local_path <- get_source_data_path("crypto", subdir = "okx", create = TRUE)
  }

  local_file_path <- file.path(local_path, sprintf("%s_%s.rds", inst_id, bar))
  if (identical(mode, "history")) {
    new_dt <- get_source_hist_data_okx_candle(inst_id, bar, before = before, limit = limit, config = config, tz = tz)
    source_utime <- NULL
  } else {
    new_dt <- get_source_data_okx_candle(inst_id, bar, limit = limit, config = config, tz = tz)
    source_utime <- get_source_utime_okx_candle(bar = bar, tz = tz)
  }

  sync_fun <- if (storage == "monthly") sync_local_data_partitioned else sync_local_data
  args <- list(
    new_data = new_dt,
    local_file_path = local_file_path,
    key_cols = "datetime",
    order_cols = "datetime",
    source_utime = source_utime
  )
  if (storage == "monthly") args$time_col <- "datetime"
  do.call(sync_fun, args)
}

#' Repair Local OKX Candle Data From Multiple History Pages
#'
#' Fetches multiple OKX historical candle pages in memory and writes the merged
#' repair result to local storage with one `sync_local_data()` call.
#'
#' @param before Character/numeric vector of OKX history pagination cursors.
#' @inheritParams sync_local_okx_candle
#'
#' @return A sync result list.
#' @export
repair_local_okx_candle_gaps <- function(inst_id, bar, before, config = NULL, local_path = NULL,
                                         limit = 100L, tz = "UTC",
                                         storage = c("single", "monthly")) {
  if (missing(before) || is.null(before) || length(before) == 0L) {
    stop("before must contain at least one OKX history pagination cursor.")
  }
  config <- .get_api_config("okx", config = config)
  storage <- match.arg(storage)
  if (is.null(local_path)) {
    local_path <- get_source_data_path("crypto", subdir = "okx", create = TRUE)
  }

  batches <- lapply(before, function(cursor) {
    get_source_hist_data_okx_candle(
      inst_id = inst_id,
      bar = bar,
      before = cursor,
      limit = limit,
      config = config,
      tz = tz
    )
  })

  combined <- data.table::rbindlist(batches, use.names = TRUE, fill = TRUE)
  sync_fun <- if (storage == "monthly") sync_local_data_partitioned else sync_local_data
  args <- list(
    new_data = combined,
    local_file_path = file.path(local_path, sprintf("%s_%s.rds", inst_id, bar)),
    key_cols = "datetime",
    order_cols = "datetime",
    source_utime = NULL
  )
  if (storage == "monthly") args$time_col <- "datetime"
  do.call(sync_fun, args)
}

#' Detect Time Gaps In OKX Candle Data
#'
#' @param dt A candle `data.table`.
#' @param bar Candle interval.
#' @param tolerance Numeric tolerance for fixed-width gap detection.
#'
#' @return A `data.table`.
#' @export
detect_time_gaps_okx_candle <- function(dt, bar = "4H", tolerance = 1e-04) {
  detect_time_gaps(dt, time_col = "datetime", frequency = bar, tolerance = tolerance)
}