Add pairwise K16 CTT dominance objective

workspace/scripts/eval_learned_dominance_selector.py

@@ -119,6 +119,22 @@ def main(argv: list[str] | None = None) -> int:
             "threshold per visible task_id bucket using calibration rows only."
         ),
     )
+    parser.add_argument(
+        "--fit-objective",
+        choices=("pointwise", "pairwise", "hybrid_pairwise"),
+        default="pointwise",
+        help=(
+            "Fit the linear utility proxy from candidate-level margins, "
+            "within-chart pairwise causal comparisons, or both. Pairwise "
+            "comparisons are built from calibration rows only."
+        ),
+    )
+    parser.add_argument(
+        "--pairwise-weight",
+        type=float,
+        default=1.0,
+        help="Relative weight for pairwise rows when --fit-objective=hybrid_pairwise.",
+    )
     parser.add_argument("--bootstrap-samples", type=int, default=1000)
     args = parser.parse_args(argv)
 
@@ -127,6 +143,8 @@ def main(argv: list[str] | None = None) -> int:
     lambdas = [float(item.strip()) for item in args.ridge_lambdas.split(",") if item.strip()]
     if not lambdas or any(value < 0.0 for value in lambdas):
         parser.error("--ridge-lambdas must contain non-negative values")
+    if args.pairwise_weight <= 0.0:
+        parser.error("--pairwise-weight must be positive")
 
     out_dir = args.out_dir
     out_dir.mkdir(parents=True, exist_ok=True)
@@ -171,6 +189,8 @@ def main(argv: list[str] | None = None) -> int:
         calibration_dataset,
         lambdas=lambdas,
         threshold_scope=args.threshold_scope,
