# CTF / MGT924 PS3: disagreement-aware ensemble with a prequential portfolio mix.
# Rules read before coding: https://jkpfactors.com/ctf/rules (2026-10-05).
# Independent R implementation. Prior-method credits are in the methodology PDF.
# All dependencies are CRAN packages; main has no file, network, or shell access.

.ctf_config <- function() list(seed=924L, min_months=36L, refit_every=12L,
  tree_months=180L, max_train_rows=100000L, nrounds=500L,
  ridge=0.10, ridge_rows=512L, daily_days=504L, min_daily=126L, risk_factors=20L,
  risk_ridge=1.0, covariance_shrink=0.50, mix_months=120L, mix_min=60L,
  uncertainty=c(0,0.5,1.0), threads=4L)

.ctf_rank <- function(x) {
  good <- is.finite(x); out <- numeric(length(x)); n <- sum(good)
  if (n > 1L) out[good] <- (rank(x[good],ties.method="average")-(n+1)/2)/n
  out
}
.ctf_z <- function(x) {
  x[!is.finite(x)] <- 0; x <- x-mean(x); s <- sqrt(mean(x^2))
  if (s > 1e-12) x/s else x*0
}
.ctf_target <- function(x) {
  good <- is.finite(x); out <- rep(NA_real_,length(x)); s <- sd(x[good])
  out[good] <- if (is.finite(s) && s > 1e-12) atan((x[good]-mean(x[good]))/s) else 0
  out
}
.ctf_flags <- function(x) !is.na(x) & tolower(trimws(as.character(x))) %in% c("1","1.0","true","t")
.ctf_month_key <- function(x) as.integer(format(x,"%Y"))*12L+as.integer(format(x,"%m"))
.ctf_next_eom <- function(x) as.Date(format(as.Date(format(x,"%Y-%m-01"))+62,"%Y-%m-01"))-1
.ctf_rank_matrix <- function(d,f) {
  z <- vapply(d[f],.ctf_rank,numeric(nrow(d)))
  matrix(z,nrow=nrow(d),ncol=length(f),dimnames=list(NULL,f))
}
.ctf_unit_gross <- function(w) {
  w[!is.finite(w)] <- 0; g <- sum(abs(w))
  if (g > 1e-12) 2*w/g else w*0
}

.ctf_tree_fit <- function(meta,dates,get_month,f,t,cfg) {
  now <- dates[t]
  # A cross-sectional target scale must not include any still-unmatured label.
  blocked_months <- unique(meta$eom[meta$mature>now])
  eligible <- which(meta$eom < now & meta$mature <= now &
                      !(meta$eom %in% blocked_months) &
                      .ctf_month_key(meta$eom) >= .ctf_month_key(now)-cfg$tree_months &
                      is.finite(meta$ret_exc_lead1m))
  # Hash sampling uses only eligible historical rows and never their returns.
  priority <- (meta$id[eligible]*104729 + .ctf_month_key(meta$eom[eligible])*13007) %% 2147483647
  take <- eligible[order(priority,meta$eom[eligible],meta$id[eligible],method="radix")]
  take <- head(take,cfg$max_train_rows)
  groups <- split(take,as.character(meta$eom[take]))
  x <- matrix(NA_real_,length(take),length(f),dimnames=list(NULL,f))
  y <- numeric(length(take)); at <- 0L
  for (month in sort(names(groups),method="radix")) {
    d <- get_month(as.Date(month)); ix <- match(meta$id[groups[[month]]],d$id)
    if (anyNA(ix)) stop("Historical sampling keys do not match characteristics")
    block <- as.matrix(d[ix,f,drop=FALSE]); block[!is.finite(block)] <- NA_real_
    out_ix <- at+seq_along(ix)
    x[out_ix,] <- block; y[out_ix] <- .ctf_target(d$ret_exc_lead1m)[ix]
    at <- at+length(ix)
  }
  if (!length(y) || any(!is.finite(y))) stop("No valid mature training targets")
  pars <- list(objective="regression",metric="l2",learning_rate=0.035,
    num_leaves=63L,min_data_in_leaf=max(20L,min(1000L,as.integer(length(y)/100))),
    max_bin=63L,lambda_l2=1.0,verbosity=-1L,num_threads=cfg$threads,
    deterministic=TRUE,force_col_wise=TRUE,seed=cfg$seed,
    data_random_seed=cfg$seed,feature_fraction_seed=cfg$seed,bagging_seed=cfg$seed)
  training <- lightgbm::lgb.Dataset(x,label=y,params=pars,free_raw_data=TRUE)
  fit <- lightgbm::lgb.train(params=pars,data=training,nrounds=cfg$nrounds,
                            verbose=-1L,reset_data=TRUE)
  list(model=fit,rows=length(y),first=min(meta$eom[take]),last=max(meta$eom[take]))
}

