Artifex/hyperscalper_feature_gap_analysis_v1.py

#!/usr/bin/env python3
"""Read-only HYPERSCALPER_FEATURE_GAP_ANALYSIS_V1 evidence runner.

This program never imports a strategy runtime, invokes an optimizer, or writes to
an input artifact.  It only summarizes frozen oracle and historical evidence.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sqlite3
from collections import Counter
from pathlib import Path
from typing import Any

import numpy as np

ARTIFACT = "HYPERSCALPER_FEATURE_GAP_ANALYSIS_V1"
EXPECTED_FEATURES = 711
MAX_SAMPLE_ROWS = 16_384
LINEAGE_ROLES = ("trend", "signal", "trigger", "confirm", "vol")
DEQ_RETURN_FIELDS = tuple(f"return_{bars}_bars_bps" for bars in (1, 2, 3, 5, 10))
DEQ_MEAN_FIELDS = ("mfe_bps", "mae_bps", "time_to_positive_bars")
DEQ_PRIMITIVE_FIELDS = DEQ_RETURN_FIELDS + (
    "mfe_bps",
    "mae_bps",
    "time_to_positive_bars",
    "time_to_25_bps_bars",
    "time_to_50_bps_bars",
    "max_adverse_before_positive_bps",
    "winner_negative_first",
    "recovery_bars",
)


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def source_hash(path: Path) -> str:
    if path.is_file():
        return sha256(path)
    digest = hashlib.sha256()
    for item in sorted(path.glob("*.npy")):
        digest.update(item.name.encode("utf-8"))
        digest.update(b"\0")
        digest.update(bytes.fromhex(sha256(item)))
    return digest.hexdigest()


def load_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"{path} must contain a JSON object")
    return value


def feature_rows(path: Path) -> list[dict[str, Any]]:
    payload = load_json(path)
    rows = payload.get("primitives") or payload.get("requests")
    if not isinstance(rows, list):
        raise ValueError("engineering map/request must contain primitives or requests")
    result = []
    seen = set()
    for row in rows:
        key = (int(row["indicator_id"]), int(row["period"]), float(row["p1"]))
        if key in seen:
            raise ValueError(f"duplicate feature primitive: {key}")
        seen.add(key)
        result.append(
            {
                "feature_key": f"{key[0]}:{key[1]}:{key[2]:g}",
                "indicator_id": key[0],
                "period": key[1],
                "p1": key[2],
                "request_id": row.get("request_id"),
            }
        )
    if len(result) != EXPECTED_FEATURES:
        raise ValueError(
            f"source map must contain exactly {EXPECTED_FEATURES} unique primitives; "
            f"found {len(result)}"
        )
    return result


def accepted_ids(manifest: dict[str, Any]) -> set[str]:
    coverage = manifest.get("coverage", {})
    values = coverage.get("feature_versions", manifest.get("accepted_request_ids", []))
    return {str(value) for value in values} if isinstance(values, list) else set()


def checkpoint_columns(
    checkpoint: Path, rows: list[dict[str, Any]]
) -> tuple[np.ndarray, list[str]]:
    with np.load(checkpoint, allow_pickle=False) as archive:
        names = list(archive.files)
        by_request = {str(row["request_id"]): row for row in rows if row["request_id"]}
        if set(names) == set(by_request):
            ordered_names = [str(row["request_id"]) for row in rows]
        elif len(names) == EXPECTED_FEATURES:
            # A frozen raw NPZ may lack request IDs; source-map order is the only asserted mapping.
            ordered_names = sorted(names)
        else:
            raise ValueError(
                "checkpoint keys do not match request IDs and do not contain 711 arrays"
            )
        arrays = [np.asarray(archive[name], dtype=np.float64).reshape(-1) for name in ordered_names]
    lengths = {array.size for array in arrays}
    if len(lengths) != 1 or not next(iter(lengths)):
        raise ValueError("checkpoint arrays must be non-empty and have equal length")
    return np.column_stack(arrays), ordered_names


def checkpoint_directory_columns(
    checkpoint_dir: Path, rows: list[dict[str, Any]]
) -> tuple[np.ndarray, list[str]]:
    files = sorted(checkpoint_dir.glob("*.npy"))
    if len(files) != EXPECTED_FEATURES:
        raise ValueError(
            f"checkpoint directory must contain exactly {EXPECTED_FEATURES} .npy files; "
            f"found {len(files)}"
        )
    by_primitive = {
        (row["indicator_id"], row["period"], row["p1"]): row for row in rows
    }
    mapped: dict[tuple[int, int, float], Path] = {}
    for path in files:
        parts = path.stem.split("_")
        if len(parts) != 4 or parts[0] != "hs22":
            raise ValueError(f"invalid checkpoint filename: {path.name}")
        try:
            primitive = (int(parts[1]), int(parts[2]), float(parts[3]))
        except ValueError as error:
            raise ValueError(f"invalid checkpoint filename: {path.name}") from error
        if primitive not in by_primitive:
            raise ValueError(f"checkpoint filename is not an engineering primitive: {path.name}")
        if primitive in mapped:
            raise ValueError(f"duplicate checkpoint primitive: {primitive}")
        mapped[primitive] = path
    if set(mapped) != set(by_primitive):
        raise ValueError("checkpoint filenames do not cover every engineering primitive")
    ordered_paths = [
        mapped[(row["indicator_id"], row["period"], row["p1"])] for row in rows
    ]
    arrays = [
        np.asarray(np.load(path, allow_pickle=False), dtype=np.float64).reshape(-1)
        for path in ordered_paths
    ]
    lengths = {array.size for array in arrays}
    if len(lengths) != 1 or not next(iter(lengths)):
        raise ValueError("checkpoint arrays must be non-empty and have equal length")
    return np.column_stack(arrays), [path.name for path in ordered_paths]


def sampled_checkpoint_columns(
    checkpoint: Path | None,
    checkpoint_dir: Path | None,
    rows: list[dict[str, Any]],
    sample_rows: int,
) -> tuple[np.ndarray, list[str], int]:
    """Read only selected observations from each checkpoint column.

    This intentionally avoids materializing the 129k by 711 source matrix.  NPZ
    members are decompressed one at a time; directory checkpoints are mmap'd.
    """
    if checkpoint_dir:
        files = sorted(checkpoint_dir.glob("*.npy"))
        if len(files) != EXPECTED_FEATURES:
            raise ValueError("checkpoint directory must contain exactly 711 .npy files")
        by_primitive = {
            (row["indicator_id"], row["period"], row["p1"]): row for row in rows
        }
        mapped: dict[tuple[int, int, float], Path] = {}
        for path in files:
            parts = path.stem.split("_")
            if len(parts) != 4 or parts[0] != "hs22":
                raise ValueError(f"invalid checkpoint filename: {path.name}")
            primitive = (int(parts[1]), int(parts[2]), float(parts[3]))
            if primitive not in by_primitive or primitive in mapped:
                raise ValueError(f"invalid or duplicate checkpoint primitive: {path.name}")
            mapped[primitive] = path
        if set(mapped) != set(by_primitive):
            raise ValueError("checkpoint filenames do not cover every engineering primitive")
        ordered = [mapped[(row["indicator_id"], row["period"], row["p1"])] for row in rows]
        first = np.load(ordered[0], allow_pickle=False, mmap_mode="r").reshape(-1)
        size = first.size
        indices = np.linspace(0, size - 1, min(sample_rows, size), dtype=int)
        arrays = []
        for path in ordered:
            values = np.load(path, allow_pickle=False, mmap_mode="r").reshape(-1)
            if values.size != size:
                raise ValueError("checkpoint arrays must be non-empty and have equal length")
            arrays.append(np.asarray(values[indices], dtype=np.float64))
        return np.column_stack(arrays), [path.name for path in ordered], size

    assert checkpoint is not None
    with np.load(checkpoint, allow_pickle=False) as archive:
        names = list(archive.files)
        by_request = {str(row["request_id"]): row for row in rows if row["request_id"]}
        if set(names) == set(by_request):
            ordered_names = [str(row["request_id"]) for row in rows]
        elif len(names) == EXPECTED_FEATURES:
            ordered_names = sorted(names)
        else:
            raise ValueError("checkpoint keys do not match request IDs and do not contain 711 arrays")
        first = np.asarray(archive[ordered_names[0]]).reshape(-1)
        size = first.size
        if not size:
            raise ValueError("checkpoint arrays must be non-empty and have equal length")
        indices = np.linspace(0, size - 1, min(sample_rows, size), dtype=int)
        arrays = []
        for name in ordered_names:
            values = np.asarray(archive[name], dtype=np.float64).reshape(-1)
            if values.size != size:
                raise ValueError("checkpoint arrays must be non-empty and have equal length")
            arrays.append(values[indices])
    return np.column_stack(arrays), ordered_names, size


def semantic_types(path: Path | None, rows: list[dict[str, Any]]) -> tuple[list[dict[str, str]], str | None]:
    """Resolve output type and domain from a metadata-only semantic map."""
    defaults = [{"output_type": "continuous", "domain": "unclassified"} for _ in rows]
    if path is None:
        return defaults, "no semantic map supplied; output types default to continuous"
    try:
        if path.suffix.lower() == ".parquet":
            import pyarrow.parquet as pq

            items = pq.read_table(path).to_pylist()
        else:
            payload = load_json(path)
            items = payload.get("primitives", payload.get("features", payload.get("rows", payload.get("semantic_map"))))
            if isinstance(items, dict):
                items = items.get("primitives", items.get("features", items.get("rows")))
    except (ImportError, OSError, ValueError) as error:
        return defaults, f"cannot read semantic map: {error}"
    if not isinstance(items, list):
        return defaults, "semantic map has no primitives/features/rows array"
    mapped: dict[str, dict[str, str]] = {}
    for item in items:
        if not isinstance(item, dict):
            continue
        key = item.get("feature_key")
        if key is None and all(name in item for name in ("indicator_id", "period", "p1")):
            key = f"{int(item['indicator_id'])}:{int(item['period'])}:{float(item['p1']):g}"
        if key is None:
            continue
        raw_type = str(item.get("output_type", item.get("semantic_type", item.get("value_type", "continuous")))).lower()
        output_type = "event" if raw_type in {"event", "detection", "binary_event"} else "state" if raw_type in {"state", "categorical", "boolean"} else "continuous"
        domain = str(item.get("domain", item.get("semantic_domain", item.get("engineering_family", item.get("family", "unclassified")))))
        mapped[str(key)] = {"output_type": output_type, "domain": domain}
    result = [mapped.get(row["feature_key"], defaults[index]) for index, row in enumerate(rows)]
    missing = sum(row["feature_key"] not in mapped for row in rows)
    return result, f"semantic map does not classify {missing} features" if missing else None


def rank_finite_columns(matrix: np.ndarray) -> np.ndarray:
    """Return deterministic ordinal ranks, leaving non-finite observations masked."""
    ranks = np.full(matrix.shape, np.nan, dtype=np.float64)
    for column in range(matrix.shape[1]):
        finite = np.isfinite(matrix[:, column])
        count = int(finite.sum())
        if count:
            order = np.argsort(matrix[finite, column], kind="mergesort")
            column_ranks = np.empty(count, dtype=np.float64)
            column_ranks[order] = np.arange(count)
            ranks[finite, column] = column_ranks
    return ranks


def pairwise_correlations(
    values: np.ndarray, finite: np.ndarray, pair_samples: np.ndarray, min_pair_samples: int
) -> np.ndarray:
    """Correlate columns using only rows finite for each pair, without pair materialization."""
    masked_values = np.where(finite, values, 0.0)
    finite_float = finite.astype(np.float64)
    counts = pair_samples.astype(np.float64)
    sums = masked_values.T @ finite_float
    squared_sums = (masked_values * masked_values).T @ finite_float
    covariance = masked_values.T @ masked_values
    with np.errstate(invalid="ignore", divide="ignore"):
        covariance -= sums * sums.T / counts
        variance = squared_sums - sums * sums / counts
        correlation = covariance / np.sqrt(variance * variance.T)
    invalid = (pair_samples < min_pair_samples) | (variance <= 0.0) | (variance.T <= 0.0)
    correlation[invalid] = np.nan
    np.fill_diagonal(correlation, np.nan)
    return correlation


def pairwise_redundancy(
    sample: np.ndarray, min_pair_samples: int
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Calculate pairwise-finite Pearson and pre-ranked ordinal correlations."""
    finite = np.isfinite(sample)
    pair_samples = finite.astype(np.int32).T @ finite.astype(np.int32)
    pearson = pairwise_correlations(sample, finite, pair_samples, min_pair_samples)
    # Ranks are computed once per feature, rather than once per feature pair.
    ranks = rank_finite_columns(sample)
    spearman = pairwise_correlations(ranks, finite, pair_samples, min_pair_samples)
    return pearson, spearman, pair_samples


