# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-prevent-latent-feature-leakage-through-correlated-public-features.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-prevent-latent-feature-leakage-through-correlated-public-features.warrant.json # Markdown SHA-256: b3aaf85b54cd8a970b46f64d31ac3fccb13b85a8223ed45069eea4e18d39e714 # Code AST SHA-256: b88fa980f0c2b70be1e536ab13e0d084df8e82601dcc9da6541217d81165c802 # Verified Classes: LeakageVerdict, FeatureSet, LeakagePolicy, LeakageManifest, LatentFeatureLeakageGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class LeakageVerdict(Enum): NO_PRIVATE_FEATURES = auto() WITHIN_CORRELATION_THRESHOLD = auto() DECORRELATION_APPLIED = auto() LEAKAGE_DETECTED = auto() @dataclass(frozen=True) class FeatureSet: featureset_id: str public_features: tuple[str, ...] private_features: tuple[str, ...] correlation_matrix: tuple[tuple[str, str, float], ...] # public, private, correlation decorrelation_applied: tuple[str, ...] @dataclass(frozen=True) class LeakagePolicy: max_acceptable_correlation: float require_decorrelation_above_threshold: bool @dataclass(frozen=True) class LeakageManifest: verdict: LeakageVerdict directive: str manifest_digest: str class LatentFeatureLeakageGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_featureset( cls, featureset: FeatureSet, policy: LeakagePolicy ) -> LeakageManifest: # Gate 1: No private features if not featureset.private_features: digest = cls._hash(json.dumps({ "featureset": featureset.featureset_id, "private": "NONE" }, sort_keys=True)) return LeakageManifest( verdict=LeakageVerdict.NO_PRIVATE_FEATURES, directive=f"PROCEED: FeatureSet {featureset.featureset_id} has no private features.", manifest_digest=digest ) # Gate 2: High correlations high_corr = [ (pub, priv, corr) for pub, priv, corr in featureset.correlation_matrix if abs(corr) > policy.max_acceptable_correlation ] # Gate 3: Decorrelation if high_corr and policy.require_decorrelation_above_threshold: undecorrelated = [ (pub, priv, corr) for pub, priv, corr in high_corr if pub not in featureset.decorrelation_applied ] if undecorrelated: return LeakageManifest( verdict=LeakageVerdict.LEAKAGE_DETECTED, directive=f"HALT: {len(undecorrelated)} public features correlate >{policy.max_acceptable_correlation:.2f} with private without decorrelation.", manifest_digest=cls._hash(f"LEAKAGE:{featureset.featureset_id}") ) digest = cls._hash(json.dumps({ "featureset": featureset.featureset_id, "decorrelated": tuple(pub for pub, _, _ in high_corr) }, sort_keys=True)) return LeakageManifest( verdict=LeakageVerdict.DECORRELATION_APPLIED, directive=f"PROCEED: {len(high_corr)} high-correlation features decorrelated.", manifest_digest=digest ) # Gate 4: Within threshold max_corr = max((abs(c) for _, _, c in featureset.correlation_matrix), default=0) digest = cls._hash(json.dumps({ "featureset": featureset.featureset_id, "max_correlation": f"{max_corr:.3f}", "threshold": f"{policy.max_acceptable_correlation:.3f}" }, sort_keys=True)) return LeakageManifest( verdict=LeakageVerdict.WITHIN_CORRELATION_THRESHOLD, directive=f"PROCEED: Max correlation {max_corr:.2f} within threshold.", manifest_digest=digest )