.ctf_daily_matrix <- function(d,ids,now,cfg) {
  d <- d[d$date <= now & d$id %in% ids,c("id","date","ret_exc"),drop=FALSE]
  if (!nrow(d)) return(NULL)
  d$date <- as.Date(d$date)
  days <- tail(sort(unique(d$date)),cfg$daily_days)
  d <- d[d$date %in% days,,drop=FALSE]
  d <- d[order(d$date,d$id,method="radix"),,drop=FALSE]
  if (anyDuplicated(d[c("id","date")])) stop("Duplicate daily security-date keys")
  x <- matrix(NA_real_,length(days),length(ids))
  x[cbind(match(d$date,days),match(d$id,ids))] <- d$ret_exc
  observed <- is.finite(x); counts <- colSums(observed)
  usable <- counts >= cfg$min_daily
  if (sum(usable) < 2L) return(NULL)
  x <- x[,usable,drop=FALSE]; observed <- observed[,usable,drop=FALSE]
  counts <- counts[usable]
  x[!observed] <- 0
  average <- colSums(x)/counts
  x <- sweep(x,2,average,"-"); x[!observed] <- 0
  # Correct the variance dilution from missing observations after mean imputation.
  x <- sweep(x,2,sqrt((nrow(x)-1)/pmax(counts-1,1)),"*")
  list(x=x,usable=usable,counts=counts)
}

.ctf_risk_books <- function(tree,ridge,ids,daily,now,cfg) {
  n <- length(ids); nt <- .ctf_z(tree); nr <- .ctf_z(ridge)
  blend <- (nt+nr)/2; disagreement <- (nt-nr)^2
  signals <- cbind(nt,blend,blend/(1+0.5*disagreement),blend/(1+disagreement))
  colnames(signals) <- c("tree","blend","uncertainty_half","uncertainty_one")
  dm <- .ctf_daily_matrix(daily,ids,now,cfg)
  if (is.null(dm)) {
    books <- apply(signals,2,function(v) .ctf_unit_gross(v-mean(v)))
    return(list(books=matrix(books,nrow=n,ncol=4,dimnames=list(NULL,colnames(signals))),
                beta=rep(0,n),eligible=n,fallback=TRUE))
  }
  x <- dm$x; sel <- dm$usable; variance <- colSums(x^2)/(nrow(x)-1)
  positive <- variance[is.finite(variance)&variance>0]
  floor_v <- if (length(positive)) median(positive)*0.05 else 1e-6
  variance <- pmax(variance,floor_v,1e-10)
  k <- min(cfg$risk_factors,as.integer(min(nrow(x),ncol(x))/3),nrow(x)-2L,ncol(x)-2L)
  b <- matrix(0,ncol(x),0)
  if (k >= 1L && sum(x^2) > 1e-12) {
    set.seed(cfg$seed+as.integer(now))
    pc <- irlba::irlba(x,nv=k,nu=0L,tol=1e-5,maxit=300L,
                      v=rep(1/sqrt(ncol(x)),ncol(x)))
    b <- sweep(pc$v,2,pc$d/sqrt(nrow(x)-1),"*")
  }
  common <- rowSums(b^2)
  residual <- pmax(variance-common,floor_v)
  diagonal <- (1-cfg$covariance_shrink)*residual + cfg$covariance_shrink*variance +
                cfg$risk_ridge*mean(variance)
  b <- b*sqrt(1-cfg$covariance_shrink)
  inv <- function(rhs) {
    dx <- rhs/diagonal
    if (!ncol(b)) return(dx)
    db <- b/diagonal
    dx-db %*% solve(diag(ncol(b))+crossprod(b,db),crossprod(b,dx))
  }
  market <- rowMeans(x); vm <- mean(market^2)
  beta <- if (vm>1e-12) drop(crossprod(x,market))/length(market)/vm else rep(1,ncol(x))
  constraints <- cbind(1,beta)
  if (qr(constraints)$rank < ncol(constraints)) constraints <- matrix(1,ncol(x),1)
  a <- inv(signals[sel,,drop=FALSE]); ic <- inv(constraints)
  a <- a-ic %*% solve(crossprod(constraints,ic),crossprod(constraints,a))
  a <- apply(a,2,.ctf_unit_gross)
  books <- matrix(0,n,4,dimnames=list(NULL,colnames(signals)))
  books[sel,] <- matrix(a,nrow=sum(sel),ncol=4)
  full_beta <- numeric(n); full_beta[sel] <- beta
  list(books=books,beta=full_beta,eligible=sum(sel),fallback=FALSE)
}