+        fit_objective=args.fit_objective,
+        pairwise_weight=args.pairwise_weight,
     )
     eval_cases = _evaluate_dataset(eval_dataset, best["weights"], best["mean"], best["std"], tau=best["tau"])
     calibration_cases = _evaluate_dataset(
@@ -205,12 +225,15 @@ def main(argv: list[str] | None = None) -> int:
         "k": args.k,
         "feature_set": args.feature_set,
         "target": args.target,
+        "fit_objective": args.fit_objective,
+        "pairwise_weight": args.pairwise_weight,
         "threshold_scope": args.threshold_scope,
         "chart_feature_mode": chart_feature_mode,
         "feature_names": _feature_names(args.feature_set),
         "ridge_lambdas": lambdas,
         "selected_lambda": best["lambda"],
         "tau": best["tau"],
+        "fit_design": best["fit_design"],
         "weights": best["weights"].tolist(),
         "feature_mean": best["mean"].tolist(),
         "feature_std": best["std"].tolist(),
@@ -248,6 +271,7 @@ def main(argv: list[str] | None = None) -> int:
     (out_dir / "report.md").write_text(_report(metrics) + "\n")
     (out_dir / "train.log").write_text(
         "trained ridge dominance calibrator on calibration measured rows only\n"
+        f"fit_objective={args.fit_objective}\n"
         f"selected_lambda={best['lambda']}\n"
         f"tau={_format_tau(best['tau'])}\n"
         f"calibration_input={args.calibration_input}\n"
@@ -628,24 +652,40 @@ def _fit_select_ridge(
     *,
     lambdas: list[float],
     threshold_scope: str = "global",
+    fit_objective: str = "pointwise",
+    pairwise_weight: float = 1.0,
 ) -> dict[str, Any]:
     samples = dataset["samples"]
     if not samples:
         raise ValueError("cannot fit learned dominance selector without candidates")
-    x = np.stack([sample["feature"] for sample in samples], axis=0)
-    y = np.asarray([float(sample["target_margin"]) for sample in samples], dtype=float)
-    mean = np.zeros(x.shape[1], dtype=float)
-    std = np.ones(x.shape[1], dtype=float)
-    mean[1:] = x[:, 1:].mean(axis=0)
-    std[1:] = x[:, 1:].std(axis=0) + 1.0e-6
-    x_norm = (x - mean) / std
-    x_norm[:, 0] = 1.0
+    if fit_objective not in {"pointwise", "pairwise", "hybrid_pairwise"}:
+        raise ValueError(f"unknown fit_objective: {fit_objective}")
+    point_x = np.stack([sample["feature"] for sample in samples], axis=0)
+    mean = np.zeros(point_x.shape[1], dtype=float)
+    std = np.ones(point_x.shape[1], dtype=float)
+    mean[1:] = point_x[:, 1:].mean(axis=0)
+    std[1:] = point_x[:, 1:].std(axis=0) + 1.0e-6
+    x_norm, y, fit_design = _fit_design_matrix(
+        dataset,
+        fit_objective=fit_objective,
+        pairwise_weight=pairwise_weight,
+        mean=mean,
+        std=std,
+    )
+    if x_norm.shape[0] == 0:
+        raise ValueError("fit objective produced no training rows")
+    candidate_x_norm = _normalized_candidate_features(dataset, mean=mean, std=std)
+    fit_design = {
+        **fit_design,
+        "num_candidate_rows": int(candidate_x_norm.shape[0]),
+        "num_fit_rows": int(x_norm.shape[0]),
+    }
     best: dict[str, Any] | None = None
     for ridge_lambda in lambdas:
         penalty = ridge_lambda * np.eye(x_norm.shape[1], dtype=float)
         penalty[0, 0] = 0.0
         weights = np.linalg.pinv(x_norm.T @ x_norm + penalty) @ (x_norm.T @ y)
-        predictions = x_norm @ weights
+        predictions = candidate_x_norm @ weights
         tau, selection = _choose_thresholds(
             dataset,
             predictions,
@@ -666,11 +706,94 @@ def _fit_select_ridge(
                 "std": std,
                 "tau": tau,
                 "selection": selection,
+                "fit_design": fit_design,
             }
     assert best is not None
     return best
 
 
+def _normalized_candidate_features(
+    dataset: dict[str, Any],
+    *,
+    mean: np.ndarray,
+    std: np.ndarray,
+) -> np.ndarray:
+    x = np.stack([sample["feature"] for sample in dataset["samples"]], axis=0)
+    x_norm = (x - mean) / std
+    x_norm[:, 0] = 1.0
+    return x_norm
+
+
+def _fit_design_matrix(
+    dataset: dict[str, Any],
+    *,
+    fit_objective: str,
+    pairwise_weight: float,
+    mean: np.ndarray,
+    std: np.ndarray,
+) -> tuple[np.ndarray, np.ndarray, dict[str, Any]]:
+    point_x = np.stack([sample["feature"] for sample in dataset["samples"]], axis=0)
+    point_y = np.asarray([float(sample["target_margin"]) for sample in dataset["samples"]], dtype=float)
+    point_x_norm = (point_x - mean) / std
+    point_x_norm[:, 0] = 1.0
+    if fit_objective == "pointwise":
+        return point_x_norm, point_y, {
+            "fit_objective": fit_objective,
+            "num_pointwise_rows": int(point_x.shape[0]),
+            "num_pairwise_rows": 0,
+            "pairwise_weight": 0.0,
+        }
+
+    pair_x, pair_y = _pairwise_design_matrix(dataset)
+    pair_x_norm = pair_x / std
+    pair_x_norm[:, 0] = 0.0
+    if fit_objective == "pairwise":
+        return pair_x_norm, pair_y, {
+            "fit_objective": fit_objective,
+            "num_pointwise_rows": 0,
+            "num_pairwise_rows": int(pair_x.shape[0]),
+            "pairwise_weight": 1.0,
+        }
+
+    if fit_objective == "hybrid_pairwise":
+        scale = math.sqrt(float(pairwise_weight))
+        x = np.concatenate([point_x_norm, pair_x_norm * scale], axis=0)
+        y = np.concatenate([point_y, pair_y * scale], axis=0)
+        return x, y, {
+            "fit_objective": fit_objective,
+            "num_pointwise_rows": int(point_x.shape[0]),
+            "num_pairwise_rows": int(pair_x.shape[0]),
+            "pairwise_weight": float(pairwise_weight),
+        }
+
+    raise ValueError(f"unknown fit_objective: {fit_objective}")
+
+
+def _pairwise_design_matrix(dataset: dict[str, Any]) -> tuple[np.ndarray, np.ndarray]:
+    x_rows: list[np.ndarray] = []
+    y_rows: list[float] = []
+    for sample_indices in dataset["by_row"].values():
+        for left_pos, left_index in enumerate(sample_indices):
+            left = dataset["samples"][left_index]
+            left_target = float(left["target_margin"])
+            for right_index in sample_indices[left_pos + 1 :]:
+                right = dataset["samples"][right_index]
+                right_target = float(right["target_margin"])
+                target_delta = left_target - right_target
+                if abs(target_delta) <= 1.0e-9:
+                    continue
+                feature_delta = np.asarray(left["feature"], dtype=float) - np.asarray(
+                    right["feature"], dtype=float
+                )
+                x_rows.append(feature_delta)
+                y_rows.append(target_delta)
+                x_rows.append(-feature_delta)
+                y_rows.append(-target_delta)
+    if not x_rows:
+        raise ValueError("pairwise objective requires at least one non-tied within-row comparison")
+    return np.stack(x_rows, axis=0), np.asarray(y_rows, dtype=float)
+
+
 def _choose_thresholds(
     dataset: dict[str, Any],
     predictions: np.ndarray,
@@ -921,6 +1044,7 @@ def _report(metrics: dict[str, Any]) -> str:
         f"Selected ridge lambda: `{metrics['selected_lambda']}`",
         f"Tau: `{_format_tau(metrics['tau'])}`",
         f"Threshold scope: `{metrics.get('threshold_scope', 'global')}`",
+        f"Fit objective: `{metrics.get('fit_objective', 'pointwise')}`",
         f"Feature set: `{metrics['feature_set']}`",
         f"Target: `{metrics['target']}`",
         "",