def type_aware_redundancy(
    sample: np.ndarray, output_types: list[str], min_pair_samples: int
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Use correlations for continuous values and overlap metrics for detections."""
    feature_count = sample.shape[1]
    finite = np.isfinite(sample)
    pair_samples = finite.astype(np.int32).T @ finite.astype(np.int32)
    pearson = np.full((feature_count, feature_count), np.nan)
    spearman = np.full((feature_count, feature_count), np.nan)
    agreement = np.full((feature_count, feature_count), np.nan)
    jaccard = np.full((feature_count, feature_count), np.nan)
    continuous = np.array([kind == "continuous" for kind in output_types])
    if continuous.any():
        indices = np.flatnonzero(continuous)
        continuous_values = sample[:, indices]
        continuous_finite = finite[:, indices]
        continuous_counts = pair_samples[np.ix_(indices, indices)]
        pearson[np.ix_(indices, indices)] = pairwise_correlations(
            continuous_values, continuous_finite, continuous_counts, min_pair_samples
        )
        ranks = rank_finite_columns(continuous_values)
        spearman[np.ix_(indices, indices)] = pairwise_correlations(
            ranks, continuous_finite, continuous_counts, min_pair_samples
        )

    discrete = ~continuous
    if discrete.any():
        indices = np.flatnonzero(discrete)
        values = sample[:, indices]
        valid = finite[:, indices]
        detected = valid & (values != 0)
        intersection = detected.astype(np.int32).T @ detected.astype(np.int32)
        detected_counts = detected.sum(axis=0)
        union = detected_counts[:, None] + detected_counts[None, :] - intersection
        discrete_counts = pair_samples[np.ix_(indices, indices)]
        with np.errstate(invalid="ignore", divide="ignore"):
            discrete_jaccard = intersection / union
        discrete_jaccard[(discrete_counts < min_pair_samples) | (union == 0)] = np.nan
        np.fill_diagonal(discrete_jaccard, np.nan)
        jaccard[np.ix_(indices, indices)] = discrete_jaccard

        binary = np.all(~valid | ((values == 0) | (values == 1)), axis=0)
        zero = valid & (values == 0)
        equal_binary = (
            zero.astype(np.int32).T @ zero.astype(np.int32) + intersection
        )
        with np.errstate(invalid="ignore", divide="ignore"):
            binary_agreement = equal_binary / discrete_counts
        binary_pairs = binary[:, None] & binary[None, :]
        binary_agreement[(discrete_counts < min_pair_samples) | ~binary_pairs] = np.nan
        np.fill_diagonal(binary_agreement, np.nan)
        agreement[np.ix_(indices, indices)] = binary_agreement

        # Categorical state outputs are uncommon; retain exact equality semantics for them.
        for left_offset, left in enumerate(indices):
            if binary[left_offset]:
                continue
            for right_offset, right in enumerate(indices):
                if (
                    right == left
                    or (not binary[right_offset] and right < left)
                    or pair_samples[left, right] < min_pair_samples
                ):
                    continue
                mask = finite[:, left] & finite[:, right]
                value = float(np.mean(sample[mask, left] == sample[mask, right]))
                agreement[left, right] = agreement[right, left] = value
    return pearson, spearman, agreement, jaccard, pair_samples


def primary_redundancy_matrix(
    pearson: np.ndarray, agreement: np.ndarray, jaccard: np.ndarray, output_types: list[str]
) -> np.ndarray:
    result = np.full_like(pearson, np.nan)
    for left in range(len(output_types)):
        for right in range(len(output_types)):
            if output_types[left] == output_types[right] == "continuous":
                result[left, right] = abs(pearson[left, right])
            elif output_types[left] == output_types[right] == "event":
                result[left, right] = jaccard[left, right]
            elif output_types[left] == output_types[right] == "state":
                result[left, right] = agreement[left, right]
    return result


def clusters(correlation: np.ndarray, threshold: float = 0.95) -> list[list[int]]:
    parent = list(range(correlation.shape[0]))

    def find(item: int) -> int:
        while parent[item] != item:
            parent[item] = parent[parent[item]]
            item = parent[item]
        return item

    def join(left: int, right: int) -> None:
        left, right = find(left), find(right)
        if left != right:
            parent[right] = left

    for left in range(correlation.shape[0]):
        for right in range(left):
            if abs(correlation[left, right]) >= threshold:
                join(left, right)
    grouped: dict[int, list[int]] = {}
    for index in range(len(parent)):
        grouped.setdefault(find(index), []).append(index)
    return [members for members in grouped.values() if len(members) > 1]


def parquet_counts(path: Path, label: str) -> tuple[Counter[str], str | None]:
    try:
        import pyarrow.parquet as pq
    except ImportError:
        return Counter(), "pyarrow unavailable"
    try:
        table = pq.read_table(path)
    except Exception as error:
        return Counter(), f"cannot read {label} parquet: {error}"
    names = set(table.column_names)
    key_column = next(
        (name for name in ("feature_key", "primitive", "indicator_id") if name in names), None
    )
    count_column = next(
        (name for name in ("usage_count", "count", "strategy_count") if name in names), None
    )
    if key_column is None:
        return Counter(), f"{label} parquet has no recognized feature key column"
    keys = table[key_column].to_pylist()
    counts = table[count_column].to_pylist() if count_column else [1] * len(keys)
    return Counter(
        {str(key): float(count or 0) for key, count in zip(keys, counts, strict=True)}
    ), None


def lineage_combo(value: Any) -> tuple[tuple[int, int, float], ...]:
    """Parse the five ordered primitive triples stored at the start of lineage combos."""
    if isinstance(value, bytes):
        value = value.decode("utf-8")
    if isinstance(value, str):
        value = json.loads(value)
    if isinstance(value, dict):
        if "combo" in value:
            return lineage_combo(value["combo"])
        if all(role in value for role in LINEAGE_ROLES):
            value = [value[role] for role in LINEAGE_ROLES]
        elif "triples" in value:
            value = value["triples"]
    if not isinstance(value, (list, tuple)):
        raise ValueError("combo is not an array")
    if len(value) < len(LINEAGE_ROLES) * 3:
        raise ValueError("combo does not contain at least five triples")
    # Historical rows append strategy parameters after the five primitives.
    triples = [value[index : index + 3] for index in range(0, len(LINEAGE_ROLES) * 3, 3)]
    result = []
    for triple in triples:
        if not isinstance(triple, (list, tuple)) or len(triple) != 3:
            raise ValueError("combo contains an invalid primitive triple")
        indicator_id, period, p1 = triple
        if isinstance(indicator_id, bool) or isinstance(period, bool):
            raise ValueError("combo primitive IDs must be integers")
        indicator_id, period, p1 = int(indicator_id), int(period), float(p1)
        if not np.isfinite(p1):
            raise ValueError("combo primitive p1 must be finite")
        result.append((indicator_id, period, p1))
    return tuple(result)


def lineage_usage(path: Path) -> tuple[Counter[str], dict[str, Any], str | None]:
    try:
        import pyarrow.parquet as pq
    except ImportError:
        return Counter(), {}, "pyarrow unavailable"
    try:
        table = pq.read_table(path)
    except Exception as error:
        return Counter(), {}, f"cannot read lineage parquet: {error}"
    if "combo_json" not in table.column_names:
        return Counter(), {}, "lineage parquet has no combo_json column"

    primitive_counts: Counter[str] = Counter()
    role_counts: Counter[str] = Counter()
    combo_counts: Counter[str] = Counter()
    strategy_ids: set[str] = set()
    invalid_rows = 0
    records = table.to_pylist()
    for record in records:
        try:
            triples = lineage_combo(record["combo_json"])
        except (TypeError, ValueError, json.JSONDecodeError, UnicodeDecodeError, OverflowError):
            invalid_rows += 1
            continue
        signature = json.dumps(triples, separators=(",", ":"))
        combo_counts[signature] += 1
        strategy_id = record.get("strategy_id")
        if strategy_id is not None:
            strategy_ids.add(str(strategy_id))
        for role, (indicator_id, period, p1) in zip(LINEAGE_ROLES, triples, strict=True):
            primitive_counts[f"{indicator_id}:{period}:{p1:g}"] += 1
            role_counts[role] += 1
    blocker = (
        f"lineage parquet has {invalid_rows} rows with an unsupported combo_json format"
        if invalid_rows
        else None
    )
    evidence = {
        "lineage_rows": len(records),
        "parsed_combo_rows": sum(combo_counts.values()),
        "invalid_combo_rows": invalid_rows,
        "strategy_ids_observed": len(strategy_ids),
        "role_counts": dict(sorted(role_counts.items())),
        "combo_counts": [
            {"combo_signature": signature, "usage_count": count}
            for signature, count in sorted(combo_counts.items(), key=lambda item: (-item[1], item[0]))
        ],
        "feature_family_counts": dict(sorted(combo_counts.items())),
        "primitive_counts": dict(sorted(primitive_counts.items())),
    }
    return primitive_counts, evidence, blocker


def deq_feature_key(value: Any) -> tuple[str, str]:
    """Return the primitive and indicator-family keys exported with a strategy genome."""
    if not isinstance(value, dict):
        raise ValueError("feature triple is not an object")
    indicator_id, period, p1 = value.get("indicator_id"), value.get("period"), value.get("p1")
    if isinstance(indicator_id, bool) or isinstance(period, bool):
        raise ValueError("feature triple IDs must be integers")
    indicator_id, period, p1 = int(indicator_id), int(period), float(p1)
    if not np.isfinite(p1):
        raise ValueError("feature triple p1 must be finite")
    return f"{indicator_id}:{period}:{p1:g}", str(indicator_id)


def deq_strategy_feature_keys(strategy: dict[str, Any]) -> list[tuple[str, str]]:
    """Read exported feature triples, falling back to the verified genome combo schema."""
    try:
        triples = strategy.get("feature_triples")
        if not isinstance(triples, list) or not triples:
            raise ValueError("strategy has no feature_triples")
        return [deq_feature_key(triple) for triple in triples]
    except (TypeError, ValueError, OverflowError) as exported_error:
        genome = strategy.get("genome")
        try:
            if not isinstance(genome, dict):
                raise ValueError("strategy has no genome object")
            return [
                (f"{indicator_id}:{period}:{p1:g}", str(indicator_id))
                for indicator_id, period, p1 in lineage_combo(genome.get("combo"))
            ]
        except (TypeError, ValueError, json.JSONDecodeError, UnicodeDecodeError, OverflowError) as genome_error:
            raise ValueError(
                f"strategy has no usable feature triples or genome combo ({exported_error}; {genome_error})"
            ) from genome_error


def deq_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
    """Summarize signed, already-recorded DEQ primitives without rerunning strategies."""
    result: dict[str, Any] = {
        "status": "AVAILABLE" if samples else "INSUFFICIENT",
        "trade_count": len(samples),
        "fold_coverage": sorted({sample["fold"] for sample in samples}),
        "scenario_coverage": sorted({sample["scenario"] for sample in samples}),
        "returns": {},
        "means": {},
        "primitives": {},
    }
    result["fold_count"] = len(result["fold_coverage"])
    result["scenario_count"] = len(result["scenario_coverage"])
    for field in DEQ_RETURN_FIELDS:
        values = [sample["deq"].get(field) for sample in samples]
        values = [float(value) for value in values if value is not None]
        result["returns"][field] = {
            "available_count": len(values),
            "mean": float(np.mean(values)) if values else None,
            # DEQ returns are signed in the strategy direction. Positive is therefore
            # directionally correct, but says nothing causal about an associated feature.
            "directional_correct_count": sum(value > 0 for value in values),
            "directional_incorrect_count": sum(value < 0 for value in values),
            "directional_neutral_count": sum(value == 0 for value in values),
            "directional_correctness_inferable": bool(values),
        }
    for field in DEQ_MEAN_FIELDS:
        values = [sample["deq"].get(field) for sample in samples]
        values = [float(value) for value in values if value is not None]
        result["means"][field] = {
            "available_count": len(values),
            "mean": float(np.mean(values)) if values else None,
        }
    for field in DEQ_PRIMITIVE_FIELDS:
        values = [sample["deq"].get(field) for sample in samples]
        available = [value for value in values if value is not None]
        if available and all(isinstance(value, bool) for value in available):
            result["primitives"][field] = {
                "available_count": len(available),
                "true_count": sum(available),
                "false_count": len(available) - sum(available),
            }
        else:
            numeric = [float(value) for value in available]
            result["primitives"][field] = {
                "available_count": len(numeric),
                "mean": float(np.mean(numeric)) if numeric else None,
            }
    return result


def cohort_deq_evidence(path: Path) -> tuple[dict[str, Any], list[dict[str, Any]], str | None]:
    """Aggregate the read-only Cohort001 DEQ export by strategy and associated genome inputs."""
    try:
        payload = load_json(path)
        if payload.get("contract") != "Cohort001-DEQ-ledger-export-v1":
            raise ValueError("unexpected contract")
        strategies = payload.get("strategies")
        if not isinstance(strategies, list) or not strategies:
            raise ValueError("strategies must be a non-empty array")
        strategy_summaries = []
        feature_samples: dict[str, list[dict[str, Any]]] = {}
        family_samples: dict[str, list[dict[str, Any]]] = {}
        for strategy in strategies:
            if not isinstance(strategy, dict) or strategy.get("strategy_version_id") is None:
                raise ValueError("strategy has no strategy_version_id")
            feature_keys = deq_strategy_feature_keys(strategy)
            samples = []
            for fold in strategy.get("folds", []):
                if not isinstance(fold, dict) or not isinstance(fold.get("fold"), str):
                    raise ValueError("strategy fold is invalid")
                for scenario in fold.get("scenarios", []):
                    if not isinstance(scenario, dict) or not isinstance(scenario.get("scenario"), str):
                        raise ValueError("strategy scenario is invalid")
                    ledger = scenario.get("deq_samples")
                    if not isinstance(ledger, list):
                        raise ValueError("scenario has no deq_samples array")
                    for row in ledger:
                        if not isinstance(row, dict) or not isinstance(row.get("deq"), dict):
                            raise ValueError("DEQ sample has no deq object")
                        samples.append({"fold": fold["fold"], "scenario": scenario["scenario"], "deq": row["deq"]})
            strategy_id = str(strategy["strategy_version_id"])
            summary = deq_summary(samples)
            strategy_summaries.append({"strategy_version_id": strategy_id, **summary})
            for feature_key, family_key in feature_keys:
                feature_samples.setdefault(feature_key, []).extend(samples)
                family_samples.setdefault(family_key, []).extend(samples)
        feature_summaries = [
            {"feature_key": key, "attribution": "ASSOCIATIVE_NOT_CAUSAL", **deq_summary(samples)}
            for key, samples in sorted(feature_samples.items())
        ]
        family_summaries = [
            {"indicator_id": key, "attribution": "ASSOCIATIVE_NOT_CAUSAL", **deq_summary(samples)}
            for key, samples in sorted(family_samples.items(), key=lambda item: int(item[0]))
        ]
        parquet_rows = [
            {"entity_type": "strategy", "entity_key": item["strategy_version_id"], **deq_parquet_row(item)}
            for item in strategy_summaries
        ] + [
            {"entity_type": "feature", "entity_key": item["feature_key"], **deq_parquet_row(item)}
            for item in feature_summaries
        ] + [
            {"entity_type": "family", "entity_key": item["indicator_id"], **deq_parquet_row(item)}
            for item in family_summaries
        ]
        return {
            "available": True,
            "status": "AVAILABLE",
            "attribution": "ASSOCIATIVE_NOT_CAUSAL",
            "attribution_note": "Multi-feature genome attribution is associative, not causal.",
            "strategy_summaries": strategy_summaries,
            "feature_summaries": feature_summaries,
            "family_summaries": family_summaries,
        }, parquet_rows, None
    except (TypeError, ValueError, json.JSONDecodeError) as error:
        return {}, [], f"cannot use cohort DEQ export: {error}"


def deq_parquet_row(summary: dict[str, Any]) -> dict[str, Any]:
    row = {
        "status": summary["status"],
        "attribution": summary.get("attribution", "STRATEGY_LEVEL"),
        "trade_count": summary["trade_count"],
        "fold_count": len(summary["fold_coverage"]),
        "scenario_count": len(summary["scenario_coverage"]),
        "fold_coverage": json.dumps(summary["fold_coverage"]),
        "scenario_coverage": json.dumps(summary["scenario_coverage"]),
    }
    for field, values in summary["returns"].items():
        row[f"{field}_count"] = values["available_count"]
        row[f"{field}_mean"] = values["mean"]
        row[f"{field}_directional_correct_count"] = values["directional_correct_count"]
        row[f"{field}_directional_incorrect_count"] = values["directional_incorrect_count"]
    for field, values in summary["means"].items():
        row[f"{field}_count"] = values["available_count"]
        row[f"{field}_mean"] = values["mean"]
    for field, values in summary["primitives"].items():
        if "true_count" in values:
            row[f"{field}_true_count"] = values["true_count"]
            row[f"{field}_false_count"] = values["false_count"]
        else:
            row.setdefault(f"{field}_count", values["available_count"])
            row.setdefault(f"{field}_mean", values["mean"])
    return row


def failure_mining(report: Path | None, database: Path | None) -> tuple[Counter[str], list[str]]:
    counts: Counter[str] = Counter()
    blockers: list[str] = []
    if report:

        def walk(value: Any) -> None:
            if isinstance(value, dict):
                for key, item in value.items():
                    if (
                        key.lower() in {"status", "outcome", "failure_reason", "reason"}
                        and isinstance(item, str)
                        and any(
                            word in item.lower() for word in ("fail", "reject", "error", "block")
                        )
                    ):
                        counts[item] += 1
                    walk(item)
            elif isinstance(value, list):
                for item in value:
                    walk(item)

        walk(load_json(report))
    if database:
        try:
            connection = sqlite3.connect(f"file:{database.resolve().as_posix()}?mode=ro", uri=True)
            with connection:
                for (table,) in connection.execute(
                    "select name from sqlite_master where type='table'"
                ):
                    columns = [
                        row[1] for row in connection.execute(f'pragma table_info("{table}")')
                    ]
                    status = next(
                        (
                            column
                            for column in columns
                            if column.lower() in {"status", "outcome", "failure_reason", "reason"}
                        ),
                        None,
                    )
                    if status:
                        for (value,) in connection.execute(
                            f'select "{status}" from "{table}" where "{status}" is not null'
                        ):
                            text = str(value)
                            if any(
                                word in text.lower()
                                for word in ("fail", "reject", "error", "block")
                            ):
                                counts[text] += 1
        except sqlite3.Error as error:
            blockers.append(f"cannot read qualification DB export read-only: {error}")
    return counts, blockers


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument("--engineering-map", type=Path)
    source.add_argument("--oracle-request", type=Path)
    checkpoint = parser.add_mutually_exclusive_group(required=True)
    checkpoint.add_argument("--oracle-checkpoint", type=Path, help="frozen .npz checkpoint")
    checkpoint.add_argument(
        "--oracle-checkpoint-dir",
        type=Path,
        help="directory of frozen hs22_{id}_{period}_{p1}.npy checkpoints",
    )
    parser.add_argument("--acceptance-manifest", type=Path, required=True)
    parser.add_argument(
        "--semantic-map", type=Path,
        help="metadata map with feature_key or primitive fields and output_type/domain",
    )
    parser.add_argument(
        "--historical-usage", type=Path, required=True, help="historical usage parquet"
    )
    parser.add_argument("--lineage", type=Path, required=True, help="lineage parquet")
    parser.add_argument("--qualification-report", type=Path)
    parser.add_argument("--qualification-db-export", type=Path)
    parser.add_argument(
        "--cohort-deq",
        type=Path,
        help="read-only Cohort001 DEQ JSON emitted by export_cohort_deq_v1.py",
    )
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--sample-rows", type=int, default=8192)
    parser.add_argument(
        "--min-pair-samples",
        type=int,
        default=2,
        help="minimum jointly finite sampled observations required for a feature pair",
    )
    args = parser.parse_args()
    if args.sample_rows < 2:
        parser.error("--sample-rows must be at least 2")
    if args.sample_rows > MAX_SAMPLE_ROWS:
        parser.error(
            f"--sample-rows may not exceed {MAX_SAMPLE_ROWS}; exact pairwise Spearman is "
            "intentionally not offered because it is not bounded for 711 x 129k inputs"
        )
    if args.min_pair_samples < 2:
        parser.error("--min-pair-samples must be at least 2")
    inputs = [
        path
        for path in (
            args.engineering_map,
            args.oracle_request,
            args.oracle_checkpoint,
            args.oracle_checkpoint_dir,
            args.acceptance_manifest,
            args.semantic_map,
            args.historical_usage,
            args.lineage,
            args.qualification_report,
            args.qualification_db_export,
            args.cohort_deq,
        )
        if path
    ]
    if any(not path.is_file() for path in inputs if path != args.oracle_checkpoint_dir):
        parser.error("all supplied input paths except --oracle-checkpoint-dir must be files")
    if args.oracle_checkpoint_dir and not args.oracle_checkpoint_dir.is_dir():
        parser.error("--oracle-checkpoint-dir must be a directory")
    rows = feature_rows(args.engineering_map or args.oracle_request)
    manifest = load_json(args.acceptance_manifest)
    sample, names, source_row_count = sampled_checkpoint_columns(
        args.oracle_checkpoint, args.oracle_checkpoint_dir, rows, args.sample_rows
    )
    blockers: list[str] = []
    semantics, semantic_blocker = semantic_types(args.semantic_map, rows)
    if semantic_blocker:
        blockers.append(semantic_blocker)
    output_types = [item["output_type"] for item in semantics]
    finite_samples = np.isfinite(sample).sum(axis=0)
    pearson, spearman, agreement, jaccard, pair_samples = type_aware_redundancy(
        sample, output_types, args.min_pair_samples
    )
    primary = primary_redundancy_matrix(pearson, agreement, jaccard, output_types)
    usage, usage_blocker = parquet_counts(args.historical_usage, "historical usage")
    lineage, lineage_evidence, lineage_blocker = lineage_usage(args.lineage)
    blockers.extend(item for item in (usage_blocker, lineage_blocker) if item)
    failures, failure_blockers = failure_mining(
        args.qualification_report, args.qualification_db_export
    )
    blockers.extend(failure_blockers)
    deq_rows: list[dict[str, Any]] = []
    if args.cohort_deq:
        deq_evidence, deq_rows, deq_blocker = cohort_deq_evidence(args.cohort_deq)
        if deq_blocker:
            blockers.append(deq_blocker)
            deq_evidence = {
                "available": False,
                "status": "BLOCKED",
                "blocker": deq_blocker,
            }
    else:
        deq_evidence = {
            "available": False,
            "status": "BLOCKED",
            "blocker": (
                "No supplied DEQ artifact or executable decision-equivalence protocol; "
                "this runner does not fabricate DEQ."
            ),
        }
    request_ids = {str(row["request_id"]) for row in rows if row["request_id"]}
    accepted = accepted_ids(manifest)
    if request_ids and not request_ids <= accepted:
        blockers.append(
            "acceptance manifest does not cover "
            f"{len(request_ids - accepted)} source-map request IDs"
        )
    acceptance_status = "covered" if request_ids and request_ids <= accepted else "not_proven"
    feature_output = []
    for index, row in enumerate(rows):
        scores = primary[index].copy()
        scores[index] = np.nan
        peer = int(np.nanargmax(scores)) if np.isfinite(scores).any() else None
        finite_count = int(finite_samples[index])
        redundancy_status = (
            "unusable_no_finite_observations"
            if finite_count == 0
            else "usable"
            if peer is not None
            else "insufficient_pairwise_observations"
        )
        feature_output.append(
            {
                **row,
                "checkpoint_key": names[index],
                **semantics[index],
                "historical_usage": usage.get(
                    row["feature_key"], usage.get(str(row["indicator_id"]), 0)
                ),
                "lineage_mentions": lineage.get(row["feature_key"], 0),
                "finite_sample_count": finite_count,
                "redundancy_status": redundancy_status,
                "redundancy_metric": (
                    "abs_pearson" if output_types[index] == "continuous" else
                    "jaccard_detection" if output_types[index] == "event" else "agreement"
                ),
                "max_redundancy": float(scores[peer]) if peer is not None else None,
                "max_abs_pearson": float(abs(pearson[index, peer])) if peer is not None and np.isfinite(pearson[index, peer]) else None,
                "max_abs_spearman": float(abs(spearman[index, peer])) if peer is not None and np.isfinite(spearman[index, peer]) else None,
                "max_agreement": float(agreement[index, peer]) if peer is not None and np.isfinite(agreement[index, peer]) else None,
                "max_jaccard_detection": float(jaccard[index, peer]) if peer is not None and np.isfinite(jaccard[index, peer]) else None,
                "most_redundant_feature_key": (
                    rows[peer]["feature_key"] if peer is not None else None
                ),
                "most_redundant_pair_sample_count": (
                    int(pair_samples[index, peer]) if peer is not None else 0
                ),
                "acceptance": acceptance_status,
            }
        )
    deq_by_feature = {
        item["feature_key"]: item for item in deq_evidence.get("feature_summaries", [])
    }
    cluster_rows = []
    for index, members in enumerate(clusters(primary), start=1):
        # Historical usage is the deterministic representative tie-breaker.
        representative = max(
            members,
            key=lambda item: (feature_output[item]["historical_usage"], rows[item]["feature_key"]),
        )
        cluster_rows.append(
            {
                "cluster_id": index,
                "metric": feature_output[representative]["redundancy_metric"],
                "representative_feature_key": rows[representative]["feature_key"],
                "members": [rows[item]["feature_key"] for item in members],
                "size": len(members),
                "domains": sorted({semantics[item]["domain"] for item in members}),
                "usage_total": sum(feature_output[item]["historical_usage"] for item in members),
                "lineage_mentions_total": sum(feature_output[item]["lineage_mentions"] for item in members),
                "representative_deq": deq_by_feature.get(rows[representative]["feature_key"]),
            }
        )
    taxonomy = load_json(Path(__file__).with_name("hyperscalper_candidate_taxonomy_v1.json"))
    result = {
        "schema_version": 1,
        "artifact": ARTIFACT,
        "read_only": True,
        "strategy_execution": "not performed",
        "source_hashes": {str(path): source_hash(path) for path in inputs},
        "information_map": {
            "expected_features": EXPECTED_FEATURES,
            "mapped_features": len(rows),
            "acceptance": acceptance_status,
            "features": feature_output,
        },
        "redundancy": {
            "method": (
                "type-aware pairwise-finite sampled metrics: Pearson and pre-ranked ordinal "
                "correlation for continuous, agreement for state, Jaccard detection for event"
            ),
            "mode": "sampled_bounded_memory",
            "exact_mode": "not offered: exact pairwise Spearman across 711 x 129k is not bounded-memory/time practical",
            "source_rows": source_row_count,
            "sample_rows_requested": args.sample_rows,
            "sample_rows_used": int(sample.shape[0]),
            "min_pair_samples": args.min_pair_samples,
            "features_without_finite_observations": int((finite_samples == 0).sum()),
            "cluster_threshold": 0.95,
            "clusters": cluster_rows,
            "map": {
                "format": "parquet",
                "path": "hyperscalper_feature_redundancy_map_v1.parquet",
                "pair_count": len(rows) * (len(rows) - 1) // 2,
            },
        },
        "lineage_usage_bias": {
            "historical_usage_available": usage_blocker is None,
            "lineage_available": lineage_blocker is None,
            "historical_search_bias_evidence": lineage_evidence,
        },
        "cohort_failure_mining": {
            "failures": dict(failures.most_common()),
            "report_supplied": bool(args.qualification_report),
            "db_export_supplied": bool(args.qualification_db_export),
        },
        "decision_equivalence": deq_evidence,
        "candidate_taxonomy": taxonomy,
        "blockers": blockers,
    }
    args.output_dir.mkdir(parents=True, exist_ok=True)
    json_path = args.output_dir / "hyperscalper_feature_gap_analysis_v1.json"
    markdown = (
        "# HYPERSCALPER Feature Gap Analysis V1\n\n"
        "Read-only: yes. Strategy execution: not performed.\n\n"
        f"- Information map: {len(rows)}/711 features\n"
        f"- Acceptance: {acceptance_status}\n"
        f"- Redundancy clusters: {len(cluster_rows)}\n"
        f"- DEQ: {deq_evidence['status']}"
    ) + (
        " (supplied Cohort001 ledger; multi-feature attribution is associative, not causal)\n"
        if deq_evidence["available"]
        else " (not supplied or invalid; not fabricated)\n"
    ) + (
        f"- Blockers: {len(blockers)}\n\n## Blockers\n"
        + "\n".join(f"- {item}" for item in blockers)
        + "\n"
    )
    (args.output_dir / "hyperscalper_feature_gap_analysis_v1.md").write_text(
        markdown, encoding="utf-8"
    )
    (args.output_dir / "hyperscalper_feature_redundancy_clusters_v1.json").write_text(
        json.dumps(
            {
                "artifact": "HYPERSCALPER_FEATURE_REDUNDANCY_CLUSTERS_V1",
                "read_only": True,
                "strategy_execution": "not performed",
                "mode": result["redundancy"]["mode"],
                "clusters": cluster_rows,
            },
            indent=2,
            sort_keys=True,
        ) + "\n",
        encoding="utf-8",
    )
    try:
        import pyarrow as pa
        import pyarrow.parquet as pq

        pq.write_table(
            pa.Table.from_pylist(feature_output),
            args.output_dir / "hyperscalper_feature_gap_analysis_v1.parquet",
            compression="zstd",
        )
        pq.write_table(
            pa.Table.from_pylist(
                [
                    {
                        "left_feature_key": rows[left]["feature_key"],
                        "right_feature_key": rows[right]["feature_key"],
                        "left_output_type": output_types[left],
                        "right_output_type": output_types[right],
                        "redundancy_metric": (
                            "abs_pearson" if output_types[left] == output_types[right] == "continuous" else
                            "jaccard_detection" if output_types[left] == output_types[right] == "event" else
                            "agreement" if output_types[left] == output_types[right] == "state" else "not_comparable"
                        ),
                        "pair_sample_count": int(pair_samples[left, right]),
                        "pearson": (
                            float(pearson[left, right])
                            if np.isfinite(pearson[left, right])
                            else None
                        ),
                        "spearman_ordinal": (
                            float(spearman[left, right])
                            if np.isfinite(spearman[left, right])
                            else None
                        ),
                        "agreement": (
                            float(agreement[left, right])
                            if np.isfinite(agreement[left, right])
                            else None
                        ),
                        "jaccard_detection": (
                            float(jaccard[left, right])
                            if np.isfinite(jaccard[left, right])
                            else None
                        ),
                        "primary_redundancy": (
                            float(primary[left, right])
                            if np.isfinite(primary[left, right])
                            else None
                        ),
                    }
                    for left in range(len(rows))
                    for right in range(left + 1, len(rows))
                ]
            ),
            args.output_dir / "hyperscalper_feature_redundancy_map_v1.parquet",
            compression="zstd",
        )
        if args.cohort_deq and deq_evidence["available"]:
            pq.write_table(
                pa.Table.from_pylist(deq_rows),
                args.output_dir / "historical_feature_deq_summary_v1.parquet",
                compression="zstd",
            )
    except ImportError:
        result["parquet"] = "not written: pyarrow unavailable"
        result["redundancy"]["map"]["status"] = "not written: pyarrow unavailable"
    json_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json_path)


if __name__ == "__main__":
    main()