"""Self-contained CCS contest implementation. No files or network are used.""" import numpy as np import pandas as pd from scipy.linalg import eigh MIN_TRAIN_MONTHS = 12 GROSS_EXPOSURE = 2.0 MISSING_FRACTION = 1.0 def _test_flags(values): """WRDS exports the flag as text; the execution service may pass bools.""" normalized = values.astype("string").str.strip().str.lower() result = normalized.map({"true": True, "t": True, "1": True, "false": False, "f": False, "0": False}) if result.isna().any(): raise ValueError("Unrecognized or missing ctff_test flag") return result.astype(bool) def _records(chars, features): required = {"id", "eom", "ctff_test", "ret_exc_lead1m"} if not required.issubset(chars.columns) or "features" not in features: raise ValueError("CTF input schema is incomplete") names = sorted(features["features"].tolist()) if not names or len(set(names)) != len(names) or any(not isinstance(c, str) for c in names): raise ValueError("Expected a nonempty, unique official feature list") if set(names) & required or not set(names).issubset(chars.columns): raise ValueError("Official features overlap metadata or are absent") data = chars[["id", "eom", "ctff_test", "ret_exc_lead1m"] + names].copy() data["eom"] = pd.to_datetime(data["eom"], errors="raise") if data[["id", "eom", "ctff_test"]].isna().any().any(): raise ValueError("Missing identifiers, dates, or test flags") data["ctff_test"] = _test_flags(data["ctff_test"]) if not data["eom"].dt.is_month_end.all(): raise ValueError("Formation dates must be month ends") numeric_ids = pd.to_numeric(data["id"], errors="raise") if not np.equal(numeric_ids, np.floor(numeric_ids)).all(): raise ValueError("Security IDs must be integers") data["id"] = numeric_ids.astype(np.int64) if data.duplicated(["id", "eom"]).any(): raise ValueError("Duplicate security-month observations") data = data.sort_values(["eom", "id"], kind="stable") records = [] for date, group in data.groupby("eom", sort=True): raw = group[names].astype(np.float64).replace([np.inf, -np.inf], np.nan) eligible = (raw.isna().mean(axis=1) <= MISSING_FRACTION).to_numpy() raw = raw.loc[eligible] count = raw.notna().sum() ranked = (raw.rank(method="average") - 1).divide((count - 1).where(count > 1)) - 0.5 z = ranked.fillna(0.0).to_numpy(dtype=np.float64) y = pd.to_numeric(group.loc[eligible, "ret_exc_lead1m"], errors="raise").to_numpy(np.float64) # y is NEVER used to select, rank, standardize, or invert characteristics. records.append(dict(date=date, ids=group["id"].to_numpy(), scored=group["ctff_test"].to_numpy(dtype=bool), eligible=eligible, z=z, y=y)) return records, names def _standardize(z): if len(z) == 0: return z.copy() sd = z.std(axis=0, ddof=0) return (z - z.mean(axis=0)) / np.where(sd > 0, sd, 1.0) def _lw_inverse(x): n, p = x.shape g = x.T @ x / n d, v = eigh((g + g.T) / 2, driver='evd') d = np.maximum(d, 0) mu = d.sum() / p norm_g2 = np.sum(d*d) # Same b2 and d2 as AIPT's stock-kernel calculation, algebraically reduced. b2 = (np.mean(np.sum(x*x, axis=1)**2) - norm_g2) / n d2 = max(norm_g2 - p*mu*mu, 0.) alpha = float(np.clip(b2/d2, 0, 1)) if d2 > 0 else 1. live = d > np.finfo(float).eps * n * d.max() denom = (1-alpha)*d[live] + alpha*mu if np.any(denom <= 0): raise ValueError('Degenerate characteristic covariance') return (v[:, live] / denom) @ v[:, live].T, alpha def _ridge_weights(f, kernel=None): """MSRR ridge with the research code's ten-point spectral LOO selector. Past factor returns only. The high-dimensional calculation stays in time-series space. Returns coefficients and the selected penalty. """ f = np.asarray(f, dtype=np.float64) n, p = f.shape if n < 2 or not np.isfinite(f).all(): raise ValueError("Ridge needs at least two finite historical returns") if p <= n: covariance = f.T @ f / n else: covariance = (f @ f.T if kernel is None else kernel) / n values, vectors = eigh((covariance + covariance.T) * 0.5, driver="evd") live = values > 1e-10 if not live.any(): return np.zeros(p), np.nan grid = np.logspace(np.log10(values[live].min()), np.log10(values[live].max()), 10) if p <= n: values, vectors = values[live], vectors[:, live] projected = f @ vectors mean_projection = projected.mean(axis=0) else: values = np.maximum(values, 0.0) projected_ones = vectors.sum(axis=0) best_score, best_penalty, best_index = -np.inf, None, None for i, penalty in enumerate(grid): if p <= n: diagonal = (projected * projected) @ (1.0 / (values + penalty)) / n row_sum = projected @ (mean_projection / (values + penalty)) else: smoother = values / (values + penalty) diagonal = (vectors * vectors) @ smoother row_sum = vectors @ (smoother * projected_ones) loo = (row_sum - diagonal) / np.maximum(1.0 - diagonal, 1e-14) second_moment = np.mean(loo * loo) score = float(loo.mean() / np.sqrt(second_moment)) if second_moment > 0 else -np.inf if np.isfinite(score) and score > best_score: best_score, best_penalty, best_index = score, float(penalty), i if best_index is None: return np.zeros(p), np.nan if p <= n: weights = vectors @ (mean_projection / (values + best_penalty)) else: dual = vectors @ (projected_ones / (values + best_penalty)) / n weights = f.T @ dual return weights, best_penalty def _output(record, eligible_weights): weights = np.zeros(len(record["ids"]), dtype=np.float64) if not np.isfinite(eligible_weights).all(): raise ValueError("Nonfinite stock weights") gross = float(np.abs(eligible_weights).sum()) if gross > 0: weights[record["eligible"]] = eligible_weights * (GROSS_EXPOSURE / gross) take = record["scored"] return pd.DataFrame({"id": record["ids"][take], "eom": record["date"], "w": weights[take]}) def _finish(chunks): if not chunks: raise ValueError("No ctff_test rows in input") output = pd.concat(chunks, ignore_index=True).sort_values(["eom", "id"], kind="stable") if output.empty or not np.isfinite(output["w"]).all(): raise ValueError("Empty or nonfinite portfolio output") return output.reset_index(drop=True) def run_ccs(chars, features, daily_ret): records, names = _records(chars, features) gram_sum = np.zeros((len(names), len(names))) gram_count = 0 history, maturity, chunks = [], [], [] for record in records: z = _standardize(record["z"]) n = len(z) w = np.zeros(n) if n >= 2: gram = z.T @ z / n inverse = _lw_inverse(z)[0] if gram_count: average = gram_sum / gram_count usable = [i for i, date in enumerate(maturity) if date <= record["date"]] if len(usable) >= MIN_TRAIN_MONTHS: coefficient, _ = _ridge_weights(np.asarray(history)[usable]) w = z @ (inverse @ (average @ coefficient)) / n # Only later dates can use this return. Incomplete return # months are omitted from training, never from formation. if np.isfinite(record["y"]).all(): history.append(average @ (inverse @ (z.T @ record["y"] / n))) maturity.append(record["date"] + pd.offsets.MonthEnd(1)) gram_sum += gram gram_count += 1 if record["scored"].any(): chunks.append(_output(record, w)) return _finish(chunks) def main(chars: pd.DataFrame, features: pd.DataFrame, daily_ret: pd.DataFrame) -> pd.DataFrame: return run_ccs(chars, features, daily_ret)