.ctf_mix <- function(realized,cfg) {
  equal <- rep(1/ncol(realized),ncol(realized))
  realized <- realized[apply(realized,1,function(z) all(is.finite(z))),,drop=FALSE]
  realized <- tail(realized,cfg$mix_months)
  if (nrow(realized)<cfg$mix_min) return(equal)
  v <- cov(realized); scale <- max(mean(diag(v)),1e-10)
  stabilized <- 0.5*v+0.5*diag(diag(v),ncol(v))+diag(0.1*scale,ncol(v))
  a <- pmax(drop(solve(stabilized,colMeans(realized))),0)
  if (sum(a)<1e-12) return(equal)
  # Half prior diversification, half learned positive mean-variance allocation.
  0.5*equal+0.5*a/sum(a)
}

.ctf_engine <- function(meta,feature_names,get_month,get_daily,cfg=.ctf_config(),callback=NULL) {
  set.seed(cfg$seed)
  meta <- as.data.frame(meta)
  meta$eom <- as.Date(meta$eom)
  if (anyNA(meta$id)||anyNA(meta$eom)||anyDuplicated(meta[c("id","eom")])) stop("Invalid monthly keys")
  meta <- meta[order(meta$eom,meta$id,method="radix"),,drop=FALSE]
  rownames(meta) <- NULL
  dates <- sort(unique(meta$eom)); month_rows <- split(seq_len(nrow(meta)),match(meta$eom,dates))
  meta$mature <- .ctf_next_eom(meta$eom)
  if ("eom_ret" %in% names(meta)) {
    given <- as.Date(meta$eom_ret)
    if (anyNA(given)) stop("Missing target realization dates")
    meta$mature <- pmax(meta$mature,given)
  }
  flags <- if ("ctff_test" %in% names(meta)) .ctf_flags(meta$ctff_test) else rep(TRUE,nrow(meta))
  validation_fallback <- !any(flags)
  p <- length(feature_names); gram <- matrix(0,p,p); rhs <- numeric(p); learned <- 0L
  pending <- vector("list",length(dates)); used <- rep(FALSE,length(dates))
  previous_books <- vector("list",length(dates)); returns <- matrix(NA_real_,length(dates),4L)
  out <- vector("list",length(dates)); model <- NULL; fit_description <- NULL
  for (t in seq_along(dates)) {
    now <- dates[t]; d <- get_month(now)
    d <- d[order(d$id,method="radix"),,drop=FALSE]
    if (anyDuplicated(d$id)) stop("Duplicate current identifiers")
    if (!all(feature_names %in% names(d))) stop("Missing supplied characteristics")
    ix <- month_rows[[t]]; alignment <- match(meta$id[ix],d$id)
    if (anyNA(alignment)||nrow(d)!=length(ix)) stop("Monthly provider keys mismatch")
    d <- d[alignment,,drop=FALSE]
    xm <- .ctf_rank_matrix(d,feature_names)
    pending[[t]] <- xm
    past <- if (t>1L) seq_len(t-1L) else integer()
    for (s in past[!used[past]]) {
      is <- month_rows[[s]]
      if (any(meta$mature[is]>now)) next
      used[s] <- TRUE
      good <- is.finite(meta$ret_exc_lead1m[is])
      if (sum(good)>=2L) {
        sx <- pending[[s]][good,,drop=FALSE]
        sy <- .ctf_rank(meta$ret_exc_lead1m[is][good])
        # A fixed, return-independent hash sample bounds the ridge update cost.
        # Every supplied feature remains available; no security is picked by performance.
        key <- (meta$id[is][good]*104729+.ctf_month_key(dates[s])*13007) %% 2147483647
        sample_ix <- head(order(key,meta$id[is][good],method="radix"),cfg$ridge_rows)
        sx <- sx[sample_ix,,drop=FALSE]; sy <- sy[sample_ix]
        sx <- sweep(sx,2,colMeans(sx),"-")
        gram <- gram+crossprod(sx)/nrow(sx)
        rhs <- rhs+drop(crossprod(sx,sy))/nrow(sx)
        learned <- learned+1L
      }
      pending[s] <- list(NULL)
      if (!is.null(previous_books[[s]])&&all(good)) {
        returns[s,] <- drop(crossprod(previous_books[[s]],meta$ret_exc_lead1m[is]))
      }
      previous_books[s] <- list(NULL)
    }
    ridge_prediction <- numeric(nrow(d)); tree_prediction <- numeric(nrow(d))
    w <- numeric(nrow(d)); baseline <- numeric(nrow(d)); mix <- rep(0.25,4)
    rb <- list(eligible=0L,fallback=TRUE,beta=rep(0,nrow(d)))
    refit <- FALSE
    if (learned>=cfg$min_months) {
      coefficient <- solve(gram/learned+diag(cfg$ridge,p),rhs/learned)
      ridge_prediction <- drop(xm %*% coefficient)
      baseline <- .ctf_unit_gross(.ctf_rank(ridge_prediction))
      if (is.null(model)||(t-cfg$min_months-1L)%%cfg$refit_every==0L) {
        fit_description <- .ctf_tree_fit(meta,dates,get_month,feature_names,t,cfg)
        model <- fit_description$model; refit <- TRUE
      }
      raw <- as.matrix(d[feature_names]); raw[!is.finite(raw)] <- NA_real_
      tree_prediction <- as.numeric(predict(model,raw))
      if (any(!is.finite(tree_prediction))) stop("Nonfinite tree forecasts")
      daily <- get_daily(now,d$id,cfg$daily_days)
      rb <- .ctf_risk_books(tree_prediction,ridge_prediction,d$id,daily,now,cfg)
      previous_books[[t]] <- rb$books
      mix <- .ctf_mix(returns[seq_len(t-1L),,drop=FALSE],cfg)
      w <- .ctf_unit_gross(drop(rb$books %*% mix))
    }
    output_id <- d$id
    # Preserve identifier values while returning native integers when safe.
    # Keep composite/64-bit identifiers in their original representation.
    if (is.double(output_id) && !inherits(output_id,"integer64") &&
        all(is.finite(output_id)) && all(output_id==trunc(output_id)) &&
        all(abs(output_id)<=.Machine$integer.max)) output_id <- as.integer(output_id)
    frame <- data.frame(id=output_id,eom=rep(now,nrow(d)),w=w)
    keep <- flags[ix] | validation_fallback
    if (any(keep)) out[[t]] <- frame[keep,,drop=FALSE]
    if (!is.null(callback)) callback(t,now,d,frame,baseline,ridge_prediction,
                                    tree_prediction,rb,mix,refit,fit_description)
    if (is.null(callback)&&(t%%12L==0L||refit)) {
      message("CTF formation ",now,"; learned months ",learned,"; refit ",refit)
    }
  }
  output <- do.call(rbind,out); rownames(output) <- NULL
  if (is.null(output)||!nrow(output)||anyNA(output)||any(!is.finite(output$w))) stop("Invalid output")
  output
}

main <- function(chars, features, daily_ret) {
  cfg <- .ctf_config()
  for (pkg in c("lightgbm","irlba","data.table")) {
    if (!requireNamespace(pkg,quietly=TRUE)) stop("Missing dependency: ",pkg)
  }
  feature_names <- sort(unique(as.character(features$features)),method="radix")
  if (!length(feature_names)||anyNA(feature_names)||any(grepl("lead",feature_names,ignore.case=TRUE))||
      any(feature_names %in% c("id","eom","eom_ret","ctff_test"))) stop("Invalid supplied feature list")
  required <- c("id","eom","ret_exc_lead1m")
  if (!all(required %in% names(chars))||!all(feature_names %in% names(chars))) stop("Invalid characteristic schema")
  # Local providers and the full-data main execute exactly the same engine.
  dt <- data.table::as.data.table(chars)
  if (!inherits(dt$eom,"Date")) dt[,eom:=as.Date(eom)]
  data.table::setindexv(dt,"eom")
  dr <- data.table::as.data.table(daily_ret)
  if (!all(c("id","date","ret_exc") %in% names(dr))) stop("Invalid daily schema")
  if (!inherits(dr$date,"Date")) dr[,date:=as.Date(date)]
  data.table::setindexv(dr,"date")
  meta_columns <- intersect(c("id","eom","eom_ret","ret_exc_lead1m","ctff_test"),names(dt))
  meta <- as.data.frame(dt[,..meta_columns])
  get_month <- function(now) as.data.frame(dt[eom==now])
  get_daily <- function(now,ids,days) {
    # Three calendar years cover the fixed 504 trading-day window.
    as.data.frame(dr[date>=now-1100 & date<=now & id %in% ids])
  }
  .ctf_engine(meta,feature_names,get_month,get_daily,cfg)
}